Persist queued delegations, and scrub secrets from what the harness stores

server/delegations.test.ts

@@ -242,7 +242,7 @@ describe("drainDelegations", () => {
       },
     );
 
-    await waitFor(() => runTargetCalls.length === 1);
+    await waitFor(() => runTargetCalls.length === 1 && _pendingCount(routineTask.threadId) === 0);
     expect(_pendingCount(routineTask.threadId)).toBe(0);
     expect(runTargetCalls[0]?.sourceThreadId).toBe(routineTask.threadId);
     expect(
@@ -401,3 +401,116 @@ describe("drainDelegations", () => {
     expect(runTargetCalls).toEqual([]);
   });
 });
+
+import { existsSync, readFileSync } from "node:fs";
+import { join } from "node:path";
+import { _loadPending, _resetPending, discardDelegations, pendingThreads } from "./delegations.ts";
+
+describe("delegations survive a restart", () => {
+  let store: Store;
+  let from: BotRecord;
+  let target: BotRecord;
+  let buses: BusPair;
+  const file = () => join(DATA_DIR, "delegations.json");
+
+  beforeEach(() => {
+    rmSync(DATA_DIR, { recursive: true, force: true });
+    _resetPending();
+    store = new Store(selection);
+    from = store.createBot();
+    target = store.createBot();
+    store.patchBot(target.id, { name: "Helper" });
+    buses = setupBuses(store);
+  });
+  afterEach(() => _resetPending());
+
+  it("writes the queue to disk on queue, and clears it on drain and discard", async () => {
+    expect(queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1)).toBe("ok");
+    expect(existsSync(file())).toBe(true);
+    const onDisk = JSON.parse(readFileSync(file(), "utf8")) as Record<string, unknown[]>;
+    expect(onDisk[from.threadId]).toHaveLength(1);
+    expect(onDisk[from.threadId][0]).toMatchObject({ toBotId: target.id, message: "do this" });
+
+    discardDelegations(buses.commsBus, from.threadId);
+    expect(JSON.parse(readFileSync(file(), "utf8"))[from.threadId]).toBeUndefined();
+
+    queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "again", depth: 0 }, 1);
+    const ran: string[] = [];
+    drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, async (_to, message) => {
+      ran.push(message);
+    });
+    await waitFor(() => ran.length === 1 && pendingThreads().length === 0);
+    expect(JSON.parse(readFileSync(file(), "utf8"))[from.threadId]).toBeUndefined();
+  });
+
+  it("keeps a handoff durable until its approval and dispatch path settles", async () => {
+    queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "wait for dispatch", depth: 0 }, 1);
+    let release!: () => void;
+    const dispatchSettled = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    let started = false;
+    drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, async () => {
+      started = true;
+      await dispatchSettled;
+    });
+
+    await waitFor(() => started);
+    expect(pendingThreads()).toEqual([from.threadId]);
+    expect(JSON.parse(readFileSync(file(), "utf8"))[from.threadId]).toHaveLength(1);
+
+    release();
+    await waitFor(() => pendingThreads().length === 0);
+    expect(JSON.parse(readFileSync(file(), "utf8"))[from.threadId]).toBeUndefined();
+  });
+
+  it("drains work queued by a later settled turn while an earlier handoff is waiting", async () => {
+    queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "first", depth: 0 }, 1);
+    let release!: () => void;
+    const firstSettled = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    const ran: string[] = [];
+    const runTarget = async (_to: string, message: string) => {
+      ran.push(message);
+      if (message.includes("first")) await firstSettled;
+    };
+    drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, runTarget);
+    await waitFor(() => ran.length === 1);
+
+    queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "second", depth: 0 }, 1);
+    drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, runTarget);
+    expect(ran).toHaveLength(1);
+
+    release();
+    await waitFor(() => ran.length === 2 && pendingThreads().length === 0);
+    expect(ran[1]).toContain("second");
+  });
+
+  it("a fresh process loads what the last one queued, and can drain it", async () => {
+    queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "left over", depth: 0 }, 1);
+    // "restart": forget memory, reload from disk
+    _resetPending();
+    expect(pendingThreads()).toEqual([]);
+    _loadPending();
+    expect(pendingThreads()).toEqual([from.threadId]);
+    const ran: string[] = [];
+    drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, async (_to, message) => {
+      ran.push(message);
+    });
+    await waitFor(() => ran.length === 1 && pendingThreads().length === 0);
+    expect(ran[0]).toContain("left over");
+    expect(pendingThreads()).toEqual([]);
+  });
+
+  it("tolerates a missing or corrupt file", () => {
+    _resetPending();
+    _loadPending(); // no file
+    expect(pendingThreads()).toEqual([]);
+    const { mkdirSync, writeFileSync } = require("node:fs") as typeof import("node:fs");
+    mkdirSync(DATA_DIR, { recursive: true });
+    writeFileSync(file(), "{not json");
+    _loadPending();
+    expect(pendingThreads()).toEqual([]);
+  });
+});

