Write mesh/check from the checker's owed state, read by a watch (R2-1)

R4e round 2 finding R2-1, ADR 0521: the delivery plan draws no edge that
orders the forge module before the forge bridge, and the forge module's
durable consumer keeps its deliver policy when its filters change, so a
check.status the bridge emitted before the forge module consumed it was lost.
An in-memory revision guard also forgot itself on a restart.

The forge module now reads mesh-delivery-checker.owed, which the checker
announces readable, by a watch: the watch hands over the latest value of every
key whenever the module starts, so a status written while it was away is
written then, and only the latest. Each mesh/check is written once per
revision, an older one dropped; a forge refusal is asked again on a bound
until written or overtaken; a refused read is said once and asked again until
granted. No tool writes mesh/check, and no check.status is consumed.

modules/gitea/check.ts

@@ -1,25 +1,26 @@
-// **mesh/check is written from the forge bridge's own event, never from a tool** (review finding K1 of R4e; novox/hq issue
+// **mesh/check is written from the checker's owed state, read by a watch** (ADR 0521; R4e round 2 finding R2-1; finding K1
-// 1301; ADR 0447 §1, ADR 0503 §3, ADR 0513 §4: main requires mesh/check alone).
+// before it; novox/hq issue 1301; ADR 0447 §1, ADR 0503 §3: main requires mesh/check alone).
 //
-// A tool is told no caller, and the console every agent session speaks through (node-tools) invokes every tool, so a tool
+// No tool writes mesh/check: a tool is told no caller, and the console every agent session speaks through (node-tools)
-// that wrote mesh/check could be called by any session to pass a merge past the checker: the fault issue 293 closed for
+// invokes every tool, so a tool that wrote it could be called by any session to merge past the checker (the fault issue
-// the two statuses before it. So the checker's forge bridge emits `check.status` instead. The bus grants the subject
+// 293 closed for the two statuses before it). Nor does an event: nothing in the delivery plan orders this module before the
-// mesh.mod.mesh-delivery-forge-bridge.event.check.status to the bridge's own account alone (its manifest's `emits`; a
+// forge bridge, and a durable consumer whose filters change keeps its deliver policy, so an event said before this module
-// module publishes under its own name only), and the bridge runs on that account, never carried by a machine's tool
+// consumed it was lost. The checker keeps what it owes the forge in its state `owed` (ADR 0484 §2), which it announces
-// runner. This module consumes the event, as it consumes mesh-controller.checked, and writes the status.
+// readable (ADR 0495 §2-§3), and only the checker writes it. This module watches it: the watch hands over the latest value of
+// every key when it starts, whenever that is, and then every change. Each key whose act is the status mesh/check is
+// written on its head, once per revision; a revision older than one already written on the key is dropped.
 //
-// The event's body: owner, repo, sha (the head, whole), state, description, target_url, key (the checker's owed key) and
+// A read refused (the grant not yet composed) is said once and asked again on a bound until it holds. A forge that refuses
-// revision (that key's revision in the checker's bucket). Its source is checked against the bridge too: the bus already
+// a status is asked again on a bound, until it writes or a newer revision of the key comes.
-// said it, and a body read from elsewhere says it again. Delivery is at-least-once and a failed write is delivered again
-// later, so a head's revision older than one already written is dropped: a stale pending never hides a verdict.
 
 import type { CommitStatus } from "./client.js";
+import { bindUntilHeld, bindWait } from "./bind.js";
 import { checkName } from "./guard.js";
 import { CHECK_STATUS } from "./pulls.js";
 
-/** The module that alone writes mesh/check, and the event it writes it with. */
+/** The checker's state this module reads, and the act of its keys this module writes. */
-export const CHECK_WRITER = "mesh-delivery-forge-bridge";
+export const CHECKER_OWED = "mesh-delivery-checker.owed";
-export const CHECK_EVENT = `${CHECK_WRITER}.check.status`;
+export const CHECK_ACT = `status=${CHECK_STATUS}`;
 
 const states = new Set(["pending", "success", "error", "failure", "warning"]);
 
@@ -28,84 +29,160 @@ export interface CheckForge {
   setCommitStatus(owner: string, repo: string, sha: string, status: CommitStatus): Promise<void>;
 }
 
