From 7a82d8b0f25fdfafbdcd2fbe40e9e8d68b981fc3 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 19 Aug 2026 05:20:33 -0700 Subject: [PATCH] fix(reply): deliver queued post-start failures (#126266) --- src/auto-reply/reply/followup-runner.test.ts | 17 ++++--- src/auto-reply/reply/followup-runner.ts | 16 +++---- .../reply/followup-turn-execution.test.ts | 44 +++++++++++++++++-- .../reply/followup-turn-execution.ts | 21 +++++++-- 4 files changed, 76 insertions(+), 22 deletions(-) diff --git a/src/auto-reply/reply/followup-runner.test.ts b/src/auto-reply/reply/followup-runner.test.ts index 6711f874bf2c..ef4ef643c6e5 100644 --- a/src/auto-reply/reply/followup-runner.test.ts +++ b/src/auto-reply/reply/followup-runner.test.ts @@ -222,23 +222,30 @@ describe("createFollowupRunner", () => { expect(turn.operation.fail).not.toHaveBeenCalled(); }); - it("consumes a turn that fails after canonical execution starts", async () => { + it("does not replay a returned execution when terminal delivery fails", async () => { const typing = createTypingController(); const turn = createTurn(); + const execution = createRejectedExecution(); + const failure = new Error("terminal delivery failed"); state.admit.mockResolvedValue({ kind: "admitted", turn }); - state.execute.mockImplementation(async ({ onExecutionStarted }) => { - onExecutionStarted?.(); - throw new Error("execution failed after start"); + state.execute.mockResolvedValue(execution); + state.account.mockResolvedValue(undefined); + state.resolveDecision.mockReturnValue({ + kind: "deliver", + payloads: [{ text: "terminal failure", isError: true }], }); + state.deliver.mockRejectedValue(failure); await createFollowupRunner({ typing, typingMode: "instant", defaultModel: "claude" })( turn.queued, ); expect(state.execute).toHaveBeenCalledOnce(); + expect(state.account).toHaveBeenCalledOnce(); + expect(state.deliver).toHaveBeenCalledOnce(); expect(state.completeLifecycle).toHaveBeenCalledWith(turn.queued); expect(state.clearRunContext).toHaveBeenCalledWith("run-1"); - expect(turn.operation.fail).toHaveBeenCalledWith("run_failed", expect.any(Error)); + expect(turn.operation.fail).toHaveBeenCalledWith("run_failed", failure); }); it("holds the reply operation through progress drain, accounting, and delivery", async () => { diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index 130db0c6a2ea..4f729a3ffb93 100644 --- a/src/auto-reply/reply/followup-runner.ts +++ b/src/auto-reply/reply/followup-runner.ts @@ -32,7 +32,6 @@ export function createFollowupRunner( let disposition: FollowupDrainDisposition = { kind: "retry", error: undefined }; let operation: ReplyOperation | undefined; let admittedRunId: string | undefined; - let executionStarted = false; let queuedFollowupAdmitted = false; const initiallyAborted = queued.abortSignal?.aborted === true || queued.queueAbortSignal?.aborted === true; @@ -79,9 +78,6 @@ export function createFollowupRunner( const execution = await executeFollowupTurn({ turn, defaults, - onExecutionStarted: () => { - executionStarted = true; - }, onToolResult: async (payload, identity) => { await deliverFollowupDecision({ decision: { kind: "deliver", payloads: [payload] }, @@ -103,6 +99,9 @@ export function createFollowupRunner( }); }, }); + // A closed execution result is terminal queue work. Commit consumption + // before accounting/delivery so their failures cannot replay model or tool effects. + disposition = { kind: "consumed" }; try { await execution.progress.drain(); } catch (error) { @@ -136,7 +135,6 @@ export function createFollowupRunner( runId: execution.execution.runId, runFollowup, }); - disposition = { kind: "consumed" }; } catch (error) { if (error instanceof FollowupRunDeferredError) { disposition = { kind: "deferred", reason: error.message }; @@ -145,15 +143,11 @@ export function createFollowupRunner( operation.result.code === "aborted_by_user" ) { disposition = { kind: "consumed" }; - } else if (executionStarted) { - // There is no durable post-execution resume record yet. Requeueing the prompt - // here can duplicate persisted turns and external tool side effects. Preserve - // the terminal failure through complete() rather than reporting success. + } else if (disposition.kind === "consumed") { defaultRuntime.error?.( - `followup queue: execution failed after start; refusing replay: ${formatErrorMessage(error)}`, + `followup queue: terminal handling failed after execution; refusing replay: ${formatErrorMessage(error)}`, ); operation?.fail("run_failed", error); - disposition = { kind: "consumed" }; } else { disposition = { kind: "retry", error }; } diff --git a/src/auto-reply/reply/followup-turn-execution.test.ts b/src/auto-reply/reply/followup-turn-execution.test.ts index e5bd4e046762..5b3c0b2410b9 100644 --- a/src/auto-reply/reply/followup-turn-execution.test.ts +++ b/src/auto-reply/reply/followup-turn-execution.test.ts @@ -2,6 +2,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { ReplyPayload } from "../types.js"; import type { AgentTurnParams } from "./agent-runner-execution.types.js"; import type { AdmittedFollowupTurn } from "./followup-turn-admission.js"; +import { markReplyOperationExecutionStarted } from "./reply-run-registry.state.js"; const state = vi.hoisted(() => ({ execute: vi.fn(), @@ -94,7 +95,6 @@ describe("executeFollowupTurn", () => { it("normalizes queued route facts into the canonical execution call", async () => { const turn = createTurn(); const typing = createTypingController(); - const onExecutionStarted = vi.fn(); const onAgentRunStart = vi.fn(); state.execute.mockImplementation(async (params: AgentTurnParams) => { params.opts?.onAgentRunStart?.("run-1"); @@ -109,7 +109,6 @@ describe("executeFollowupTurn", () => { defaultModel: "claude", opts: { onAgentRunStart }, }, - onExecutionStarted, onToolResult: vi.fn(async () => {}), onCompactionNoticePayload: vi.fn(async () => {}), }); @@ -135,7 +134,6 @@ describe("executeFollowupTurn", () => { SenderId: "user-1", }); expect(call.sessionCtx.media).toEqual([{ kind: "audio", contentType: "audio/ogg" }]); - expect(onExecutionStarted).toHaveBeenCalledOnce(); expect(onAgentRunStart).toHaveBeenCalledWith("run-1", undefined); }); @@ -987,6 +985,46 @@ describe("executeFollowupTurn", () => { expect(order).toEqual(["progress"]); }); + it("normalizes a post-start execution failure after draining detached progress", async () => { + const failure = new Error("execution failed after start"); + const onItemEvent = vi.fn(async () => {}); + const fail = vi.fn(); + const operation = { + abortSignal: new AbortController().signal, + fail, + } as unknown as AdmittedFollowupTurn["operation"]; + const turn = createTurn({ operation }); + turn.queued.originatingChatType = "direct"; + state.execute.mockImplementation(async (params: AgentTurnParams) => { + void params.opts?.onItemEvent?.({ progressText: "working" }); + markReplyOperationExecutionStarted(operation); + throw failure; + }); + const pending = executeFollowupTurn({ + turn, + defaults: { + typing: createTypingController(), + typingMode: "never", + defaultModel: "claude", + opts: { onItemEvent }, + }, + onToolResult: vi.fn(async () => {}), + onCompactionNoticePayload: vi.fn(async () => {}), + }); + + await expect(pending).resolves.toMatchObject({ + execution: { + runId: "run-1", + outcome: { + kind: "rejected", + payload: { isError: true }, + }, + }, + }); + expect(onItemEvent).toHaveBeenCalledOnce(); + expect(fail).toHaveBeenCalledWith("run_failed", failure); + }); + it("waits for every pending task before propagating a drain failure", async () => { const failure = new Error("tool task failed"); let releaseSlowTask!: () => void; diff --git a/src/auto-reply/reply/followup-turn-execution.ts b/src/auto-reply/reply/followup-turn-execution.ts index 7bb1f423a073..6734e3e55dfc 100644 --- a/src/auto-reply/reply/followup-turn-execution.ts +++ b/src/auto-reply/reply/followup-turn-execution.ts @@ -6,11 +6,13 @@ import type { VerboseLevel } from "../thinking.js"; import type { ReplyPayload } from "../types.js"; import { executeAgentTurn } from "./agent-runner-execution.js"; import type { AgentTurnExecutionResult } from "./agent-runner-execution.types.js"; +import { buildTerminalAgentRunFailureReplyPayload } from "./agent-runner-failure-reply.js"; import { resetReplyRunSession } from "./agent-runner-session-reset.js"; import { resolveTurnCommentaryProgressOwner } from "./commentary-progress-owner.js"; import { requiresDurableToolResultDelivery } from "./dispatch-from-config.payloads.js"; import type { AdmittedFollowupTurn, FollowupRunnerParams } from "./followup-turn-admission.js"; import type { InternalGetReplyOptions } from "./get-reply.types.js"; +import { hasReplyOperationExecutionStarted } from "./reply-run-registry.js"; import { createTypingSignaler, type TypingSignaler } from "./typing-mode.js"; export type FollowupExecutionResult = { @@ -67,7 +69,6 @@ function buildFollowupTemplateContext(turn: AdmittedFollowupTurn): TemplateConte export async function executeFollowupTurn(params: { turn: AdmittedFollowupTurn; defaults: FollowupRunnerParams; - onExecutionStarted?: () => void; onToolResult: (payload: ReplyPayload, execution: { runId: string }) => Promise; onCompactionNoticePayload: (payload: ReplyPayload, execution: { runId: string }) => Promise; }): Promise { @@ -196,7 +197,6 @@ export async function executeFollowupTurn(params: { commentaryPayloadsEnabled, runId: turn.runId, onAgentRunStart: (runId, executionIdentityToken) => { - params.onExecutionStarted?.(); sourceOpts?.onAgentRunStart?.(runId, executionIdentityToken); }, onBlockReply: undefined, @@ -418,7 +418,22 @@ export async function executeFollowupTurn(params: { ...pendingToolTaskWatchers, ]); } - throw error; + if (!hasReplyOperationExecutionStarted(turn.operation)) { + throw error; + } + turn.operation.fail("run_failed", error); + execution = { + runId: turn.runId, + outcome: { + kind: "rejected", + payload: buildTerminalAgentRunFailureReplyPayload({ + isHeartbeat: sourceOpts?.isHeartbeat, + visibleReplyDelivered: false, + sessionCtx, + cfg: turn.config, + }), + }, + }; } } return {