docs(hub): name run startup as the ambiguous boundary

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/hub.md

@@ -61,6 +61,10 @@ The daemon journals the assigned agent ID before creation. Concurrent retries sh
 and a retry after a lost acknowledgement or restart returns the durable agent without creating a
 workspace or starting a turn. Deleting that agent does not make its old creation key reusable.
 Confirmed message delivery is also journaled, so retrying its message ID does not submit it again.
+A direct send (one without `sourceAgentId`) rejected before its prompt is handed to a run, for
+example because the agent could not be loaded or the requested mode could not be set, leaves no
+receipt, so retrying its message ID submits it. A failure during or after run startup is treated
+as an unfinished delivery.
 An unfinished delivery after restart reports `agent_request_outcome_unknown`: a provider can have
 accepted a prompt before the daemon recorded success, so automatic resubmission could duplicate
 work. Inspect the agent before choosing a new message ID. Receipts contain identity and request

packages/server/src/server/agent/agent-prompt.ts

@@ -210,6 +210,8 @@ export async function unarchiveAgentState(
 }
 
 export interface SendPromptToAgentParams {
+  /** Awaited after the agent is loaded and its mode is set, before the prompt is handed to a run. */
+  beforeDispatch?: () => Promise<void>;
   onDispatch?: StartAgentRunOptions["onDispatch"];
   agentManager: AgentManager;
   agentStorage: AgentStorage;
@@ -357,6 +359,7 @@ export async function sendPreparedPromptToAgent(
     await params.agentManager.setAgentMode(params.agentId, params.sessionMode);
   }
 
+  await params.beforeDispatch?.();
   return await startAgentRun(params.agentManager, params.agentId, params.prompt, params.logger, {
     replaceRunning: true,
     onDispatch: params.onDispatch,

packages/server/src/server/hub/execution-session.websocket.test.ts

@@ -182,6 +182,46 @@ test("ordinary Hub create and message retries do not duplicate agents or prompts
   ).toMatchObject({ payload: { accepted: false } });
 });
 
+test("Hub retries a message whose send failed before reaching the provider", async () => {
+  const hub = await launchRelationship();
+  const agentId = await hub.createUnrelatedLocalAgent();
+  const message = {
+    type: "send_agent_message_request",
+    agentId,
+    messageId: "mode-change-failed",
+    text: "retry me",
+    sessionMode: "full-access",
+  };
+  hub.failNextProviderModeChange();
+  expect(await hub.requestOrdinary({ ...message, requestId: "first-attempt" })).toMatchObject({
+    payload: { accepted: false, error: expect.stringContaining("mode change failure") },
+  });
+  expect(hub.providerPromptTexts().filter((text) => text === "retry me")).toHaveLength(0);
+
+  expect(await hub.requestOrdinary({ ...message, requestId: "retry" })).toMatchObject({
+    payload: { accepted: true, error: null },
+  });
+  expect(hub.providerPromptTexts().filter((text) => text === "retry me")).toHaveLength(1);
+});
+
+test("Hub never resubmits a message whose send failed after reaching the provider", async () => {
+  const hub = await launchRelationship();
+  const agentId = await hub.createUnrelatedLocalAgent();
+  const message = {
+    type: "send_agent_message_request",
+    agentId,
+    messageId: "turn-start-failed",
+    text: "do not repeat me",
+  };
+  hub.failProviderPromptStart("do not repeat me");
+  await hub.requestOrdinary({ ...message, requestId: "first-attempt" });
+
+  expect(await hub.requestOrdinary({ ...message, requestId: "retry" })).toMatchObject({
+    payload: { accepted: false, error: "agent_request_outcome_unknown" },
+  });
+  expect(hub.providerPromptTexts().filter((text) => text === "do not repeat me")).toHaveLength(0);
+});
+
 test("ordinary Hub requests survive daemon restart and restore an archived workspace", async () => {
   const hub = await launchRelationship();
   const create = {

packages/server/src/server/hub/test-utils/relationship-harness.ts

@@ -458,6 +458,7 @@ export class HubRelationshipHarness {
   private readonly claimedCliSockets = new Set<WebSocket>();
   private readonly promptsToFail = new Set<string>();
   private failNextSessionClose = false;
+  private failNextModeChange = false;
   private observedEnrollments = 0;
   private observedSockets = 0;
   private readonly codex: ControlledAgentClient;
@@ -474,6 +475,11 @@ export class HubRelationshipHarness {
           this.failNextSessionClose = false;
           throw new Error("Requested provider session close failure");
         },
+        beforeSetMode: async () => {
+          if (!this.failNextModeChange) return;
+          this.failNextModeChange = false;
+          throw new Error("Requested provider mode change failure");
+        },
         onStartTurn: (prompt) => {
           const promptText = typeof prompt === "string" ? prompt : JSON.stringify(prompt);
           if (this.promptsToFail.delete(promptText)) {
@@ -674,6 +680,10 @@ export class HubRelationshipHarness {
     this.promptsToFail.add(prompt);
   }
 
+  failNextProviderModeChange(): void {
+    this.failNextModeChange = true;
+  }
+
   failNextProviderSessionClose(): void {
     this.failNextSessionClose = true;
   }

packages/server/src/server/message-receipts/index.test.ts

@@ -23,7 +23,8 @@ test("message retries survive reconstruction without submitting twice", async ()
     agentId: "agent",
     messageId: "arrival",
     request: { text: "hello" },
-    send: async () => {
+    send: async (beforeDispatch: () => Promise<void>) => {
+      await beforeDispatch();
       deliveries++;
     },
   };
@@ -41,7 +42,8 @@ test("ambiguous provider delivery is never blindly replayed after restart", asyn
     agentId: "agent",
     messageId: "arrival",
     request: {},
-    send: async () => {
+    send: async (beforeDispatch: () => Promise<void>) => {
+      await beforeDispatch();
       deliveries++;
       throw new Error("connection lost");
     },
@@ -53,7 +55,7 @@ test("ambiguous provider delivery is never blindly replayed after restart", asyn
   expect(deliveries).toBe(1);
 });
 
-test("failed local message preparation does not leave an ambiguous receipt", async () => {
+test("a send that fails before beforeDispatch can be retried with the same message", async () => {
   const { requests, directory } = await fixture();
   let available = false;
   let sends = 0;
@@ -61,14 +63,13 @@ test("failed local message preparation does not leave an ambiguous receipt", asy
     agentId: "agent",
     messageId: "message",
     request: {},
-    prepare: async () => {
-      if (!available) throw new Error("load failed");
-    },
-    send: async () => {
+    send: async (beforeDispatch: () => Promise<void>) => {
+      if (!available) throw new Error("mode change failed");
+      await beforeDispatch();
       sends++;
     },
   };
-  await expect(requests.send(input)).rejects.toThrow("load failed");
+  await expect(requests.send(input)).rejects.toThrow("mode change failed");
   available = true;
   await new MessageReceipts(directory).send(input);
   available = false;

packages/server/src/server/message-receipts/index.ts

@@ -13,8 +13,11 @@ interface SendMessageInput {
   agentId: string;
   messageId: string;
   request: unknown;
-  send: () => Promise<void>;
-  prepare?: () => Promise<void>;
+  /**
+   * Delivers the message. It awaits `beforeDispatch` immediately before the provider can receive
+   * the message; a failure before that call proves nothing was submitted and leaves no receipt.
+   */
+  send: (beforeDispatch: () => Promise<void>) => Promise<void>;
 }
 
 /** Owns message delivery receipts; creation is owned by CreationService. */
@@ -48,10 +51,12 @@ export class MessageReceipts {
       // A provider may have accepted the message before its receipt was committed.
       throw new Error("agent_request_outcome_unknown");
     }
-    await input.prepare?.();
     const receipt = { fingerprint, agentId: input.agentId };
-    await writeJsonFileAtomic(file, { ...receipt, state: "pending" });
-    await input.send();
+    let pending: Promise<void> | null = null;
+    await input.send(() => {
+      pending ??= writeJsonFileAtomic(file, { ...receipt, state: "pending" });
+      return pending;
+    });
     await writeJsonFileAtomic(file, { ...receipt, state: "completed" });
   }
 }

packages/server/src/server/session.ts

@@ -8917,8 +8917,9 @@ export class Session {
         queued = acceptAdmission;
       });
       let deliveryStatus: "queued" | undefined;
-      const send = async () => {
+      const send = async (beforeDispatch: () => Promise<void>) => {
         if (msg.sourceAgentId) {
+          await beforeDispatch();
           const receipt = this.agentMessages.enqueue({
             recipientId: agentId,
             messageId: msg.messageId,
@@ -8949,6 +8950,7 @@ export class Session {
           activeTurnBehavior: msg.activeTurnBehavior ?? "interrupt",
           sessionMode: msg.sessionMode,
           clearPendingPermissions: true,
+          beforeDispatch,
           logger: this.sessionLogger,
         });
         if (result.disposition === "turn_started") {
@@ -8971,12 +8973,12 @@ export class Session {
               activeTurnBehavior:
                 msg.activeTurnBehavior ?? (msg.sourceAgentId ? "steer" : "interrupt"),
             },
-            prepare: async () => {
+            send: async (beforeDispatch) => {
               if (!msg.sourceAgentId) await this.prepareAgentMessage(agentId, msg.text);
+              await send(beforeDispatch);
             },
-            send,
           })
-        : send();
+        : send(async () => {});
       void sent.catch((error) => {
         if (deliveryStatus === "queued") {
           this.handleAgentRunError(agentId, error, "Failed to dispatch queued agent message");

packages/server/src/server/test-utils/fake-agent-client.ts

@@ -84,6 +84,7 @@ interface FakeAgentSessionOptions {
   beforeTurnComplete?: (prompt: AgentPromptInput) => Promise<void>;
   steerActiveTurn?: AgentSession["steerActiveTurn"];
   onInterrupt?: () => void;
+  beforeSetMode?: (modeId: string) => Promise<void>;
   historyEvents?: readonly AgentStreamEvent[];
 }
 
@@ -95,6 +96,7 @@ export interface TestAgentClientOptions {
   beforeTurnComplete?: (prompt: AgentPromptInput) => Promise<void>;
   steerActiveTurn?: AgentSession["steerActiveTurn"];
   onInterrupt?: () => void;
+  beforeSetMode?: (modeId: string) => Promise<void>;
   supportsMcpServers?: boolean;
   historyEvents?: readonly AgentStreamEvent[];
 }
@@ -369,6 +371,7 @@ class FakeAgentSession implements AgentSession {
 
   readonly steerActiveTurn: AgentSession["steerActiveTurn"];
   private readonly onInterrupt: (() => void) | undefined;
+  private readonly beforeSetMode: ((modeId: string) => Promise<void>) | undefined;
   private readonly closeSession: (() => Promise<void>) | undefined;
   private readonly onStartTurn: TestAgentClientOptions["onStartTurn"];
   private readonly beforeTurnComplete: ((prompt: AgentPromptInput) => Promise<void>) | undefined;
@@ -385,6 +388,7 @@ class FakeAgentSession implements AgentSession {
     this.memoryMarker = options.memoryMarker ?? null;
     this.steerActiveTurn = options.steerActiveTurn;
     this.onInterrupt = options.onInterrupt;
+    this.beforeSetMode = options.beforeSetMode;
     this.closeSession = options.closeSession;
     this.onStartTurn = options.onStartTurn;
     this.beforeTurnComplete = options.beforeTurnComplete;
@@ -891,6 +895,7 @@ class FakeAgentSession implements AgentSession {
   }
 
   async setMode(modeId: string): Promise<void> {
+    await this.beforeSetMode?.(modeId);
     this.config.modeId = modeId;
   }
 
@@ -1264,6 +1269,7 @@ class FakeAgentClient implements AgentClient {
       beforeTurnComplete: this.options.beforeTurnComplete,
       steerActiveTurn: this.options.steerActiveTurn,
       onInterrupt: this.options.onInterrupt,
+      beforeSetMode: this.options.beforeSetMode,
       historyEvents: this.options.historyEvents,
     });
   }
@@ -1294,6 +1300,7 @@ class FakeAgentClient implements AgentClient {
       beforeTurnComplete: this.options.beforeTurnComplete,
       steerActiveTurn: this.options.steerActiveTurn,
       onInterrupt: this.options.onInterrupt,
+      beforeSetMode: this.options.beforeSetMode,
       historyEvents: this.options.historyEvents,
     });
   }

packages/server/src/server/test-utils/session-stubs.ts

@@ -306,7 +306,7 @@ export function createProviderSnapshotManagerStub(): {
 
 export function createMessageReceiptsStub(): SessionOptions["messageReceipts"] {
   return {
-    send: (input) => input.send(),
+    send: (input) => input.send(async () => {}),
   };
 }