server/delegations.ts

@@ -11,7 +11,13 @@
 // time, never at queue time, because the user might have just turned
 // approvePeerComms on between queueing and draining.
 
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+
+import { writeFileAtomic } from "./atomic.ts";
 import { getOrCreateChannel, mirrorExchange, type CommsBus } from "./comms-visibility.ts";
+import { DATA_DIR } from "./config.ts";
+import { newId } from "./contracts.ts";
 import { requestPeerApproval, type ApprovalBus } from "./peer-approval.ts";
 import type { BotRecord, GroupRecord } from "./store.ts";
 
@@ -26,12 +32,64 @@ export interface DelegationItem {
   depth: number;
 }
 
+interface PendingDelegationItem extends DelegationItem {
+  /** Stable acknowledgement key for crash-safe removal from the queue. */
+  id: string;
+}
+
 export type QueueResult = "ok" | "no_target" | "self" | "too_deep" | "too_many";
 
-/** Per source-thread queue. Persisted nowhere — a server restart drops
- * delegations the same way provider permissions drop, which is honest:
- * nobody can answer for an unattended bot. */
-const pendingDelegations = new Map<string, DelegationItem[]>();
+/** Per source-thread queue. Persisted to delegations.json on every change
+ * and reloaded at boot: a handoff queued right before a restart runs after
+ * it. (Provider PERMISSIONS still die with the process — nobody can answer
+ * for an unattended bot — but queued work is not a permission; the target
+ * and approvePeerComms are re-checked at drain time as always.) */
+const pendingDelegations = new Map<string, PendingDelegationItem[]>();
+const drainingThreads = new Set<string>();
+const DELEGATIONS_FILE = join(DATA_DIR, "delegations.json");
+
+function savePending(): void {
+  try {
+    writeFileAtomic(DELEGATIONS_FILE, JSON.stringify(Object.fromEntries(pendingDelegations), null, 2), { mode: 0o600 });
+  } catch (error) {
+    console.error("delegations: could not persist queue", error);
+  }
+}
+
+/** Load what a previous process left queued. Missing or corrupt → empty. */
+export function _loadPending(): void {
+  pendingDelegations.clear();
+  try {
+    const raw = JSON.parse(readFileSync(DELEGATIONS_FILE, "utf8")) as Record<string, unknown>;
+    for (const [threadId, list] of Object.entries(raw)) {
+      if (!Array.isArray(list)) continue;
+      const items = list.flatMap((value): PendingDelegationItem[] => {
+        if (!value || typeof value !== "object") return [];
+        const item = value as Partial<PendingDelegationItem>;
+        if (
+          typeof item.toBotId !== "string" ||
+          typeof item.message !== "string" ||
+          !Number.isFinite(item.depth)
+        ) return [];
+        return [{
+          id: typeof item.id === "string" && item.id ? item.id : newId(),
+          toBotId: item.toBotId,
+          message: item.message,
+          ...(typeof item.reason === "string" ? { reason: item.reason } : {}),
+          depth: Math.max(0, Math.trunc(item.depth!)),
+        }];
+      });
+      if (items.length) pendingDelegations.set(threadId, items);
+    }
+  } catch {
+    /* fresh install, or unreadable — start empty */
+  }
+}
+
+/** Source threads with something queued — what a boot drain iterates. */
+export function pendingThreads(): string[] {
+  return [...pendingDelegations.keys()];
+}
 
 /** How many handoffs one turn may queue. Small on purpose: this is the only
  * thing standing between a confused bot and a fan-out of real turns. */
