fix(acp): persist confirmed output from cancelled turns (#110300)

* fix(acp): persist cancelled partial replies

* fix(acp): persist delivered output for cancelled turns

Normalize terminal status at the ACP manager boundary and settle routed/direct delivery outcomes before persisting cancelled bound turns.

Co-authored-by: shaoohh <150606856+shaoohh@users.noreply.github.com>

* fix(acp): require confirmed cancelled-turn delivery

Persist cancelled-turn output only when the core dispatcher reports successful delivery. Keep canonical ACP history independent of outbound-only hook rewrites and prove backend cancellation with the real dispatcher.

Co-authored-by: shaoohh <150606856+shaoohh@users.noreply.github.com>

* test(acp): satisfy cancellation proof lint

Keep the pending-delivery race assertion behavior while avoiding a return value from the Promise executor.

Co-authored-by: shaoohh <150606856+shaoohh@users.noreply.github.com>

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

src/acp/control-plane/manager.cancel-session.test.ts

@@ -1,4 +1,5 @@
 /** Tests ACP manager cancellation of active turns and idle sessions. */
+import type { AcpRuntimeEvent } from "@openclaw/acp-core/runtime/types";
 import { describe, expect, it, vi } from "vitest";
 import {
   requireTaskByRunId,
@@ -45,6 +46,7 @@ describe("AcpSessionManager cancelSession", () => {
       });
 
       const manager = new AcpSessionManager();
+      const events: AcpRuntimeEvent[] = [];
       const runPromise = manager.runTurn({
         provenance: "system",
         cfg: baseCfg,
@@ -52,6 +54,9 @@ describe("AcpSessionManager cancelSession", () => {
         text: "long task",
         mode: "prompt",
         requestId: "run-1",
+        onEvent: (event) => {
+          events.push(event);
+        },
       });
       await vi.waitFor(
         () => {
@@ -76,6 +81,11 @@ describe("AcpSessionManager cancelSession", () => {
         childSessionKey: "agent:codex:acp:child-1",
         status: "cancelled",
       });
+      expect(events.at(-1)).toEqual({
+        type: "done",
+        status: "cancelled",
+        stopReason: "cancel",
+      });
       const states = extractStatesFromUpserts();
       expect(states).toContain("running");
       expect(states).toContain("idle");

src/acp/control-plane/manager.turn-stream.ts

@@ -24,6 +24,13 @@ function isCancellationStopReason(stopReason: string | undefined): boolean {
   return stopReason === "cancel" || stopReason === "cancelled" || stopReason === "manual-cancel";
 }
 
+/** Resolves legacy and current done events to the manager's canonical terminal status. */
+function resolveAcpTurnTerminalStatus(
+  event: Extract<AcpRuntimeEvent, { type: "done" }>,
+): "completed" | "cancelled" {
+  return event.status ?? (isCancellationStopReason(event.stopReason) ? "cancelled" : "completed");
+}
+
 async function consumeAcpTurnEvents(params: {
   events: AsyncIterable<AcpRuntimeEvent>;
   eventGate: AcpTurnEventGate;
@@ -40,10 +47,11 @@ async function consumeAcpTurnEvents(params: {
     if (!params.eventGate.open) {
       continue;
     }
+    let forwardedEvent = event;
     if (event.type === "done") {
       // Legacy runTurn adapters may omit status but retain the cancellation reason.
-      terminalStatus =
-        event.status ?? (isCancellationStopReason(event.stopReason) ? "cancelled" : "completed");
+      terminalStatus = resolveAcpTurnTerminalStatus(event);
+      forwardedEvent = { ...event, status: terminalStatus };
     } else if (event.type === "error") {
       streamError = new AcpRuntimeError(
         normalizeAcpErrorCode(event.code),
@@ -54,7 +62,7 @@ async function consumeAcpTurnEvents(params: {
       sawOutput = true;
       await params.onOutputEvent?.(event);
     }
-    await params.onEvent?.(event);
+    await params.onEvent?.(forwardedEvent);
   }
 
   if (params.eventGate.open && streamError) {

src/auto-reply/reply/dispatch-acp-delivery.test.ts

@@ -321,12 +321,99 @@ describe("createAcpDispatchDeliveryCoordinator", () => {
     expect(delivered).toEqual([{ text: "hello" }]);
     expect(deliverySettled).toBe(true);
 
+    let transcriptSettled = false;
+    const transcriptPromise = coordinator
+      .resolveAccumulatedDeliveredTranscriptText()
+      .then((text) => {
+        transcriptSettled = true;
+        return text;
+      });
+    await Promise.resolve();
+    expect(transcriptSettled).toBe(false);
+
     releaseDelivery?.();
     await expect(deliveryPromise).resolves.toBe(true);
+    await expect(transcriptPromise).resolves.toBe("hello");
     expect(deliverySettled).toBe(true);
     await dispatcher.waitForIdle();
   });
 
+  it("excludes direct output cancelled by a core before-delivery hook", async () => {
+    const dispatcher = createReplyDispatcher({ deliver: vi.fn(async () => {}) });
+    dispatcher.appendBeforeDeliver?.(() => null);
+    const coordinator = createAcpDispatchDeliveryCoordinator({
+      cfg: createAcpTestConfig(),
+      ctx: buildTestCtx({
+        Provider: "visiblechat",
+        Surface: "visiblechat",
+        SessionKey: "agent:codex-acp:session-1",
+      }),
+      dispatcher,
+      inboundAudio: false,
+      shouldRouteToOriginating: false,
+    });
+
+    await expect(
+      coordinator.deliver("block", { text: "cancelled output" }, { skipTts: true }),
+    ).resolves.toBe(true);
+    await dispatcher.waitForIdle();
+
+    await expect(coordinator.resolveAccumulatedDeliveredTranscriptText()).resolves.toBe("");
+  });
+
+  it("keeps canonical ACP text after an outbound hook rewrites the payload", async () => {
+    const delivered: unknown[] = [];
+    const dispatcher = createReplyDispatcher({
+      deliver: async (payload) => {
+        delivered.push(payload);
+      },
+    });
+    dispatcher.appendBeforeDeliver?.((payload) => ({ ...payload, text: "transport rewrite" }));
+    const coordinator = createAcpDispatchDeliveryCoordinator({
+      cfg: createAcpTestConfig(),
+      ctx: buildTestCtx({
+        Provider: "visiblechat",
+        Surface: "visiblechat",
+        SessionKey: "agent:codex-acp:session-1",
+      }),
+      dispatcher,
+      inboundAudio: false,
+      shouldRouteToOriginating: false,
+    });
+
+    await expect(
+      coordinator.deliver("block", { text: "canonical runtime text" }, { skipTts: true }),
+    ).resolves.toBe(true);
+    await dispatcher.waitForIdle();
+
+    expect(delivered).toEqual([{ text: "transport rewrite" }]);
+    await expect(coordinator.resolveAccumulatedDeliveredTranscriptText()).resolves.toBe(
+      "canonical runtime text",
+    );
+  });
+
+  it("does not treat custom dispatcher enqueue acceptance as confirmed delivery", async () => {
+    const coordinator = createCoordinator();
+
+    await expect(
+      coordinator.deliver("block", { text: "unconfirmed" }, { skipTts: true }),
+    ).resolves.toBe(true);
+
+    await expect(coordinator.resolveAccumulatedDeliveredTranscriptText()).resolves.toBe("");
+  });
+
+  it("excludes status notices from delivered transcript text", async () => {
+    const coordinator = createCoordinator();
+
+    await coordinator.deliver(
+      "block",
+      { text: "runtime status", isStatusNotice: true },
+      { skipTts: true },
+    );
+
+    await expect(coordinator.resolveAccumulatedDeliveredTranscriptText()).resolves.toBe("");
+  });
+
   it("waits for pending direct block delivery before resolving tool delivery", async () => {
     const delivered: unknown[] = [];
     let releaseDelivery: (() => void) | undefined;
@@ -946,6 +1033,41 @@ describe("createAcpDispatchDeliveryCoordinator", () => {
     expect(coordinator.hasDeliveredVisibleText()).toBe(true);
     expect(coordinator.hasFailedVisibleTextDelivery()).toBe(false);
     expect(coordinator.getRoutedCounts().block).toBe(1);
+    await expect(coordinator.resolveAccumulatedDeliveredTranscriptText()).resolves.toBe("hello");
+  });
+
+  it("passes caller cancellation through routed ACP delivery", async () => {
+    const controller = new AbortController();
+    controller.abort();
+    deliveryMocks.routeReply.mockImplementationOnce(async (paramsUnknown: unknown) => {
+      const params = paramsUnknown as { abortSignal?: AbortSignal };
+      return params.abortSignal?.aborted
+        ? { ok: false, error: "Reply routing aborted" }
+        : { ok: true, messageId: "unexpected" };
+    });
+    const coordinator = createAcpDispatchDeliveryCoordinator({
+      cfg: createAcpTestConfig(),
+      ctx: buildTestCtx({
+        Provider: "visiblechat",
+        Surface: "visiblechat",
+        SessionKey: "agent:codex-acp:session-1",
+      }),
+      dispatcher: createDispatcher(),
+      inboundAudio: false,
+      shouldRouteToOriginating: true,
+      originatingChannel: "visiblechat",
+      originatingTo: "channel:thread-1",
+      abortSignal: controller.signal,
+    });
+
+    await expect(coordinator.deliver("final", { text: "late" })).resolves.toBe(false);
+
+    const [routeParams] = expectDefined(
+      (deliveryMocks.routeReply.mock.calls as unknown as Array<[{ abortSignal?: AbortSignal }]>)[0],
+      "route call",
+    );
+    expect(routeParams.abortSignal).toBe(controller.signal);
+    await expect(coordinator.resolveAccumulatedDeliveredTranscriptText()).resolves.toBe("");
   });
 
   it("treats hook-suppressed routed ACP block text as handled", async () => {
@@ -974,5 +1096,6 @@ describe("createAcpDispatchDeliveryCoordinator", () => {
     expect(coordinator.hasDeliveredVisibleText()).toBe(true);
     expect(coordinator.hasFailedVisibleTextDelivery()).toBe(false);
     expect(coordinator.getRoutedCounts().block).toBe(0);
+    await expect(coordinator.resolveAccumulatedDeliveredTranscriptText()).resolves.toBe("");
   });
 });

src/auto-reply/reply/dispatch-acp-delivery.ts

@@ -16,7 +16,10 @@ import { resolveConfiguredTtsMode, shouldCleanTtsDirectiveText } from "../../tts
 import { isReplyPayloadStatusNotice } from "../reply-payload.js";
 import type { FinalizedMsgContext } from "../templating.js";
 import type { ReplyPayload } from "../types.js";
-import { waitForReplyDispatcherIdle } from "./reply-dispatcher.js";
+import {
+  captureReplyDispatchDeliveryOutcome,
+  waitForReplyDispatcherIdle,
+} from "./reply-dispatcher.js";
 import type { ReplyDispatchKind, ReplyDispatcher } from "./reply-dispatcher.types.js";
 import { readDispatcherFailedCounts } from "./reply-dispatcher.types.js";
 import {
@@ -152,9 +155,12 @@ async function maybeApplyAcpTts(params: {
 type AcpDispatchDeliveryState = {
   startedReplyLifecycle: boolean;
   accumulatedBlockText: string;
+  accumulatedDeliveredBlockText: string;
   accumulatedVisibleBlockText: string;
   accumulatedBlockTtsText: string;
   accumulatedFinalText: string;
+  accumulatedDeliveredFinalText: string;
+  pendingTranscriptOutcomes: Promise<void>[];
   cleanBlockTtsDirectiveText?: ReturnType<typeof createTtsDirectiveTextStreamCleaner>;
   blockCount: number;
   deliveredFinalReply: boolean;
@@ -178,6 +184,8 @@ export type AcpDispatchDeliveryCoordinator = {
   getAccumulatedVisibleBlockText: () => string;
   getAccumulatedBlockTtsText: () => string;
   getAccumulatedFinalText: () => string;
+  getAccumulatedTranscriptText: () => string;
+  resolveAccumulatedDeliveredTranscriptText: () => Promise<string>;
   settleVisibleText: () => Promise<void>;
   hasDeliveredFinalReply: () => boolean;
   hasDeliveredVisibleText: () => boolean;
@@ -232,9 +240,12 @@ export function createAcpDispatchDeliveryCoordinator(params: {
   const state: AcpDispatchDeliveryState = {
     startedReplyLifecycle: false,
     accumulatedBlockText: "",
+    accumulatedDeliveredBlockText: "",
     accumulatedVisibleBlockText: "",
     accumulatedBlockTtsText: "",
     accumulatedFinalText: "",
+    accumulatedDeliveredFinalText: "",
+    pendingTranscriptOutcomes: [],
     cleanBlockTtsDirectiveText: shouldCleanTtsDirectiveText({
       cfg: params.cfg,
       ttsAuto: params.sessionTtsAuto,
@@ -258,6 +269,25 @@ export function createAcpDispatchDeliveryCoordinator(params: {
     toolMessageByCallId: new Map(),
   };
   let hasPendingDirectBlockReplyDelivery = false;
+
+  const appendDeliveredTranscriptText = (
+    kind: ReplyDispatchKind,
+    blockText: string | undefined,
+    finalText: string | undefined,
+  ) => {
+    // ACP history keeps canonical runtime text, while delivery hooks may render
+    // transport-specific text. Only the delivery outcome gates this snapshot.
+    if (kind === "block" && blockText) {
+      state.accumulatedDeliveredBlockText = state.accumulatedDeliveredBlockText
+        ? `${state.accumulatedDeliveredBlockText}\n${blockText}`
+        : blockText;
+    }
+    if (kind === "final" && finalText) {
+      state.accumulatedDeliveredFinalText = state.accumulatedDeliveredFinalText
+        ? `${state.accumulatedDeliveredFinalText}\n${finalText}`
+        : finalText;
+    }
+  };
   const waitForPendingDirectBlockReplyDelivery = async () => {
     if (!hasPendingDirectBlockReplyDelivery) {
       return;
@@ -350,12 +380,14 @@ export function createAcpDispatchDeliveryCoordinator(params: {
     meta?: AcpDispatchDeliveryMeta,
   ): Promise<boolean> => {
     let visiblePayload = payload;
-    const rawBlockText = kind === "block" ? normalizeOptionalString(payload.text) : undefined;
-    if (rawBlockText) {
-      const isStatusNotice = isReplyPayloadStatusNotice(payload);
+    const isStatusNotice = isReplyPayloadStatusNotice(payload);
+    const rawBlockPayloadText =
+      kind === "block" ? normalizeOptionalString(payload.text) : undefined;
+    const rawBlockText = isStatusNotice ? undefined : rawBlockPayloadText;
+    if (rawBlockPayloadText) {
       const joinsBufferedTtsDirective =
         state.cleanBlockTtsDirectiveText?.hasBufferedDirectiveText() === true;
-      if (!isStatusNotice) {
+      if (rawBlockText) {
         if (state.accumulatedBlockText.length > 0) {
           state.accumulatedBlockText += "\n";
         }
@@ -367,8 +399,8 @@ export function createAcpDispatchDeliveryCoordinator(params: {
         state.blockCount += 1;
       }
 
-      if (state.cleanBlockTtsDirectiveText && !isStatusNotice) {
-        const text = state.cleanBlockTtsDirectiveText.push(rawBlockText);
+      if (state.cleanBlockTtsDirectiveText && rawBlockText) {
+        const text = state.cleanBlockTtsDirectiveText.push(rawBlockPayloadText);
         visiblePayload = { ...payload, text: text.trim() ? text : undefined };
       }
       if (visiblePayload.text) {
@@ -378,7 +410,6 @@ export function createAcpDispatchDeliveryCoordinator(params: {
         state.accumulatedVisibleBlockText += visiblePayload.text;
       }
     }
-    const isStatusNotice = isReplyPayloadStatusNotice(payload);
     const rawFinalText =
       kind === "final" && !isStatusNotice ? normalizeOptionalString(payload.text) : undefined;
     if (rawFinalText) {
@@ -448,6 +479,7 @@ export function createAcpDispatchDeliveryCoordinator(params: {
         threadId,
         replyDelivery: routedReplyDelivery,
         cfg: params.cfg,
+        abortSignal: params.abortSignal,
         mirror: false,
         replyKind: kind,
         runId: params.runId,
@@ -479,6 +511,7 @@ export function createAcpDispatchDeliveryCoordinator(params: {
           messageId: result.messageId,
         });
       }
+      appendDeliveredTranscriptText(kind, rawBlockText, rawFinalText);
       if (kind === "final") {
         state.deliveredFinalReply = true;
       }
@@ -499,12 +532,25 @@ export function createAcpDispatchDeliveryCoordinator(params: {
       text: ttsPayload.text,
       routed: false,
     });
+    const transcriptOutcome =
+      rawBlockText || rawFinalText ? captureReplyDispatchDeliveryOutcome(ttsPayload) : undefined;
     const delivered =
       kind === "tool"
         ? params.dispatcher.sendToolResult(ttsPayload)
         : kind === "block"
           ? params.dispatcher.sendBlockReply(ttsPayload)
           : params.dispatcher.sendFinalReply(ttsPayload);
+    if (delivered && transcriptOutcome) {
+      if (transcriptOutcome.isTracked()) {
+        state.pendingTranscriptOutcomes.push(
+          transcriptOutcome.promise.then((outcome) => {
+            if (outcome === "delivered") {
+              appendDeliveredTranscriptText(kind, rawBlockText, rawFinalText);
+            }
+          }),
+        );
+      }
+    }
     if (kind === "final" && delivered) {
       state.deliveredFinalReply = true;
     }
@@ -528,6 +574,11 @@ export function createAcpDispatchDeliveryCoordinator(params: {
     getAccumulatedVisibleBlockText: () => state.accumulatedVisibleBlockText,
     getAccumulatedBlockTtsText: () => state.accumulatedBlockTtsText,
     getAccumulatedFinalText: () => state.accumulatedFinalText,
+    getAccumulatedTranscriptText: () => state.accumulatedFinalText || state.accumulatedBlockText,
+    resolveAccumulatedDeliveredTranscriptText: async () => {
+      await Promise.all(state.pendingTranscriptOutcomes.splice(0));
+      return state.accumulatedDeliveredFinalText || state.accumulatedDeliveredBlockText;
+    },
     settleVisibleText: settleDirectVisibleText,
     hasDeliveredFinalReply: () => state.deliveredFinalReply,
     hasDeliveredVisibleText: () => state.deliveredVisibleText,

src/auto-reply/reply/dispatch-acp.test.ts

@@ -15,10 +15,12 @@ import {
   resolveInlineAgentImageAttachments,
 } from "./agent-turn-attachments.js";
 import { tryDispatchAcpReply } from "./dispatch-acp.js";
+import { createAbortAwareDispatcher } from "./dispatch-from-config.abort.js";
 import {
   appendRecentHistoryImageContext,
   resolveRecentInboundHistoryImages,
 } from "./history-media.js";
+import { createReplyDispatcher } from "./reply-dispatcher.js";
 import type { ReplyDispatcher } from "./reply-dispatcher.types.js";
 import { buildTestCtx } from "./test-ctx.js";
 import { createAcpSessionMeta, createAcpTestConfig } from "./test-fixtures/acp-runtime.js";
@@ -338,6 +340,11 @@ async function runDispatch(params: {
   suppressReplyLifecycle?: boolean;
   sourceReplyDeliveryMode?: "automatic" | "message_tool_only";
   toolsAllow?: string[];
+  recordProcessed?: (
+    outcome: "completed" | "skipped" | "error",
+    opts?: { reason?: string; error?: string },
+  ) => void;
+  markIdle?: (reason: string) => void;
 }) {
   const targetSessionKey = params.sessionKeyOverride ?? sessionKey;
   return tryDispatchAcpReply({
@@ -369,8 +376,8 @@ async function runDispatch(params: {
     bypassForCommand: false,
     toolsAllow: params.toolsAllow,
     ...(params.onReplyStart ? { onReplyStart: params.onReplyStart } : {}),
-    recordProcessed: vi.fn(),
-    markIdle: vi.fn(),
+    recordProcessed: params.recordProcessed ?? vi.fn(),
+    markIdle: params.markIdle ?? vi.fn(),
   });
 }
 
@@ -821,31 +828,203 @@ describe("tryDispatchAcpReply", () => {
     expect(auditMocks.emitAcpLifecycleError).not.toHaveBeenCalled();
   });
 
-  it("records cancellation only after ACP output flushing", async () => {
+  it("persists delivered ACP output for backend cancellation without a caller abort", async () => {
     setReadyAcpResolution();
-    const abortController = new AbortController();
+    const deliveredPayloads: unknown[] = [];
+    const dispatcher = createReplyDispatcher({
+      deliver: async (payload) => {
+        deliveredPayloads.push(payload);
+      },
+    });
+    const recordProcessed = vi.fn();
+    const markIdle = vi.fn();
     managerMocks.runTurn.mockImplementationOnce(
       async ({ onEvent }: { onEvent: (event: unknown) => Promise<void> }) => {
         await onEvent({ type: "text_delta", text: "partial", tag: "agent_message_chunk" });
-        await onEvent({ type: "done", status: "cancelled" });
-        abortController.abort();
+        await onEvent({ type: "done", status: "cancelled", stopReason: "cancelled" });
       },
     );
 
-    await runDispatch({
+    const result = await runDispatch({
       bodyForAgent: "cancel this turn",
-      abortSignal: abortController.signal,
+      dispatcher,
+      recordProcessed,
+      markIdle,
     });
 
+    expect(result?.queuedFinal).toBe(true);
+    expect(deliveredPayloads).toEqual([{ text: "partial" }]);
+    expect(transcriptMocks.persistAcpDispatchTranscript).toHaveBeenCalledTimes(1);
+    const transcript = requireRecord(
+      mockArg(transcriptMocks.persistAcpDispatchTranscript, 0, 0, "transcript call"),
+      "transcript call",
+    );
+    expect(transcript.sessionKey).toBe(sessionKey);
+    expect(transcript.promptText).toBe("cancel this turn");
+    expect(transcript.finalText).toBe("partial");
+    expect(recordProcessed).toHaveBeenCalledWith("completed", { reason: "acp_aborted" });
+    expect(markIdle).toHaveBeenCalledWith("message_aborted");
+    expect(transcriptMocks.persistAcpDispatchTranscript.mock.invocationCallOrder[0]).toBeLessThan(
+      recordProcessed.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY,
+    );
     expect(auditMocks.emitAcpLifecycleEnd).toHaveBeenCalledWith(
       expect.objectContaining({
-        abortSignal: abortController.signal,
         resultStatus: "cancelled",
       }),
     );
     expect(auditMocks.emitAcpLifecycleError).not.toHaveBeenCalled();
   });
 
+  it("does not persist final-only output rejected after caller cancellation", async () => {
+    setReadyAcpResolution();
+    const abortController = new AbortController();
+    const base = createDispatcher();
+    const dispatcher = createAbortAwareDispatcher({
+      dispatcher: base.dispatcher,
+      isAborted: () => abortController.signal.aborted,
+    });
+    managerMocks.runTurn.mockImplementationOnce(
+      async ({ onEvent }: { onEvent: (event: unknown) => Promise<void> }) => {
+        await onEvent({ type: "text_delta", text: "not delivered", tag: "agent_message_chunk" });
+        abortController.abort();
+        await onEvent({ type: "done", status: "cancelled" });
+      },
+    );
+
+    const result = await runDispatch({
+      bodyForAgent: "cancel before delivery",
+      abortSignal: abortController.signal,
+      dispatcher,
+    });
+
+    expect(result?.queuedFinal).toBe(false);
+    expect(base.dispatcher.sendFinalReply).not.toHaveBeenCalled();
+    const transcript = requireRecord(
+      mockArg(transcriptMocks.persistAcpDispatchTranscript, 0, 0, "transcript call"),
+      "transcript call",
+    );
+    expect(transcript.promptText).toBe("cancel before delivery");
+    expect(transcript.finalText).toBe("");
+  });
+
+  it("persists live ACP output delivered before caller cancellation", async () => {
+    setReadyAcpResolution();
+    const abortController = new AbortController();
+    const deliveredPayloads: Array<Record<string, unknown>> = [];
+    let markDeliveryStarted!: () => void;
+    let releaseDelivery!: () => void;
+    const deliveryStarted = new Promise<void>((resolve) => {
+      markDeliveryStarted = resolve;
+    });
+    const deliveryGate = new Promise<void>((resolve) => {
+      releaseDelivery = resolve;
+    });
+    const coreDispatcher = createReplyDispatcher({
+      deliver: async (payload) => {
+        deliveredPayloads.push(requireRecord(payload, "delivered payload"));
+        markDeliveryStarted();
+        await deliveryGate;
+      },
+    });
+    const dispatcher = createAbortAwareDispatcher({
+      dispatcher: coreDispatcher,
+      isAborted: () => abortController.signal.aborted,
+    });
+    const partial = "Visible before cancellation. ".repeat(4);
+    let markTurnReady!: () => void;
+    let finishTurn!: () => void;
+    let markTurnDone!: () => void;
+    const turnReady = new Promise<void>((resolve) => {
+      markTurnReady = resolve;
+    });
+    const finishTurnGate = new Promise<void>((resolve) => {
+      finishTurn = resolve;
+    });
+    const turnDone = new Promise<void>((resolve) => {
+      markTurnDone = resolve;
+    });
+    managerMocks.runTurn.mockImplementationOnce(
+      async ({ onEvent }: { onEvent: (event: unknown) => Promise<void> }) => {
+        await onEvent({ type: "text_delta", text: partial, tag: "agent_message_chunk" });
+        markTurnReady();
+        await finishTurnGate;
+        await onEvent({ type: "done", status: "cancelled" });
+        markTurnDone();
+      },
+    );
+
+    const dispatchPromise = runDispatch({
+      bodyForAgent: "cancel after delivery",
+      abortSignal: abortController.signal,
+      cfg: createAcpTestConfig({
+        acp: {
+          enabled: true,
+          stream: { deliveryMode: "live", coalesceIdleMs: 0, maxChunkChars: 64 },
+        },
+      }),
+      dispatcher,
+    });
+
+    await turnReady;
+    await deliveryStarted;
+    abortController.abort();
+    finishTurn();
+    await turnDone;
+    const earlyOutcome = await Promise.race([
+      dispatchPromise.then(() => "settled" as const),
+      new Promise<"pending">((resolve) => {
+        setTimeout(() => resolve("pending"), 10);
+      }),
+    ]);
+    expect(earlyOutcome).toBe("pending");
+    expect(transcriptMocks.persistAcpDispatchTranscript).not.toHaveBeenCalled();
+
+    releaseDelivery();
+    await dispatchPromise;
+
+    const deliveredText = deliveredPayloads.map((payload) => String(payload.text)).join("\n");
+    expect(deliveredText).not.toBe("");
+    expect(partial).toContain(deliveredText.replaceAll("\n", ""));
+    const transcript = requireRecord(
+      mockArg(transcriptMocks.persistAcpDispatchTranscript, 0, 0, "transcript call"),
+      "transcript call",
+    );
+    expect(transcript.finalText).toBe(deliveredText);
+  });
+
+  it("keeps caller abort authoritative until completed output settles", async () => {
+    setReadyAcpResolution();
+    const abortController = new AbortController();
+    const { dispatcher } = createDispatcher();
+    const recordProcessed = vi.fn();
+    const markIdle = vi.fn();
+    managerMocks.runTurn.mockImplementationOnce(
+      async ({ onEvent }: { onEvent: (event: unknown) => Promise<void> }) => {
+        await onEvent({ type: "text_delta", text: "complete", tag: "agent_message_chunk" });
+        await onEvent({ type: "done", status: "completed" });
+        abortController.abort();
+      },
+    );
+
+    const result = await runDispatch({
+      bodyForAgent: "finish first",
+      abortSignal: abortController.signal,
+      dispatcher,
+      recordProcessed,
+      markIdle,
+    });
+
+    expect(result?.queuedFinal).toBe(true);
+    expect(recordProcessed).toHaveBeenCalledWith("completed", { reason: "acp_aborted" });
+    expect(markIdle).toHaveBeenCalledWith("message_aborted");
+    expect(auditMocks.emitAcpLifecycleEnd).toHaveBeenCalledWith(
+      expect.objectContaining({
+        abortSignal: abortController.signal,
+        resultStatus: "completed",
+      }),
+    );
+  });
+
   it("records an ACP error when output finalization fails", async () => {
     setReadyAcpResolution();
     mockVisibleTextTurn("visible output");

src/auto-reply/reply/dispatch-acp.ts

@@ -548,6 +548,7 @@ export async function tryDispatchAcpReply(params: {
   let auditTerminalOutcome: "blocked" | undefined;
   let auditStopReason: string | undefined;
   let auditResultStatus: "completed" | "cancelled" | undefined;
+  let runtimeTurnWasCancelled = false;
   const emitAuditStart = () => {
     if (auditStarted) {
       return;
@@ -771,13 +772,18 @@ export async function tryDispatchAcpReply(params: {
         if (event.type === "done") {
           auditStopReason = event.stopReason;
           auditResultStatus = event.status;
+          runtimeTurnWasCancelled = event.status === "cancelled";
         }
         await projector.onEvent(event);
       },
     });
 
     await projector.flush(true);
-    if (params.abortSignal?.aborted) {
+    if (runtimeTurnWasCancelled || params.abortSignal?.aborted) {
+      // A cancelled runtime can return normally after the projector has already
+      // delivered partial output. Keep the bound transcript aligned with it.
+      await persistTranscript(await delivery.resolveAccumulatedDeliveredTranscriptText());
+      queuedFinal = delivery.hasDeliveredFinalReply() || queuedFinal;
       const counts = params.dispatcher.getQueuedCounts();
       delivery.applyRoutedCounts(counts);
       params.recordProcessed("completed", { reason: "acp_aborted" });
@@ -800,9 +806,7 @@ export async function tryDispatchAcpReply(params: {
 
     // Persist once the turn's outcome is settled. Writing before finalization
     // would leave a finalizer failure recorded as a clean success.
-    await persistTranscript(
-      delivery.getAccumulatedFinalText() || delivery.getAccumulatedBlockText(),
-    );
+    await persistTranscript(delivery.getAccumulatedTranscriptText());
 
     const result = finishAttempt({
       queuedFinal,
@@ -825,7 +829,7 @@ export async function tryDispatchAcpReply(params: {
     const errorText = formatAcpRuntimeErrorText(acpError);
     // Snapshot streamed output before delivering the error: delivery accumulates
     // what it sends, so reading after would fold the error text in twice.
-    const partialText = delivery.getAccumulatedFinalText() || delivery.getAccumulatedBlockText();
+    const partialText = delivery.getAccumulatedTranscriptText();
     const delivered = await delivery.deliver("final", {
       text: errorText,
       isError: true,