fix(worker): reconcile terminal ACK loss exactly

vinci/test/lib/worker-fixture.mjs

@@ -285,8 +285,13 @@ export class WorkerTestFixture {
     // When > 0, /v1/version answers only after this many ms (to exercise the daemon's timeout).
     this.versionDelayMs = 0;
     this.versionRequests = 0;
+    this.identityRequests = 0;
     this.busServer = null;
     this.busPort = 0;
+    // Tests that attack credential/--id binding may set these before startBus. Ordinary worker
+    // fixtures model the correctly provisioned worker token by stamping the supplied worker name.
+    this.busPrincipal = null;
+    this.busPrincipalRole = "worker";
     mkdirSync(this.reposDir, { recursive: true });
   }
 
@@ -385,6 +390,16 @@ process.exit(r.status ?? 1);
     this.getRequests = [];
     this.evidencePosts = [];
     this.contractRequests = [];
+    // One bearer represents one worker. Most fixtures carry a handoff addressed to that worker;
+    // empty-bus startup cases use w1 unless the test sets busPrincipal explicitly.
+    if (this.busPrincipal === null) {
+      const addressed = [...new Set(
+        handoffs
+          .map((message) => message.to_agent)
+          .filter((principal) => typeof principal === "string" && principal.startsWith("worker:")),
+      )];
+      this.busPrincipal = addressed.length === 1 ? addressed[0] : "worker:w1";
+    }
 
     const server = createServer((request, response) => {
       if (request.method === "GET" && request.url === "/v1/version") {
@@ -408,11 +423,30 @@ process.exit(r.status ?? 1);
         return;
       }
       const url = new URL(request.url, "http://fixture.invalid");
+      if (request.method === "GET" && url.pathname === "/v1/worker-principal") {
+        this.identityRequests += 1;
+        if (this.busPrincipalRole !== "worker") {
+          response.writeHead(403, { "content-type": "application/json" });
+          response.end(JSON.stringify({ detail: "worker bearer required for authenticated worker identity" }));
+          return;
+        }
+        response.writeHead(200, { "content-type": "application/json" });
+        response.end(JSON.stringify({ worker_principal: this.busPrincipal }));
+        return;
+      }
       if (request.method === "GET" && url.pathname === "/v1/messages") {
         const limit = Number(url.searchParams.get("limit") ?? 100);
         const offset = Number(url.searchParams.get("offset") ?? 0);
         this.getRequests.push({ limit, offset });
+        const fromAgent = url.searchParams.get("from");
+        const kind = url.searchParams.get("kind");
+        const since = url.searchParams.get("since");
+        const postedBy = url.searchParams.get("posted_by");
         const messages = this.busMessages
+          .filter((message) => fromAgent === null || message.from_agent === fromAgent)
+          .filter((message) => kind === null || message.kind === kind)
+          .filter((message) => since === null || message.ts >= since)
+          .filter((message) => postedBy === null || message.posted_by === postedBy)
           .slice()
           .sort((left, right) => left.ts.localeCompare(right.ts) || left.message_id.localeCompare(right.message_id));
         response.writeHead(200, { "content-type": "application/json" });
@@ -427,6 +461,17 @@ process.exit(r.status ?? 1);
         });
         request.on("end", () => {
           const message = JSON.parse(body);
+          const authenticatedPrincipal = this.busPrincipal ?? message.from_agent;
+          if (
+            this.busPrincipalRole === "worker"
+            && message.from_agent !== undefined
+            && message.from_agent !== authenticatedPrincipal
+          ) {
+            this.rejectedPosts.push(message);
+            response.writeHead(400, { "content-type": "application/json" });
+            response.end(JSON.stringify({ error: "from_agent must match authenticated worker principal" }));
+            return;
+          }
           const invalidRefs = (message.refs ?? []).filter((ref) => !LEDGER_REF.test(ref));
           if (invalidRefs.length > 0) {
             this.rejectedPosts.push(message);
@@ -440,9 +485,25 @@ process.exit(r.status ?? 1);
             response.end(JSON.stringify({ error: "fixture: post refused" }));
             return;
           }
+          const record = {
+            message_id: `msg_fixture_${this.postedMessages.length + 1}`,
+            ts: new Date().toISOString(),
+            from_agent: this.busPrincipalRole === "worker" ? authenticatedPrincipal : message.from_agent,
+            posted_by: authenticatedPrincipal,
+            to_agent: message.to_agent ?? null,
+            kind: message.kind,
+            subject: message.subject ?? "",
+            body: message.body ?? "",
+            outcome: message.outcome ?? null,
+            in_reply_to: message.in_reply_to ?? null,
+            refs: message.refs ?? [],
+          };
+          // Preserve the existing fixture contract: assertions over postedMessages inspect the
+          // client payload. busMessages models the server's durable, authenticated representation.
           this.postedMessages.push(message);
+          this.busMessages.push(record);
           response.writeHead(200, { "content-type": "application/json" });
-          response.end(JSON.stringify({ ok: true }));
+          response.end(JSON.stringify({ message_id: record.message_id, ts: record.ts }));
         });
         return;
       }

vinci/test/worker-lease-loop.mjs

