diff --git a/tools/bus/subscribe.test.ts b/tools/bus/subscribe.test.ts
new file mode 100644
index 0000000000..1fe63015bb
--- /dev/null
+++ b/tools/bus/subscribe.test.ts
@@ -0,0 +1,104 @@
+import { describe, expect, test, mock } from "bun:test";
+import { subscribeOnce } from "./subscribe";
+import type { MessageEnvelope } from "./types";
+import { chmodSync, rmSync, writeFileSync } from "node:fs";
+import { join } from "node:path";
+import { BUS_DIR, ensureDir } from "./bus";
+
+describe("bus subscribeOnce (B-0459 slice 1)", () => {
+  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: MessageEnvelope) => {});
+
+    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("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();
+
+    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..84d8ee0430
--- /dev/null
+++ b/tools/bus/subscribe.ts
@@ -0,0 +1,66 @@
+import { join } from "node:path";
+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
+ * in a surface-specific seen.json file.
+ */
+export async function subscribeOnce<T extends Topic>(
+  topic: T,
+  surface: string,
+  handler: (envelope: MessageEnvelope & { topic: T }) => Promise<void> | void,
+  adapters = { list }
+): Promise<void> {
+  // 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<string>;
+
+  // 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 (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)
+  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) {
+    // 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));
+  }
+}