diff --git a/src/acp/control-plane/manager.cancel-session.test.ts b/src/acp/control-plane/manager.cancel-session.test.ts index 447d76357af5..0951be999537 100644 --- a/src/acp/control-plane/manager.cancel-session.test.ts +++ b/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"); diff --git a/src/acp/control-plane/manager.turn-stream.ts b/src/acp/control-plane/manager.turn-stream.ts index df94ba69f948..abc1c04fc1d2 100644 --- a/src/acp/control-plane/manager.turn-stream.ts +++ b/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, +): "completed" | "cancelled" { + return event.status ?? (isCancellationStopReason(event.stopReason) ? "cancelled" : "completed"); +} + async function consumeAcpTurnEvents(params: { events: AsyncIterable; 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) { diff --git a/src/auto-reply/reply/dispatch-acp-delivery.test.ts b/src/auto-reply/reply/dispatch-acp-delivery.test.ts index 1456775fdb46..7256aa918c46 100644 --- a/src/auto-reply/reply/dispatch-acp-delivery.test.ts +++ b/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(""); }); }); diff --git a/src/auto-reply/reply/dispatch-acp-delivery.ts b/src/auto-reply/reply/dispatch-acp-delivery.ts index 11e08130bcac..e9346c6a2833 100644 --- a/src/auto-reply/reply/dispatch-acp-delivery.ts +++ b/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[]; cleanBlockTtsDirectiveText?: ReturnType; blockCount: number; deliveredFinalReply: boolean; @@ -178,6 +184,8 @@ export type AcpDispatchDeliveryCoordinator = { getAccumulatedVisibleBlockText: () => string; getAccumulatedBlockTtsText: () => string; getAccumulatedFinalText: () => string; + getAccumulatedTranscriptText: () => string; + resolveAccumulatedDeliveredTranscriptText: () => Promise; settleVisibleText: () => Promise; 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 => { 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, diff --git a/src/auto-reply/reply/dispatch-acp.test.ts b/src/auto-reply/reply/dispatch-acp.test.ts index 38e9e0bf3370..ba6f5f1aba53 100644 --- a/src/auto-reply/reply/dispatch-acp.test.ts +++ b/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 }) => { 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 }) => { + 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> = []; + let markDeliveryStarted!: () => void; + let releaseDelivery!: () => void; + const deliveryStarted = new Promise((resolve) => { + markDeliveryStarted = resolve; + }); + const deliveryGate = new Promise((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((resolve) => { + markTurnReady = resolve; + }); + const finishTurnGate = new Promise((resolve) => { + finishTurn = resolve; + }); + const turnDone = new Promise((resolve) => { + markTurnDone = resolve; + }); + managerMocks.runTurn.mockImplementationOnce( + async ({ onEvent }: { onEvent: (event: unknown) => Promise }) => { + 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 }) => { + 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"); diff --git a/src/auto-reply/reply/dispatch-acp.ts b/src/auto-reply/reply/dispatch-acp.ts index 6dbe283f8227..41295b3ab60b 100644 --- a/src/auto-reply/reply/dispatch-acp.ts +++ b/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,