diff --git a/docs/hub.md b/docs/hub.md index b4df317e76..1fbe6b92d5 100644 --- a/docs/hub.md +++ b/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 diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index dde5c02bbd..d178c72f6c 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/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; 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, diff --git a/packages/server/src/server/hub/execution-session.websocket.test.ts b/packages/server/src/server/hub/execution-session.websocket.test.ts index 266c55e196..7cc2db0480 100644 --- a/packages/server/src/server/hub/execution-session.websocket.test.ts +++ b/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 = { diff --git a/packages/server/src/server/hub/test-utils/relationship-harness.ts b/packages/server/src/server/hub/test-utils/relationship-harness.ts index a0c02bbb12..f2e0e4dff6 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -458,6 +458,7 @@ export class HubRelationshipHarness { private readonly claimedCliSockets = new Set(); private readonly promptsToFail = new Set(); 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; } diff --git a/packages/server/src/server/message-receipts/index.test.ts b/packages/server/src/server/message-receipts/index.test.ts index 10c1615a6a..f77ced4d6e 100644 --- a/packages/server/src/server/message-receipts/index.test.ts +++ b/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) => { + 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) => { + 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) => { + 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; diff --git a/packages/server/src/server/message-receipts/index.ts b/packages/server/src/server/message-receipts/index.ts index 6103160b12..f76d1381ec 100644 --- a/packages/server/src/server/message-receipts/index.ts +++ b/packages/server/src/server/message-receipts/index.ts @@ -13,8 +13,11 @@ interface SendMessageInput { agentId: string; messageId: string; request: unknown; - send: () => Promise; - prepare?: () => Promise; + /** + * 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) => Promise; } /** 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 | null = null; + await input.send(() => { + pending ??= writeJsonFileAtomic(file, { ...receipt, state: "pending" }); + return pending; + }); await writeJsonFileAtomic(file, { ...receipt, state: "completed" }); } } diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index d6567da4ca..ae19cdc0d4 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -8918,8 +8918,9 @@ export class Session { queued = acceptAdmission; }); let deliveryStatus: "queued" | undefined; - const send = async () => { + const send = async (beforeDispatch: () => Promise) => { if (msg.sourceAgentId) { + await beforeDispatch(); const receipt = this.agentMessages.enqueue({ recipientId: agentId, messageId: msg.messageId, @@ -8950,6 +8951,7 @@ export class Session { activeTurnBehavior: msg.activeTurnBehavior ?? "interrupt", sessionMode: msg.sessionMode, clearPendingPermissions: true, + beforeDispatch, logger: this.sessionLogger, }); if (result.disposition === "turn_started") { @@ -8972,12 +8974,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"); diff --git a/packages/server/src/server/test-utils/fake-agent-client.ts b/packages/server/src/server/test-utils/fake-agent-client.ts index a7a237b9de..1be2a405e8 100644 --- a/packages/server/src/server/test-utils/fake-agent-client.ts +++ b/packages/server/src/server/test-utils/fake-agent-client.ts @@ -84,6 +84,7 @@ interface FakeAgentSessionOptions { beforeTurnComplete?: (prompt: AgentPromptInput) => Promise; steerActiveTurn?: AgentSession["steerActiveTurn"]; onInterrupt?: () => void; + beforeSetMode?: (modeId: string) => Promise; historyEvents?: readonly AgentStreamEvent[]; } @@ -95,6 +96,7 @@ export interface TestAgentClientOptions { beforeTurnComplete?: (prompt: AgentPromptInput) => Promise; steerActiveTurn?: AgentSession["steerActiveTurn"]; onInterrupt?: () => void; + beforeSetMode?: (modeId: string) => Promise; 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) | undefined; private readonly closeSession: (() => Promise) | undefined; private readonly onStartTurn: TestAgentClientOptions["onStartTurn"]; private readonly beforeTurnComplete: ((prompt: AgentPromptInput) => Promise) | 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 { + 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, }); } diff --git a/packages/server/src/server/test-utils/session-stubs.ts b/packages/server/src/server/test-utils/session-stubs.ts index 0c156ae5d9..7546da1427 100644 --- a/packages/server/src/server/test-utils/session-stubs.ts +++ b/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 () => {}), }; }