diff --git a/vinci/test/lib/worker-fixture.mjs b/vinci/test/lib/worker-fixture.mjs index ec749fe97..ef553a8df 100644 --- a/vinci/test/lib/worker-fixture.mjs +++ b/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; } diff --git a/vinci/test/worker-lease-loop.mjs b/vinci/test/worker-lease-loop.mjs index 105c52f70..927beb84e 100644 --- a/vinci/test/worker-lease-loop.mjs +++ b/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 = ''; diff --git a/vinci/test/worker-lifecycle-integration.mjs b/vinci/test/worker-lifecycle-integration.mjs index af1969094..e131838e3 100644 --- a/vinci/test/worker-lifecycle-integration.mjs +++ b/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 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; } diff --git a/vinci/test/worker-lock-integration.mjs b/vinci/test/worker-lock-integration.mjs index 2ed65bf35..90cba21d1 100644 --- a/vinci/test/worker-lock-integration.mjs +++ b/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 = ""; diff --git a/vinci/test/worker-terminal-outbox.mjs b/vinci/test/worker-terminal-outbox.mjs index 28726d7c1..c16a49a32 100644 --- a/vinci/test/worker-terminal-outbox.mjs +++ b/vinci/test/worker-terminal-outbox.mjs @@ -15,8 +15,14 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import test from "node:test"; -import { BusClient } from "../worker/bus.mjs"; -import { DEFAULT_OUTBOX_DIR, listPending, replayPending } from "../worker/outbox.mjs"; +import { BusClient, WorkerIdentityRefusal } from "../worker/bus.mjs"; +import { + DEFAULT_OUTBOX_DIR, + DUPLICATE_DELIVERY, + listPending, + recordPending, + replayPending, +} from "../worker/outbox.mjs"; function scratch() { return mkdtempSync(join(tmpdir(), "vinci-outbox-")); @@ -24,29 +30,51 @@ function scratch() { // A bus that cannot reach anything: 127.0.0.1:9 is the discard port. function unreachableBus(dir) { - return new BusClient("http://127.0.0.1:9/nope", "t", 100, dir); + return new BusClient("http://127.0.0.1:9/nope", "t", 100, dir, "worker:test"); } -test("a terminal post that FAILS leaves a durable record", async () => { +function recordTerminal(dir, subject, outcome, inReplyTo = null) { + return recordPending({ + kind: "status", + subject, + body: "body", + options: { outcome, ...(inReplyTo === null ? {} : { inReplyTo }) }, + configured_worker_principal: "worker:test", + expected_posted_by: "worker:test", + }, dir); +} + +test("a terminal identity outage preserves the terminal evidence before refusing", async () => { const dir = join(scratch(), "outbox"); const bus = unreachableBus(dir); await assert.rejects( () => bus.postTerminal("status", "task X failed", "body", { outcome: "FAILED" }), + (error) => error instanceof WorkerIdentityRefusal + && error.refused === true + && error.code === "worker_identity_unavailable", ); - const pending = listPending(dir); - assert.equal(pending.length, 1, "the undelivered terminal must be on disk"); - assert.equal(pending[0].entry.options.outcome, "FAILED"); - assert.equal(pending[0].entry.subject, "task X failed"); - assert.equal(pending[0].entry.kind, "status"); + const [pending] = listPending(dir); + assert.equal(pending.entry.configured_worker_principal, "worker:test"); + assert.equal(pending.entry.expected_posted_by, undefined, "an unavailable server cannot be a principal source"); }); -test("a terminal post that SUCCEEDS leaves nothing behind", async () => { +test("a terminal post with no configured worker identity writes nothing", async () => { const dir = join(scratch(), "outbox"); - const bus = unreachableBus(dir); - // Replace the transport, keeping postTerminal's own logic under test. - bus.post = async () => ({ ok: true }); - await bus.postTerminal("status", "task Y done", "body", { outcome: "COMPLETED" }); - assert.equal(listPending(dir).length, 0, "a delivered record must not linger"); + const bus = new BusClient("http://127.0.0.1:9/nope", "t", 100, dir); + await assert.rejects( + () => bus.postTerminal("status", "task X failed", "body", { outcome: "FAILED" }), + (error) => error instanceof WorkerIdentityRefusal + && error.code === "worker_identity_unconfigured", + ); + assert.equal(listPending(dir).length, 0); +}); + +test("a pending terminal record carries the authenticated principal binding", () => { + const dir = join(scratch(), "outbox"); + recordTerminal(dir, "task Y done", "COMPLETED"); + const [pending] = listPending(dir); + assert.equal(pending.entry.expected_posted_by, "worker:test"); + assert.equal(pending.entry.options.outcome, "COMPLETED"); }); test("an invalid outcome is refused BEFORE anything is written", async () => { @@ -63,13 +91,15 @@ test("an invalid outcome is refused BEFORE anything is written", async () => { test("replay delivers what was undelivered, then clears it", async () => { const dir = join(scratch(), "outbox"); - const bus = unreachableBus(dir); - await assert.rejects(() => bus.postTerminal("status", "s1", "b", { outcome: "BLOCKED" })); - await assert.rejects(() => bus.postTerminal("status", "s2", "b", { outcome: "UNVERIFIED" })); + recordTerminal(dir, "s1", "BLOCKED"); + recordTerminal(dir, "s2", "UNVERIFIED"); assert.equal(listPending(dir).length, 2); const delivered = []; - const good = { post: async (k, s, b, o) => { delivered.push([s, o.outcome]); } }; + const good = { + findTerminalDeliveries: async () => [], + deliverPendingTerminal: async (entry) => { delivered.push([entry.subject, entry.options.outcome]); }, + }; const summary = await replayPending(good, dir, { warn() {}, error() {} }); assert.equal(summary.delivered, 2); @@ -80,10 +110,12 @@ test("replay delivers what was undelivered, then clears it", async () => { test("replay that STILL fails keeps the record rather than dropping it", async () => { const dir = join(scratch(), "outbox"); - const bus = unreachableBus(dir); - await assert.rejects(() => bus.postTerminal("status", "s", "b", { outcome: "FAILED" })); + recordTerminal(dir, "s", "FAILED"); - const stillBroken = { post: async () => { throw new Error("bus down"); } }; + const stillBroken = { + findTerminalDeliveries: async () => [], + deliverPendingTerminal: async () => { throw new Error("bus down"); }, + }; const summary = await replayPending(stillBroken, dir, { warn() {}, error() {} }); assert.equal(summary.failed, 1); @@ -95,14 +127,15 @@ test("a corrupt record is reported, never silently dropped", async () => { // A record we cannot read is still evidence that something terminal went // unannounced. Deleting it would destroy the only trace. const dir = join(scratch(), "outbox"); - const bus = unreachableBus(dir); - await assert.rejects(() => bus.postTerminal("status", "s", "b", { outcome: "FAILED" })); + recordTerminal(dir, "s", "FAILED"); const [name] = readdirSync(dir); writeFileSync(join(dir, name), "{ this is not json"); const errors = []; const summary = await replayPending( - { post: async () => {} }, dir, { warn() {}, error: (m) => errors.push(m) }, + { findTerminalDeliveries: async () => [], deliverPendingTerminal: async () => {} }, + dir, + { warn() {}, error: (m) => errors.push(m) }, ); assert.equal(summary.corrupt, 1); assert.equal(summary.delivered, 0); @@ -116,8 +149,95 @@ test("the bus records into ITS OWN directory, not the process cwd", async () => // replayed from another is an inert fix that looks like a working one. const dir = join(scratch(), "outbox"); const bus = unreachableBus(dir); - await assert.rejects(() => bus.postTerminal("status", "s", "b", { outcome: "FAILED" })); + recordTerminal(bus.outboxDir, "s", "FAILED"); assert.equal(bus.outboxDir, dir); assert.equal(listPending(dir).length, 1); assert.notEqual(dir, DEFAULT_OUTBOX_DIR); }); + +test("one exact committed row reconciles ACK loss without a second POST", async () => { + const dir = join(scratch(), "outbox"); + recordTerminal(dir, "task exact done", "COMPLETED", "msg_exact"); + + let posts = 0; + const summary = await replayPending({ + findTerminalDeliveries: async () => ["msg_terminal_1"], + deliverPendingTerminal: async () => { posts += 1; }, + }, dir, { warn() {}, error() {} }); + + assert.equal(summary.reconciled, 1); + assert.equal(summary.delivered, 0); + assert.equal(summary.duplicate, 0); + assert.equal(posts, 0, "an observed exact terminal effect must not be posted again"); + assert.equal(listPending(dir).length, 0); +}); + +test("two exact committed rows preserve a typed duplicate condition and post no third row", async () => { + const dir = join(scratch(), "outbox"); + recordTerminal(dir, "task duplicate done", "COMPLETED", "msg_duplicate"); + + let posts = 0; + const errors = []; + const summary = await replayPending({ + findTerminalDeliveries: async () => ["msg_terminal_1", "msg_terminal_2"], + deliverPendingTerminal: async () => { posts += 1; }, + }, dir, { warn() {}, error: (message) => errors.push(message) }); + + assert.equal(summary.duplicate, 1); + assert.equal(summary.reconciled, 0); + assert.equal(summary.delivered, 0); + assert.equal(posts, 0, "an existing duplicate must never grow to three rows"); + assert.equal(summary.conditions[0].type, DUPLICATE_DELIVERY); + assert.equal(summary.conditions[0].exact_match_count, 2); + assert.match(errors.join("\n"), /DUPLICATE_DELIVERY/); + const [pending] = listPending(dir); + assert.equal(pending.entry.delivery_condition.type, DUPLICATE_DELIVERY); + assert.deepEqual(pending.entry.delivery_condition.message_ids, ["msg_terminal_1", "msg_terminal_2"]); +}); + +test("reconciliation failure retains the entry and posts nothing", async () => { + const dir = join(scratch(), "outbox"); + recordTerminal(dir, "task uncertain", "FAILED", "msg_uncertain"); + + let posts = 0; + const summary = await replayPending({ + findTerminalDeliveries: async () => { throw new Error("pagination incomplete"); }, + deliverPendingTerminal: async () => { posts += 1; }, + }, dir, { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(posts, 0, "an incomplete observation is not proof of absence"); + assert.equal(listPending(dir).length, 1); +}); + +test("a legacy pending record without authenticated provenance is retained, never adopted", async () => { + const dir = join(scratch(), "outbox"); + const bus = unreachableBus(dir); + recordTerminal(dir, "legacy", "FAILED", "msg_legacy"); + const [pending] = listPending(dir); + delete pending.entry.configured_worker_principal; + delete pending.entry.expected_posted_by; + writeFileSync(pending.path, JSON.stringify(pending.entry)); + + let posts = 0; + const summary = await replayPending(bus, dir, { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(posts, 0); + assert.equal(listPending(dir).length, 1); +}); + +test("a pending record bound to another worker principal is retained", async () => { + const dir = join(scratch(), "outbox"); + const bus = unreachableBus(dir); + recordTerminal(dir, "other worker", "FAILED", "msg_other"); + const [pending] = listPending(dir); + pending.entry.configured_worker_principal = "worker:somebody-else"; + pending.entry.expected_posted_by = "worker:somebody-else"; + writeFileSync(pending.path, JSON.stringify(pending.entry)); + + const summary = await replayPending(bus, dir, { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(listPending(dir).length, 1); +}); diff --git a/vinci/test/worker-terminal-reconcile-integration.mjs b/vinci/test/worker-terminal-reconcile-integration.mjs new file mode 100644 index 000000000..f0b05a339 --- /dev/null +++ b/vinci/test/worker-terminal-reconcile-integration.mjs @@ -0,0 +1,995 @@ +// Consumer-boundary proof for terminal outbox reconciliation. The direct cases drive the real +// HTTP BusClient; the final case drives two fresh `vinci worker` processes and can be pointed at +// an unpacked/installed candidate with VINCI_TEST_WORKER_LAUNCHER. +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { createServer } from "node:http"; +import { mkdirSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { dirname, join, resolve } from "node:path"; +import test from "node:test"; +import { fileURLToPath } from "node:url"; + +import { + BusClient, + WorkerIdentityRefusal, + WorkerTerminalAuthorityConflict, +} from "../worker/bus.mjs"; +import { + DUPLICATE_DELIVERY, + listPending, + recordPending, + replayPending, +} from "../worker/outbox.mjs"; + +const ROOT = resolve(dirname(fileURLToPath(import.meta.url)), "../.."); +const WORKER_ID = "lane-b-worker"; +const POSTED_BY = `worker:${WORKER_ID}`; +const EXPECTED_WORKER_HEADER = "x-vgc-expected-worker-principal"; +const SERVER_STRIP_EDGE = /^[\u0009-\u000d\u001c-\u0020\u0085\u00a0\u1680\u2000-\u200a\u2028\u2029\u202f\u205f\u3000]+|[\u0009-\u000d\u001c-\u0020\u0085\u00a0\u1680\u2000-\u200a\u2028\u2029\u202f\u205f\u3000]+$/gu; +const TERMINAL = Object.freeze({ + kind: "status", + subject: "task msg_lane_b completed", + body: "state=COMPLETED contract=wo_lane_b@01234567 attempt=msg_lane_b/1 economics_sha256=e".concat( + "1".repeat(63), + " evidence_uri=s3://evidence/msg_lane_b/1 evidence_sha256=", + "a".repeat(64), + ), + options: { + outcome: "COMPLETED", + inReplyTo: "msg_lane_b", + refs: ["job_lane_b"], + }, + configured_worker_principal: POSTED_BY, + expected_posted_by: POSTED_BY, +}); + +function terminalRow(id, overrides = {}) { + return { + message_id: id, + ts: "2026-09-13T12:00:00.000Z", + from_agent: POSTED_BY, + posted_by: POSTED_BY, + posted_role: "worker", + to_agent: null, + kind: TERMINAL.kind, + subject: TERMINAL.subject, + body: TERMINAL.body, + outcome: TERMINAL.options.outcome, + in_reply_to: TERMINAL.options.inReplyTo, + refs: TERMINAL.options.refs, + ...overrides, + }; +} + +function serverStrip(value) { + return value.replace(SERVER_STRIP_EDGE, ""); +} + +class TerminalBusFixture { + constructor(messages = [], { + authenticatedPrincipal = POSTED_BY, + principalRole = "worker", + postAuthenticatedPrincipal = authenticatedPrincipal, + postPrincipalRole = principalRole, + forcedBoundRefusalStatus = null, + } = {}) { + this.messages = messages.slice(); + this.posts = []; + this.nextId = 1; + this.authenticatedPrincipal = authenticatedPrincipal; + this.principalRole = principalRole; + this.postAuthenticatedPrincipal = postAuthenticatedPrincipal; + this.postPrincipalRole = postPrincipalRole; + this.forcedBoundRefusalStatus = forcedBoundRefusalStatus; + this.identityStatus = principalRole === "worker" ? 200 : 403; + this.identityPayload = { worker_principal: authenticatedPrincipal }; + this.identityRaw = null; + this.identityDelayMs = 0; + this.identityRequests = 0; + this.identityRequestUrls = []; + this.messageGetRequests = 0; + this.messagePostRequests = []; + this.dropAckSubject = null; + this.droppedAck = false; + this.breakSecondPage = false; + this.identityReadbackTransform = null; + this.terminalReadbackTransform = null; + this.afterMessageGet = null; + this.server = null; + this.url = null; + this.terminalCommitted = null; + this.resolveTerminalCommitted = null; + } + + async start() { + this.terminalCommitted = new Promise((resolveCommitted) => { + this.resolveTerminalCommitted = resolveCommitted; + }); + this.server = createServer((request, response) => { + const url = new URL(request.url, "http://fixture.invalid"); + if (request.method === "GET" && url.pathname === "/v1/version") { + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ component: "lane-b-fixture", git_sha: "f".repeat(40), dirty: false })); + return; + } + if (request.headers.authorization !== "Bearer test-token") { + response.writeHead(401); + response.end(); + return; + } + if (request.method === "GET" && url.pathname === "/v1/worker-principal") { + this.identityRequests += 1; + this.identityRequestUrls.push(request.url); + const sendIdentity = () => { + response.writeHead(this.identityStatus, { "content-type": "application/json" }); + response.end(this.identityRaw ?? JSON.stringify(this.identityPayload)); + }; + if (this.identityDelayMs > 0) setTimeout(sendIdentity, this.identityDelayMs); + else sendIdentity(); + return; + } + if (request.method === "GET" && url.pathname === "/v1/messages") { + this.messageGetRequests += 1; + const limit = Number(url.searchParams.get("limit") ?? 100); + const offset = Number(url.searchParams.get("offset") ?? 0); + const fromAgent = url.searchParams.get("from"); + const postedBy = url.searchParams.get("posted_by"); + const kind = url.searchParams.get("kind"); + const since = url.searchParams.get("since"); + const filtered = this.messages.filter((message) => + (fromAgent === null || message.from_agent === fromAgent) + && (postedBy === null || message.posted_by === postedBy) + && (kind === null || message.kind === kind) + && (since === null || message.ts >= since)); + let page = this.breakSecondPage && offset > 0 ? [] : filtered.slice(offset, offset + limit); + if (this.identityReadbackTransform !== null && fromAgent !== null && since !== null) { + page = page.map((message) => this.identityReadbackTransform({ ...message })); + } + if (this.terminalReadbackTransform !== null && postedBy !== null && kind !== null) { + page = page.map((message) => this.terminalReadbackTransform({ ...message })); + } + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ messages: page, total: filtered.length, limit, offset })); + this.afterMessageGet?.(); + return; + } + if (request.method === "POST" && url.pathname === "/v1/messages") { + let raw = ""; + request.setEncoding("utf8"); + request.on("data", (chunk) => { raw += chunk; }); + request.on("end", () => { + const payload = JSON.parse(raw); + const expectedHeaders = []; + for (let index = 0; index < request.rawHeaders.length; index += 2) { + if (request.rawHeaders[index].toLowerCase() === EXPECTED_WORKER_HEADER) { + expectedHeaders.push(request.rawHeaders[index + 1]); + } + } + this.messagePostRequests.push({ expectedHeaders, payload }); + if ( + expectedHeaders.length > 0 + && ( + expectedHeaders.length !== 1 + || !/^worker:[A-Za-z0-9][A-Za-z0-9._-]{0,56}$/.test(expectedHeaders[0]) + ) + ) { + response.writeHead(422, { "content-type": "application/json" }); + response.end(JSON.stringify({ detail: "expected worker principal must be supplied exactly once" })); + return; + } + if (expectedHeaders.length === 1 && this.forcedBoundRefusalStatus !== null) { + response.writeHead(this.forcedBoundRefusalStatus, { "content-type": "application/json" }); + response.end(JSON.stringify({ detail: "forced bound publication refusal" })); + return; + } + if ( + expectedHeaders.length === 1 + && ( + this.postPrincipalRole !== "worker" + || this.postAuthenticatedPrincipal !== expectedHeaders[0] + ) + ) { + const status = this.postPrincipalRole === "ambiguous" ? 403 : 412; + response.writeHead(status, { "content-type": "application/json" }); + response.end(JSON.stringify({ detail: "authenticated POST principal does not satisfy expected worker" })); + return; + } + if (typeof payload.subject !== "string" || serverStrip(payload.subject).length === 0) { + response.writeHead(422, { "content-type": "application/json" }); + response.end(JSON.stringify({ detail: "subject must be a non-empty string" })); + return; + } + const storedSubject = serverStrip(payload.subject); + if ([...storedSubject].length > 200) { + response.writeHead(422, { "content-type": "application/json" }); + response.end(JSON.stringify({ detail: "subject must be <= 200 characters" })); + return; + } + if (payload.body !== null && payload.body !== undefined && typeof payload.body !== "string") { + response.writeHead(422, { "content-type": "application/json" }); + response.end(JSON.stringify({ detail: "body must be a string" })); + return; + } + if (typeof payload.body === "string" && payload.body.length > 0 && [...payload.body].length > 8_000) { + response.writeHead(422, { "content-type": "application/json" }); + response.end(JSON.stringify({ detail: "body must be <= 8000 characters" })); + return; + } + const storedBody = typeof payload.body === "string" ? serverStrip(payload.body) || null : null; + const row = { + message_id: `msg_server_${this.nextId++}`, + ts: new Date().toISOString(), + from_agent: this.postPrincipalRole === "worker" ? this.postAuthenticatedPrincipal : payload.from_agent, + posted_by: this.postAuthenticatedPrincipal, + posted_role: this.postPrincipalRole === "worker" ? "worker" : null, + to_agent: payload.to_agent ?? null, + kind: payload.kind, + subject: storedSubject, + body: storedBody, + outcome: payload.outcome ?? null, + in_reply_to: payload.in_reply_to ?? null, + refs: payload.refs ?? [], + }; + this.messages.push(row); + this.posts.push(row); + if (row.subject === TERMINAL.subject) this.resolveTerminalCommitted?.(row); + if (row.subject === this.dropAckSubject && !this.droppedAck) { + this.droppedAck = true; + request.socket.destroy(); + return; + } + response.writeHead(201, { "content-type": "application/json" }); + response.end(JSON.stringify({ message_id: row.message_id, ts: row.ts })); + }); + return; + } + response.writeHead(404); + response.end(); + }); + await new Promise((resolveListen) => this.server.listen(0, "127.0.0.1", resolveListen)); + this.url = `http://127.0.0.1:${this.server.address().port}`; + } + + async close() { + if (!this.server) return; + this.server.closeAllConnections?.(); + await new Promise((resolveClose) => this.server.close(resolveClose)); + } +} + +function scratch(name) { + return mkdtempSync(join(tmpdir(), `vinci-terminal-${name}-`)); +} + +function recordTerminal(dir) { + return recordPending(TERMINAL, join(dir, "outbox")); +} + +async function authenticatedBus(fixture, dir, pageSize = 100) { + const bus = new BusClient(fixture.url, "test-token", pageSize, join(dir, "outbox"), POSTED_BY); + await bus.establishAuthenticatedPostingPrincipal(); + return bus; +} + +function terminalPostCount(fixture) { + return fixture.posts.filter((row) => row.subject === TERMINAL.subject).length; +} + +function runWorker(launcher, serverUrl, stateDir, { once = true } = {}) { + const args = [ + "worker", "start", + "--id", WORKER_ID, + "--server", serverUrl, + "--state-dir", stateDir, + "--poll-seconds", "300", + ]; + if (once) args.push("--once"); + const child = spawn(launcher, args, { + env: { + ...process.env, + HOME: join(stateDir, "home"), + PATH: `${dirname(launcher)}:${process.env.PATH}`, + VINCI_BUS_TOKEN: "test-token", + VINCI_NO_BOOTSTRAP_HEAL: "1", + VINCI_UPDATE_DISABLED: "1", + }, + stdio: ["ignore", "pipe", "pipe"], + }); + let stderr = ""; + child.stderr.setEncoding("utf8"); + child.stderr.on("data", (chunk) => { stderr += chunk; }); + return { child, stderr: () => stderr }; +} + +function waitForExit(child, timeoutMs = 15_000) { + return new Promise((resolveExit, rejectExit) => { + const timeout = setTimeout(() => { + child.kill("SIGKILL"); + rejectExit(new Error(`worker did not exit within ${timeoutMs} ms`)); + }, timeoutMs); + child.once("close", (code, signal) => { + clearTimeout(timeout); + resolveExit({ code, signal }); + }); + }); +} + +async function withTimeout(promise, message, timeoutMs = 15_000) { + let timeout; + try { + return await Promise.race([ + promise, + new Promise((_, rejectTimeout) => { + timeout = setTimeout(() => rejectTimeout(new Error(message)), timeoutMs); + }), + ]); + } finally { + clearTimeout(timeout); + } +} + +test("different raw posted_by with worker role is filter-inconsistent and cannot publish or reconcile", async (t) => { + const dir = scratch("provenance"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([terminalRow("msg_filter_inconsistent")]); + fixture.terminalReadbackTransform = (message) => ({ ...message, posted_by: "worker:other" }); + await fixture.start(); + t.after(() => fixture.close()); + recordTerminal(dir); + + const bus = await authenticatedBus(fixture, dir); + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(summary.reconciled, 0); + assert.equal(summary.delivered, 0, "a filter-inconsistent row is not evidence of absence"); + assert.equal(fixture.messagePostRequests.length, 0); + assert.equal(listPending(join(dir, "outbox")).length, 1); +}); + +test("identity binding comes from the worker-only server endpoint without a publication probe", async (t) => { + const dir = scratch("identity-readback"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture(); + await fixture.start(); + t.after(() => fixture.close()); + const bus = await authenticatedBus(fixture, dir, 1); + assert.equal(bus.authenticatedPostingPrincipal, POSTED_BY); + assert.equal(fixture.identityRequests, 1); + assert.deepEqual(fixture.identityRequestUrls, ["/v1/worker-principal"]); + assert.equal(fixture.posts.length, 0, "identity resolution itself must be side-effect free"); +}); + +test("every bound worker POST carries the exact configured atomic precondition", async (t) => { + const dir = scratch("bound-post-header"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture(); + await fixture.start(); + t.after(() => fixture.close()); + const bus = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + + await bus.post("status", "bound non-terminal status", "still running"); + + assert.equal(fixture.identityRequests, 0, "the POST precondition is not derived from identity discovery"); + assert.equal(fixture.posts.length, 1); + assert.deepEqual(fixture.messagePostRequests[0].expectedHeaders, [POSTED_BY]); + assert.equal(fixture.messagePostRequests[0].payload.from_agent, undefined); + assert.equal(fixture.messagePostRequests[0].payload.posted_by, undefined); + assert.equal(fixture.posts[0].posted_by, POSTED_BY, "the server stamps publication identity"); +}); + +test("TOCTOU identity discovery cannot authorize a differently authenticated terminal POST", async (t) => { + const dir = scratch("atomic-toctou"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([], { + authenticatedPrincipal: POSTED_BY, + principalRole: "worker", + postAuthenticatedPrincipal: "george", + postPrincipalRole: "admin", + }); + await fixture.start(); + t.after(() => fixture.close()); + const bus = await authenticatedBus(fixture, dir); + + await assert.rejects( + () => bus.postTerminal("status", TERMINAL.subject, TERMINAL.body, TERMINAL.options), + (error) => error instanceof WorkerIdentityRefusal + && error.code === "worker_publication_precondition_refused" + && /failed: 412/.test(error.message), + ); + + assert.equal(fixture.identityRequests, 2, "both discovery reads still report the claimed worker"); + assert.equal(fixture.messagePostRequests.length, 1, "there is no unbound downgrade retry"); + assert.deepEqual(fixture.messagePostRequests[0].expectedHeaders, [POSTED_BY]); + assert.equal(fixture.posts.length, 0, "the server-style guard refuses before row insertion"); + assert.equal(fixture.messages.length, 0); + assert.equal(listPending(join(dir, "outbox")).length, 1, "refusal retains terminal evidence"); +}); + +for (const variant of [ + { name: "body with trailing newline", subject: TERMINAL.subject, body: `${TERMINAL.body}\n`, storedBody: TERMINAL.body }, + { name: "subject with surrounding server whitespace", subject: `\u0085 ${TERMINAL.subject} \u001c`, body: TERMINAL.body, storedBody: TERMINAL.body }, + { name: "whitespace-only body stored as null", subject: TERMINAL.subject, body: " \n\u0085", storedBody: null }, +]) { + test(`ACK loss reconciles the canonical server row for ${variant.name}`, async (t) => { + const dir = scratch(`canonical-${variant.name.replaceAll(" ", "-")}`); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture(); + fixture.dropAckSubject = serverStrip(variant.subject); + await fixture.start(); + t.after(() => fixture.close()); + const beforeCrash = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + + await assert.rejects( + () => beforeCrash.postTerminal("status", variant.subject, variant.body, TERMINAL.options), + /fetch failed/, + ); + assert.equal(fixture.posts.length, 1, "the first bound POST committed before its ACK was lost"); + assert.equal(fixture.posts[0].subject, TERMINAL.subject); + assert.equal(fixture.posts[0].body, variant.storedBody); + assert.equal(fixture.posts[0].posted_by, POSTED_BY); + assert.equal(fixture.posts[0].posted_role, "worker"); + assert.equal(listPending(join(dir, "outbox")).length, 1); + + const restarted = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + const summary = await replayPending(restarted, join(dir, "outbox"), { warn() {}, error() {} }); + const terminalRequests = fixture.messagePostRequests.filter(({ payload }) => + serverStrip(payload.subject) === TERMINAL.subject); + + assert.equal(summary.reconciled, 1); + assert.equal(summary.delivered, 0); + assert.equal(terminalRequests.length, 1, "canonical reconciliation must not append a duplicate"); + assert.equal(fixture.posts.length, 1); + assert.equal(listPending(join(dir, "outbox")).length, 0); + }); +} + +test("one exact worker-authoritative row reconciles and clears pending without POST", async (t) => { + const dir = scratch("worker-authority-positive"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([terminalRow("msg_worker_authoritative")]); + await fixture.start(); + t.after(() => fixture.close()); + recordTerminal(dir); + const bus = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.reconciled, 1); + assert.equal(summary.failed, 0); + assert.equal(fixture.messagePostRequests.length, 0); + assert.equal(listPending(join(dir, "outbox")).length, 0); +}); + +for (const role of [ + { name: "admin", value: "admin" }, + { name: "agent", value: "agent" }, + { name: "collector", value: "collector" }, + { name: "missing", omit: true }, + { name: "null", value: null }, + { name: "malformed number", value: 123 }, + { name: "malformed boolean", value: true }, + { name: "malformed object", value: { role: "worker" } }, + { name: "malformed array", value: ["worker"] }, +]) { + test(`exact semantic row with ${role.name} posted_role is a typed authority conflict`, async (t) => { + const dir = scratch(`role-conflict-${role.name.replaceAll(" ", "-")}`); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([terminalRow("msg_role_conflict")]); + fixture.terminalReadbackTransform = (message) => { + const transformed = { ...message }; + if (role.omit) delete transformed.posted_role; + else transformed.posted_role = role.value; + return transformed; + }; + await fixture.start(); + t.after(() => fixture.close()); + recordTerminal(dir); + const bus = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + + await assert.rejects( + () => bus.findTerminalDeliveries(TERMINAL), + (error) => error instanceof WorkerTerminalAuthorityConflict + && error.code === "terminal_delivery_authority_conflict" + && error.conflict === true + && error.messageIds.length === 1 + && error.messageIds[0] === "msg_role_conflict", + ); + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(summary.reconciled, 0); + assert.equal(summary.delivered, 0, "an authority conflict is not ordinary cardinality zero"); + assert.equal(fixture.messagePostRequests.length, 0); + assert.equal(fixture.messages.length, 1); + assert.equal(listPending(join(dir, "outbox")).length, 1); + }); +} + +test("no semantic row publishes exactly once with the bound worker precondition", async (t) => { + const dir = scratch("authority-no-row"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture(); + await fixture.start(); + t.after(() => fixture.close()); + recordTerminal(dir); + const bus = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.delivered, 1); + assert.equal(summary.reconciled, 0); + assert.equal(summary.failed, 0); + assert.equal(fixture.messagePostRequests.length, 1); + assert.deepEqual(fixture.messagePostRequests[0].expectedHeaders, [POSTED_BY]); + assert.equal(fixture.posts[0].posted_by, POSTED_BY); + assert.equal(fixture.posts[0].posted_role, "worker"); + assert.equal(listPending(join(dir, "outbox")).length, 0); +}); + +for (const malformedBody of [123, true, { unexpected: "object" }, ["unexpected array"]]) { + test(`malformed terminal reconciliation body ${JSON.stringify(malformedBody)} cannot erase pending evidence`, async (t) => { + const dir = scratch("malformed-body"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([ + terminalRow("msg_different_body", { body: "DIFFERENT NONEMPTY BODY" }), + ]); + fixture.terminalReadbackTransform = (message) => ({ ...message, body: malformedBody }); + await fixture.start(); + t.after(() => fixture.close()); + recordPending({ ...TERMINAL, body: " \n\u0085" }, join(dir, "outbox")); + const bus = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(summary.reconciled, 0); + assert.equal(summary.delivered, 0); + assert.equal(fixture.messagePostRequests.length, 0, "a malformed readback cannot authorize a POST or deletion"); + assert.equal(fixture.messages.length, 1); + assert.equal(fixture.messages[0].body, "DIFFERENT NONEMPTY BODY"); + assert.equal(listPending(join(dir, "outbox")).length, 1); + }); +} + +for (const malformedSubject of [123, true, { unexpected: "object" }, ["unexpected array"]]) { + test(`malformed terminal reconciliation subject ${JSON.stringify(malformedSubject)} fails closed`, async (t) => { + const dir = scratch("malformed-subject"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([terminalRow("msg_exact_with_malformed_readback")]); + fixture.terminalReadbackTransform = (message) => ({ ...message, subject: malformedSubject }); + await fixture.start(); + t.after(() => fixture.close()); + recordTerminal(dir); + const bus = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(summary.reconciled, 0); + assert.equal(summary.delivered, 0); + assert.equal(fixture.messagePostRequests.length, 0, "malformed readback must not downgrade to publication"); + assert.equal(fixture.messages.length, 1); + assert.equal(listPending(join(dir, "outbox")).length, 1); + }); +} + +test("terminal canonicalization preserves server length and empty-subject refusals", async (t) => { + const dir = scratch("canonical-validation"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture(); + await fixture.start(); + t.after(() => fixture.close()); + const bus = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + + await assert.rejects( + () => bus.postTerminal("status", TERMINAL.subject, " ".repeat(8_001), TERMINAL.options), + /body must be <= 8000 characters before canonical storage/, + ); + await assert.rejects( + () => bus.postTerminal("status", " \n\u0085", TERMINAL.body, TERMINAL.options), + /subject must be a non-empty string/, + ); + + assert.equal(fixture.identityRequests, 0); + assert.equal(fixture.messagePostRequests.length, 0); + assert.equal(fixture.posts.length, 0); + assert.equal(listPending(join(dir, "outbox")).length, 2, "both refused terminal records remain durable"); +}); + +test("bearer and configured identity are immutable after a successful lookup", async (t) => { + const dir = scratch("identity-immutable"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture(); + await fixture.start(); + t.after(() => fixture.close()); + const bus = await authenticatedBus(fixture, dir); + + assert.throws(() => { bus.token = "replacement-token"; }, TypeError); + assert.throws(() => { bus.expectedPostingPrincipal = "worker:other"; }, TypeError); + assert.equal(bus.token, "test-token"); + assert.equal(bus.expectedPostingPrincipal, POSTED_BY); +}); + +test("a server-side identity change after prior success blocks terminal publication and preserves evidence", async (t) => { + const dir = scratch("identity-changed-before-post"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture(); + await fixture.start(); + t.after(() => fixture.close()); + const bus = await authenticatedBus(fixture, dir); + fixture.identityPayload = { worker_principal: "worker:other" }; + + await assert.rejects( + () => bus.postTerminal("status", TERMINAL.subject, TERMINAL.body, TERMINAL.options), + (error) => error instanceof WorkerIdentityRefusal && error.code === "worker_identity_mismatch", + ); + assert.equal(fixture.identityRequests, 2, "terminal publication must not reuse the earlier lookup"); + assert.equal(fixture.posts.length, 0); + const [pending] = listPending(join(dir, "outbox")); + assert.equal(pending.entry.configured_worker_principal, POSTED_BY); + assert.equal(pending.entry.expected_posted_by, undefined); +}); + +test("an identity outage after prior success blocks reconciliation before any message read", async (t) => { + const dir = scratch("identity-outage-before-reconcile"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + recordTerminal(dir); + const fixture = new TerminalBusFixture([terminalRow("msg_exact")]); + await fixture.start(); + t.after(() => fixture.close()); + const bus = await authenticatedBus(fixture, dir); + fixture.identityStatus = 503; + + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(fixture.identityRequests, 2, "reconciliation must freshly resolve identity"); + assert.equal(fixture.messageGetRequests, 0); + assert.equal(fixture.posts.length, 0); + assert.equal(bus.authenticatedPostingPrincipal, null, "a failed refresh must erase stale identity state"); + assert.equal(listPending(join(dir, "outbox")).length, 1); +}); + +test("identity is checked again between cardinality zero and the replay POST", async (t) => { + const dir = scratch("identity-change-after-scan"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + recordTerminal(dir); + const fixture = new TerminalBusFixture(); + await fixture.start(); + t.after(() => fixture.close()); + const bus = await authenticatedBus(fixture, dir); + fixture.afterMessageGet = () => { + fixture.identityPayload = { worker_principal: "worker:other" }; + fixture.afterMessageGet = null; + }; + + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(summary.delivered, 0); + assert.equal(fixture.identityRequests, 3, "startup, reconciliation, and publication each require identity"); + assert.equal(fixture.messageGetRequests, 1, "absence was observed before identity changed"); + assert.equal(fixture.posts.length, 0, "changed identity must block the terminal POST"); + assert.equal(listPending(join(dir, "outbox")).length, 1); +}); + +test("exact old A/B ACK-loss sequence now refuses both attempts and retains evidence", async (t) => { + const dir = scratch("ir-block-closed"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([], { + authenticatedPrincipal: "worker:other", + principalRole: "worker", + }); + fixture.dropAckSubject = TERMINAL.subject; + await fixture.start(); + t.after(() => fixture.close()); + + const beforeCrash = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + await assert.rejects( + () => beforeCrash.postTerminal("status", TERMINAL.subject, TERMINAL.body, TERMINAL.options), + (error) => error instanceof WorkerIdentityRefusal && error.code === "worker_identity_mismatch", + ); + assert.equal(terminalPostCount(fixture), 0, "the mismatched bearer cannot create the first terminal effect"); + assert.equal(listPending(join(dir, "outbox")).length, 1); + + const restarted = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + const summary = await replayPending(restarted, join(dir, "outbox"), { warn() {}, error() {} }); + assert.equal(summary.failed, 1); + assert.equal(summary.reconciled, 0); + assert.equal(summary.delivered, 0); + assert.equal(fixture.messageGetRequests, 0); + assert.equal(terminalPostCount(fixture), 0, "restart cannot duplicate under the wrong authenticated principal"); + assert.equal(listPending(join(dir, "outbox")).length, 1); +}); + +for (const failure of [ + { name: "missing bearer", token: "", code: "worker_identity_forbidden", pattern: /failed: 401/ }, + { name: "unknown bearer", token: "unknown-token", code: "worker_identity_forbidden", pattern: /failed: 401/ }, + { name: "ambiguous bearer", status: 403, raw: '{"detail":"operator error: overlapping credential"}', code: "worker_identity_forbidden", pattern: /failed: 403/ }, + { name: "malformed JSON", raw: "{not json", code: "worker_identity_malformed", pattern: /returned invalid JSON/ }, + { name: "null response", raw: "null", code: "worker_identity_malformed", pattern: /must be exactly/ }, + { name: "array response", raw: '[]', code: "worker_identity_malformed", pattern: /must be exactly/ }, + { name: "primitive response", raw: '"worker:lane-b-worker"', code: "worker_identity_malformed", pattern: /must be exactly/ }, + { name: "missing principal", payload: {}, code: "worker_identity_malformed", pattern: /must be exactly/ }, + { name: "extra property", payload: { worker_principal: POSTED_BY, source: "client" }, code: "worker_identity_malformed", pattern: /must be exactly/ }, + { name: "non-string principal", payload: { worker_principal: 42 }, code: "worker_identity_malformed", pattern: /string worker_principal/ }, + { name: "invalid worker form", payload: { worker_principal: "admin:lane-b-worker" }, code: "worker_identity_malformed", pattern: /invalid worker principal/ }, +]) { + test(`identity ${failure.name} fails closed before reconciliation or terminal POST`, async (t) => { + const dir = scratch(`identity-${failure.name.replaceAll(" ", "-")}`); + t.after(() => rmSync(dir, { recursive: true, force: true })); + recordTerminal(dir); + const fixture = new TerminalBusFixture(); + if (failure.status !== undefined) fixture.identityStatus = failure.status; + if (failure.raw !== undefined) fixture.identityRaw = failure.raw; + if (failure.payload !== undefined) fixture.identityPayload = failure.payload; + await fixture.start(); + t.after(() => fixture.close()); + const bus = new BusClient( + fixture.url, + failure.token ?? "test-token", + 100, + join(dir, "outbox"), + POSTED_BY, + ); + + await assert.rejects( + () => bus.establishAuthenticatedPostingPrincipal(), + (error) => error instanceof WorkerIdentityRefusal + && error.refused === true + && error.code === failure.code + && failure.pattern.test(error.message), + ); + assert.equal(bus.authenticatedPostingPrincipal, null); + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + assert.equal(summary.failed, 1); + await assert.rejects(() => bus.postTerminal("status", TERMINAL.subject, TERMINAL.body, TERMINAL.options)); + assert.equal(terminalPostCount(fixture), 0); + assert.equal(fixture.messageGetRequests, 0, "identity refusal must precede reconciliation reads"); + assert.equal(listPending(join(dir, "outbox")).length, 2, "both pre-existing and new terminal evidence remain"); + }); +} + +test("identity timeout fails closed with pending retained and no terminal POST", async (t) => { + const dir = scratch("identity-timeout"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + recordTerminal(dir); + const fixture = new TerminalBusFixture(); + fixture.identityDelayMs = 200; + await fixture.start(); + t.after(() => fixture.close()); + const bus = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY, 20); + + await assert.rejects( + () => bus.establishAuthenticatedPostingPrincipal(), + (error) => error instanceof WorkerIdentityRefusal && error.code === "worker_identity_unavailable", + ); + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + assert.equal(summary.failed, 1); + assert.equal(terminalPostCount(fixture), 0); + assert.equal(fixture.messageGetRequests, 0); + assert.equal(listPending(join(dir, "outbox")).length, 1); +}); + +test("identity network failure fails closed with pending retained and no terminal POST", async (t) => { + const dir = scratch("identity-network"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + recordTerminal(dir); + const fixture = new TerminalBusFixture(); + await fixture.start(); + const unreachableUrl = fixture.url; + await fixture.close(); + const bus = new BusClient(unreachableUrl, "test-token", 100, join(dir, "outbox"), POSTED_BY, 100); + + await assert.rejects( + () => bus.establishAuthenticatedPostingPrincipal(), + (error) => error instanceof WorkerIdentityRefusal && error.code === "worker_identity_unavailable", + ); + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + assert.equal(summary.failed, 1); + assert.equal(listPending(join(dir, "outbox")).length, 1); +}); + +test("two exact server-stamped rows are a retained duplicate condition with no third POST", async (t) => { + const dir = scratch("duplicate"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([ + terminalRow("msg_exact_1"), + terminalRow("msg_exact_2", { ts: "2026-09-13T12:00:01.000Z" }), + ]); + await fixture.start(); + t.after(() => fixture.close()); + recordTerminal(dir); + + const bus = await authenticatedBus(fixture, dir, 1); + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.duplicate, 1); + assert.equal(summary.conditions[0].type, DUPLICATE_DELIVERY); + assert.equal(summary.conditions[0].exact_match_count, 2); + assert.equal(terminalPostCount(fixture), 0, "duplicate detection must not append a third row"); + assert.equal(listPending(join(dir, "outbox"))[0].entry.delivery_condition.type, DUPLICATE_DELIVERY); + assert.equal(fixture.identityRequests, 2, "duplicate classification uses a fresh identity lookup"); +}); + +test("a near match and a later exact match are distinguished across complete pagination", async (t) => { + const dir = scratch("pagination-positive"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([ + terminalRow("msg_near_body", { body: `${TERMINAL.body} mutated` }), + terminalRow("msg_near_subject", { subject: `${TERMINAL.subject} mutated` }), + terminalRow("msg_near_outcome", { outcome: "FAILED" }), + terminalRow("msg_near_reply", { in_reply_to: "msg_other_attempt" }), + terminalRow("msg_near_refs", { refs: ["job_other"] }), + terminalRow("msg_near_recipient", { to_agent: POSTED_BY }), + terminalRow("msg_exact"), + ]); + await fixture.start(); + t.after(() => fixture.close()); + recordTerminal(dir); + + const bus = await authenticatedBus(fixture, dir, 1); + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.reconciled, 1); + assert.equal(summary.delivered, 0); + assert.equal(terminalPostCount(fixture), 0); + assert.equal(listPending(join(dir, "outbox")).length, 0); +}); + +test("premature pagination is incomplete observation: retain pending and POST zero", async (t) => { + const dir = scratch("pagination-fail-closed"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([ + terminalRow("msg_near", { body: `${TERMINAL.body} mutated` }), + terminalRow("msg_exact"), + ]); + fixture.breakSecondPage = true; + await fixture.start(); + t.after(() => fixture.close()); + recordTerminal(dir); + + const bus = await authenticatedBus(fixture, dir, 1); + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(terminalPostCount(fixture), 0); + assert.equal(listPending(join(dir, "outbox")).length, 1); + assert.equal(fixture.identityRequests, 2, "pagination starts only after a fresh identity lookup"); +}); + +for (const refusal of [ + { name: "wrong worker", postAuthenticatedPrincipal: "worker:other", postPrincipalRole: "worker", code: "worker_publication_precondition_refused" }, + { name: "admin", postAuthenticatedPrincipal: "george", postPrincipalRole: "admin", code: "worker_publication_precondition_refused" }, + { name: "collector", postAuthenticatedPrincipal: "collector:ci", postPrincipalRole: "collector", code: "worker_publication_precondition_refused" }, + { name: "agent", postAuthenticatedPrincipal: "agent:configured", postPrincipalRole: "agent", code: "worker_publication_precondition_refused" }, + { name: "ambiguous", postAuthenticatedPrincipal: "worker:ambiguous", postPrincipalRole: "ambiguous", code: "worker_publication_forbidden" }, + { name: "malformed precondition response", forcedBoundRefusalStatus: 422, code: "worker_publication_precondition_malformed" }, +]) { + test(`atomic publication keeps pending when the POST authenticates as ${refusal.name}`, async (t) => { + const dir = scratch(`atomic-${refusal.name.replaceAll(" ", "-")}`); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const fixture = new TerminalBusFixture([], refusal); + await fixture.start(); + t.after(() => fixture.close()); + const bus = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + + await assert.rejects( + () => bus.postTerminal("status", TERMINAL.subject, TERMINAL.body, TERMINAL.options), + (error) => error instanceof WorkerIdentityRefusal && error.code === refusal.code, + ); + + assert.equal(fixture.identityRequests, 1, "discovery succeeds but does not authorize the POST"); + assert.equal(fixture.messagePostRequests.length, 1, "refusal must not trigger an unbound retry"); + assert.deepEqual(fixture.messagePostRequests[0].expectedHeaders, [POSTED_BY]); + assert.equal(fixture.posts.length, 0); + assert.equal(fixture.messages.length, 0); + assert.equal(listPending(join(dir, "outbox")).length, 1); + }); +} + +for (const invalidBinding of [ + { name: "missing", entry: { ...TERMINAL, configured_worker_principal: undefined, expected_posted_by: undefined } }, + { name: "malformed", entry: { ...TERMINAL, configured_worker_principal: "worker:bad/id" } }, +]) { + test(`${invalidBinding.name} pending worker binding is retained without any POST`, async (t) => { + const dir = scratch(`binding-${invalidBinding.name}`); + t.after(() => rmSync(dir, { recursive: true, force: true })); + const entry = { ...invalidBinding.entry }; + if (invalidBinding.name === "missing") { + delete entry.configured_worker_principal; + delete entry.expected_posted_by; + } + recordPending(entry, join(dir, "outbox")); + const fixture = new TerminalBusFixture(); + await fixture.start(); + t.after(() => fixture.close()); + const bus = new BusClient(fixture.url, "test-token", 100, join(dir, "outbox"), POSTED_BY); + + const summary = await replayPending(bus, join(dir, "outbox"), { warn() {}, error() {} }); + + assert.equal(summary.failed, 1); + assert.equal(fixture.identityRequests, 0); + assert.equal(fixture.messageGetRequests, 0); + assert.equal(fixture.messagePostRequests.length, 0); + assert.equal(fixture.posts.length, 0); + assert.equal(listPending(join(dir, "outbox")).length, 1); + }); +} + +for (const mismatch of [ + { name: "wrong worker", authenticatedPrincipal: "worker:other", principalRole: "worker" }, + { name: "admin", authenticatedPrincipal: "george", principalRole: "admin" }, + { name: "collector", authenticatedPrincipal: "collector:ci", principalRole: "collector" }, +]) { + test(`IR BLOCK control: ${mismatch.name} bearer is refused before terminal replay or publication`, async (t) => { + const dir = scratch(`identity-${mismatch.principalRole}`); + t.after(() => rmSync(dir, { recursive: true, force: true })); + mkdirSync(join(dir, "home"), { recursive: true }); + recordTerminal(dir); + const fixture = new TerminalBusFixture([], mismatch); + await fixture.start(); + t.after(() => fixture.close()); + const launcher = resolve(process.env.VINCI_TEST_WORKER_LAUNCHER ?? join(ROOT, "vinci/bin/vinci")); + + const run = runWorker(launcher, fixture.url, dir); + const exit = await waitForExit(run.child); + + assert.equal(exit.code, 1, run.stderr()); + assert.equal(terminalPostCount(fixture), 0, "identity refusal must occur before terminal replay"); + assert.equal(listPending(join(dir, "outbox")).length, 1, "identity refusal must retain terminal debt"); + if (mismatch.principalRole === "worker") { + assert.match(run.stderr(), /bearer authenticates as worker:other, but --id requires worker:lane-b-worker/); + } else { + assert.match(run.stderr(), /worker identity GET .* failed: 403/); + } + assert.equal(fixture.posts.length, 0, "identity refusal must precede every publication"); + }); +} + +test("installed worker survives commit-then-ACK-loss and process loss without a duplicate", async (t) => { + const dir = scratch("installed-restart"); + t.after(() => rmSync(dir, { recursive: true, force: true })); + mkdirSync(join(dir, "home"), { recursive: true }); + recordTerminal(dir); + const fixture = new TerminalBusFixture(); + fixture.dropAckSubject = TERMINAL.subject; + await fixture.start(); + t.after(() => fixture.close()); + const launcher = resolve(process.env.VINCI_TEST_WORKER_LAUNCHER ?? join(ROOT, "vinci/bin/vinci")); + + const first = runWorker(launcher, fixture.url, dir, { once: false }); + await withTimeout(fixture.terminalCommitted, "terminal was not committed"); + first.child.kill("SIGKILL"); + const firstExit = await waitForExit(first.child); + assert.equal(firstExit.signal, "SIGKILL", first.stderr()); + assert.equal(listPending(join(dir, "outbox")).length, 1, "ACK loss must leave the debt durable"); + + const second = runWorker(launcher, fixture.url, dir); + const secondExit = await waitForExit(second.child); + assert.equal(secondExit.code, 0, second.stderr()); + const terminalRows = fixture.messages.filter((row) => row.subject === TERMINAL.subject); + const terminalRequests = fixture.messagePostRequests.filter(({ payload }) => payload.subject === TERMINAL.subject); + assert.equal(terminalRows.length, 1, "restart must reconcile, not append a duplicate"); + assert.equal(terminalRequests.length, 1, "ACK loss must not cause another terminal POST"); + assert.deepEqual(terminalRequests[0].expectedHeaders, [POSTED_BY]); + assert.equal(terminalRequests[0].payload.from_agent, undefined); + assert.equal(terminalRequests[0].payload.posted_by, undefined); + assert.equal(terminalRows[0].posted_by, POSTED_BY); + assert.equal(terminalRows[0].posted_role, "worker"); + assert.equal(terminalRows[0].in_reply_to, TERMINAL.options.inReplyTo); + assert.deepEqual(terminalRows[0].refs, TERMINAL.options.refs); + assert.match(terminalRows[0].body, /contract=wo_lane_b@01234567/); + assert.match(terminalRows[0].body, /attempt=msg_lane_b\/1/); + assert.match(terminalRows[0].body, /economics_sha256=e[1]{63}/); + assert.match(terminalRows[0].body, /evidence_sha256=a{64}/); + assert.equal(listPending(join(dir, "outbox")).length, 0); + assert.match(second.stderr(), /reconciled 1/); + assert.equal( + fixture.identityRequests, + 5, + "both startups, both reconciliations, and the first replay POST each authenticate independently", + ); +}); diff --git a/vinci/test/worker-typed-terminals.mjs b/vinci/test/worker-typed-terminals.mjs index 113abb232..e4bc903a1 100644 --- a/vinci/test/worker-typed-terminals.mjs +++ b/vinci/test/worker-typed-terminals.mjs @@ -225,7 +225,7 @@ exit 0 assert.doesNotMatch(okCalls, /--title Worker task/, "the opaque title must never be emitted"); }); -test("UNVERIFIED is a terminal and carries a type like any other", async () => { +test("UNVERIFIED is a terminal and carries a type like any other", async (t) => { // finalState's DEFAULT is UNVERIFIED -- "anything else, incl. evidence: none, exit 0 alone". // So this is the most common non-success terminal, not an edge case. It used to post through // the untyped `bus.post`, which meant the commonest way for a run to end badly produced a @@ -233,8 +233,17 @@ test("UNVERIFIED is a terminal and carries a type like any other", async () => { // // The bus has no server here, so a VALID outcome gets past validation and then fails on the // network. That difference is the assertion: valid values must fail LATER than invalid ones. + const tempDir = mkdtempSync(join(tmpdir(), "typed-terminal-unverified-")); + t.after(() => rmSync(tempDir, { recursive: true, force: true })); + const boundBus = new BusClient( + "https://example.invalid", + "t", + 100, + join(tempDir, "outbox"), + "worker:test", + ); await assert.rejects( - () => bus().postTerminal("status", "task t", "b", { outcome: "UNVERIFIED" }), + () => boundBus.postTerminal("status", "task t", "b", { outcome: "UNVERIFIED" }), (err) => !/terminal record must carry a typed outcome/.test(err.message), "UNVERIFIED must pass validation and fail only at the network", ); diff --git a/vinci/worker/README.md b/vinci/worker/README.md index d7973887c..e4a339179 100644 --- a/vinci/worker/README.md +++ b/vinci/worker/README.md @@ -540,6 +540,42 @@ configured nothing changes (no downgrade), so soak boxes may run without it. - If `terminal=true`: skip (already done) - If `terminal=false` (`PENDING`/`RUNNING`): increment `attempt`, keep same `session_id`, resume +### Terminal outbox reconciliation + +The lifecycle transition and the terminal bus POST are not atomic. Before every terminal POST, +the worker writes the exact payload plus its configured worker principal to +`/outbox/`. It then resolves the bearer through the worker-only +`GET /v1/worker-principal` contract, requires exact equality with `worker:`, and +atomically adds that authenticated principal to the pending record before POSTing. It removes the +record only after the POST is acknowledged. On startup, after taking the state-directory daemon +lock and before polling new work, the worker settles each pending record against +`GET /v1/messages`: + +- Identity is fetched again immediately before every reconciliation and again before a + cardinality-zero replay POST. The bearer and configured principal are immutable within the + client. A 401/403, timeout, connection failure, malformed response, non-worker principal, or + configured-id mismatch is a typed refusal: no reconciliation read or terminal POST occurs and + the pending evidence remains. There is no cached-identity or locally-derived fallback. +- The lookup is filtered and rechecked against the server-stamped `posted_by=worker:`. + `from_agent` is never accepted as authenticated provenance. +- Kind, subject, body, outcome, `in_reply_to`, refs, and broadcast recipient must all match exactly. + The contract/economics/evidence fields in the body therefore remain bound to the same handoff and + artifact set; the economics digest binds the attempt label in `economics-summary.json`. +- Zero exact matches proves the effect is unobserved, so the worker POSTs once and clears only after + the acknowledgement. One exact match is the commit-then-ACK-loss case and clears without POSTing. +- More than one exact match is `duplicate_terminal_delivery`: the worker POSTs no additional row, + retains the outbox entry, persists the typed condition and matching message ids, and reports it. +- A failed, malformed, changing, repeated, or prematurely-ended pagination scan is not proof of + absence. The worker retains the entry and POSTs nothing. + +Records written by a pre-reconciliation worker do not carry either worker-principal binding. They +are retained and reported as unbound rather than replayed under an identity the original record +did not prove. A new record may temporarily carry only `configured_worker_principal` when the +identity endpoint refuses or is unavailable; a later restart must freshly establish that exact +identity before it can inspect or deliver the record. +This is process-loss safety for one state directory under its daemon lock, not server-wide atomic +idempotency for copied outboxes or independently running workers. + ## Credentials - `VINCI_BUS_TOKEN`: Bearer auth to bus (/v1/messages) @@ -566,6 +602,7 @@ No new npm dependencies introduced; uses only node:* and global APIs. ``` / cursor.json # High-water mark per worker + outbox/*.json # pending terminal effects and retained duplicate conditions tasks/ .json # Lifecycle record sessions// # vinci JSONL read for outcomes and usage (outside every tree) diff --git a/vinci/worker/bus.mjs b/vinci/worker/bus.mjs index 23849d5db..6266e4325 100644 --- a/vinci/worker/bus.mjs +++ b/vinci/worker/bus.mjs @@ -11,6 +11,73 @@ const LEDGER_REF = /^(?:job|exp|bk)_[A-Za-z0-9][A-Za-z0-9._-]*$/; // case, so leaving it untyped left the most COMMON non-success terminal with a null outcome // and therefore invisible to a consumer that keys attention on `outcome !== "COMPLETED"`. const TERMINAL_OUTCOMES = new Set(["COMPLETED", "FAILED", "BLOCKED", "REFUSED", "UNVERIFIED"]); +const WORKER_PRINCIPAL = /^worker:[A-Za-z0-9][A-Za-z0-9._-]{0,56}$/; +const SERVER_STRIP_EDGE = /^[\u0009-\u000d\u001c-\u0020\u0085\u00a0\u1680\u2000-\u200a\u2028\u2029\u202f\u205f\u3000]+|[\u0009-\u000d\u001c-\u0020\u0085\u00a0\u1680\u2000-\u200a\u2028\u2029\u202f\u205f\u3000]+$/gu; +const SERVER_MAX_SUBJECT_CODE_POINTS = 200; +const SERVER_MAX_BODY_CODE_POINTS = 8_000; +const DEFAULT_IDENTITY_TIMEOUT_MS = 10_000; + +export class WorkerIdentityRefusal extends Error { + constructor(code, message, options = {}) { + super(message, options); + this.name = "WorkerIdentityRefusal"; + this.code = code; + this.refused = true; + } +} + +export class WorkerTerminalAuthorityConflict extends Error { + constructor(messageIds) { + super(`terminal reconciliation found exact semantic rows without worker authority (${messageIds.join(", ")})`); + this.name = "WorkerTerminalAuthorityConflict"; + this.code = "terminal_delivery_authority_conflict"; + this.conflict = true; + this.messageIds = messageIds; + } +} + +function sameStrings(left, right) { + return left.length === right.length && left.every((value, index) => value === right[index]); +} + +function serverStrip(value) { + return value.replace(SERVER_STRIP_EDGE, ""); +} + +function canonicalTerminalEntry(entry) { + if (typeof entry?.subject !== "string") { + throw new Error("terminal subject must be a non-empty string"); + } + const subject = serverStrip(entry.subject); + if (subject.length === 0) throw new Error("terminal subject must be a non-empty string"); + if ([...subject].length > SERVER_MAX_SUBJECT_CODE_POINTS) { + throw new Error(`terminal subject must be <= ${SERVER_MAX_SUBJECT_CODE_POINTS} characters`); + } + if (entry.body !== null && entry.body !== undefined && typeof entry.body !== "string") { + throw new Error("terminal body must be a string"); + } + if (typeof entry.body === "string" && entry.body.length > 0 && [...entry.body].length > SERVER_MAX_BODY_CODE_POINTS) { + throw new Error(`terminal body must be <= ${SERVER_MAX_BODY_CODE_POINTS} characters before canonical storage`); + } + return { + ...entry, + subject, + body: typeof entry.body === "string" ? serverStrip(entry.body) : "", + }; +} + +function isExactTerminalSemantics(message, entry) { + const options = entry.options ?? {}; + const expectedRefs = options.refs ?? []; + return message.to_agent === null + && message.kind === entry.kind + && message.subject === entry.subject + && message.body === entry.body + && (message.outcome ?? null) === (options.outcome ?? null) + && (message.in_reply_to ?? null) === (options.inReplyTo ?? null) + && Array.isArray(message.refs) + && sameStrings(message.refs, expectedRefs); +} export function isLedgerRef(value) { return typeof value === "string" && LEDGER_REF.test(value); @@ -45,14 +112,39 @@ export function normaliseMessage(message) { } export class BusClient { - constructor(serverUrl, token, pageSize = 100, outboxDir = null) { + #authenticatedPostingPrincipal; + #expectedPostingPrincipal; + #token; + + constructor( + serverUrl, + token, + pageSize = 100, + outboxDir = null, + expectedPostingPrincipal = null, + identityTimeoutMs = DEFAULT_IDENTITY_TIMEOUT_MS, + ) { const url = new URL(serverUrl); if (url.protocol !== "http:" && url.protocol !== "https:") throw new Error("server must use http or https"); if (url.username || url.password) throw new Error("server URL must not contain credentials"); if (!Number.isInteger(pageSize) || pageSize <= 0) throw new Error("bus page size must be a positive integer"); this.serverUrl = url.href.replace(/\/$/, ""); - this.token = token; + this.#token = token; this.pageSize = pageSize; + if ( + expectedPostingPrincipal !== null + && (typeof expectedPostingPrincipal !== "string" || !WORKER_PRINCIPAL.test(expectedPostingPrincipal)) + ) { + throw new Error("expected posting principal must be a worker: principal"); + } + if (!Number.isInteger(identityTimeoutMs) || identityTimeoutMs <= 0) { + throw new Error("worker identity timeout must be a positive integer"); + } + // A configured expectation is not authenticated identity. Only the server's worker-only + // identity endpoint can assign authenticatedPostingPrincipal. + this.#expectedPostingPrincipal = expectedPostingPrincipal; + this.identityTimeoutMs = identityTimeoutMs; + this.#authenticatedPostingPrincipal = null; // Where undelivered terminal records are parked. Settable because the // worker keeps its durable state under --state-dir, and a default that // wrote to the process cwd would park records somewhere the replay at @@ -61,6 +153,18 @@ export class BusClient { this.outboxDir = outboxDir ?? DEFAULT_OUTBOX_DIR; } + get authenticatedPostingPrincipal() { + return this.#authenticatedPostingPrincipal; + } + + get expectedPostingPrincipal() { + return this.#expectedPostingPrincipal; + } + + get token() { + return this.#token; + } + async poll(workerId, cursor = null) { const messagesById = new Map(); let offset = 0; @@ -68,7 +172,7 @@ export class BusClient { const url = new URL(`${this.serverUrl}/v1/messages`); url.searchParams.set("limit", String(this.pageSize)); url.searchParams.set("offset", String(offset)); - const response = await fetch(url, { headers: { authorization: `Bearer ${this.token}` } }); + const response = await fetch(url, { headers: { authorization: `Bearer ${this.#token}` } }); if (!response.ok) throw new Error(`bus GET ${url} failed: ${response.status} ${await response.text()}`); const payload = await response.json(); if ( @@ -114,7 +218,240 @@ export class BusClient { .sort((left, right) => left.ts.localeCompare(right.ts) || left.message_id.localeCompare(right.message_id)); } - async post(kind, subject, body, options = {}) { + // Establish which worker principal the server authenticated for this bearer. The endpoint is + // worker-only and resolves the bearer at the server boundary; no client identity field is sent. + // No terminal publication or replay is allowed until this exact value matches --id. + async #requireAuthenticatedPostingPrincipal() { + const expected = this.#expectedPostingPrincipal; + this.#authenticatedPostingPrincipal = null; + if (typeof expected !== "string") { + throw new WorkerIdentityRefusal( + "worker_identity_unconfigured", + "worker principal authentication requires an expected worker: principal", + ); + } + + const url = `${this.serverUrl}/v1/worker-principal`; + const token = this.#token; + let response; + try { + response = await fetch(url, { + headers: { authorization: `Bearer ${token}` }, + signal: AbortSignal.timeout(this.identityTimeoutMs), + }); + } catch (error) { + throw new WorkerIdentityRefusal( + "worker_identity_unavailable", + `worker identity GET ${url} failed: ${error.message}`, + { cause: error }, + ); + } + if (!response.ok) { + let detail = ""; + try { + detail = await response.text(); + } catch (error) { + throw new WorkerIdentityRefusal( + "worker_identity_unavailable", + `worker identity GET ${url} failed while reading ${response.status}: ${error.message}`, + { cause: error }, + ); + } + throw new WorkerIdentityRefusal( + response.status === 401 || response.status === 403 + ? "worker_identity_forbidden" + : "worker_identity_unavailable", + `worker identity GET ${url} failed: ${response.status} ${detail}`, + ); + } + let payload; + try { + payload = await response.json(); + } catch (error) { + throw new WorkerIdentityRefusal( + "worker_identity_malformed", + `worker identity GET ${url} returned invalid JSON: ${error.message}`, + { cause: error }, + ); + } + if ( + !payload + || typeof payload !== "object" + || Array.isArray(payload) + || Object.keys(payload).length !== 1 + || !Object.hasOwn(payload, "worker_principal") + ) { + throw new WorkerIdentityRefusal( + "worker_identity_malformed", + "worker identity GET response must be exactly {worker_principal: string}", + ); + } + const principal = payload.worker_principal; + if (typeof principal !== "string") { + throw new WorkerIdentityRefusal( + "worker_identity_malformed", + "worker identity GET response must contain a string worker_principal", + ); + } + if (!WORKER_PRINCIPAL.test(principal)) { + throw new WorkerIdentityRefusal( + "worker_identity_malformed", + `worker identity GET returned invalid worker principal ${principal}`, + ); + } + if (principal !== expected) { + throw new WorkerIdentityRefusal( + "worker_identity_mismatch", + `worker bearer authenticates as ${principal}, but --id requires ${expected}`, + ); + } + this.#authenticatedPostingPrincipal = principal; + return { principal, token }; + } + + async establishAuthenticatedPostingPrincipal() { + return (await this.#requireAuthenticatedPostingPrincipal()).principal; + } + + #configuredPrincipalForEntry(entry) { + const configured = entry?.configured_worker_principal; + if (typeof configured !== "string" || !WORKER_PRINCIPAL.test(configured)) { + throw new WorkerIdentityRefusal( + "worker_identity_outbox_unbound", + "pending terminal record has no configured worker-principal binding", + ); + } + if (configured !== this.#expectedPostingPrincipal) { + throw new WorkerIdentityRefusal( + "worker_identity_outbox_mismatch", + `pending terminal record belongs to ${configured}, configured worker is ${this.#expectedPostingPrincipal ?? "unbound"}`, + ); + } + if (entry?.expected_posted_by !== undefined && entry.expected_posted_by !== configured) { + throw new WorkerIdentityRefusal( + "worker_identity_outbox_mismatch", + `pending terminal record has conflicting configured and authenticated principals (${configured} != ${entry.expected_posted_by})`, + ); + } + return configured; + } + + // Classify whether the exact terminal effect represented by an outbox entry is already visible + // on the bus. Every page is read before the caller is allowed to POST. A partial or shifting + // offset scan is not evidence of absence, so response-shape, total, offset and duplicate-id + // inconsistencies are hard errors; replay retains the entry and posts nothing on any such error. + async findTerminalDeliveries(entry) { + const expectedPostedBy = this.#configuredPrincipalForEntry(entry); + const canonicalEntry = canonicalTerminalEntry(entry); + const { token } = await this.#requireAuthenticatedPostingPrincipal(); + + const messages = []; + const rawAuthorityByMessageId = new Map(); + const messageIds = new Set(); + let expectedTotal = null; + let offset = 0; + while (true) { + const url = new URL(`${this.serverUrl}/v1/messages`); + url.searchParams.set("posted_by", expectedPostedBy); + url.searchParams.set("kind", canonicalEntry.kind); + url.searchParams.set("limit", String(this.pageSize)); + url.searchParams.set("offset", String(offset)); + const response = await fetch(url, { headers: { authorization: `Bearer ${token}` } }); + if (!response.ok) throw new Error(`terminal reconciliation GET ${url} failed: ${response.status} ${await response.text()}`); + let payload; + try { + payload = await response.json(); + } catch (error) { + throw new Error(`terminal reconciliation GET ${url} returned invalid JSON: ${error.message}`); + } + if ( + !payload + || !Array.isArray(payload.messages) + || !Number.isInteger(payload.total) + || payload.total < 0 + || !Number.isInteger(payload.limit) + || payload.limit !== this.pageSize + || !Number.isInteger(payload.offset) + || payload.offset !== offset + || payload.messages.length > payload.limit + ) { + throw new Error("terminal reconciliation GET response has an invalid or incomplete pagination shape"); + } + if (expectedTotal === null) expectedTotal = payload.total; + else if (payload.total !== expectedTotal) { + throw new Error(`terminal reconciliation GET total changed during pagination (${expectedTotal} -> ${payload.total})`); + } + + for (const raw of payload.messages) { + if ( + !raw + || typeof raw !== "object" + || !Object.hasOwn(raw, "posted_by") + || !Object.hasOwn(raw, "to_agent") + || !Object.hasOwn(raw, "subject") + || !Object.hasOwn(raw, "body") + || typeof raw.subject !== "string" + || (raw.body !== null && typeof raw.body !== "string") + || !Object.hasOwn(raw, "outcome") + || !Object.hasOwn(raw, "in_reply_to") + || !Object.hasOwn(raw, "refs") + ) { + throw new Error("terminal reconciliation GET returned a malformed or filter-inconsistent message"); + } + const message = normaliseMessage(raw); + if ( + message === null + || raw.posted_by !== expectedPostedBy + || message.kind !== canonicalEntry.kind + || typeof message.subject !== "string" + || typeof message.body !== "string" + || !Array.isArray(message.refs) + || message.refs.some((ref) => typeof ref !== "string") + || (message.outcome !== null && message.outcome !== undefined && typeof message.outcome !== "string") + || (message.in_reply_to !== null && message.in_reply_to !== undefined && typeof message.in_reply_to !== "string") + ) { + throw new Error("terminal reconciliation GET returned a malformed or filter-inconsistent message"); + } + if (messageIds.has(message.message_id)) { + throw new Error(`terminal reconciliation GET repeated message ${message.message_id}; pagination is incomplete`); + } + messageIds.add(message.message_id); + messages.push(message); + rawAuthorityByMessageId.set(message.message_id, { + hasPostedRole: Object.hasOwn(raw, "posted_role"), + postedBy: raw.posted_by, + postedRole: raw.posted_role, + }); + } + + offset += payload.messages.length; + if (offset === expectedTotal) break; + if (offset > expectedTotal || payload.messages.length === 0) { + throw new Error(`terminal reconciliation GET ended at ${offset} of ${expectedTotal} messages`); + } + } + + if (messages.length !== expectedTotal) { + throw new Error(`terminal reconciliation GET returned ${messages.length} unique messages for total ${expectedTotal}`); + } + const semanticMatches = messages.filter((message) => isExactTerminalSemantics(message, canonicalEntry)); + const authorityConflicts = semanticMatches.filter((message) => { + const authority = rawAuthorityByMessageId.get(message.message_id); + return authority.postedBy === expectedPostedBy + && (!authority.hasPostedRole || authority.postedRole !== "worker"); + }); + if (authorityConflicts.length > 0) { + throw new WorkerTerminalAuthorityConflict(authorityConflicts.map((message) => message.message_id)); + } + return semanticMatches + .filter((message) => { + const authority = rawAuthorityByMessageId.get(message.message_id); + return authority.postedBy === expectedPostedBy && authority.postedRole === "worker"; + }) + .map((message) => message.message_id); + } + + async #post(kind, subject, body, options, token, postingPrincipal) { if (kind !== "status" && kind !== "finding" && kind !== "blocker") { throw new Error(`worker cannot post message kind ${kind}`); } @@ -129,19 +466,63 @@ export class BusClient { } const url = `${this.serverUrl}/v1/messages`; const payload = { kind, subject, body }; + if (postingPrincipal !== null && this.#expectedPostingPrincipal === null) { + payload.from_agent = postingPrincipal; + } if (options.outcome !== undefined) payload.outcome = options.outcome; if (options.refs !== undefined) payload.refs = options.refs; if (options.inReplyTo !== undefined) payload.in_reply_to = options.inReplyTo; + const headers = { authorization: `Bearer ${token}`, "content-type": "application/json" }; + if (this.#expectedPostingPrincipal !== null) { + // This is an opt-in, same-request server precondition. Its value comes only from the + // configured worker binding: neither from_agent nor any prior identity discovery is + // authorization for the POST that follows. + headers["X-VGC-Expected-Worker-Principal"] = this.#expectedPostingPrincipal; + } const response = await fetch(url, { method: "POST", - headers: { authorization: `Bearer ${this.token}`, "content-type": "application/json" }, + headers, body: JSON.stringify(payload), }); - if (!response.ok) throw new Error(`bus POST ${url} failed: ${response.status} ${await response.text()}`); + if (!response.ok) { + const detail = await response.text(); + if (this.#expectedPostingPrincipal !== null && [401, 403, 412, 422].includes(response.status)) { + const code = response.status === 412 + ? "worker_publication_precondition_refused" + : response.status === 422 + ? "worker_publication_precondition_malformed" + : "worker_publication_forbidden"; + throw new WorkerIdentityRefusal(code, `bound worker bus POST ${url} failed: ${response.status} ${detail}`); + } + throw new Error(`bus POST ${url} failed: ${response.status} ${detail}`); + } const text = await response.text(); return text ? JSON.parse(text) : undefined; } + async post(kind, subject, body, options = {}) { + return this.#post(kind, subject, body, options, this.#token, this.#authenticatedPostingPrincipal); + } + + async deliverPendingTerminal(entry) { + this.#configuredPrincipalForEntry(entry); + if (!TERMINAL_OUTCOMES.has(entry?.options?.outcome)) { + throw new Error( + `a pending terminal record must carry a typed outcome (${[...TERMINAL_OUTCOMES].join(", ")}); got ${entry?.options?.outcome}`, + ); + } + const canonicalEntry = canonicalTerminalEntry(entry); + const { principal, token } = await this.#requireAuthenticatedPostingPrincipal(); + return this.#post( + canonicalEntry.kind, + canonicalEntry.subject, + canonicalEntry.body, + canonicalEntry.options ?? {}, + token, + principal, + ); + } + // The ONLY sanctioned way to announce that a task has ended. Requires the typed outcome, so a // terminal record cannot be posted without one by construction rather than by convention. async postTerminal(kind, subject, body, options = {}) { @@ -150,15 +531,37 @@ export class BusClient { `a terminal record must carry a typed outcome (${[...TERMINAL_OUTCOMES].join(", ")}); got ${options.outcome}`, ); } - // RECORDED BEFORE THE ATTEMPT, cleared only after it succeeds, so anything - // left on disk is by definition undelivered. The worker transitions its + if (typeof this.#expectedPostingPrincipal !== "string") { + throw new WorkerIdentityRefusal( + "worker_identity_unconfigured", + "terminal posts require a configured worker-principal binding", + ); + } + // RECORDED BEFORE THE ATTEMPT, cleared only after it succeeds or an exact + // server-stamped delivery is reconciled. The worker transitions its // lifecycle to terminal and THEN announces it, and those two steps are not // atomic: without this, a transient bus failure left the task terminal and // unannounced, and a restart skipped it precisely because it was already // terminal. A typed terminal outcome exists so a failure is VISIBLE without // being an open decision -- undelivered, it is neither. - const pendingId = recordPending({ kind, subject, body, options }, this.outboxDir); - const result = await this.post(kind, subject, body, options); + const entry = { + kind, + subject, + body, + options, + configured_worker_principal: this.#expectedPostingPrincipal, + }; + const pendingId = recordPending(entry, this.outboxDir); + const canonicalEntry = canonicalTerminalEntry(entry); + const { principal, token } = await this.#requireAuthenticatedPostingPrincipal(); + const result = await this.#post( + canonicalEntry.kind, + canonicalEntry.subject, + canonicalEntry.body, + canonicalEntry.options, + token, + principal, + ); clearPending(pendingId, this.outboxDir); return result; } diff --git a/vinci/worker/outbox.mjs b/vinci/worker/outbox.mjs index 8f5b521b7..cbb8836c1 100644 --- a/vinci/worker/outbox.mjs +++ b/vinci/worker/outbox.mjs @@ -9,8 +9,9 @@ // decision. Undelivered, it is neither. // // This is the durable half. A terminal record is written to disk BEFORE the post -// is attempted and removed only after it succeeds, so anything left on disk is -// by definition undelivered and is replayed at startup. +// is attempted and removed only after it succeeds. Anything left on disk has an +// UNKNOWN delivery state: the server may have committed it before the ACK was +// lost. Startup reconciles that state before deciding whether another POST is safe. // // WHY AT postTerminal AND NOT AT THE CALL SITES: there are eleven terminal post // sites in worker.mjs and there will be more. Wrapping the single choke point @@ -30,6 +31,7 @@ import { randomBytes } from "node:crypto"; export const DEFAULT_OUTBOX_DIR = process.env.VINCI_WORKER_OUTBOX || join(process.cwd(), ".vinci-worker-outbox"); +export const DUPLICATE_DELIVERY = "duplicate_terminal_delivery"; export function recordPending(entry, dir = DEFAULT_OUTBOX_DIR) { mkdirSync(dir, { recursive: true }); @@ -49,6 +51,20 @@ export function clearPending(id, dir = DEFAULT_OUTBOX_DIR) { if (existsSync(path)) rmSync(path); } +export function bindPendingPrincipal(id, principal, dir = DEFAULT_OUTBOX_DIR) { + const path = join(dir, `${id}.json`); + const entry = JSON.parse(readFileSync(path, "utf8")); + const temporary = `${path}.tmp-${process.pid}`; + writeFileSync(temporary, JSON.stringify({ ...entry, expected_posted_by: principal })); + renameSync(temporary, path); +} + +function preserveDeliveryCondition(path, entry, condition) { + const temporary = `${path}.tmp-${process.pid}`; + writeFileSync(temporary, JSON.stringify({ ...entry, delivery_condition: condition })); + renameSync(temporary, path); +} + export function listPending(dir = DEFAULT_OUTBOX_DIR) { if (!existsSync(dir)) return []; return readdirSync(dir) @@ -67,12 +83,20 @@ export function listPending(dir = DEFAULT_OUTBOX_DIR) { }); } -// Replay every undelivered terminal record. Returns a summary rather than -// throwing: a bus that is still unreachable must not stop the worker from -// starting, and the records stay on disk for the next attempt. +// Settle every pending terminal record. The reconciliation read is part of the delivery attempt: +// absence is trustworthy only after a complete scan. Returns a summary rather than throwing so a +// bus that is unreachable cannot erase the record or prevent the daemon from reporting its debt. export async function replayPending(bus, dir = DEFAULT_OUTBOX_DIR, log = console) { const pending = listPending(dir); - const summary = { attempted: 0, delivered: 0, failed: 0, corrupt: 0 }; + const summary = { + attempted: 0, + delivered: 0, + reconciled: 0, + duplicate: 0, + failed: 0, + corrupt: 0, + conditions: [], + }; for (const { path, entry, corrupt } of pending) { if (corrupt || !entry) { summary.corrupt += 1; @@ -81,13 +105,41 @@ export async function replayPending(bus, dir = DEFAULT_OUTBOX_DIR, log = console } summary.attempted += 1; try { - await bus.post(entry.kind, entry.subject, entry.body, entry.options ?? {}); + if (typeof bus.findTerminalDeliveries !== "function" || typeof bus.deliverPendingTerminal !== "function") { + throw new Error("bus does not support interruption-safe terminal reconciliation"); + } + const matches = await bus.findTerminalDeliveries(entry); + if (!Array.isArray(matches) || matches.some((id) => typeof id !== "string")) { + throw new Error("bus returned an invalid terminal reconciliation result"); + } + if (matches.length === 1) { + rmSync(path); + summary.reconciled += 1; + log.warn(`worker outbox: reconciled ACK-loss terminal record ${entry.id} as ${matches[0]} (${entry.options?.outcome})`); + continue; + } + if (matches.length > 1) { + const condition = { + type: DUPLICATE_DELIVERY, + detected_at: new Date().toISOString(), + exact_match_count: matches.length, + message_ids: matches, + }; + preserveDeliveryCondition(path, entry, condition); + summary.duplicate += 1; + summary.conditions.push({ pending_id: entry.id, ...condition }); + log.error( + `worker outbox: DUPLICATE_DELIVERY for ${entry.id}: ${matches.length} exact terminal rows (${matches.join(", ")}); kept on disk and posted nothing`, + ); + continue; + } + await bus.deliverPendingTerminal(entry); rmSync(path); summary.delivered += 1; - log.warn(`worker outbox: replayed undelivered terminal record ${entry.id} (${entry.options?.outcome})`); + log.warn(`worker outbox: delivered previously unobserved terminal record ${entry.id} (${entry.options?.outcome})`); } catch (error) { summary.failed += 1; - log.error(`worker outbox: replay FAILED for ${entry.id}, kept on disk: ${error.message}`); + log.error(`worker outbox: reconciliation/delivery FAILED for ${entry.id}, kept on disk and posted nothing unless absence was proven: ${error.message}`); } } return summary; diff --git a/vinci/worker/worker.mjs b/vinci/worker/worker.mjs index ee0bcd839..51a5455db 100644 --- a/vinci/worker/worker.mjs +++ b/vinci/worker/worker.mjs @@ -703,6 +703,11 @@ async function postFinal(bus, message, envelope, state, evidence, economics = nu : []; const details = [ `state=${state.state}`, + // The terminal payload is the durable delivery identity reconciled after ACK loss. Carry the + // lifecycle's exact attempt label in that byte-exact payload so a terminal from another + // attempt of the same WorkOrder can never satisfy reconciliation merely because its subject, + // contract and outcome agree. + `attempt=${message.message_id}/${state.attempt}`, `exit_code=${state.exit_code}`, `cost_usd=${Number(state.cost_usd).toFixed(6)}`, state.limit_tripped ? `limit=${state.limit_tripped}` : undefined, @@ -1681,13 +1686,23 @@ async function main() { process.once("SIGTERM", handleSignal); process.once("SIGINT", handleSignal); try { - const bus = new BusClient(options.server, options.token, 100, join(options.stateDir, "outbox")); + const bus = new BusClient( + options.server, + options.token, + 100, + join(options.stateDir, "outbox"), + `worker:${options.id}`, + ); // W0.5: record the server's build next to our own and announce both ONCE per daemon start, // before the first poll. A failed /v1/version fetch is recorded, never fatal: the bus // token check and the first poll already gate startup. serverBuild = await fetchServerBuild(options.server); // #18: and the version of the `vinci` binary this daemon will spawn (never fatal either). vinciBinary = vinciBinaryVersion(); + // The configured --id is only an expectation. Resolve the bearer at the server's worker-only + // identity boundary before any publication or terminal reconciliation; no declared or locally + // persisted identity is accepted as a fallback. + await bus.establishAuthenticatedPostingPrincipal(); await bus.post( "status", `worker ${options.id} online`, @@ -1726,7 +1741,8 @@ async function main() { if (replayed.attempted || replayed.corrupt) { process.stderr.write( `vinci worker: terminal outbox -- attempted ${replayed.attempted}, ` - + `delivered ${replayed.delivered}, failed ${replayed.failed}, ` + + `delivered ${replayed.delivered}, reconciled ${replayed.reconciled}, ` + + `duplicate ${replayed.duplicate}, failed ${replayed.failed}, ` + `unreadable ${replayed.corrupt}\n`, ); }