From aa8a424403df826a5b00d4b9ecaa7a02182f0567 Mon Sep 17 00:00:00 2001 From: Aaron Stainback Date: Sat, 16 May 2026 02:17:00 -0400 Subject: [PATCH 1/6] feat(bus): add subscribeOnce helper (decomposed from B-0459) --- tools/bus/subscribe.test.ts | 69 +++++++++++++++++++++++++++++++++++++ tools/bus/subscribe.ts | 57 ++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+) create mode 100644 tools/bus/subscribe.test.ts create mode 100644 tools/bus/subscribe.ts diff --git a/tools/bus/subscribe.test.ts b/tools/bus/subscribe.test.ts new file mode 100644 index 0000000000..f6cc284835 --- /dev/null +++ b/tools/bus/subscribe.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, test, mock } from "bun:test"; +import { subscribeOnce } from "./subscribe"; +import type { MessageEnvelope, Topic } from "./types"; +import { rmSync } from "node:fs"; +import { join } from "node:path"; +import { BUS_DIR } from "./bus"; + +describe("bus subscribeOnce (B-0449 slice 5)", () => { + const seenFile = join(BUS_DIR, "seen-test-surface.json"); + + // Helper to clear state + function clearState() { + try { rmSync(seenFile); } catch {} + } + + test("calls handler for unseen envelopes and records seen state", async () => { + clearState(); + + const env1: MessageEnvelope = { + id: "env-1", + from: "otto", + to: "test-surface" as any, + timestamp: new Date().toISOString(), + expiresAt: new Date(Date.now() + 10000).toISOString(), + topic: "work-assignment", + payload: { rowId: "B-1234", priority: "P1", rationale: "test" }, + }; + + const fakeList = mock(() => { + return [env1]; + }); + + const handler = mock(async (env) => {}); + + await subscribeOnce("work-assignment", "test-surface", handler, { list: fakeList as any }); + + expect(handler).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenCalledWith(env1); + + // Call again, should not trigger handler because it was recorded in seen-test-surface.json + await subscribeOnce("work-assignment", "test-surface", handler, { list: fakeList as any }); + expect(handler).toHaveBeenCalledTimes(1); // Still 1 + }); + + test("does not mark as seen if handler throws", async () => { + clearState(); + + const env2: MessageEnvelope = { + id: "env-2", + from: "otto", + to: "test-surface" as any, + timestamp: new Date().toISOString(), + expiresAt: new Date(Date.now() + 10000).toISOString(), + topic: "work-assignment", + payload: { rowId: "B-2222", priority: "P2", rationale: "test2" }, + }; + + const fakeList = mock(() => [env2]); + const handlerFailing = mock(async () => { throw new Error("fail"); }); + + await subscribeOnce("work-assignment", "test-surface", handlerFailing, { list: fakeList as any }); + + expect(handlerFailing).toHaveBeenCalledTimes(1); + + // Call again, should retry because it failed and wasn't marked seen + await subscribeOnce("work-assignment", "test-surface", handlerFailing, { list: fakeList as any }); + expect(handlerFailing).toHaveBeenCalledTimes(2); + }); +}); diff --git a/tools/bus/subscribe.ts b/tools/bus/subscribe.ts new file mode 100644 index 0000000000..928f85efd0 --- /dev/null +++ b/tools/bus/subscribe.ts @@ -0,0 +1,57 @@ +import { join } from "node:path"; +import { existsSync, readFileSync, writeFileSync } from "node:fs"; +import { BUS_DIR, ensureDir, list } from "./bus"; +import type { MessageEnvelope, Topic } from "./types"; + +/** + * Reads envelopes from the bus matching the given topic and recipient, + * calls the handler for each unseen envelope, and marks them as seen + * in a surface-specific seen.json file. + */ +export async function subscribeOnce( + topic: T, + surface: string, + handler: (envelope: MessageEnvelope & { topic: T }) => Promise | void, + adapters = { list } +): Promise { + ensureDir(); + const seenFile = join(BUS_DIR, `seen-${surface}.json`); + let seenIds: Set; + + try { + if (existsSync(seenFile)) { + const data = JSON.parse(readFileSync(seenFile, "utf8")); + seenIds = new Set(Array.isArray(data) ? data : []); + } else { + seenIds = new Set(); + } + } catch { + seenIds = new Set(); + } + + // Get all envelopes matching topic and targeted at this surface (or broadcast) + const envelopes = adapters.list({ topic, to: surface as any }); + + let newlySeen = false; + + for (const envelope of envelopes) { + if (!seenIds.has(envelope.id)) { + try { + await handler(envelope as MessageEnvelope & { topic: T }); + seenIds.add(envelope.id); + newlySeen = true; + } catch (err) { + // If handler fails, we do NOT mark as seen, so it can be retried next tick + console.error(`[subscribeOnce] Handler for ${envelope.id} failed:`, err); + } + } + } + + if (newlySeen) { + try { + writeFileSync(seenFile, JSON.stringify(Array.from(seenIds), null, 2)); + } catch (err) { + console.error(`[subscribeOnce] Failed to write seen file:`, err); + } + } +} From 13c497bd0f8c7fbe5305692d4918bee00c490324 Mon Sep 17 00:00:00 2001 From: Aaron Stainback Date: Sat, 16 May 2026 17:02:40 -0400 Subject: [PATCH 2/6] fix(bus): remove unused Topic import and rename unused mock param (#3775) --- tools/bus/subscribe.test.ts | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/tools/bus/subscribe.test.ts b/tools/bus/subscribe.test.ts index f6cc284835..bf2bb9757a 100644 --- a/tools/bus/subscribe.test.ts +++ b/tools/bus/subscribe.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test, mock } from "bun:test"; import { subscribeOnce } from "./subscribe"; -import type { MessageEnvelope, Topic } from "./types"; +import type { MessageEnvelope } from "./types"; import { rmSync } from "node:fs"; import { join } from "node:path"; import { BUS_DIR } from "./bus"; @@ -15,7 +15,7 @@ describe("bus subscribeOnce (B-0449 slice 5)", () => { test("calls handler for unseen envelopes and records seen state", async () => { clearState(); - + const env1: MessageEnvelope = { id: "env-1", from: "otto", @@ -25,15 +25,15 @@ describe("bus subscribeOnce (B-0449 slice 5)", () => { topic: "work-assignment", payload: { rowId: "B-1234", priority: "P1", rationale: "test" }, }; - + const fakeList = mock(() => { return [env1]; }); - const handler = mock(async (env) => {}); - + const handler = mock(async (_env: MessageEnvelope) => {}); + await subscribeOnce("work-assignment", "test-surface", handler, { list: fakeList as any }); - + expect(handler).toHaveBeenCalledTimes(1); expect(handler).toHaveBeenCalledWith(env1); @@ -44,7 +44,7 @@ describe("bus subscribeOnce (B-0449 slice 5)", () => { test("does not mark as seen if handler throws", async () => { clearState(); - + const env2: MessageEnvelope = { id: "env-2", from: "otto", @@ -54,14 +54,14 @@ describe("bus subscribeOnce (B-0449 slice 5)", () => { topic: "work-assignment", payload: { rowId: "B-2222", priority: "P2", rationale: "test2" }, }; - + const fakeList = mock(() => [env2]); const handlerFailing = mock(async () => { throw new Error("fail"); }); - + await subscribeOnce("work-assignment", "test-surface", handlerFailing, { list: fakeList as any }); - + expect(handlerFailing).toHaveBeenCalledTimes(1); - + // Call again, should retry because it failed and wasn't marked seen await subscribeOnce("work-assignment", "test-surface", handlerFailing, { list: fakeList as any }); expect(handlerFailing).toHaveBeenCalledTimes(2); From 5618091c2d57b6cecea03ff64e322f3ad083342d Mon Sep 17 00:00:00 2001 From: Aaron Stainback Date: Sat, 16 May 2026 17:03:13 -0400 Subject: [PATCH 3/6] fix(bus): remove existsSync to eliminate TOCTOU race (#3775) --- tools/bus/subscribe.ts | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/tools/bus/subscribe.ts b/tools/bus/subscribe.ts index 928f85efd0..8591dfd47d 100644 --- a/tools/bus/subscribe.ts +++ b/tools/bus/subscribe.ts @@ -1,5 +1,5 @@ import { join } from "node:path"; -import { existsSync, readFileSync, writeFileSync } from "node:fs"; +import { readFileSync, writeFileSync } from "node:fs"; import { BUS_DIR, ensureDir, list } from "./bus"; import type { MessageEnvelope, Topic } from "./types"; @@ -17,21 +17,20 @@ export async function subscribeOnce( ensureDir(); const seenFile = join(BUS_DIR, `seen-${surface}.json`); let seenIds: Set; - + + // Rely on a single readFileSync + catch instead of existsSync+read; this + // avoids a TOCTOU race (CodeQL js/file-system-race) and is functionally + // equivalent for our local-only seen.json. try { - if (existsSync(seenFile)) { - const data = JSON.parse(readFileSync(seenFile, "utf8")); - seenIds = new Set(Array.isArray(data) ? data : []); - } else { - seenIds = new Set(); - } + const data = JSON.parse(readFileSync(seenFile, "utf8")); + seenIds = new Set(Array.isArray(data) ? data : []); } catch { seenIds = new Set(); } // Get all envelopes matching topic and targeted at this surface (or broadcast) const envelopes = adapters.list({ topic, to: surface as any }); - + let newlySeen = false; for (const envelope of envelopes) { From c538a8365feeb254b0eb5f39d0c7c20034df64ea Mon Sep 17 00:00:00 2001 From: Aaron Stainback Date: Sat, 16 May 2026 17:26:01 -0400 Subject: [PATCH 4/6] =?UTF-8?q?fix(bus):=20subscribeOnce=20=E2=80=94=20sur?= =?UTF-8?q?face=20allowlist=20+=20ENOENT-aware=20seen=20read?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Copilot/Codex P1 threads on #3775: - Path-traversal: surface interpolated into `seen-${surface}.json` is now guarded by SURFACE_RE = /^[a-z0-9_-]{1,64}$/. Throws on invalid input; defense-in-depth even though current callers pass hard-coded surface ids. - Broad catch: distinguish ENOENT (no seen-file yet) from real errors. Re-throw permission / corruption errors so seen-state is never silently dropped. Validated locally: `bun build` clean; `bun test tools/bus/subscribe.test.ts` passes (2/2). --- tools/bus/subscribe.ts | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/tools/bus/subscribe.ts b/tools/bus/subscribe.ts index 8591dfd47d..62c6cc394a 100644 --- a/tools/bus/subscribe.ts +++ b/tools/bus/subscribe.ts @@ -3,6 +3,8 @@ import { readFileSync, writeFileSync } from "node:fs"; import { BUS_DIR, ensureDir, list } from "./bus"; import type { MessageEnvelope, Topic } from "./types"; +const SURFACE_RE = /^[a-z0-9_-]{1,64}$/; + /** * Reads envelopes from the bus matching the given topic and recipient, * calls the handler for each unseen envelope, and marks them as seen @@ -14,18 +16,28 @@ export async function subscribeOnce( handler: (envelope: MessageEnvelope & { topic: T }) => Promise | void, adapters = { list } ): Promise { + // P1 defense-in-depth: surface is interpolated into a filename; constrain + // to a safe charset so a stray "../" cannot escape BUS_DIR. + if (!SURFACE_RE.test(surface)) { + throw new Error(`subscribeOnce: invalid surface "${surface}" (must match ${SURFACE_RE})`); + } + ensureDir(); const seenFile = join(BUS_DIR, `seen-${surface}.json`); let seenIds: Set; - // Rely on a single readFileSync + catch instead of existsSync+read; this - // avoids a TOCTOU race (CodeQL js/file-system-race) and is functionally - // equivalent for our local-only seen.json. + // Single readFileSync + ENOENT-aware catch. Other errors (permission, + // JSON corruption) re-throw so we never silently drop seen-state and + // re-deliver already-handled envelopes. try { const data = JSON.parse(readFileSync(seenFile, "utf8")); seenIds = new Set(Array.isArray(data) ? data : []); - } catch { - seenIds = new Set(); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === "ENOENT") { + seenIds = new Set(); + } else { + throw err; + } } // Get all envelopes matching topic and targeted at this surface (or broadcast) From 36b6a35b0b5f36176125fcc8ff7d948dd3010e87 Mon Sep 17 00:00:00 2001 From: Aaron Stainback Date: Sat, 16 May 2026 17:26:11 -0400 Subject: [PATCH 5/6] =?UTF-8?q?chore(bus):=20correct=20describe=20label=20?= =?UTF-8?q?B-0449=20=E2=86=92=20B-0459=20in=20subscribe.test.ts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Copilot P2 thread on #3775. PR title + description say "B-0459 slice 1" but the describe string still read "B-0449 slice 5". --- tools/bus/subscribe.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tools/bus/subscribe.test.ts b/tools/bus/subscribe.test.ts index bf2bb9757a..19fc7d03aa 100644 --- a/tools/bus/subscribe.test.ts +++ b/tools/bus/subscribe.test.ts @@ -5,7 +5,7 @@ import { rmSync } from "node:fs"; import { join } from "node:path"; import { BUS_DIR } from "./bus"; -describe("bus subscribeOnce (B-0449 slice 5)", () => { +describe("bus subscribeOnce (B-0459 slice 1)", () => { const seenFile = join(BUS_DIR, "seen-test-surface.json"); // Helper to clear state From 5b490ba21397bfce0f4d2a5d13a34d7034003eb6 Mon Sep 17 00:00:00 2001 From: Aaron Stainback Date: Sat, 16 May 2026 20:14:07 -0400 Subject: [PATCH 6/6] fix(bus): propagate seen-file write failures from subscribeOnce MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Codex P2 thread on #3775. The read-side already re-throws non-ENOENT errors so we "never silently drop seen-state and re-deliver already-handled envelopes" — the write-side was breaking the same invariant in the opposite direction by logging-and-swallowing. Now writeFileSync errors propagate to the caller. Adds a test that pre-creates the seen file read-only so the read succeeds and the write fails with EACCES; subscribeOnce rejects rather than silently continuing. Co-Authored-By: Claude --- tools/bus/subscribe.test.ts | 39 +++++++++++++++++++++++++++++++++++-- tools/bus/subscribe.ts | 8 +++----- 2 files changed, 40 insertions(+), 7 deletions(-) diff --git a/tools/bus/subscribe.test.ts b/tools/bus/subscribe.test.ts index 19fc7d03aa..1fe63015bb 100644 --- a/tools/bus/subscribe.test.ts +++ b/tools/bus/subscribe.test.ts @@ -1,9 +1,9 @@ import { describe, expect, test, mock } from "bun:test"; import { subscribeOnce } from "./subscribe"; import type { MessageEnvelope } from "./types"; -import { rmSync } from "node:fs"; +import { chmodSync, rmSync, writeFileSync } from "node:fs"; import { join } from "node:path"; -import { BUS_DIR } from "./bus"; +import { BUS_DIR, ensureDir } from "./bus"; describe("bus subscribeOnce (B-0459 slice 1)", () => { const seenFile = join(BUS_DIR, "seen-test-surface.json"); @@ -42,6 +42,41 @@ describe("bus subscribeOnce (B-0459 slice 1)", () => { expect(handler).toHaveBeenCalledTimes(1); // Still 1 }); + test("propagates seen-file write failures to caller", async () => { + const surface = "test-write-fail"; + const failSeenFile = join(BUS_DIR, `seen-${surface}.json`); + + // Pre-create the seen file as readable-but-not-writable so the read + // succeeds (empty array) and the write throws EACCES. This catches + // the failure mode where a write error would have been silently + // logged, leaving subscribeOnce believing persistence succeeded. + ensureDir(); + writeFileSync(failSeenFile, "[]"); + chmodSync(failSeenFile, 0o444); + + try { + const env: MessageEnvelope = { + id: "env-write-fail", + from: "otto", + to: surface as any, + timestamp: new Date().toISOString(), + expiresAt: new Date(Date.now() + 10000).toISOString(), + topic: "work-assignment", + payload: { rowId: "B-3333", priority: "P2", rationale: "test3" }, + }; + + const fakeList = mock(() => [env]); + const handler = mock(async (_env: MessageEnvelope) => {}); + + await expect( + subscribeOnce("work-assignment", surface, handler, { list: fakeList as any }), + ).rejects.toThrow(); + } finally { + chmodSync(failSeenFile, 0o644); + try { rmSync(failSeenFile); } catch {} + } + }); + test("does not mark as seen if handler throws", async () => { clearState(); diff --git a/tools/bus/subscribe.ts b/tools/bus/subscribe.ts index 62c6cc394a..84d8ee0430 100644 --- a/tools/bus/subscribe.ts +++ b/tools/bus/subscribe.ts @@ -59,10 +59,8 @@ export async function subscribeOnce( } if (newlySeen) { - try { - writeFileSync(seenFile, JSON.stringify(Array.from(seenIds), null, 2)); - } catch (err) { - console.error(`[subscribeOnce] Failed to write seen file:`, err); - } + // Symmetric with read-side: surface persistence failures to the caller so + // already-handled envelopes don't silently re-deliver on the next poll. + writeFileSync(seenFile, JSON.stringify(Array.from(seenIds), null, 2)); } }