@@ -523,6 +523,7 @@ await test('release failure is logged and never changes the state', async () =>
     assert.equal(governor.releases.length, 1);
     const final = fixture.getPostedMessages().find((p) => p.subject === 'task 112 completed');
     assert(final, 'final post still happens');
+    assert.match(final.body, /(?:^| )attempt=112\/1(?: |$)/, 'terminal binds the exact lifecycle attempt');
   } finally {
     await governor.close();
     await fixture.cleanup();
@@ -1175,6 +1176,7 @@ function declarationPosts(fixture, workerId) {
 
 // A daemon (no --once) so the interval can actually fire.
 function spawnDaemon(fixture, workerId, extraArgs, envOverrides = {}) {
+  fixture.busPrincipal = `worker:${workerId}`;
   const env = fixture.getEnv({ VINCI_GOVERNOR_TOKEN: 'gov-token', ...envOverrides });
   const proc = spawn('node', [WORKER, 'start', '--id', workerId, '--server', fixture.busUrl(), '--state-dir', fixture.tempDir, ...extraArgs], { env, stdio: 'pipe' });
   let stderr = '';

vinci/test/worker-lifecycle-integration.mjs

@@ -29,6 +29,16 @@ function envelope(overrides = {}) {
 async function fakeBus(body) {
   const posts = [];
   const onlinePosts = [];
+  const messages = [{
+    message_id: "1",
+    to_agent: "worker:t1",
+    kind: "handoff",
+    subject: "lifecycle task",
+    body,
+    ts: "2026-08-26T10:00:00Z",
+    posted_by: "scheduler",
+  }];
+  let nextMessage = 2;
   const server = createServer((request, response) => {
     // W0.5: GET /v1/version is unauthenticated by contract; this bus does not serve it, so the
     // daemon records `server_build={error}` and still starts.
@@ -38,21 +48,28 @@ async function fakeBus(body) {
       return;
     }
     assert.equal(request.headers.authorization, "Bearer test-token");
+    if (request.method === "GET" && request.url === "/v1/worker-principal") {
+      response.setHeader("content-type", "application/json");
+      response.end(JSON.stringify({ worker_principal: "worker:t1" }));
+      return;
+    }
     if (request.method === "GET" && request.url?.startsWith("/v1/messages")) {
+      const url = new URL(request.url, "http://fixture.invalid");
+      const fromAgent = url.searchParams.get("from");
+      const kind = url.searchParams.get("kind");
+      const since = url.searchParams.get("since");
+      const limit = Number(url.searchParams.get("limit") ?? 100);
+      const offset = Number(url.searchParams.get("offset") ?? 0);
+      const filtered = messages.filter((message) =>
+        (fromAgent === null || message.from_agent === fromAgent)
+        && (kind === null || message.kind === kind)
+        && (since === null || message.ts >= since));
       response.setHeader("content-type", "application/json");
       response.end(JSON.stringify({
-        messages: [{
-          message_id: "1",
-          to_agent: "worker:t1",
-          kind: "handoff",
-          subject: "lifecycle task",
-          body,
-          ts: "2026-08-26T10:00:00Z",
-          posted_by: "scheduler",
-        }],
-        total: 1,
-        limit: 100,
-        offset: 0,
+        messages: filtered.slice(offset, offset + limit),
+        total: filtered.length,
+        limit,
+        offset,
       }));
       return;
     }
@@ -64,12 +81,26 @@ async function fakeBus(body) {
       });
       request.on("end", () => {
         const post = JSON.parse(raw);
+        const row = {
+          message_id: String(nextMessage++),
+          ts: new Date().toISOString(),
+          from_agent: "worker:t1",
+          posted_by: "worker:t1",
+          to_agent: null,
+          kind: post.kind,
+          subject: post.subject ?? "",
+          body: post.body ?? "",
+          outcome: post.outcome ?? null,
+          in_reply_to: post.in_reply_to ?? null,
+          refs: post.refs ?? [],
+        };
+        messages.push(row);
         // The per-start `worker <id> online` status (W0.5) is not a task post; keep the
         // per-task kind sequences below exact by recording it separately.
         if (/ online$/.test(post.subject)) onlinePosts.push(post);
         else posts.push(post);
         response.setHeader("content-type", "application/json");
-        response.end("{}");
+        response.end(JSON.stringify({ message_id: row.message_id, ts: row.ts }));
       });
       return;
     }

vinci/test/worker-lock-integration.mjs

@@ -21,10 +21,16 @@ const fixture = new WorkerTestFixture("lock");
 try {
   fixture.createRepo("test", "repo");
   fixture.linkTools(TOOLS);
+  fixture.busPrincipal = "worker:locked";
   await fixture.startBus([]);
   const args = [join(ROOT, "vinci/worker/worker.mjs"), "start", "--id", "locked", "--server", fixture.busUrl(), "--state-dir", fixture.tempDir];
   const first = spawn("node", [...args, "--poll-seconds", "60"], { env: fixture.getEnv(), stdio: "pipe" });
-  await waitFor(() => existsSync(join(fixture.tempDir, "daemon.lock")) && fixture.getRequests.length === 1, "first daemon lock and poll");
+  await waitFor(
+    () => existsSync(join(fixture.tempDir, "daemon.lock"))
+      && fixture.identityRequests === 1
+      && fixture.getRequests.length === 1,
+    "first daemon identity lookup and poll",
+  );
   const getsBeforeSecond = fixture.getRequests.length;
   const second = spawn("node", [...args, "--once"], { env: fixture.getEnv(), stdio: ["ignore", "pipe", "pipe"] });
   let secondStderr = "";