fix(reply): deliver queued post-start failures (#126266)

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 () => {

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 };
       }

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;

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<void>;
   onCompactionNoticePayload: (payload: ReplyPayload, execution: { runId: string }) => Promise<void>;
 }): Promise<FollowupExecutionResult> {
@@ -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 {