-/** An event as the SDK's `on` hands it. */
+/** One change of the checker's owed, as the SDK's state watch hands it. */
-export interface CheckEvent {
+export interface OwedChange {
-  source: string;
+  key: string;
-  id?: string;
+  op: "put" | "delete";
-  body: unknown;
+  value?: unknown;
+  revision: number;
+  current?: boolean;
 }
 
-/** The status a check.status asks for, checked; or why it is not one. */
+/** A key part as mesh-sdk go/delivery EscapeName writes it, read back: `=` and two lowercase hex digits per byte. */
-export function checkStatusOf(body: unknown): { owner: string; repo: string; sha: string; revision: number; status: CommitStatus } {
+function unescapeName(part: string): string {
-  if (!body || typeof body !== "object") throw new Error("the event carries no body");
+  if (!/^([-/_a-zA-Z0-9]|=[0-9a-f]{2})+$/.test(part)) throw new Error(`${JSON.stringify(part)} is not an escaped name`);
-  const b = body as Record<string, unknown>;
+  return Buffer.from(part.replace(/=([0-9a-f]{2})/g, (_, h) => String.fromCharCode(parseInt(h, 16))), "latin1").toString("utf8");
-  const owner = checkName("owner", b.owner), repo = checkName("name", b.repo);
+}
-  const sha = String(b.sha ?? "");
+
-  if (!/^([0-9a-f]{40}|[0-9a-f]{64})$/i.test(sha)) throw new Error(`${JSON.stringify(b.sha)} is not a head's whole sha`);
+/** The pull request and head of the checker's owed key (mesh-sdk go/delivery CheckerOwedKey):
-  if (b.context !== undefined && b.context !== null && b.context !== CHECK_STATUS) {
+ *  `<owner>/<name>.<number>.<head>.status=mesh/check`, owner and name escaped. */
-    throw new Error(`${JSON.stringify(b.context)}: check.status writes ${CHECK_STATUS} alone`);
+export function owedKeyOf(key: string): { owner: string; repo: string; number: number; sha: string } {
-  }
+  const parts = String(key ?? "").split(".");
-  const state = String(b.state ?? "");
+  if (parts.length !== 4) throw new Error(`${JSON.stringify(key)} is not an owed key, <repository>.<number>.<head>.<act>`);
-  if (!states.has(state)) throw new Error(`${JSON.stringify(b.state)} is not a status the forge keeps`);
+  const [repository, number, sha, act] = parts;
-  const revision = b.revision;
+  if (act !== CHECK_ACT) throw new Error(`${JSON.stringify(key)} owes ${JSON.stringify(act)}, not ${CHECK_ACT}`);
+  const m = /^([^/]+)\/([^/]+)$/.exec(repository);
+  if (!m) throw new Error(`${JSON.stringify(key)} names no <owner>/<name>`);
+  if (!/^(0|[1-9][0-9]*)$/.test(number)) throw new Error(`${JSON.stringify(key)} names no pull request number`);
+  if (!/^([0-9a-f]{40}|[0-9a-f]{64})$/i.test(sha)) throw new Error(`${JSON.stringify(key)} names no head's whole sha`);
+  return { owner: checkName("owner", unescapeName(m[1])), repo: checkName("name", unescapeName(m[2])), number: Number(number), sha };
+}
+
+/** The status an owed change asks for, checked as mesh-sdk go/delivery DecodeCheckerOwed checks it; or why it is not one. */
+export function checkStatusOf(change: OwedChange): { owner: string; repo: string; sha: string; revision: number; status: CommitStatus } {
+  if (change.op !== "put") throw new Error(`a ${change.op} owes nothing`);
+  const { owner, repo, sha } = owedKeyOf(change.key);
+  const revision = change.revision;
   if (typeof revision !== "number" || !Number.isSafeInteger(revision) || revision < 0) {
-    throw new Error(`${JSON.stringify(revision)} is not a revision of the checker's owed key`);
+    throw new Error(`${JSON.stringify(revision)} is not a revision`);
   }
-  let description = String(b.description ?? "").replace(/\s+/g, " ").trim();
+  const v = change.value;
+  if (!v || typeof v !== "object") throw new Error("the value is no owed act");
+  const o = v as Record<string, unknown>;
+  if (o.v !== 1) throw new Error(`the owed act's version ${JSON.stringify(o.v)} is not one this module reads`);
+  if (o.act !== "status" || o.context !== CHECK_STATUS) {
+    throw new Error(`the owed act is ${JSON.stringify(o.act)} ${JSON.stringify(o.context)}, not the status ${CHECK_STATUS}`);
+  }
+  const state = String(o.state ?? "");
+  if (!states.has(state)) throw new Error(`${JSON.stringify(o.state)} is not a status the forge keeps`);
+  let description = String(o.text ?? "").replace(/\s+/g, " ").trim();
   if (description.length > 140) description = description.slice(0, 139) + "…";
-  const target = typeof b.target_url === "string" && /^https?:\/\//.test(b.target_url) ? b.target_url : undefined;
+  const target = typeof o.target === "string" && /^https?:\/\//.test(o.target) ? o.target : undefined;
   const status: CommitStatus = { state: state as CommitStatus["state"], context: CHECK_STATUS, description,
     ...(target ? { target_url: target } : {}) };
   return { owner, repo, sha, revision, status };
 }
 
-/** The one writer of mesh/check: each check.status the bridge emits, in revision order per head. */
+/** The one writer of mesh/check: each owed revision of the checker's, in revision order per key. */
 export class CheckStatuses {
-  /** The newest revision written on each head, owner/repo@sha; the oldest heads go past what any forge holds open. */
+  /** The newest revision written of each key, and the newest seen; the oldest keys go past what any forge holds open. */
   private written = new Map<string, number>();
-  /** The turn each head's writes wait on. */
+  private seen = new Map<string, number>();
-  private turns = new Map<string, Promise<void>>();
+  /** The turn each key's writes wait on, and the asks again still running. */
-
+  private turns = new Map<string, Promise<unknown>>();
+  private again = new Set<Promise<void>>();
   private forge: CheckForge;
   private say: (line: string) => void;
+  private sleep: (ms: number) => Promise<void>;
 
-  constructor(forge: CheckForge, say: (line: string) => void = (l) => console.error(l)) {
+  constructor(forge: CheckForge, say: (line: string) => void = (l) => console.error(l),
+    sleep: (ms: number) => Promise<void> = (ms) => new Promise((r) => setTimeout(r, ms))) {
     this.forge = forge;
     this.say = say;
+    this.sleep = sleep;
   }
 
-  /** Write the event's status: "written", or "dropped" for one that is not the bridge's, malformed or stale. A forge
+  /** Write the change's status: "written", or "dropped" for one that owes no mesh/check or is older than one written. A
-   *  that refuses throws, so the bus delivers the event again. */
+   *  forge that refuses throws. */
-  async apply(event: CheckEvent): Promise<"written" | "dropped"> {
+  async apply(change: OwedChange): Promise<"written" | "dropped"> {
-    if (event.source !== CHECK_WRITER) {
-      this.say(`[gitea] a check.status from ${JSON.stringify(event.source)}, not the forge bridge, is dropped: mesh/check is the bridge's alone`);
-      return "dropped";
-    }
     let asked;
     try {
-      asked = checkStatusOf(event.body);
+      asked = checkStatusOf(change);
     } catch (err) {
-      this.say(`[gitea] the forge bridge's check.status ${event.id ?? ""} is dropped: ${err instanceof Error ? err.message : err}`);
+      this.say(`[gitea] ${CHECKER_OWED} ${change?.key} at revision ${change?.revision} writes nothing: ${err instanceof Error ? err.message : err}`);
       return "dropped";
     }
-    const head = `${asked.owner}/${asked.repo}@${asked.sha}`.toLowerCase();
+    const key = change.key;
-    // One head's writes take turns, so two deliveries at once are weighed one after the other.
+    const seen = this.seen.get(key);
-    const before = this.turns.get(head) ?? Promise.resolve();
+    if (seen === undefined || asked.revision > seen) this.remember(this.seen, key, asked.revision);
+    // One key's writes take turns, so two at once are weighed one after the other.
+    const before = this.turns.get(key) ?? Promise.resolve();
     const turn = before.then(async (): Promise<"written" | "dropped"> => {
-      const newest = this.written.get(head);
+      const newest = this.written.get(key);
       if (newest !== undefined && asked.revision < newest) {
         this.say(`[gitea] ${asked.owner}/${asked.repo}@${asked.sha.slice(0, 12)}: mesh/check at revision ${asked.revision} is older than ${newest}, already written, and is dropped`);
         return "dropped";
       }
       await this.forge.setCommitStatus(asked.owner, asked.repo, asked.sha, asked.status);
-      this.written.delete(head);
+      this.remember(this.written, key, asked.revision);
-      this.written.set(head, asked.revision);
-      if (this.written.size > 2000) this.written.delete(this.written.keys().next().value as string);
       return "written";
     });
-    const settled = turn.then(() => {}, () => {});
+    const settled = turn.catch(() => {});
-    this.turns.set(head, settled);
+    this.turns.set(key, settled);
-    void settled.then(() => { if (this.turns.get(head) === settled) this.turns.delete(head); });
+    void settled.then(() => { if (this.turns.get(key) === settled) this.turns.delete(key); });
     return turn;
   }
+
+  /** What the watch hands over: written, or when the forge refuses, asked again on a bound in the background until it is
+   *  written or a newer revision of the key was seen. Never throws: a watch's handler that throws loses the change. */
+  async take(change: OwedChange): Promise<void> {
+    try {
+      await this.apply(change);
+    } catch (err) {
+      const run = this.askAgain(change, err);
+      this.again.add(run);
+      void run.finally(() => this.again.delete(run));
+    }
+  }
+
+  /** Resolves when no ask again is running. */
+  async settled(): Promise<void> {
+    while (this.again.size) await Promise.all([...this.again]);
+  }
+
+  private async askAgain(change: OwedChange, first: unknown): Promise<void> {
+    let err = first;
+    for (let attempt = 1; ; attempt++) {
+      const wait = bindWait(attempt);
+      this.say(`[gitea] ${change.key}: mesh/check at revision ${change.revision} could not be written (${err instanceof Error ? err.message : err}); asked again in ${wait / 1000}s`);
+      await this.sleep(wait);
+      const newest = this.seen.get(change.key);
+      if (newest !== undefined && newest > change.revision) return;
+      try {
+        await this.apply(change);
+        return;
+      } catch (e) {
+        err = e;
+      }
+    }
+  }
+
+  private remember(map: Map<string, number>, key: string, revision: number): void {
+    map.delete(key);
+    map.set(key, revision);
+    if (map.size > 4000) map.delete(map.keys().next().value as string);
+  }
+}
+
+/**
+ * Watch the checker's owed with `watch` (the SDK's `state(CHECKER_OWED).watch`), handing each change to `checks`. A watch
+ * refused — the read not granted yet — is said once and asked again on a bound until it holds (bind.ts). Never rejects.
+ */
+export async function watchOwed(watch: (handler: (change: OwedChange) => Promise<void>) => Promise<void>, checks: CheckStatuses,
+  sleep?: (ms: number) => Promise<void>, say: (line: string) => void = (l) => console.error(l)): Promise<void> {
+  let refusals = 0;
+  await bindUntilHeld(CHECKER_OWED, () => watch((change) => checks.take(change)), sleep, (line) => {
+    if (/could not be bound/.test(line) && refusals++ > 0) return;
+    say(line);
+  });
 }

modules/gitea/delivery.ts

@@ -101,10 +101,10 @@ export function deliveryStatus(context: string, state: string, description: stri
   if (!/^mesh\/[a-z-]+$/.test(context)) throw new Error(`${context} is not a status of the mesh's own`);
   if (!states.has(state)) throw new Error(`${state} is not a status the forge keeps`);
   // mesh/check is the checker's verdict and the one status main requires (ADR 0447 §1): no hand sets it in any
-  // state, as issue 293 closed for the two before it. No tool writes it, since a tool is told no caller: the forge
+  // state, as issue 293 closed for the two before it. No tool writes it, since a tool is told no caller (R4e finding K1):
-  // bridge emits check.status on a subject only its own account may publish, and check.ts writes it (R4e finding K1).
+  // check.ts writes it from the checker's owed state (ADR 0521).
   if (context === CHECK_STATUS) {
-    throw new Error(`${CHECK_STATUS} is the checker's status: it is written from the forge bridge's check.status event alone, never by a tool`);
+    throw new Error(`${CHECK_STATUS} is the checker's status: it is written from mesh-delivery-checker.owed alone, never by a tool`);
   }
   // The merge check's statuses are its verdict's, set when the controller says it, and a branch may require
   // them: never set to a verdict by hand, so nothing passes — or fails — a merge past the check (novox/hq
@@ -136,7 +136,7 @@ export interface StatusWriters {
 export function commitStatusTool(writers: StatusWriters): ToolDefinition {
   return {
     name: "gitea_commit_status",
-    description: "Set one of the mesh's statuses on a commit (mesh/delivery, mesh/delivery-group): pending, success, error, failure or warning, a short description, and the page it links to. mesh/check, the status main requires, is refused: no tool writes it, only the checker's forge bridge's check.status event. hq's mesh/repo-check (and mesh/merge-gate, kept for a revert) take pending only — a recheck putting them back until the fresh verdict sets them — on the head's whole sha, in their turn after the check's earlier writes.",
+    description: "Set one of the mesh's statuses on a commit (mesh/delivery, mesh/delivery-group): pending, success, error, failure or warning, a short description, and the page it links to. mesh/check, the status main requires, is refused: no tool writes it: it is written from the checker's own state. hq's mesh/repo-check (and mesh/merge-gate, kept for a revert) take pending only — a recheck putting them back until the fresh verdict sets them — on the head's whole sha, in their turn after the check's earlier writes.",
     input: {
       owner: { type: "string", description: "the repository owner" },
       repo: { type: "string", description: "the repository name" },

modules/gitea/index.ts

@@ -16,8 +16,9 @@
 // still requires are written (issue 1301): main requires `mesh/check` — the checker's — everywhere but hq, whose main
 // requires `mesh/repo-check` alone (ADR 0503 §5).
 //
-// Consumes mesh-delivery-forge-bridge.check.status — the checker's forge bridge's mesh/check on a head — and writes it
+// Reads mesh-delivery-checker.owed — what the checker owes the forge, mesh/check on each head it judged — by a watch, and
-// (check.ts). No tool writes mesh/check: the bus lets the bridge's own account alone publish that event (R4e finding K1).
+// writes it (check.ts; ADR 0521, R4e finding R2-1). No tool writes mesh/check (finding K1), and no event: the watch hands
+// over the latest value of every key whenever this module starts.
 //
 // Consumes node-build-agent.started — a build machine took an ask — and says, on the head a pending check is of,
 // which machine builds it since when (novox/hq issue 446). This module is the one writer of mesh/merge-gate.
@@ -45,7 +46,8 @@ import { emit, on } from "@novox/mesh-sdk/events";
 import { registerModuleTools } from "@novox/mesh-sdk/tools";
 import { statesAnnouncement } from "./kept.js";
 import { commitStatusTool } from "./delivery.js";
-import { CHECK_EVENT, CheckStatuses } from "./check.js";
+import { CHECKER_OWED, CheckStatuses, watchOwed, type OwedChange } from "./check.js";
+import { state } from "@novox/mesh-sdk/state";
 import { announce, keepWatermark, readWatermark } from "./announce.js";
 import { mergeTool } from "./merge.js";
 import { GiteaAdmin, GiteaClient, movedSince } from "./client.js";
@@ -479,7 +481,7 @@ const gates = new GateWriter({
   headChecks: (owner, repo, sha) => (gitea as GiteaClient).headChecks(owner, repo, sha),
 });
 
-// The one writer of mesh/check, from the forge bridge's check.status (check.ts, R4e finding K1).
+// The one writer of mesh/check, from the checker's owed (check.ts; ADR 0521).
 const checks = new CheckStatuses({
   setCommitStatus: (owner, repo, sha, status) => (gitea as GiteaClient).setCommitStatus(owner, repo, sha, status),
 });
@@ -627,8 +629,8 @@ if (gitea) {
       return setVerdict(client, event);
     })],
     ["node-build-agent.started", () => on("node-build-agent.started", buildStarted)],
-    // The checker's mesh/check, from the one account the bus lets publish it (check.ts): a forge that refuses throws,
-    // and the event is delivered again.
-    [CHECK_EVENT, () => on(CHECK_EVENT, async (event) => { await checks.apply(event); })],
   ], () => console.log("[gitea] watching for new repositories and merged pull requests"));
+  // The checker's mesh/check, from its owed state (check.ts; ADR 0521): a read not granted yet is said once and asked again,
+  // apart from the binds above, so neither waits for the other.
+  void watchOwed((handler) => state<OwedChange["value"]>(CHECKER_OWED).watch((change) => handler(change)), checks);
 }

modules/gitea/module.json

@@ -87,8 +87,7 @@
   ],
   "consumes": [
     "mesh-controller.checked",
-    "node-build-agent.started",
+    "node-build-agent.started"
-    "mesh-delivery-forge-bridge.check.status"
   ],
   "listens": [
     {
@@ -544,6 +543,7 @@
     "ssh-client.keys",
     "mesh-build-go.keys",
     "mesh-build-node.keys",
-    "mesh-build-image.keys"
+    "mesh-build-image.keys",
+    "mesh-delivery-checker.owed"
   ]
 }

modules/gitea/pulls.ts

@@ -19,8 +19,8 @@ export const CHECK_CONTEXT = "mesh/merge-gate";
 /** The status context of the repository's own merge-check.sh, the check's second layer (novox/hq ADR 0237). */
 export const REPO_CHECK_CONTEXT = "mesh/repo-check";
 
-/** The one status main requires (novox/hq ADR 0447 §1, issue 1301): the checker's verdict, written by its forge bridge
+/** The one status main requires (novox/hq ADR 0447 §1, issue 1301): the checker's verdict, written by this module from the
- *  through gitea_check_status. Its line names the head and the trunk commit it judged: "pass <head12> on <trunk12> …". */
+ *  checker's owed state (check.ts, ADR 0521). Its line names the head and the trunk commit it judged: "pass <head12> on <trunk12> …". */
 export const CHECK_STATUS = "mesh/check";
 
 /** The statuses a branch's rule requires, from the rule the forge applies to it: none when it is unprotected or checks

modules/gitea/test/bundle-import.test.ts

@@ -37,10 +37,12 @@ test("importing the bundle resolves while the broker keeps refusing the binds",
   assert.match(out, /imported/, `the import did not resolve within 3 s while a bind was refused; stderr: ${err.slice(0, 600)}`);
 });
 
-// **The bundle binds the forge bridge's check.status and serves no tool that writes mesh/check** (review finding K1 of R4e;
+// **The bundle writes mesh/check from the checker's owed state, and serves no tool that writes it** (ADR 0521; R4e round 2
-// issue 1301 S4; ADR 0503 §3). The bundle is imported whole, as the runtime imports it; the consumer it binds is handed the
+// finding R2-1; finding K1). The bundle is imported whole, as the runtime imports it, while its read of the checker's owed
-// bridge's event and one from another source, and the forge stand-in sees mesh/check written for the bridge's alone.
+// is refused (the grant not yet composed): the watch is asked again, and once granted it is handed the bucket as the bus
-test("the bundle writes mesh/check from the bridge's check.status alone, and serves no gitea_check_status", async () => {
+// keeps it, the latest revision of each key, then an older revision delivered late. The forge stand-in sees the latest
+// written, and nothing after it.
+test("a bundle whose read of the checker's owed is granted late writes the latest mesh/check, and serves no gitea_check_status", async () => {
   const index = pathToFileURL(join(here, "..", "dist", "index.js")).href;
   const sha = "0123456789abcdef0123456789abcdef01234567";
   const script = `
@@ -54,23 +56,35 @@ test("the bundle writes mesh/check from the bridge's check.status alone, and ser
     });
     await new Promise((r) => forge.listen(0, "127.0.0.1", r));
     process.env.MESH_GITEA_URL = "http://127.0.0.1:" + forge.address().port;
-    const bound = {};
+    const bound = [];
     const { useBroker } = await import("@novox/mesh-sdk/messaging");
     useBroker(() => ({
-      subscribe: async (pattern, handler) => { bound[pattern] = handler; return () => {}; },
+      subscribe: async (pattern) => { bound.push(pattern); return () => {}; },
       publish: async () => {}, request: async () => { throw new Error("no responders"); },
       handle: async () => () => {}, close: async () => {},
     }));
+    const key = "novox/mesh-sdk.12.${sha}.status=mesh/check";
+    const owed = (state, text) => ({ v: 1, act: "status", context: "mesh/check", state, text, target: "https://view.example/c" });
+    const watched = [];
+    let granted = 0;
+    const { useStateChannel } = await import("@novox/mesh-sdk/state");
+    useStateChannel({
+      ask: async () => null,
+      watch: async (params, deliver) => {
+        watched.push(params.state);
+        if (watched.length === 1) throw new Error('gitea keeps and reads no state called "mesh-delivery-checker.owed"');
+        granted++;
+        // Two revisions were written while the read was refused; the bucket hands the latest, then a late older one.
+        await deliver({ state: params.state, key, op: "put", value: owed("success", "pass at 4"), revision: 4, current: true });
+        await deliver({ state: params.state, key, op: "put", value: owed("pending", "pending at 3"), revision: 3, current: false });
+      },
+    });
     await import(${JSON.stringify(index)});
     const { collectTools } = await import("@novox/mesh-sdk/tools");
     const tools = collectTools().flatMap((r) => r.tools.map((t) => t.name));
-    for (let i = 0; i < 100 && !bound["mesh-delivery-forge-bridge.check.status"]; i++) await new Promise((r) => setTimeout(r, 20));
+    for (let i = 0; i < 300 && !(granted && statuses.length); i++) await new Promise((r) => setTimeout(r, 20));
-    const h = bound["mesh-delivery-forge-bridge.check.status"];
+    await new Promise((r) => setTimeout(r, 200));
-    const env = (source, state) => ({ key: "check.status", node: "n1",
+    console.log(JSON.stringify({ tools, bound, watched, statuses }));
-      headers: { "x-event-id": "e", "x-source": source, "x-node": "n1", "x-time": "", "content-type": "application/json" },
-      body: { owner: "novox", repo: "mesh-sdk", sha: ${JSON.stringify(sha)}, state, description: "by " + source, revision: 1 } });
-    if (h) { await h(env("node-tools", "success")); await h(env("mesh-delivery-forge-bridge", "pending")); }
-    console.log(JSON.stringify({ tools, bound: Object.keys(bound), statuses }));
     process.exit(0);`;
   const env = { ...process.env, MESH_GITEA_TOKEN: "x" };
   for (const k of ["MESH_GITEA_STATE_DIR", "MESH_GITEA_ADMIN_USER", "MESH_GITEA_ADMIN_PASSWORD", "MESH_SERVED_MODULE", "MESH_GITEA_URL"]) delete env[k];
@@ -80,16 +94,18 @@ test("the bundle writes mesh/check from the bridge's check.status alone, and ser
   child.stdout.on("data", (d) => (out += d));
   child.stderr.on("data", (d) => (err += d));
   const ended = new Promise<void>((r) => child.on("exit", () => r()));
-  const timer = setTimeout(() => child.kill("SIGKILL"), 8000);
+  const timer = setTimeout(() => child.kill("SIGKILL"), 15000);
   await ended;
   clearTimeout(timer);
   const line = out.split("\n").find((l) => l.startsWith("{"));
   assert.ok(line, `the bundle said nothing; stdout: ${out.slice(0, 400)} stderr: ${err.slice(0, 600)}`);
-  const got = JSON.parse(line!) as { tools: string[]; bound: string[]; statuses: string[] };
+  const got = JSON.parse(line!) as { tools: string[]; bound: string[]; watched: string[]; statuses: string[] };
   assert.ok(got.tools.includes("gitea_commit_status"), `the bundle's tools: ${got.tools}`);
   assert.ok(!got.tools.includes("gitea_check_status"), "no tool writes mesh/check");
-  assert.ok(got.bound.includes("mesh-delivery-forge-bridge.check.status"), `bound: ${got.bound}`);
+  assert.ok(!got.bound.some((p) => p.includes("check.status")), `bound: ${got.bound}`);
+  assert.deepEqual(got.watched, ["mesh-delivery-checker.owed", "mesh-delivery-checker.owed"], "refused, then asked again");
   assert.equal(got.statuses.length, 1, `statuses written: ${got.statuses}`);
   assert.match(got.statuses[0], new RegExp(`/repos/novox/mesh-sdk/statuses/${sha} .*"context":"mesh/check"`));
-  assert.match(got.statuses[0], /"state":"pending".*by mesh-delivery-forge-bridge/);
+  assert.match(got.statuses[0], /"state":"success".*pass at 4/);
+  assert.match(err, /mesh-delivery-checker\.owed could not be bound/, "the refusal is said");
 });

modules/gitea/test/check-event.test.ts

@@ -1,102 +0,0 @@
-// **mesh/check is written from the forge bridge's own event, never from a tool** (review finding K1 of R4e; issue 1301 S4;
-// ADR 0503 §3, ADR 0513 §4). A tool cannot know its caller, and node-tools invokes every tool ("*"), so any session could
-// have set mesh/check through gitea_check_status. The bus grants the subject mesh.mod.mesh-delivery-forge-bridge.event.
-// check.status to the bridge's own account alone; this module consumes it and writes the status.
-
-import { test } from "node:test";
-import assert from "node:assert/strict";
-import { readFileSync } from "node:fs";
-
-const sha = "0123456789abcdef0123456789abcdef01234567";
-const body = (over: Record<string, unknown> = {}) => ({ owner: "novox", repo: "mesh-sdk", sha, state: "success",
-  description: "pass 0123456789ab on aaaaaaaaaaaa", target_url: "https://view.example/checks/g.1",
-  key: `novox/mesh-sdk.12.${sha}.status=mesh~check`, revision: 7, ...over });
-const event = (b: unknown, source = "mesh-delivery-forge-bridge") => ({ type: "check.status", id: "x", source, node: "n1", at: "", body: b });
-
-function forge() {
-  const sent: string[] = [];
-  let fail = false;
-  return {
-    sent,
-    failNext: () => { fail = true; },
-    setCommitStatus: async (o: string, r: string, s: string, st: any) => {
-      if (fail) { fail = false; throw new Error("Gitea API /repos/x/statuses: 502 bad gateway"); }
-      sent.push(`${o}/${r}@${s.slice(0, 8)} ${st.context} ${st.state} ${st.description}${st.target_url ? " " + st.target_url : ""}`);
-    },
-  };
-}
-
-test("the bridge's check.status event is the subject this module consumes, and the bridge is its one source", async () => {
-  const { CHECK_EVENT, CHECK_WRITER } = await import("../check.ts");
-  assert.equal(CHECK_EVENT, "mesh-delivery-forge-bridge.check.status");
-  assert.equal(CHECK_WRITER, "mesh-delivery-forge-bridge");
-  const manifest = JSON.parse(readFileSync(new URL("../module.json", import.meta.url), "utf8"));
-  assert.ok(manifest.consumes.includes(CHECK_EVENT), "gitea's manifest consumes the bridge's check.status");
-});
-
-test("a check.status from the bridge writes mesh/check on the head, its state, its line cut and its link", async () => {
-  const { CheckStatuses } = await import("../check.ts");
-  const f = forge();
-  const checks = new CheckStatuses(f, () => {});
-  assert.equal(await checks.apply(event(body())), "written");
-  assert.equal(await checks.apply(event(body({ sha: sha.replace(/0/g, "f"), state: "pending", description: "y".repeat(300), target_url: "javascript:x", revision: 8 }))), "written");
-  assert.equal(f.sent[0], "novox/mesh-sdk@01234567 mesh/check success pass 0123456789ab on aaaaaaaaaaaa https://view.example/checks/g.1");
-  assert.ok(f.sent[1].startsWith("novox/mesh-sdk@f1234567 mesh/check pending yyyy"), f.sent[1]);
-  assert.ok(f.sent[1].length < 200, "the line is cut as every status's is");
-  assert.ok(!f.sent[1].includes("javascript"), "a link that is no web page is not written");
-});
-
-test("an event whose source is not the bridge writes nothing", async () => {
-  const { CheckStatuses } = await import("../check.ts");
-  const f = forge();
-  const said: string[] = [];
-  const checks = new CheckStatuses(f, (l) => said.push(l));
-  for (const source of ["node-tools", "gitea", "unknown", ""]) assert.equal(await checks.apply(event(body(), source)), "dropped", source);
-  assert.deepEqual(f.sent, []);
-  assert.match(said[0], /not the forge bridge/);
-});
-
-test("a malformed event writes nothing and is not asked for again", async () => {
-  const { CheckStatuses } = await import("../check.ts");
-  const f = forge();
-  const checks = new CheckStatuses(f, () => {});
-  for (const b of [body({ sha: sha.slice(0, 12) }), body({ sha: "../../admin" }), body({ state: "green" }), body({ owner: "../x" }),
-    body({ repo: ".." }), body({ revision: undefined }), body({ revision: -1 }), body({ revision: "7" }), null, "text", body({ context: "mesh/delivery" })]) {
-    assert.equal(await checks.apply(event(b)), "dropped", JSON.stringify(b));
-  }
-  assert.deepEqual(f.sent, []);
-  // The context may be named, and is mesh/check then.
-  assert.equal(await checks.apply(event(body({ context: "mesh/check" }))), "written");
-});
-
-test("a revision older than one written on the head is dropped; the same one is written again (a redelivery)", async () => {
-  const { CheckStatuses } = await import("../check.ts");
-  const f = forge();
-  const checks = new CheckStatuses(f, () => {});
-  await checks.apply(event(body({ state: "success", revision: 10 })));
-  assert.equal(await checks.apply(event(body({ state: "pending", revision: 9 }))), "dropped", "a stale pending never hides the verdict");
-  assert.equal(await checks.apply(event(body({ state: "success", revision: 10 }))), "written");
-  // Another head is its own.
-  assert.equal(await checks.apply(event(body({ sha: sha.replace(/1/g, "2"), state: "pending", revision: 3 }))), "written");
-  assert.equal(await checks.apply(event(body({ state: "pending", revision: 11 }))), "written");
-  assert.deepEqual(f.sent.map((s) => s.split(" ").slice(0, 3).join(" ")), ["novox/mesh-sdk@01234567 mesh/check success",
-    "novox/mesh-sdk@01234567 mesh/check success", "novox/mesh-sdk@02234567 mesh/check pending", "novox/mesh-sdk@01234567 mesh/check pending"]);
-});
-
-test("a forge that refuses throws, so the bus delivers the event again, and the revision is not taken as written", async () => {
-  const { CheckStatuses } = await import("../check.ts");
-  const f = forge();
-  const checks = new CheckStatuses(f, () => {});
-  f.failNext();
-  await assert.rejects(checks.apply(event(body({ revision: 5 }))), /502/);
-  assert.equal(await checks.apply(event(body({ state: "pending", revision: 4 }))), "written", "nothing of 5 was kept");
-  assert.equal(await checks.apply(event(body({ revision: 5 }))), "written");
-});
-
-test("no tool writes mesh/check: gitea_check_status is gone and gitea_commit_status refuses it", async () => {
-  const delivery = await import("../delivery.ts");
-  assert.equal((delivery as Record<string, unknown>).checkStatusTool, undefined, "delivery.ts serves no gitea_check_status");
-  assert.throws(() => delivery.deliveryStatus("mesh/check", "success", "x"), /check\.status/);
-  const index = readFileSync(new URL("../index.ts", import.meta.url), "utf8");
-  assert.ok(!/gitea_check_status|checkStatusTool/.test(index), "index.ts serves no gitea_check_status");
-});

modules/gitea/test/check-owed.test.ts

@@ -0,0 +1,135 @@
+// **mesh/check is written from the checker's owed state, read by a watch** (ADR 0521; R4e round 2 finding R2-1; finding K1
+// before it). No tool writes mesh/check, since a tool is told no caller and node-tools invokes every tool. Nor does an event:
+// nothing in the delivery plan orders this module before the forge bridge, and a durable consumer whose filters change keeps
+// its deliver policy, so an event said before this module consumed it was lost. A watch of the checker's state hands over
+// the latest value of every key when it starts, whenever that is.
+
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { readFileSync } from "node:fs";
+
+const sha = "0123456789abcdef0123456789abcdef01234567";
+const key = (s = sha, repo = "novox/mesh-sdk", n = 12) => `${repo}.${n}.${s}.status=mesh/check`;
+const owed = (over: Record<string, unknown> = {}) => ({ v: 1, act: "status", context: "mesh/check", state: "success",
+  text: "pass 0123456789ab on aaaaaaaaaaaa", target: "https://view.example/checks/g.1", ...over });
+const put = (k: string, value: unknown, revision: number, current = false) =>
+  ({ state: "mesh-delivery-checker.owed", key: k, op: "put" as const, value, revision, current });
+
+function forge() {
+  const sent: string[] = [];
+  let failing = 0;
+  return {
+    sent,
+    fail: (n: number) => { failing = n; },
+    setCommitStatus: async (o: string, r: string, s: string, st: any) => {
+      if (failing > 0) { failing--; throw new Error("Gitea API /repos/x/statuses: 502 bad gateway"); }
+      sent.push(`${o}/${r}@${s.slice(0, 8)} ${st.context} ${st.state} ${st.description}${st.target_url ? " " + st.target_url : ""}`);
+    },
+  };
+}
+const noSleep = async () => {};
+
+test("the checker's owed is what this module reads, and it consumes no check.status", async () => {
+  const { CHECKER_OWED } = await import("../check.ts");
+  assert.equal(CHECKER_OWED, "mesh-delivery-checker.owed");
+  const manifest = JSON.parse(readFileSync(new URL("../module.json", import.meta.url), "utf8"));
+  assert.ok(manifest.reads.includes(CHECKER_OWED), "gitea's manifest reads the checker's owed");
+  assert.ok(!manifest.consumes.some((c: string) => c.includes("check.status")), "no check.status is consumed");
+});
+
+test("an owed mesh/check writes the status on its head: its state, its line cut and its link", async () => {
+  const { CheckStatuses } = await import("../check.ts");
+  const f = forge();
+  const checks = new CheckStatuses(f, () => {}, noSleep);
+  assert.equal(await checks.apply(put(key(), owed(), 7, true)), "written");
+  const other = sha.replace(/0/g, "f");
+  assert.equal(await checks.apply(put(key(other, "novox/my=2erepo"), owed({ state: "pending", text: "y".repeat(300), target: "javascript:x" }), 8)), "written");
+  assert.equal(f.sent[0], "novox/mesh-sdk@01234567 mesh/check success pass 0123456789ab on aaaaaaaaaaaa https://view.example/checks/g.1");
+  assert.ok(f.sent[1].startsWith("novox/my.repo@f1234567 mesh/check pending yyyy"), f.sent[1]);
+  assert.ok(f.sent[1].length < 200, "the line is cut as every status's is");
+  assert.ok(!f.sent[1].includes("javascript"), "a link that is no web page is not written");
+});
+
+test("a key or value that is no owed mesh/check, and a delete, write nothing", async () => {
+  const { CheckStatuses } = await import("../check.ts");
+  const f = forge();
+  const checks = new CheckStatuses(f, () => {}, noSleep);
+  const cases = [
+    put(key(sha.slice(0, 12)), owed(), 1), put(`novox/mesh-sdk.12.${sha}.status=mesh/merge-gate`, owed(), 1),
+    put(`novox/mesh-sdk.12.${sha}.view`, owed(), 1), put(`nokey`, owed(), 1), put(`../x/y.12.${sha}.status=mesh/check`, owed(), 1),
+    put(key(), owed({ context: "mesh/delivery" }), 1), put(key(), owed({ act: "view" }), 1), put(key(), owed({ state: "green" }), 1),
+    put(key(), owed({ v: 2 }), 1), put(key(), null, 1), put(key(), "text", 1), put(key(), owed(), -1),
+    { state: "mesh-delivery-checker.owed", key: key(), op: "delete" as const, revision: 2, current: false },
+  ];
+  for (const c of cases) assert.equal(await checks.apply(c as any), "dropped", JSON.stringify(c));
+  assert.deepEqual(f.sent, []);
+});
+
+test("a revision older than one written on the key is dropped; the same one is written again", async () => {
+  const { CheckStatuses } = await import("../check.ts");
+  const f = forge();
+  const checks = new CheckStatuses(f, () => {}, noSleep);
+  await checks.apply(put(key(), owed({ state: "success" }), 10));
+  assert.equal(await checks.apply(put(key(), owed({ state: "pending" }), 9)), "dropped", "a stale pending never hides the verdict");
+  assert.equal(await checks.apply(put(key(), owed({ state: "success" }), 10)), "written");
+  assert.equal(await checks.apply(put(key(sha.replace(/1/g, "2")), owed({ state: "pending" }), 3)), "written", "another key is its own");
+  assert.equal(await checks.apply(put(key(), owed({ state: "pending" }), 11)), "written");
+  assert.deepEqual(f.sent.map((s) => s.split(" ").slice(0, 3).join(" ")), ["novox/mesh-sdk@01234567 mesh/check success",
+    "novox/mesh-sdk@01234567 mesh/check success", "novox/mesh-sdk@02234567 mesh/check pending", "novox/mesh-sdk@01234567 mesh/check pending"]);
+});
+
+test("a forge that refuses is asked again until it writes, unless a newer revision of the key came meanwhile", async () => {
+  const { CheckStatuses } = await import("../check.ts");
+  const f = forge();
+  const said: string[] = [];
+  const waits: number[] = [];
+  const checks = new CheckStatuses(f, (l) => said.push(l), async (ms) => { waits.push(ms); });
+  f.fail(2);
+  await checks.take(put(key(), owed({ state: "success" }), 5));
+  await checks.settled();
+  assert.deepEqual(f.sent.map((s) => s.split(" ")[2]), ["success"], "written on the third ask");
+  assert.deepEqual(waits, [2_000, 5_000]);
+  assert.match(said[0], /502.*asked again/);
+  // A refusal whose key moved on meanwhile is not asked again: the newer revision is written, the older never after it.
+  const g = forge();
+  let release: () => void = () => {};
+  const gate = new Promise<void>((r) => { release = r; });
+  const later = new CheckStatuses(g, () => {}, () => gate);
+  g.fail(1);
+  await later.take(put(key(), owed({ state: "pending" }), 6));
+  await later.take(put(key(), owed({ state: "success" }), 7));
+  release();
+  await later.settled();
+  assert.deepEqual(g.sent.map((s) => s.split(" ")[2]), ["success"]);
+});
+
+test("a watch refused at first is asked again, said once, and once granted writes the latest revision of each key", async () => {
+  const { CheckStatuses, watchOwed } = await import("../check.ts");
+  const f = forge();
+  const said: string[] = [];
+  const checks = new CheckStatuses(f, () => {}, noSleep);
+  // The checker's bucket as the bus keeps it: two revisions of one key written while the read was refused.
+  const kept = new Map<string, { value: unknown; revision: number }>();
+  kept.set(key(), { value: owed({ state: "pending" }), revision: 3 });
+  kept.set(key(), { value: owed({ state: "success" }), revision: 4 });
+  let asked = 0;
+  const watch = async (handler: (c: any) => Promise<void>) => {
+    asked++;
+    if (asked < 3) throw new Error("gitea keeps and reads no state called \"mesh-delivery-checker.owed\"");
+    for (const [k, e] of kept) await handler(put(k, e.value, e.revision, true));
+  };
+  await watchOwed(watch, checks, noSleep, (l) => said.push(l));
+  await checks.settled();
+  assert.equal(asked, 3);
+  assert.deepEqual(f.sent.map((s) => s.split(" ")[2]), ["success"], "only the latest revision is written");
+  assert.equal(said.filter((l) => /could not be bound/.test(l)).length, 1, `said once: ${said.join(" | ")}`);
+  assert.ok(said.some((l) => /bound after 3 attempts/.test(l)), said.join(" | "));
+});
+
+test("no tool writes mesh/check: gitea_check_status is gone and gitea_commit_status refuses it", async () => {
+  const delivery = await import("../delivery.ts");
+  assert.equal((delivery as Record<string, unknown>).checkStatusTool, undefined, "delivery.ts serves no gitea_check_status");
+  assert.throws(() => delivery.deliveryStatus("mesh/check", "success", "x"), /mesh-delivery-checker\.owed/);
+  const index = readFileSync(new URL("../index.ts", import.meta.url), "utf8");
+  assert.ok(!/gitea_check_status|checkStatusTool/.test(index), "index.ts serves no gitea_check_status");
+});

modules/gitea/test/delivery.test.ts

@@ -62,9 +62,9 @@ test("one view per pull request, found by its marker; only the mesh's statuses",
   assert.equal(deliveryStatus("mesh/delivery", "success", "x", "javascript:alert(1)").target_url, undefined);
 });
 
-// Issue 1301-S4 (ADR 0447 §1; R4e finding K1): main requires mesh/check, so no hand may set it. The status tool every
+// Issue 1301-S4 (ADR 0447 §1; R4e findings K1 and R2-1): main requires mesh/check, so no hand may set it. The status tool
-// caller may use refuses it in every state; no tool writes it at all: it is written from the forge bridge's check.status
+// every caller may use refuses it in every state; no tool writes it at all: it is written from the checker's owed state
-// event (check.ts, test/check-event.test.ts).
+// (check.ts, test/check-owed.test.ts; ADR 0521).
 test("mesh/check is refused by the status tool in every state", async () => {
   const { commitStatusTool, deliveryStatus, CHECK_STATUS } = await import("../delivery.ts");
   assert.equal(CHECK_STATUS, "mesh/check");
@@ -75,7 +75,7 @@ test("mesh/check is refused by the status tool in every state", async () => {
   for (const state of ["pending", "success", "failure", "error", "warning"]) {
     assert.throws(() => deliveryStatus("mesh/check", state, "x"), /mesh\/check is the checker's status/);
     await assert.rejects(commitStatusTool({ setCommitStatus, gates } as any).run({ owner: "novox", repo: "mesh-sdk", sha,
-      context: "mesh/check", state, description: "by hand" }), /check\.status/);
+      context: "mesh/check", state, description: "by hand" }), /mesh-delivery-checker\.owed/);
   }
   assert.deepEqual(sent, [], "nothing was sent for a hand-set mesh/check");
 });