@@ -55,8 +113,9 @@ export function queueDelegation(
   // making the caller wait. Without a cap, one turn can queue unboundedly
   // and fan out into as many real turns on the next settle.
   if (list.length >= MAX_QUEUED_PER_THREAD) return "too_many";
-  list.push(item);
+  list.push({ ...item, id: newId() });
   pendingDelegations.set(sourceThreadId, list);
+  savePending();
   const label = `Delegated to @${target.name}${item.reason ? `: ${item.reason}` : ""}`;
   bus.store.appendMessage(sourceThreadId, {
     role: "bot",
@@ -84,25 +143,55 @@ export function drainDelegations(
     channel?: GroupRecord,
   ) => void | Promise<void>,
 ): void {
+  if (drainingThreads.has(threadId)) return;
   const list = pendingDelegations.get(threadId);
   if (!list?.length) return;
-  pendingDelegations.delete(threadId);
   const from = bus.store.botByThread(threadId);
-  if (!from) return;
-  for (const item of list) {
-    void processOne(bus, approvalBus, from, threadId, item, runTarget).catch((error) => {
-      const why = error instanceof Error ? error.message : String(error);
+  if (!from) {
+    pendingDelegations.delete(threadId);
+    savePending();
+    return;
+  }
+  const snapshot = [...list];
+  drainingThreads.add(threadId);
+  void (async () => {
+    for (const item of snapshot) {
       try {
-        bus.store.appendMessage(threadId, {
-          role: "bot",
-          kind: "activity",
-          tool: { name: `error: delegation failed — ${why.slice(0, 120)}`, ok: false },
-        });
-      } catch (reportError) {
-        console.error("delegation failed and could not be reported", reportError);
+        await processOne(bus, approvalBus, from, threadId, item, runTarget);
+      } catch (error) {
+        const why = error instanceof Error ? error.message : String(error);
+        try {
+          bus.store.appendMessage(threadId, {
+            role: "bot",
+            kind: "activity",
+            tool: { name: `error: delegation failed — ${why.slice(0, 120)}`, ok: false },
+          });
+        } catch (reportError) {
+          console.error("delegation failed and could not be reported", reportError);
+        }
+      } finally {
+        acknowledgeDelegation(threadId, item.id);
       }
-    });
-  }
+    }
+  })().finally(() => {
+    drainingThreads.delete(threadId);
+    // A later turn may have queued and settled while this thread was
+    // waiting for approval. Its items were not in our snapshot, so start a
+    // fresh drain instead of leaving them parked until another restart.
+    if (pendingDelegations.get(threadId)?.length) {
+      drainDelegations(bus, approvalBus, threadId, runTarget);
+    }
+  });
+}
+
+/** Remove one terminal handoff only after approval/dispatch has settled. */
+function acknowledgeDelegation(threadId: string, itemId: string): void {
+  const current = pendingDelegations.get(threadId);
+  if (!current) return;
+  const remaining = current.filter((item) => item.id !== itemId);
+  if (remaining.length) pendingDelegations.set(threadId, remaining);
+  else pendingDelegations.delete(threadId);
+  savePending();
 }
 
 /** Drop a thread's queued handoffs without running them, telling the user
@@ -111,6 +200,7 @@ export function discardDelegations(bus: CommsBus, threadId: string): void {
   const list = pendingDelegations.get(threadId);
   if (!list?.length) return;
   pendingDelegations.delete(threadId);
+  savePending();
   const from = bus.store.botByThread(threadId);
   if (!from) return;
   bus.store.appendMessage(threadId, {
@@ -198,3 +288,9 @@ async function processOne(
 export function _pendingCount(threadId: string): number {
   return pendingDelegations.get(threadId)?.length ?? 0;
 }
+
+/** Test helper: forget the in-memory queue (a simulated restart). */
+export function _resetPending(): void {
+  pendingDelegations.clear();
+  drainingThreads.clear();
+}

server/harness/bus.test.ts

@@ -73,6 +73,20 @@ describe("EventBus", () => {
     expect(logged[0].type).toBe("turn.started");
   });
 
+  it("redacts credential-shaped content before writing the NDJSON log", () => {
+    const key = `sk-ant-api03-${"abcdefghijklmnopqrstuvwxyz0123456789"}`;
+    const bus = new EventBus();
+    bus.publish(testEvent({
+      threadId: "redacted-log",
+      type: "runtime.error",
+      message: `provider returned ${key}`,
+    }));
+
+    const logged = readFileSync(join(EVENTS_DIR, "redacted-log.ndjson"), "utf8");
+    expect(logged).not.toContain(key);
+    expect(logged).toContain("«redacted");
+  });
+
   it("still delivers when the NDJSON log cannot be written", () => {
     rmSync(EVENTS_DIR, { recursive: true, force: true });
     const bus = new EventBus();

server/harness/bus.ts

@@ -8,6 +8,7 @@ import { appendFileSync } from "node:fs";
 import { join } from "node:path";
 
 import { EVENTS_DIR } from "../config.ts";
+import { redactSecrets } from "../redact.ts";
 import type { ProviderInstance, RuntimeEvent, RuntimeEventListener } from "../contracts.ts";
 
 export class EventBus {
@@ -31,7 +32,14 @@ export class EventBus {
 
   publish(event: RuntimeEvent) {
     try {
-      appendFileSync(join(EVENTS_DIR, `${event.threadId}.ndjson`), JSON.stringify(event) + "\n");
+      // the canonical log is a file people paste into bug reports; scrub
+      // credential-shaped content (tool titles, request summaries, reply
+      // text) the same way the native tee does
+      appendFileSync(
+        join(EVENTS_DIR, `${event.threadId}.ndjson`),
+        JSON.stringify(redactSecrets(event)) + "\n",
+        { mode: 0o600 },
+      );
     } catch {
       /* logging must never take down the stream */
     }