diff --git a/src/auto-reply/get-reply-options.types.ts b/src/auto-reply/get-reply-options.types.ts index 0947439fdb8a..5b0922c416ba 100644 --- a/src/auto-reply/get-reply-options.types.ts +++ b/src/auto-reply/get-reply-options.types.ts @@ -91,6 +91,12 @@ export type GetReplyOptions = { imageOrder?: PromptImageOrderEntry[]; /** Notifies when an agent run actually starts (useful for webchat command handling). */ onAgentRunStart?: (runId: string) => void; + /** + * Called after the restart-recovery delivery-context persist attempt + * completes (context may be absent when source delivery is suppressed). + * Channels may complete ingress ownership here without waiting for settle. + */ + onTurnAdopted?: () => void | Promise; /** Shared lifecycle owner for the current user-turn transcript append. */ userTurnTranscriptRecorder?: UserTurnTranscriptRecorder; onReplyStart?: () => Promise | void; diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts index 56622122d0b9..be81f109b5bd 100644 --- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts +++ b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts @@ -743,6 +743,105 @@ describe("runReplyAgent pending final delivery capture", () => { expect(stored.restartRecoveryDeliveryRunId).toBeUndefined(); }); + it("fires onTurnAdopted after restart recovery delivery context persist completes", async () => { + const sessionEntry: SessionEntry = { + sessionId: "session", + updatedAt: Date.now(), + }; + const sessionStore = { main: sessionEntry }; + const storePath = await createSessionStoreFile(sessionEntry); + const events: string[] = []; + const onTurnAdopted = vi.fn(async () => { + const storedAtAdoption = await readStoredMainSession(storePath); + expect(storedAtAdoption.restartRecoveryDeliveryContext).toEqual({ + channel: "discord", + to: "channel:24680", + accountId: "work", + threadId: "1503645939964055592", + }); + expect(typeof storedAtAdoption.restartRecoveryDeliveryRunId).toBe("string"); + events.push("adopted"); + }); + state.runEmbeddedAgentMock.mockImplementationOnce(async () => { + events.push("agent-run"); + return { + payloads: [{ text: "visible final" }], + meta: {}, + }; + }); + + const { run } = createMinimalRun({ + opts: { onTurnAdopted }, + sessionCtx: { + Provider: "discord", + OriginatingChannel: "discord", + OriginatingTo: "channel:24680", + AccountId: "work", + MessageSid: "1503645939964055592", + MessageThreadId: "1503645939964055592", + }, + runOverrides: { messageProvider: "discord" }, + sessionEntry, + sessionStore, + sessionKey: "main", + storePath, + }); + + await run(); + + expect(onTurnAdopted).toHaveBeenCalledOnce(); + expect(events).toEqual(["adopted", "agent-run"]); + }); + + it("fires onTurnAdopted for suppressed-delivery runs before the agent turn", async () => { + const sessionEntry: SessionEntry = { + sessionId: "session", + updatedAt: Date.now(), + }; + const sessionStore = { main: sessionEntry }; + const storePath = await createSessionStoreFile(sessionEntry); + const events: string[] = []; + const onTurnAdopted = vi.fn(async () => { + const storedAtAdoption = await readStoredMainSession(storePath); + expect(storedAtAdoption.restartRecoveryDeliveryContext).toBeUndefined(); + expect(storedAtAdoption.restartRecoveryDeliveryRunId).toBeUndefined(); + events.push("adopted"); + }); + state.runEmbeddedAgentMock.mockImplementationOnce(async () => { + events.push("agent-run"); + return { + payloads: [{ text: "ambient final" }], + meta: {}, + }; + }); + + const { run } = createMinimalRun({ + opts: { + onTurnAdopted, + sourceReplyDeliveryMode: "message_tool_only", + }, + sessionCtx: { + Provider: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "telegram:123", + AccountId: "default", + MessageSid: "42", + InboundEventKind: "room_event", + }, + runOverrides: { messageProvider: "telegram" }, + sessionEntry, + sessionStore, + sessionKey: "main", + storePath, + currentInboundEventKind: "room_event", + }); + + await run(); + + expect(onTurnAdopted).toHaveBeenCalledOnce(); + expect(events).toEqual(["adopted", "agent-run"]); + }); + it("keeps heartbeat replies with real content in pending final delivery", async () => { const sessionEntry: SessionEntry = { sessionId: "session", diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts index 88acd4303138..e773677bf81b 100644 --- a/src/auto-reply/reply/agent-runner.ts +++ b/src/auto-reply/reply/agent-runner.ts @@ -1728,6 +1728,10 @@ export async function runReplyAgent(params: { replyOperation.setPhase("running"); const runStartedAt = Date.now(); await persistRestartRecoveryDeliveryContext(); + // Adoption marks run start and must never be spool-replayed (would re-run tools). + // Suppressed delivery has no recovery state to persist; crashed suppressed runs die + // silently. When a delivery context is resolvable, this still runs after its persist. + await opts?.onTurnAdopted?.(); const runOutcome = await traceAgentPhase("reply.run_agent_turn", () => runAgentTurnWithFallback({ commandBody, diff --git a/src/channels/turn/kernel.test.ts b/src/channels/turn/kernel.test.ts index 1d0ecfdf7840..11d148829e9a 100644 --- a/src/channels/turn/kernel.test.ts +++ b/src/channels/turn/kernel.test.ts @@ -1087,6 +1087,49 @@ describe("channel turn kernel", () => { expect(events).toEqual(["record", "afterRecord", "dispatch"]); }); + it("threads onTurnAdopted into assembled reply options and fires after recovery persist attempt", async () => { + const events: string[] = []; + const onTurnAdopted = vi.fn(async () => { + events.push("adopted"); + }); + const dispatchReplyWithBufferedBlockDispatcher = vi.fn( + async (params: Parameters[0]) => { + events.push("dispatch-start"); + // Persist attempt completes before adoption (agent-runner contract). + events.push("recovery-persist"); + await params.replyOptions?.onTurnAdopted?.(); + events.push("settle"); + return { + queuedFinal: true, + counts: { tool: 0, block: 0, final: 1 }, + }; + }, + ) as DispatchReplyWithBufferedBlockDispatcher; + + await dispatchAssembledChannelTurn({ + cfg, + channel: "test", + agentId: "main", + routeSessionKey: "agent:main:test:peer", + storePath: "/tmp/sessions.json", + ctxPayload: createCtx(), + recordInboundSession: createRecordInboundSession(events), + dispatchReplyWithBufferedBlockDispatcher, + delivery: { + deliver: vi.fn(async () => undefined), + }, + onTurnAdopted, + }); + + expect(onTurnAdopted).toHaveBeenCalledOnce(); + expect(events).toEqual(["record", "dispatch-start", "recovery-persist", "adopted", "settle"]); + expect(dispatchReplyWithBufferedBlockDispatcher).toHaveBeenCalledWith( + expect.objectContaining({ + replyOptions: expect.objectContaining({ onTurnAdopted }), + }), + ); + }); + it("does not run afterRecord when session recording fails", async () => { const recordError = new Error("session store failed"); const afterRecord = vi.fn(); diff --git a/src/channels/turn/kernel.ts b/src/channels/turn/kernel.ts index 5dcfd1581de3..884edc0b91e3 100644 --- a/src/channels/turn/kernel.ts +++ b/src/channels/turn/kernel.ts @@ -241,10 +241,11 @@ export const recordDroppedChannelInboundHistory = recordDroppedChannelTurnHistor function resolveAssembledReplyPipeline( params: AssembledChannelTurn, ): Pick { + const onTurnAdopted = params.onTurnAdopted ?? params.replyOptions?.onTurnAdopted; if (!params.replyPipeline) { return { dispatcherOptions: params.dispatcherOptions, - replyOptions: params.replyOptions, + replyOptions: onTurnAdopted ? { ...params.replyOptions, onTurnAdopted } : params.replyOptions, }; } const { onModelSelected, ...replyPipeline } = createChannelReplyPipeline({ @@ -262,6 +263,7 @@ function resolveAssembledReplyPipeline( replyOptions: { onModelSelected, ...params.replyOptions, + ...(onTurnAdopted ? { onTurnAdopted } : {}), }, }; } @@ -762,20 +764,27 @@ export async function runChannelTurn< const admission = resolved.admission ?? preflightAdmission ?? ({ kind: "dispatch" } as const); let result: ChannelTurnResult; try { + // Prepared runDispatch was assembled earlier and ignores late options (including onTurnAdopted). const dispatchResult = await dispatchResolvedChannelTurn( - admission.kind === "observeOnly" + "runDispatch" in resolved ? { ...resolved, - delivery: createNoopChannelEventDeliveryAdapter(), + ...(admission.kind === "observeOnly" + ? { delivery: createNoopChannelEventDeliveryAdapter() } + : {}), admission, log: params.log, messageId: input.id, } : { ...resolved, + ...(admission.kind === "observeOnly" + ? { delivery: createNoopChannelEventDeliveryAdapter() } + : {}), admission, log: params.log, messageId: input.id, + ...(params.onTurnAdopted ? { onTurnAdopted: params.onTurnAdopted } : {}), }, ); result = dispatchResult.dispatched ? { ...dispatchResult, admission } : dispatchResult; diff --git a/src/channels/turn/types.ts b/src/channels/turn/types.ts index 16fa13c9a71f..8ed69e651279 100644 --- a/src/channels/turn/types.ts +++ b/src/channels/turn/types.ts @@ -370,6 +370,11 @@ export type AssembledChannelTurn = { botLoopProtection?: ChannelBotLoopProtectionFacts; log?: (event: ChannelTurnLogEvent) => void; messageId?: string; + /** + * Observes turn adoption without waiting for settle. Threaded into + * replyOptions for the agent runner (after recovery persist attempt). + */ + onTurnAdopted?: () => void | Promise; }; /** Channel turn with dispatch runner already prepared. */ @@ -473,4 +478,10 @@ export type RunChannelTurnParams; log?: (event: ChannelTurnLogEvent) => void; + /** + * Observes turn adoption without waiting for settle. Fired after the + * recovery-context persist attempt (context may be absent when source + * delivery is suppressed). Default callers still await full settle. + */ + onTurnAdopted?: () => void | Promise; };