diff --git a/docs/design/web-shell-cancel-duration.md b/docs/design/web-shell-cancel-duration.md new file mode 100644 index 00000000000..1e2058a53a7 --- /dev/null +++ b/docs/design/web-shell-cancel-duration.md @@ -0,0 +1,90 @@ +# Persistent cancellation elapsed time + +[English](web-shell-cancel-duration.md) | [简体中文](web-shell-cancel-duration.zh-CN.md) + +## Problem and scope + +Web Shell renders a live cancellation event, but loading persisted history loses +that marker. The marker should survive refresh and read “You cancelled this +request after X seconds” (Chinese: “你在 X 秒后取消了请求”). Preserve the existing +right-aligned presentation. No new routes, storage system, or settings. + +## Design + +Reuse the existing `system/turn_result` record. Capture optional `cancelledAt` +on the turn controller's first explicit user abort, using the same agent clock +as `startedAt`. Keep `endedAt` as settlement time. The displayed whole seconds +are the nonnegative, rounded-up difference between cancellation and execution +start; queue wait and cancellation cleanup are excluded. A cancellation terminal +waits for the pinned recorder to flush, so immediate refresh cannot overtake its +write. Recording failures retain the existing degraded-recording behavior. + +Carry this authoritative cancellation timing through the prompt response and +bridge terminal event. Normalize live completion and persisted replay into the +existing `prompt.cancelled` event. Match by `promptId` so authoritative timing +updates the provisional live marker without duplication. Preserve record ids +and timestamps through replay. Overlapping history pages merge the persisted +marker into its existing position, including provisional markers and repair +checkpoints, without replaying cancellation against the active tools. The Web Shell adapter passes duration to the +existing system-message component, which formats localized text. The processed +summary uses the same cancellation duration and rounds up with the same rule, +including an explicit 0s. Direct SDK updates that omit a timestamp preserve the +previous timestamp. + +Existing records without explicit cancellation timing retain existing behavior; +do not invent user attribution or duration for session disposal, superseded +prompts, permission denial, or failed forwarding. Existing provisional cancel +markers remain a fallback for older servers. Fork-history semantics are out of +scope. + +Interrupted model streams must also retain the thinking and text already emitted. +The shared stream processor saves those partial parts in both in-memory history +and the existing deferred assistant recorder from its cancellation cleanup, +including when the consumer closes the generator. Ordinary transport errors keep +their retry/discard behavior. MAX_TOKENS escalation and continuation retries skip +rollback when the request is cancelled, and recovery rethrows cancellation instead +of synthesizing normal completion. The outer generator cleanup merges retained +continuations and removes internal recovery prompts, including cancellation at a +yield or before continuation content arrives. Synthetic tool-call chunks are +recorded before they are yielded, so closing there preserves recovered tool calls +rather than their original XML. During transport continuation backoff and stream +establishment, the outer generator retains the accumulated text prefix for +cancellation. Once the response processor takes over, or a fresh retry discards +the prefix, this temporary ownership is cleared to prevent duplicate recording. +If an established text continuation fails into reactive compression, the outer +generator takes ownership back until a fresh retry discards the visible text. +Restoring the assistant content also restores the +existing processed summary; no summary-visibility override is needed. + +## Affected areas + +CLI session turn recording; core stream finalization and turn-result payload +validation; ACP bridge +terminal forwarding and shared transcript replay; SDK normalization and +transcript reduction; Web Shell message adapter, component, and translations. +All data remains within the existing owning session and persisted workspace. + +## Validation and acceptance + +Attempt baseline with global `qwen`, then verify the local bundle. Cancel an +active request, await settlement, reload from persisted history, and verify one +marker with unchanged elapsed seconds. Cancel after thinking and text have both +arrived and verify their contents and the processed summary survive reload. +Also test cancellation before the first +second, slow cleanup, repeated cancel, a second cancelled turn, and exclusion +of ordinary completion, errors, direct disposal, and supersession. Exercise malformed +metadata and legacy records. Cover cancellation in escalation, the first and +later recovery attempts, empty partial output, consumer closure, and synthetic +recovered-tool-call output. Cancel at the transport continuation retry yield, +backoff, and stream establishment, including a later continuation with overlapping +text; confirm one matching history/record entry and no duplicate after the next +processor takes over. Verify failed partials are still discarded when a +supplied signal is not aborted, and that no internal recovery user prompt remains. Run focused package tests, build, typecheck, +bundle, two self-audit passes, and independent review. + +## Open questions + +Old history has no reliable explicit-cancel timestamp and is not backfilled. +The existing protocol also maps bridge deadlines and some close paths to its +user-cancel signal. This change preserves that classification; introducing a +cancellation-reason protocol is outside scope. diff --git a/docs/design/web-shell-cancel-duration.zh-CN.md b/docs/design/web-shell-cancel-duration.zh-CN.md new file mode 100644 index 00000000000..93dacc7e68f --- /dev/null +++ b/docs/design/web-shell-cancel-duration.zh-CN.md @@ -0,0 +1,63 @@ +# 持久化取消请求耗时 + +[English](web-shell-cancel-duration.md) | [简体中文](web-shell-cancel-duration.zh-CN.md) + +## 问题与范围 + +Web Shell 会显示实时取消事件，但加载持久化历史时会丢失该提示。提示应在刷新后 +保留，并显示“你在 X 秒后取消了请求”（英文：“You cancelled this request after +X seconds”）。保留现有右对齐样式，不增加路由、存储系统或设置。 + +## 设计 + +复用现有 `system/turn_result` 记录。回合控制器首次因用户显式取消而中止时， +记录可选的 `cancelledAt`，与 `startedAt` 使用同一代理时钟。`endedAt` 仍表示 +收尾结束时间。显示秒数为取消时间减去执行开始时间，向上取整且不小于零； +不包含排队等待和取消清理耗时。取消终态返回前等待绑定的 recorder 写入完成， +避免立即刷新抢在写入之前；记录失败沿用现有的持久化降级行为。 + +通过 prompt 响应及 bridge 终止事件传递权威取消时间。将实时完成事件与持久化 +回放统一转换为现有 `prompt.cancelled` 事件。按 `promptId` 匹配，更新临时的 +实时提示而不重复新增。回放保留记录标识和时间戳。重叠历史分页将持久化提示合并到 +现有位置，同时处理临时提示及重放修复检查点，不重新取消当前正在运行的工具。Web Shell 适配器把耗时传给 +现有系统消息组件，按语言格式化文案。“已处理”统计使用同一个取消耗时， +并采用相同的向上取整规则，包括明确显示 0s。SDK 直接更新省略时间戳时，保留原有时间戳。 + +没有显式取消时间的旧记录保持现有行为；不为会话销毁、新请求替换旧请求、 +权限拒绝或转发失败虚构用户归因或耗时。保留现有临时取消提示以兼容旧服务端。 +分支会话的历史继承语义不在本次范围内。 + +被中断的模型流也必须保留已输出的思考和正文。共享流处理器在取消清理时， +将部分输出同时保存到内存历史和现有的 assistant 延迟记录队列；消费者直接关闭 +生成器也要执行此步骤。普通传输错误继续沿用重试及丢弃失败尝试的行为。 +MAX_TOKENS 升级和续写重试在请求取消时跳过回滚；恢复流程直接重抛取消，不合成正常完成。 +外层生成器收尾时合并保留的续写并删除内部恢复提示，覆盖在 yield 处取消和续写尚无内容的情况。 +合成工具调用先记录再输出，确保此时关闭生成器仍保留恢复后的调用，而非原始 XML。网络断流续写在退避等待或建流期间，由外层生成器暂存累计正文前缀， +取消时负责保存；新流处理器接管或普通重试丢弃前缀时清除暂存，避免重复记录。 +已建立的正文续写流失败并进入响应式压缩时，由外层生成器重新负责保存，直到普通重试明确丢弃可见正文。 +恢复 assistant 内容后，现有“已处理”统计栏也随之恢复，无需修改统计栏显示条件。 + +## 影响范围 + +CLI 会话回合记录；core 流式输出收尾及回合结果载荷校验；ACP bridge 终止事件转发及共享历史 +回放；SDK 事件归一化及消息归约；Web Shell 消息适配器、组件和翻译。 +所有数据仍位于现有所属会话及持久化工作区内。 + +## 验证与验收 + +先使用全局 `qwen` 尝试基线验证，再验证本地 bundle。取消执行中的请求，等待 +收尾后从持久化历史重载，确认只有一条提示且秒数不变。在思考和正文都已输出后 +取消，确认两类内容和“已处理”统计在重载后保留。另测一秒内取消、缓慢 +清理、重复取消、连续两轮取消，以及普通完成、错误、直接销毁和请求替换不误显示。 +覆盖畸形元数据和旧记录。测试升级、首次及后续恢复、空输出、消费者关闭生成器和合成工具 +调用处的取消，验证未取消的 signal 仍正常丢弃失败尝试，且不留下内部恢复用户提示。运行相关包定向测试、构建、类型检查、打包、两轮 +自查及独立审查。 + +还需在网络断流续写的 RETRY yield、退避等待和建流期间取消，覆盖含重叠正文的后续 +续写；确认 history 与记录只各保存一份，新流处理器接管后取消也不会重复记录。 + +## 待决问题 + +旧历史缺少可靠的显式取消时间，不进行回填。 +现有协议也会将 bridge 超时及部分关闭路径映射为用户取消信号。本次保留该 +分类方式；增加取消原因协议不在范围内。 diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 4a545abf5ad..32959dc0110 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -18643,6 +18643,54 @@ describe('createAcpSessionBridge', () => { }); }); + describe('turn_complete promptCancelled', () => { + it.each([ + ['cancelled', { cancelledAt: 4_500, elapsedMs: 3_500 }, true], + ['end_turn', { cancelledAt: 4_500, elapsedMs: 3_500 }, false], + ['cancelled', { cancelledAt: NaN, elapsedMs: 3_500 }, false], + ['cancelled', { cancelledAt: 4_500, elapsedMs: -1 }, false], + ['cancelled', { cancelledAt: 4_500, elapsedMs: Infinity }, false], + ])( + 'validates %s cancellation metadata %j', + async (stopReason, metadata, valid) => { + const events: BridgeEvent[] = []; + const handle = makeChannel({ + promptImpl: () => + ({ + stopReason, + _meta: { 'qwen.promptCancelled': metadata }, + }) as PromptResponse, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const sub = (async () => { + for await (const event of bridge.subscribeEvents(session.sessionId)) { + events.push(event); + } + })(); + sub.catch(() => {}); + await bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'cancel me' }], + }); + await vi.waitFor(() => + expect(events.some((event) => event.type === 'turn_complete')).toBe( + true, + ), + ); + const terminal = events.find((event) => event.type === 'turn_complete'); + if (valid) { + expect(terminal?.data).toHaveProperty('promptCancelled', metadata); + } else { + expect(terminal?.data).not.toHaveProperty('promptCancelled'); + } + await bridge.shutdown(); + }, + ); + }); + describe('turn_complete branchPoint', () => { it('publishes a validated completed-turn branch point', async () => { const events: BridgeEvent[] = []; @@ -20886,6 +20934,7 @@ describe('createAcpSessionBridge', () => { const terms = terminalsFor(events, 'prompt-b'); expect(terms).toHaveLength(1); expect(terms[0]?.type).toBe('turn_complete'); + expect(terms[0]?.data).not.toHaveProperty('promptCancelled'); expect((terms[0]?.data as { stopReason?: string }).stopReason).toBe( 'cancelled', ); diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 36ac9ccb98f..10d299ac46e 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -1869,6 +1869,23 @@ function broadcastTurnComplete( checkpointUuid: rawBranchPoint['checkpointUuid'], } : undefined; + const rawPromptCancelled = + meta?.['qwen.promptCancelled'] && + typeof meta['qwen.promptCancelled'] === 'object' + ? (meta['qwen.promptCancelled'] as Record) + : undefined; + const promptCancelled = + promptResult.stopReason === 'cancelled' && + typeof rawPromptCancelled?.['cancelledAt'] === 'number' && + Number.isFinite(rawPromptCancelled['cancelledAt']) && + typeof rawPromptCancelled['elapsedMs'] === 'number' && + Number.isFinite(rawPromptCancelled['elapsedMs']) && + rawPromptCancelled['elapsedMs'] >= 0 + ? { + cancelledAt: rawPromptCancelled['cancelledAt'], + elapsedMs: rawPromptCancelled['elapsedMs'], + } + : undefined; const published = entry.events.publish({ type: 'turn_complete', ...(promptId ? { promptId } : {}), @@ -1877,6 +1894,7 @@ function broadcastTurnComplete( stopReason: promptResult.stopReason ?? 'end_turn', ...(promptId ? { promptId } : {}), ...(branchPoint ? { branchPoint } : {}), + ...(promptCancelled ? { promptCancelled } : {}), }, ...(originatorClientId ? { originatorClientId } : {}), }); diff --git a/packages/acp-bridge/src/transcript-replay.ts b/packages/acp-bridge/src/transcript-replay.ts index f3b99d34f73..6f842fc8186 100644 --- a/packages/acp-bridge/src/transcript-replay.ts +++ b/packages/acp-bridge/src/transcript-replay.ts @@ -981,6 +981,36 @@ class DefaultTranscriptReplayMachine implements TranscriptReplayMachine { emit: (update: SessionUpdate) => TranscriptReplayEmission, meta: UpdateMetaOptions, ): Iterable { + if (record.subtype === 'turn_result') { + const payload = isObjectRecord(record.systemPayload) + ? record.systemPayload + : undefined; + const cancelledAt = finiteNumber(payload?.['cancelledAt']); + const startedAt = finiteNumber(payload?.['startedAt']); + const promptId = payload?.['promptId']; + if ( + payload?.['state'] !== 'cancelled' || + cancelledAt === undefined || + typeof promptId !== 'string' || + !promptId || + (payload['startedAt'] !== undefined && startedAt === undefined) + ) + return; + const elapsedMs = Math.max(0, cancelledAt - (startedAt ?? cancelledAt)); + if (!Number.isFinite(elapsedMs)) return; + yield emit( + createTranscriptMessageUpdate({ + role: 'assistant', + text: '', + ...meta, + extra: { + qwenDiscreteMessage: true, + promptCancelled: { promptId, cancelledAt, elapsedMs }, + }, + }), + ); + return; + } if (record.subtype === 'agent_session_ready') { const payload = isObjectRecord(record.systemPayload) ? record.systemPayload diff --git a/packages/cli/src/acp-integration/session/Session.test.ts b/packages/cli/src/acp-integration/session/Session.test.ts index 34e0f6955e9..51dc831badb 100644 --- a/packages/cli/src/acp-integration/session/Session.test.ts +++ b/packages/cli/src/acp-integration/session/Session.test.ts @@ -9427,6 +9427,8 @@ describe('Session', () => { }); it('records a cancelled turn when admission aborts before dispatch', async () => { + let now = 1_000; + vi.spyOn(Date, 'now').mockImplementation(() => now); let releaseAdmission!: () => void; const admission = new Promise((resolve) => { releaseAdmission = resolve; @@ -9445,15 +9447,27 @@ describe('Session', () => { await vi.waitFor(() => expect(mockConfig.assertCanStartTurn).toHaveBeenCalledOnce(), ); + now = 4_500; cancellation.abort(); + now = 9_500; releaseAdmission(); - await expect(prompt).resolves.toEqual({ stopReason: 'cancelled' }); + await expect(prompt).resolves.toMatchObject({ + stopReason: 'cancelled', + _meta: { + 'qwen.promptCancelled': { + cancelledAt: 4_500, + elapsedMs: 0, + }, + }, + }); expect(mockChatRecordingService.recordTurnResult).toHaveBeenCalledWith( expect.objectContaining({ promptId: 'daemon-prompt-id', state: 'cancelled', stopReason: 'cancelled', + cancelledAt: 4_500, + endedAt: 9_500, }), ); }); @@ -9536,10 +9550,14 @@ describe('Session', () => { ); }); - it('records a cancelled turn when user cancel races an abort-shaped stream error', async () => { + it('records cancellation time before stream wind-down completes', async () => { + let now = 1_000; + vi.spyOn(Date, 'now').mockImplementation(() => now); mockChat.sendMessageStream = vi.fn().mockResolvedValue( createFailingStream('Request was aborted.', () => { + now = 4_500; void session.cancelPendingPrompt(); + now = 9_500; }), ); @@ -9551,16 +9569,64 @@ describe('Session', () => { }, trustedContext, ), - ).resolves.toEqual({ stopReason: 'cancelled' }); + ).resolves.toMatchObject({ + stopReason: 'cancelled', + _meta: { + 'qwen.promptCancelled': { cancelledAt: 4_500, elapsedMs: 3_500 }, + }, + }); expect(mockChatRecordingService.recordTurnResult).toHaveBeenCalledWith( expect.objectContaining({ promptId: 'daemon-prompt-id', state: 'cancelled', + startedAt: 1_000, + cancelledAt: 4_500, + endedAt: 9_500, }), ); }); + it('waits for the cancellation record to flush before returning', async () => { + let releaseFlush!: () => void; + const flushed = new Promise((resolve) => { + releaseFlush = resolve; + }); + mockChatRecordingService.flush.mockReturnValue(flushed); + mockChat.sendMessageStream = vi.fn().mockResolvedValue( + createFailingStream('Request was aborted.', () => { + void session.cancelPendingPrompt(); + }), + ); + let settled = false; + const prompt = session + .prompt( + { + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'cancel me' }], + }, + trustedContext, + ) + .then((response) => { + settled = true; + return response; + }); + await vi.waitFor(() => + expect(mockChatRecordingService.flush).toHaveBeenCalled(), + ); + expect(mockChatRecordingService.recordTurnResult).toHaveBeenCalledWith( + expect.objectContaining({ state: 'cancelled' }), + ); + expect(settled).toBe(false); + releaseFlush(); + await expect(prompt).resolves.toMatchObject({ + stopReason: 'cancelled', + _meta: { + 'qwen.promptCancelled': { cancelledAt: expect.any(Number) }, + }, + }); + }); + it('records a cancelled turn when user cancel races a plain stream error', async () => { mockChat.sendMessageStream = vi.fn().mockResolvedValue( createFailingStream('model exploded', () => { @@ -9576,7 +9642,12 @@ describe('Session', () => { }, trustedContext, ), - ).resolves.toEqual({ stopReason: 'cancelled' }); + ).resolves.toMatchObject({ + stopReason: 'cancelled', + _meta: { + 'qwen.promptCancelled': { cancelledAt: expect.any(Number) }, + }, + }); expect(mockChatRecordingService.recordTurnResult).toHaveBeenCalledWith( expect.objectContaining({ @@ -9769,6 +9840,7 @@ describe('Session', () => { stopReason: 'cancelled', promptText: 'first prompt', }); + expect(firstPayload).not.toHaveProperty('cancelledAt'); expect(mockChatRecordingService.recordTurnResult).toHaveBeenCalledTimes( 2, ); @@ -9916,7 +9988,15 @@ describe('Session', () => { releaseFirst(); await first; - await expect(second).resolves.toEqual({ stopReason: 'cancelled' }); + await expect(second).resolves.toMatchObject({ + stopReason: 'cancelled', + _meta: { + 'qwen.promptCancelled': { + cancelledAt: expect.any(Number), + elapsedMs: 0, + }, + }, + }); expect(mockChatRecordingService.recordTurnResult).toHaveBeenCalledTimes( 2, @@ -40301,7 +40381,7 @@ describe('Session', () => { await session.cancelPendingPrompt(); releaseWait(); - await expect(prompt).resolves.toEqual({ stopReason: 'cancelled' }); + await expect(prompt).resolves.toMatchObject({ stopReason: 'cancelled' }); }); it('lets cancellation win while a loop-detected Stop continuation is preserved', async () => { @@ -40352,7 +40432,7 @@ describe('Session', () => { await session.cancelPendingPrompt(); releaseDrain(); - await expect(prompt).resolves.toEqual({ stopReason: 'cancelled' }); + await expect(prompt).resolves.toMatchObject({ stopReason: 'cancelled' }); }); it('rejects a foreground turn whose Stop continuation trips loop protection', async () => { @@ -43820,7 +43900,7 @@ describe('Session', () => { }), ], }); - expect(firstResult).toEqual({ stopReason: 'cancelled' }); + expect(firstResult).toMatchObject({ stopReason: 'cancelled' }); }, ); diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index 3bcc07d3138..82fe75a5a3e 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/packages/cli/src/acp-integration/session/Session.ts @@ -1217,6 +1217,7 @@ interface InFlightTurnRecording { originatorClientId?: string; abortController?: AbortController; startedAt?: number; + cancelledAt?: number; promptText: string; promptTextTruncated: boolean; finalAnswer: { finalText: string }; @@ -4649,6 +4650,17 @@ export class Session implements SessionContext { ); } const turnRecording = this.#beginTurnRecording(params, invocationContext); + const recordAdmissionCancellation = () => { + if (turnRecording) turnRecording.cancelledAt ??= Date.now(); + }; + admissionCancellation?.addEventListener( + 'abort', + recordAdmissionCancellation, + { + once: true, + }, + ); + if (admissionCancellation?.aborted) recordAdmissionCancellation(); try { const result = await this.#promptWithTurnRecording( params, @@ -4658,7 +4670,7 @@ export class Session implements SessionContext { scheduledGoalTurn, turnRecording, ); - this.#settleTurnRecording( + await this.#settleTurnRecording( result.stopReason === 'cancelled' ? 'cancelled' : 'completed', turnRecording, result, @@ -4683,11 +4695,16 @@ export class Session implements SessionContext { (abortReason === NEW_PROMPT_ABORT_REASON && this.#isAbortError(error)); if (controlledAbort) { const result = { stopReason: 'cancelled' as const }; - this.#settleTurnRecording('cancelled', turnRecording, result); + await this.#settleTurnRecording('cancelled', turnRecording, result); return result; } - this.#settleTurnRecording('error', turnRecording, undefined, error); + await this.#settleTurnRecording('error', turnRecording, undefined, error); throw error; + } finally { + admissionCancellation?.removeEventListener( + 'abort', + recordAdmissionCancellation, + ); } } @@ -4787,6 +4804,17 @@ export class Session implements SessionContext { this.pendingPrompt?.abort(NEW_PROMPT_ABORT_REASON); const pendingSend = goalTurn?.controller ?? new AbortController(); if (turnRecording) turnRecording.abortController = pendingSend; + const recordCancellation = () => { + if ( + turnRecording && + pendingSend.signal.reason === USER_CANCEL_ABORT_REASON + ) { + turnRecording.cancelledAt ??= Date.now(); + } + }; + pendingSend.signal.addEventListener('abort', recordCancellation, { + once: true, + }); const cancelPendingSend = () => pendingSend.abort(USER_CANCEL_ABORT_REASON); if (admissionCancellation) { admissionCancellation.addEventListener('abort', cancelPendingSend, { @@ -4797,6 +4825,7 @@ export class Session implements SessionContext { this.pendingPrompt = pendingSend; const releasePendingSend = () => { admissionCancellation?.removeEventListener('abort', cancelPendingSend); + pendingSend.signal.removeEventListener('abort', recordCancellation); if (this.pendingPrompt === pendingSend) { this.pendingPrompt = null; } @@ -7856,13 +7885,27 @@ export class Session implements SessionContext { }; } - #settleTurnRecording( + async #settleTurnRecording( state: 'completed' | 'cancelled' | 'error', recording: InFlightTurnRecording | null, response?: PromptResponse, error?: unknown, - ): void { + ): Promise { if (recording === null) return; + const cancelledAt = + state === 'cancelled' ? recording.cancelledAt : undefined; + if (cancelledAt !== undefined && response) { + response._meta = { + ...response._meta, + 'qwen.promptCancelled': { + cancelledAt, + elapsedMs: Math.max( + 0, + cancelledAt - (recording.startedAt ?? cancelledAt), + ), + }, + }; + } const finalAnswer = truncateTurnText(recording.finalAnswer.finalText); const stopReason = response?.stopReason ?? (state === 'cancelled' ? 'cancelled' : undefined); @@ -7874,6 +7917,7 @@ export class Session implements SessionContext { ...(recording.startedAt !== undefined ? { startedAt: recording.startedAt } : {}), + ...(cancelledAt !== undefined ? { cancelledAt } : {}), endedAt: Date.now(), promptText: recording.promptText, ...(recording.promptTextTruncated ? { promptTextTruncated: true } : {}), @@ -7890,6 +7934,9 @@ export class Session implements SessionContext { }; try { recording.recordingService?.recordTurnResult(payload); + if (cancelledAt !== undefined) { + await recording.recordingService?.flush(); + } } catch (recordError) { debugLogger.warn( `Failed to record turn result: ${this.#formatError(recordError)}`, diff --git a/packages/core/src/core/llm-chat.test.ts b/packages/core/src/core/llm-chat.test.ts index 15c5a4aa288..b12fbae80c3 100644 --- a/packages/core/src/core/llm-chat.test.ts +++ b/packages/core/src/core/llm-chat.test.ts @@ -1794,6 +1794,194 @@ describe('LlmChat', async () => { expect(functionCallPart?.functionCall?.id).toBe('call_thinking_tool_use'); }); + it.each(['throw', 'end', 'close'] as const)( + 'persists cancelled thinking and text when the stream exits via %s', + async (exitMode) => { + const controller = new AbortController(); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + const parts: Part[] = [ + { text: 'Thinking ', thought: true }, + { text: 'first.', thought: true }, + { thought: true, thoughtSignature: 'signature' }, + { text: 'Partial ' }, + { text: 'answer.' }, + ]; + vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue( + (async function* () { + for (const part of parts) { + yield { + candidates: [{ content: { role: 'model', parts: [part] } }], + } as GenerateContentResponse; + } + if (exitMode === 'throw') throw controller.signal.reason; + })(), + ); + const stream = await recordingChat.sendMessageStream( + 'test-model', + { message: 'hello', config: { abortSignal: controller.signal } }, + 'cancelled-partial', + ); + for (let i = 0; i < parts.length; i++) { + expect((await stream.next()).done).toBe(false); + } + controller.abort(new DOMException('Cancelled', 'AbortError')); + if (exitMode === 'close') { + await stream.return(undefined); + } else { + await expect(stream.next()).rejects.toBe(controller.signal.reason); + } + const expectedParts = [ + { + text: 'Thinking first.', + thought: true, + thoughtSignature: 'signature', + }, + { text: 'Partial answer.' }, + ]; + expect(recordingChat.getHistory()).toEqual([ + { role: 'user', parts: [{ text: 'hello' }] }, + { role: 'model', parts: expectedParts }, + ]); + expect(recordAssistantTurn).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + model: 'test-model', + message: expectedParts, + }), + ); + expect( + mockContentGenerator.generateContentStream, + ).toHaveBeenCalledTimes(1); + }, + ); + + it.each(['abort', 'backend'] as const)( + 'preserves the upstream %s error and partial output during supersession', + async (kind) => { + const controller = new AbortController(); + const originalError = + kind === 'abort' + ? new DOMException('The operation was aborted.', 'AbortError') + : new Error('model backend failed'); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + const parts = [ + { text: 'Partial thought', thought: true }, + { text: 'Partial body' }, + ]; + vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue( + (async function* () { + yield { + candidates: [{ content: { role: 'model', parts } }], + } as GenerateContentResponse; + controller.abort('qwen:new-prompt'); + throw originalError; + })(), + ); + const stream = await recordingChat.sendMessageStream( + 'test-model', + { message: 'hello', config: { abortSignal: controller.signal } }, + 'superseded-partial', + ); + expect((await stream.next()).done).toBe(false); + await expect(stream.next()).rejects.toBe(originalError); + expect(recordingChat.getHistory()).toEqual([ + { role: 'user', parts: [{ text: 'hello' }] }, + { role: 'model', parts }, + ]); + expect(recordAssistantTurn).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ message: parts }), + ); + expect( + mockContentGenerator.generateContentStream, + ).toHaveBeenCalledTimes(1); + }, + ); + + it.each(['throw', 'close'] as const)( + 'retains signed reasoning episodes in order when cancellation exits via %s', + async (exitMode) => { + const controller = new AbortController(); + const abortError = new DOMException('Cancelled', 'AbortError'); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + const firstCall = { + functionCall: { id: 'call1', name: 'tool', args: {} }, + }; + const secondCall = { + functionCall: { id: 'call2', name: 'tool', args: {} }, + }; + vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue( + (async function* () { + yield { + candidates: [ + { + content: { + role: 'model', + parts: [ + { text: 'First thought', thought: true }, + { thought: true, thoughtSignature: 'sigA' }, + firstCall, + { text: 'Second thought', thought: true }, + { thought: true, thoughtSignature: 'sigB' }, + secondCall, + { text: 'Partial body' }, + ], + }, + }, + ], + } as GenerateContentResponse; + throw abortError; + })(), + ); + const stream = await recordingChat.sendMessageStream( + 'test-model', + { message: 'hello', config: { abortSignal: controller.signal } }, + 'cancelled-episodes', + ); + expect((await stream.next()).done).toBe(false); + controller.abort('qwen:user-cancel'); + if (exitMode === 'close') await stream.return(undefined); + else await expect(stream.next()).rejects.toBe(abortError); + const parts = [ + { text: 'First thought', thought: true, thoughtSignature: 'sigA' }, + firstCall, + { text: 'Second thought', thought: true, thoughtSignature: 'sigB' }, + secondCall, + { text: 'Partial body' }, + ]; + expect(recordingChat.getHistory()).toEqual([ + { role: 'user', parts: [{ text: 'hello' }] }, + { role: 'model', parts }, + ]); + expect(recordAssistantTurn).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ message: parts }), + ); + }, + ); + + it('does not record an empty assistant when cancelled before any content', async () => { + const controller = new AbortController(); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue( + (async function* () { + yield* []; + controller.abort('qwen:user-cancel'); + })(), + ); + const stream = await recordingChat.sendMessageStream( + 'test-model', + { message: 'hello', config: { abortSignal: controller.signal } }, + 'empty-cancel', + ); + await expect(stream.next()).rejects.toBe('qwen:user-cancel'); + expect(recordingChat.getHistory()).toEqual([ + { role: 'user', parts: [{ text: 'hello' }] }, + ]); + expect(recordAssistantTurn).not.toHaveBeenCalled(); + }); + it('does NOT persist partial assistant turn when stream throws before any tool_use chunk', async () => { // Plain-text partial responses are deliberately dropped on stream // error: the Retry path pops the trailing user prompt and re-issues @@ -9767,6 +9955,86 @@ describe('LlmChat', async () => { expect(fallbackBGenerateContentStream).not.toHaveBeenCalled(); }); + it('retains tool calls and recording when a fallback is cancelled with an ACP reason', async () => { + const controller = new AbortController(); + const abortError = new DOMException( + 'The operation was aborted.', + 'AbortError', + ); + const record = vi.fn(); + const chatWithRecording = chatWithRecorder(record); + vi.mocked(mockConfig.getContentGeneratorConfig).mockReturnValue({ + authType: AuthType.USE_GEMINI, + model: 'test-model', + maxRetries: 0, + }); + vi.mocked(mockConfig.getModelFallbacks).mockReturnValue(['fallback-a']); + vi.mocked( + mockContentGenerator.generateContentStream, + ).mockRejectedValueOnce( + Object.assign(new Error('capacity'), { status: 503 }), + ); + const fallback = { + ...mockContentGenerator, + generateContentStream: vi.fn().mockResolvedValue( + (async function* () { + yield { + candidates: [ + { + content: { + role: 'model', + parts: [ + { text: 'thinking', thought: true }, + { + functionCall: { + id: 'call-1', + name: 'read_file', + args: { path: 'foo' }, + }, + }, + ], + }, + }, + ], + } as GenerateContentResponse; + controller.abort('qwen:user-cancel'); + throw abortError; + })(), + ), + }; + vi.mocked(mockConfig.getBaseLlmClient).mockReturnValue({ + resolveForModel: vi.fn().mockResolvedValue({ + contentGenerator: fallback, + model: 'fallback-a', + retryAuthType: AuthType.USE_GEMINI, + }), + } as never); + const stream = await chatWithRecording.sendMessageStream( + 'test-model', + { message: 'test', config: { abortSignal: controller.signal } }, + 'test', + ); + await expect( + (async () => { + for await (const _ of stream) { + /* consume */ + } + })(), + ).rejects.toBe(abortError); + expect(chatWithRecording.getHistory()).toEqual([ + expect.objectContaining({ role: 'user' }), + expect.objectContaining({ + role: 'model', + parts: expect.arrayContaining([ + expect.objectContaining({ + functionCall: expect.objectContaining({ id: 'call-1' }), + }), + ]), + }), + ]); + expect(record).toHaveBeenCalledOnce(); + }); + it('does not fallback on non-eligible primary auth errors', async () => { vi.mocked(mockConfig.getContentGeneratorConfig).mockReturnValue({ authType: AuthType.USE_GEMINI, @@ -10493,6 +10761,305 @@ describe('LlmChat', async () => { ]![0].contents as Content[]; } + it.each([ + ['retry yield', 1], + ['retry delay', 1], + ['stream establishment', 1], + ['resumed output', 1], + ['retry yield', 2], + ['retry delay', 2], + ['stream establishment', 2], + ['resumed output', 2], + ] as const)( + 'persists the prefix when cancelled at %s of continuation %s', + async (phase, continuation) => { + vi.useFakeTimers(); + try { + const controller = new AbortController(); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + const generate = vi.mocked( + mockContentGenerator.generateContentStream, + ); + generate.mockResolvedValueOnce( + cutAfter([textChunk('first half ')]), + ); + if (continuation === 2) { + generate.mockResolvedValueOnce( + cutAfter([textChunk(' half second part ')]), + ); + } + let establishing = false; + generate.mockImplementationOnce(async () => { + establishing = true; + if (phase === 'resumed output') { + return (async function* () { + yield textChunk('resumed tail', 'STOP'); + })(); + } + return new Promise>( + (_resolve, reject) => { + controller.signal.addEventListener( + 'abort', + () => reject(controller.signal.reason), + { once: true }, + ); + }, + ); + }); + const stream = await recordingChat.sendMessageStream( + 'test-model', + { + message: 'write answer', + config: { abortSignal: controller.signal }, + }, + 'cancel-transport-gap', + ); + let retries = 0; + const delivered: string[] = []; + while (retries < continuation) { + const next = stream.next(); + if (retries > 0) await vi.advanceTimersByTimeAsync(5_000); + const event = await next; + expect(event.done).toBe(false); + if (event.done) break; + if (event.value.type === StreamEventType.RETRY) { + expect(event.value.isContinuation).toBe(true); + retries++; + } else if (event.value.type === StreamEventType.CHUNK) { + delivered.push( + (event.value.value.candidates?.[0]?.content?.parts ?? []) + .filter((part) => !part.thought) + .map((part) => part.text ?? '') + .join(''), + ); + } + } + expect(delivered.join('')).toContain('first half '); + expect(retries).toBe(continuation); + expect(establishing).toBe(false); + if (phase === 'retry yield') { + controller.abort('qwen:user-cancel'); + await stream.return(undefined); + } else if (phase === 'resumed output') { + const next = stream.next(); + await vi.advanceTimersByTimeAsync(5_000); + expect(await next).toMatchObject({ + done: false, + value: { type: StreamEventType.CHUNK }, + }); + controller.abort('qwen:user-cancel'); + await stream.return(undefined); + } else { + const next = stream.next(); + if (phase === 'stream establishment') { + await vi.advanceTimersByTimeAsync(5_000); + expect(establishing).toBe(true); + } + controller.abort('qwen:user-cancel'); + await expect(next).rejects.toBe('qwen:user-cancel'); + } + const message = [ + { + text: + (continuation === 1 + ? 'first half ' + : 'first half second part ') + + (phase === 'resumed output' ? 'resumed tail' : ''), + }, + ]; + expect(recordingChat.getHistory()).toEqual([ + { role: 'user', parts: [{ text: 'write answer' }] }, + { role: 'model', parts: message }, + ]); + expect(recordAssistantTurn).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ model: 'test-model', message }), + ); + expect(generate).toHaveBeenCalledTimes( + continuation + + (phase === 'stream establishment' || phase === 'resumed output' + ? 1 + : 0), + ); + } finally { + vi.useRealTimers(); + } + }, + ); + + it.each(['waiting', 'compressed notification'] as const)( + 'preserves delivered text when cancelled at reactive compression %s', + async (phase) => { + vi.useFakeTimers(); + try { + const controller = new AbortController(); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + let compressing = false; + vi.spyOn(ChatCompressionService.prototype, 'compress') + .mockResolvedValueOnce({ + newHistory: null, + info: { + originalTokenCount: 0, + newTokenCount: 0, + compressionStatus: CompressionStatus.NOOP, + }, + }) + .mockImplementationOnce(async () => { + compressing = true; + if (phase === 'waiting') { + return new Promise((_resolve, reject) => { + controller.signal.addEventListener( + 'abort', + () => reject(controller.signal.reason), + { once: true }, + ); + }); + } + return { + newHistory: [{ role: 'user', parts: [{ text: 'summary' }] }], + info: { + originalTokenCount: 135_000, + newTokenCount: 40_000, + compressionStatus: CompressionStatus.COMPRESSED, + }, + }; + }); + vi.mocked(mockContentGenerator.generateContentStream) + .mockResolvedValueOnce(cutAfter([textChunk('first half ')])) + .mockResolvedValueOnce( + (async function* () { + yield textChunk('second half'); + throw new StreamContentError( + 'prompt is too long: 135000 tokens > 128000 maximum', + ); + })(), + ); + const stream = await recordingChat.sendMessageStream( + 'test-model', + { + message: 'write answer', + config: { abortSignal: controller.signal }, + }, + 'cancel-reactive-compression', + ); + expect(await stream.next()).toMatchObject({ + value: { type: StreamEventType.CHUNK }, + }); + expect(await stream.next()).toMatchObject({ + value: { type: StreamEventType.RETRY, isContinuation: true }, + }); + const resumed = stream.next(); + await vi.advanceTimersByTimeAsync(5_000); + expect(await resumed).toMatchObject({ + value: { type: StreamEventType.CHUNK }, + }); + const compress = stream.next(); + const outcome = compress.catch((error) => error); + await vi.advanceTimersByTimeAsync(0); + expect(compressing).toBe(true); + if (phase === 'compressed notification') { + expect(await outcome).toMatchObject({ + value: { type: StreamEventType.COMPRESSED }, + }); + } + controller.abort('qwen:user-cancel'); + if (phase === 'waiting') + expect(await outcome).toBe(controller.signal.reason); + else await stream.return(undefined); + const parts = [{ text: 'first half second half' }]; + expect( + recordingChat + .getHistory() + .filter((turn) => turn.role === 'model'), + ).toEqual([{ role: 'model', parts }]); + expect(recordAssistantTurn).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ message: parts }), + ); + expect( + mockContentGenerator.generateContentStream, + ).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }, + ); + + it.each(['rate limit', 'compression'])( + 'does not restore a discarded prefix when cancelled at a fresh %s retry', + async (retry) => { + vi.useFakeTimers(); + try { + const controller = new AbortController(); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + if (retry === 'compression') { + vi.spyOn(ChatCompressionService.prototype, 'compress') + .mockResolvedValueOnce({ + newHistory: null, + info: { + originalTokenCount: 0, + newTokenCount: 0, + compressionStatus: CompressionStatus.NOOP, + }, + }) + .mockResolvedValueOnce({ + newHistory: [{ role: 'user', parts: [{ text: 'summary' }] }], + info: { + originalTokenCount: 135_000, + newTokenCount: 40_000, + compressionStatus: CompressionStatus.COMPRESSED, + }, + }); + } + vi.mocked(mockContentGenerator.generateContentStream) + .mockResolvedValueOnce(cutAfter([textChunk('discarded prefix')])) + .mockRejectedValueOnce( + retry === 'rate limit' + ? Object.assign(new Error('rate limit'), { status: 429 }) + : new Error( + 'prompt is too long: 135000 tokens > 128000 maximum', + ), + ); + const stream = await recordingChat.sendMessageStream( + 'test-model', + { + message: 'write answer', + config: { abortSignal: controller.signal }, + }, + 'cancel-fresh-retry', + ); + expect(await stream.next()).toMatchObject({ + value: { type: StreamEventType.CHUNK }, + }); + expect(await stream.next()).toMatchObject({ + value: { type: StreamEventType.RETRY, isContinuation: true }, + }); + const next = stream.next(); + await vi.advanceTimersByTimeAsync(5_000); + let result = await next; + if (result.value?.type === StreamEventType.COMPRESSED) { + result = await stream.next(); + } + expect(result.done).toBe(false); + expect(result.value).toMatchObject({ type: StreamEventType.RETRY }); + if (result.done || result.value.type !== StreamEventType.RETRY) { + throw new Error('Expected a fresh retry'); + } + expect(result.value.isContinuation).not.toBe(true); + result.value.retryInfo?.skipDelay?.(); + controller.abort('qwen:user-cancel'); + await stream.return(undefined); + expect(JSON.stringify(recordingChat.getHistory())).not.toContain( + 'discarded prefix', + ); + expect(recordAssistantTurn).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }, + ); + it('continues from the delivered text instead of failing the send', async () => { vi.useFakeTimers(); try { @@ -13327,69 +13894,86 @@ describe('LlmChat', async () => { }); }); - it('should discard valid partial content from a failed attempt upon retry', async () => { - // Mock the stream to fail on the first attempt after yielding some valid content. - vi.mocked(mockContentGenerator.generateContentStream) - .mockImplementationOnce(async () => - // First attempt: yields one valid chunk, then one invalid chunk - (async function* () { - yield { - candidates: [ - { - content: { - parts: [{ text: 'This valid part should be discarded' }], + it.each([false, true])( + 'discards failed partials on retry with an un-aborted signal present: %s', + async (withSignal) => { + const controller = new AbortController(); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + // Mock the stream to fail on the first attempt after yielding some valid content. + vi.mocked(mockContentGenerator.generateContentStream) + .mockImplementationOnce(async () => + // First attempt: yields one valid chunk, then one invalid chunk + (async function* () { + yield { + candidates: [ + { + content: { + parts: [{ text: 'This valid part should be discarded' }], + }, }, - }, - ], - } as unknown as GenerateContentResponse; - yield { - candidates: [{ content: { parts: [{ text: '' }] } }], // Invalid chunk triggers retry - } as unknown as GenerateContentResponse; - })(), - ) - .mockImplementationOnce(async () => - // Second attempt (the retry): succeeds - (async function* () { - yield { - candidates: [ - { - content: { - parts: [{ text: 'Successful final response' }], + ], + } as unknown as GenerateContentResponse; + yield { + candidates: [{ content: { parts: [{ text: '' }] } }], // Invalid chunk triggers retry + } as unknown as GenerateContentResponse; + })(), + ) + .mockImplementationOnce(async () => + // Second attempt (the retry): succeeds + (async function* () { + yield { + candidates: [ + { + content: { + parts: [{ text: 'Successful final response' }], + }, + finishReason: 'STOP', }, - finishReason: 'STOP', - }, - ], - } as unknown as GenerateContentResponse; - })(), + ], + } as unknown as GenerateContentResponse; + })(), + ); + + // Send a message and consume the stream + const stream = await recordingChat.sendMessageStream( + 'test-model', + { + message: 'test', + ...(withSignal ? { config: { abortSignal: controller.signal } } : {}), + }, + 'prompt-id-discard-test', ); + const events: StreamEvent[] = []; + for await (const event of stream) { + events.push(event); + } - // Send a message and consume the stream - const stream = await chat.sendMessageStream( - 'test-model', - { message: 'test' }, - 'prompt-id-discard-test', - ); - const events: StreamEvent[] = []; - for await (const event of stream) { - events.push(event); - } + expect(controller.signal.aborted).toBe(false); + expect(recordAssistantTurn).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + message: [{ text: 'Successful final response' }], + }), + ); + // Check that a retry happened + expect(mockContentGenerator.generateContentStream).toHaveBeenCalledTimes( + 2, + ); + expect(events.some((e) => e.type === StreamEventType.RETRY)).toBe(true); - // Check that a retry happened - expect(mockContentGenerator.generateContentStream).toHaveBeenCalledTimes(2); - expect(events.some((e) => e.type === StreamEventType.RETRY)).toBe(true); + // Check the final recorded history + const history = recordingChat.getHistory(); + expect(history.length).toBe(2); // user turn + final model turn - // Check the final recorded history - const history = chat.getHistory(); - expect(history.length).toBe(2); // user turn + final model turn - - const modelTurn = history[1]!; - // The model turn should only contain the text from the successful attempt - expect(modelTurn!.parts![0]!.text).toBe('Successful final response'); - // It should NOT contain any text from the failed attempt - expect(modelTurn!.parts![0]!.text).not.toContain( - 'This valid part should be discarded', - ); - }); + const modelTurn = history[1]!; + // The model turn should only contain the text from the successful attempt + expect(modelTurn!.parts![0]!.text).toBe('Successful final response'); + // It should NOT contain any text from the failed attempt + expect(modelTurn!.parts![0]!.text).not.toContain( + 'This valid part should be discarded', + ); + }, + ); it('discards a completed protocol-tagged response and retries before persistence', async () => { const recordAssistantTurn = vi.fn(); @@ -15590,6 +16174,145 @@ describe('LlmChat', async () => { expect(recovery2Config.maxOutputTokens).toBeGreaterThanOrEqual(4_000); }); + it.each([ + ['escalation', 'throw'], + ['escalation', 'close'], + ['recovery', 'throw'], + ['recovery', 'close'], + ['later recovery', 'throw'], + ['later recovery', 'close'], + ] as const)( + 'preserves cancelled %s output via %s without internal user messages', + async (phase, exitMode) => { + const controller = new AbortController(); + const abortError = new DOMException('Cancelled', 'AbortError'); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + const streams = [ + makeStream([makeChunk([{ text: 'INITIAL' }], 'MAX_TOKENS')]), + ]; + if (phase !== 'escalation') + streams.push( + makeStream([makeChunk([{ text: 'BASE' }], 'MAX_TOKENS')]), + ); + if (phase === 'later recovery') + streams.push( + makeStream([ + makeChunk([{ text: 'FIRST CONTINUATION' }], 'MAX_TOKENS'), + ]), + ); + streams.push( + (async function* () { + yield makeChunk([{ text: 'CANCELLED THOUGHT', thought: true }]); + yield makeChunk([{ text: 'CANCELLED BODY' }]); + throw abortError; + })(), + ); + let calls = 0; + vi.mocked( + mockContentGenerator.generateContentStream, + ).mockImplementation(async () => streams[calls++]!); + const stream = await recordingChat.sendMessageStream( + 'gemini-pro', + { + message: 'write long answer', + config: { abortSignal: controller.signal }, + }, + 'cancel-output', + ); + let receivedBody = false; + for (let i = 0; i < 20; i++) { + const next = await stream.next(); + expect(next.done).toBe(false); + if ( + !next.done && + next.value.type === StreamEventType.CHUNK && + next.value.value.candidates?.[0]?.content?.parts?.some( + (part) => part.text === 'CANCELLED BODY', + ) + ) { + receivedBody = true; + break; + } + } + expect(receivedBody).toBe(true); + expect(calls).toBe(streams.length); + controller.abort('qwen:user-cancel'); + if (exitMode === 'close') await stream.return(undefined); + else await expect(stream.next()).rejects.toBe(abortError); + const history = recordingChat.getHistory(); + expect(history.map((entry) => entry.role)).toEqual(['user', 'model']); + expect(history[0]?.parts).toEqual([{ text: 'write long answer' }]); + expect(JSON.stringify(history[1])).toContain('CANCELLED THOUGHT'); + expect(JSON.stringify(history[1])).toContain('CANCELLED BODY'); + if (phase !== 'escalation') + expect(JSON.stringify(history[1])).toContain('BASE'); + if (phase === 'later recovery') + expect(JSON.stringify(history[1])).toContain('FIRST CONTINUATION'); + const cancelledRecords = recordAssistantTurn.mock.calls.filter( + ([record]) => + JSON.stringify(record.message).includes('CANCELLED BODY'), + ); + expect(cancelledRecords).toHaveLength(1); + expect(JSON.stringify(cancelledRecords[0])).toContain( + 'CANCELLED THOUGHT', + ); + }, + ); + + it.each(['escalation', 'recovery'] as const)( + 'removes internal recovery prompts when %s is cancelled before content', + async (phase) => { + const controller = new AbortController(); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + const streams = [ + makeStream([makeChunk([{ text: 'INITIAL' }], 'MAX_TOKENS')]), + ]; + if (phase === 'recovery') + streams.push( + makeStream([makeChunk([{ text: 'BASE' }], 'MAX_TOKENS')]), + ); + streams.push( + (async function* () { + yield* []; + controller.abort('qwen:user-cancel'); + })(), + ); + let calls = 0; + vi.mocked( + mockContentGenerator.generateContentStream, + ).mockImplementation(async () => streams[calls++]!); + const stream = await recordingChat.sendMessageStream( + 'gemini-pro', + { + message: 'write long answer', + config: { abortSignal: controller.signal }, + }, + 'empty-recovery', + ); + await expect( + (async () => { + for await (const _ of stream) { + /* consume */ + } + })(), + ).rejects.toBe('qwen:user-cancel'); + expect(calls).toBe(streams.length); + expect(recordingChat.getHistory()).toEqual([ + { role: 'user', parts: [{ text: 'write long answer' }] }, + ...(phase === 'recovery' + ? [{ role: 'model', parts: [{ text: 'BASE' }] }] + : []), + ]); + expect( + recordAssistantTurn.mock.calls.every( + ([record]) => record.message.length > 0, + ), + ).toBe(true); + }, + ); + it('should enter recovery loop when escalated response is also truncated', async () => { // Three streams: initial (MAX_TOKENS) → escalated (MAX_TOKENS) → // recovery (STOP). @@ -19117,6 +19840,88 @@ describe('LlmChat', async () => { } as unknown as GenerateContentResponse; } + it.each(['xml', 'buffered-json'])( + 'preserves a %s tool call when cancelled at its synthetic chunk', + async (kind) => { + const controller = new AbortController(); + const recordAssistantTurn = vi.fn(); + const recordingChat = chatWithRecorder(recordAssistantTurn); + vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue( + (async function* () { + yield { + candidates: [ + { + content: { + role: 'model', + parts: [{ text: 'Thinking', thought: true }], + }, + }, + ], + } as GenerateContentResponse; + if (kind === 'xml') { + yield xmlChunk( + 'a.ts', + 'STOP', + ); + } else { + yield { + candidates: [ + { + content: { + role: 'model', + parts: [ + { text: '{"ok":true}' }, + { + functionCall: { + id: 'call-pending', + name: 'read_file', + args: { file_path: 'a.ts' }, + }, + }, + ], + }, + }, + ], + } as GenerateContentResponse; + } + })(), + ); + const stream = await recordingChat.sendMessageStream( + 'gemini-pro', + { + message: 'read the file', + config: { abortSignal: controller.signal }, + }, + 'cancel-synthetic', + ); + let call: Part['functionCall']; + for (let i = 0; i < 10; i++) { + const next = await stream.next(); + expect(next.done).toBe(false); + if (!next.done && next.value.type === StreamEventType.CHUNK) + call = next.value.value.functionCalls?.[0]; + if (call) break; + } + expect(call?.name).toBe('read_file'); + controller.abort('qwen:user-cancel'); + await stream.return(undefined); + expect(recordingChat.getHistory()[1]?.parts).toEqual( + expect.arrayContaining([ + { text: 'Thinking', thought: true }, + { functionCall: call }, + ]), + ); + expect(recordAssistantTurn).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + message: expect.arrayContaining([ + { text: 'Thinking', thought: true }, + { functionCall: call }, + ]), + }), + ); + }, + ); + it('recovers XML tool calls from plain text content and updates history', async () => { const xml = 'a.ts'; diff --git a/packages/core/src/core/llm-chat.ts b/packages/core/src/core/llm-chat.ts index f53849999f7..26e38ba95e1 100644 --- a/packages/core/src/core/llm-chat.ts +++ b/packages/core/src/core/llm-chat.ts @@ -340,6 +340,107 @@ export function redactStructuredOutputArgsForRecording( }; } +function consolidateModelResponseParts(allModelParts: Part[]): Part[] { + // A turn can legitimately contain multiple distinct reasoning episodes + // separated by tool calls (Anthropic interleaved thinking, OpenAI + // Responses reasoning items on parallel function calls). Each episode + // must keep its own signature and its own position relative to the + // tool calls it preceded -- merging every thought-flagged part into one + // blob and keeping only the first signature silently discards every + // other episode's replayable payload and destroys the interleaving. + // + // Both wires terminate an episode with a text-less, signature-only + // chunk (anthropicContentGenerator.ts's signature_delta handling; + // responses-converter.ts's output_item.done for a reasoning item), so a + // thought part carrying fresh non-empty text while the open episode + // already has both accumulated text and a signature can only be the + // start of a new episode -- no legitimate continuation of the same + // episode reintroduces text after its signature is set. The + // `openEpisodeText.length > 0` guard additionally protects against a + // non-compliant proxy emitting a signature before any thinking text for + // its episode. Signature fragments are concatenated (not "first seen") + // because a long signature can legitimately arrive split across + // multiple signature_delta events. + // + // Known limitation: two back-to-back thought parts with NO signature at + // all and no intervening non-thought part still merge into one episode + // -- neither boundary condition above can fire without a signature to + // test. This is consistent with both wires' documented invariant that + // every episode ends in a signature-only chunk; it is not reachable via + // Anthropic interleaved thinking or OpenAI Responses reasoning items as + // implemented, but would misattribute text across episodes if a + // non-compliant proxy ever dropped a signature entirely. + // + // Responses emits one complete JSON {id, encrypted_content} payload at + // output_item.done. Close that episode immediately, even without summary + // text; unlike Anthropic signature_delta fragments, it must never be + // concatenated with the next reasoning item's payload. + const consolidatedHistoryParts: Part[] = []; + let openEpisodeText = ''; + let openEpisodeSignature = ''; + let hasOpenEpisode = false; + + const flushThoughtEpisode = () => { + if (!hasOpenEpisode) return; + const text = openEpisodeText.trim(); + // A signature-only episode (no text) is kept, not dropped: it is + // still potentially replayable per Anthropic's spec, and this is + // the ACTIVE (latest) turn's thinking, which must replay byte-exact + // -- unlike converter.ts's dropEmptyTextThinkingBlocks, which drops + // this same empty-text shape but only from non-latest turns, where + // the rationale is that prior-turn thinking is disposable, not that + // an empty-text signed block is inherently invalid. + if (text !== '' || openEpisodeSignature !== '') { + const episodePart: Part = { text, thought: true }; + if (openEpisodeSignature) { + episodePart.thoughtSignature = openEpisodeSignature; + } + consolidatedHistoryParts.push(episodePart); + } + openEpisodeText = ''; + openEpisodeSignature = ''; + hasOpenEpisode = false; + }; + + for (const part of allModelParts) { + if (part.thought) { + const partText = typeof part.text === 'string' ? part.text : ''; + if ( + hasOpenEpisode && + partText !== '' && + openEpisodeText.length > 0 && + openEpisodeSignature !== '' + ) { + flushThoughtEpisode(); + } + hasOpenEpisode = true; + openEpisodeText += partText; + if (part.thoughtSignature) { + openEpisodeSignature += part.thoughtSignature; + if (isCompleteResponsesReasoningSignature(part.thoughtSignature)) { + flushThoughtEpisode(); + } + } + continue; + } + flushThoughtEpisode(); + const lastPart = + consolidatedHistoryParts[consolidatedHistoryParts.length - 1]; + if ( + lastPart?.text && + isValidNonThoughtTextPart(lastPart) && + isValidNonThoughtTextPart(part) + ) { + lastPart.text += part.text; + } else if (isValidContentPart(part)) { + consolidatedHistoryParts.push(part); + } + } + flushThoughtEpisode(); + + return consolidatedHistoryParts; +} + function shouldStopAfterHardRescue( shouldForceFromHard: boolean, hardLimit: number, @@ -3210,6 +3311,9 @@ export class LlmChat { // eslint-disable-next-line @typescript-eslint/no-this-alias const self = this; return (async function* () { + let successfulRecoveries = 0; + let activeRecoveryUser: Content | undefined; + let pendingTransportPrefix = ''; const sleepInhibitorHandle = acquireSleepInhibitor( self.config, 'Qwen Code is streaming a model response', @@ -3367,6 +3471,7 @@ export class LlmChat { transportContinuationText = ''; transportAttemptText = ''; transportContinuationPrefix = ''; + pendingTransportPrefix = ''; }; // Fold the running attempt's text into the accumulated buffer, @@ -3435,6 +3540,8 @@ export class LlmChat { acceptQuietToolResultCompletion, ); + // The processor now owns cancellation persistence for this prefix. + pendingTransportPrefix = ''; lastFinishReason = undefined; for await (const chunk of stream) { if (hasCandidateOutput(chunk)) { @@ -3471,6 +3578,7 @@ export class LlmChat { break; } catch (error) { lastError = error; + if (params.config?.abortSignal?.aborted) throw error; // This attempt is over; fold what it delivered into the running // buffer before any branch below reads it. Doing this here rather // than per chunk keeps the overlap scan anchored at the attempt @@ -3547,6 +3655,7 @@ export class LlmChat { delayMs, params.config?.abortSignal, ); + resetTransportContinuation(); yield { type: StreamEventType.RETRY, retryInfo: { @@ -3658,6 +3767,7 @@ export class LlmChat { // was folded in at the catch above with its replayed overlap // stripped, so this carries no fragment twice. transportContinuationPrefix = transportContinuationText; + pendingTransportPrefix = transportContinuationPrefix; const delayMs = TRANSPORT_STREAM_RETRY_CONFIG.initialDelayMs * transportContinuationCount; @@ -3760,6 +3870,10 @@ export class LlmChat { ? 'Request body rejected with HTTP 413; attempting reactive compression.' : 'Context length exceeded; attempting reactive compression.', ); + // The failed text stream no longer owns cancellation recording. + if (!streamYieldedFunctionCall) { + pendingTransportPrefix = transportContinuationText; + } try { const reactiveInfo = await self.tryCompress( prompt_id, @@ -3824,12 +3938,10 @@ export class LlmChat { type: StreamEventType.COMPRESSED, info: reactiveInfo, }; - yield { type: StreamEventType.RETRY }; - // Compression rebuilt `requestContents` from scratch, so - // any continuation staged against the old contents is - // stale — and the RETRY above already told the UI to drop - // the delivered text. + // Drop the stale continuation before telling the UI to + // clear it: the consumer may cancel at the RETRY yield. resetTransportContinuation(); + yield { type: StreamEventType.RETRY }; suppressNextRetryEvent = true; rearmQuietAcceptanceIfBudgetSpent(); continue; @@ -4099,6 +4211,7 @@ export class LlmChat { } return; } catch (error) { + if (attemptState.params.config?.abortSignal?.aborted) throw error; attemptState.rollback(); if (!(error instanceof InvalidStreamError)) throw error; @@ -4216,7 +4329,6 @@ export class LlmChat { // response in history and inject a recovery message so the model can // continue from where it left off. let recoveryCount = 0; - let successfulRecoveries = 0; while ( recoveryFinishReason === FinishReason.MAX_TOKENS && recoveryCount < MAX_OUTPUT_RECOVERY_ATTEMPTS @@ -4323,6 +4435,7 @@ export class LlmChat { for await (const event of streamWithInvalidStreamRetries( () => { self.history.push(recoveryUserContent); + activeRecoveryUser = recoveryUserContent; return { requestContents: self.getRequestHistoryForRoute( currentUserContent, @@ -4346,7 +4459,9 @@ export class LlmChat { // the model continuation turn are now in history and can be // coalesced back into the preceding model entry after the loop. successfulRecoveries++; + activeRecoveryUser = undefined; } catch (recoveryError) { + if (params.config?.abortSignal?.aborted) throw recoveryError; rollbackRecoveryAttempt(); debugLogger.warn( `Recovery attempt ${recoveryCount} failed: ${recoveryError}`, @@ -4370,15 +4485,6 @@ export class LlmChat { break; } } - - // Coalesce completed recovery pairs back into the preceding model - // turn so the OUTPUT_RECOVERY_MESSAGE control prompt does not - // persist as a synthetic user turn in durable history. The user - // never sent that message, and leaving it in history would bias - // later turns and pollute compression / replay / export. - if (successfulRecoveries > 0) { - self.coalesceRecoveryPairs(successfulRecoveries); - } } if (lastError) { @@ -4551,7 +4657,12 @@ export class LlmChat { ); return; } catch (fallbackError) { - if (isAbortError(fallbackError)) throw fallbackError; + if ( + params.config?.abortSignal?.aborted || + isAbortError(fallbackError) + ) { + throw fallbackError; + } lastError = fallbackError; if (currentFallbackYieldedAnyChunk) { @@ -4637,6 +4748,36 @@ export class LlmChat { } } } finally { + // Between attempts there is no response processor to save visible text. + if (params.config?.abortSignal?.aborted && pendingTransportPrefix) { + const parts = [{ text: pendingTransportPrefix }]; + self.history.push({ role: 'model', parts }); + self.pendingPartialAssistantTurnIndex = self.history.length - 1; + self.pendingPartialAssistantRecord = { + model, + message: parts, + contextWindowSize: + self.config.getContentGeneratorConfig()?.contextWindowSize, + ...(turnGoalContext ? { goalContext: { ...turnGoalContext } } : {}), + }; + } + // Also clean up a recovery abandoned at a yield, before its success + // counter advances. Preserve the continuation, not its control prompt. + const recoveryIndex = activeRecoveryUser + ? self.history.indexOf(activeRecoveryUser) + : -1; + if (recoveryIndex === self.history.length - 1 && recoveryIndex >= 0) { + self.history.pop(); + } else if ( + recoveryIndex >= 0 && + recoveryIndex === self.history.length - 2 && + self.history.at(-1)?.role === 'model' + ) { + successfulRecoveries++; + } + if (successfulRecoveries > 0) { + self.coalesceRecoveryPairs(successfulRecoveries); + } sleepInhibitorHandle.release(); streamDoneResolver!(); // Flush any deferred partial-tool_use record. Covers both the @@ -4771,6 +4912,7 @@ export class LlmChat { goalContext, transportContinuationPrefix, acceptQuietToolResultCompletion, + params.config?.abortSignal, ); } @@ -5316,6 +5458,7 @@ export class LlmChat { goalContext?: GoalTurnPermit, transportContinuationPrefix?: string, acceptQuietToolResultCompletion = false, + abortSignal?: AbortSignal, ): AsyncGenerator { // Collect ALL parts from the model response (including thoughts for recording) const allModelParts: Part[] = []; @@ -5583,8 +5726,47 @@ export class LlmChat { } } catch (e) { streamError = e; + } finally { + // Cancellation can close the generator at a yield, skipping everything + // after this finally. Keep the delivered partial in both history and JSONL. + if (abortSignal?.aborted) { + const parts = consolidateModelResponseParts(allModelParts); + dropDanglingUnsignedTrailingThought(parts, hasToolCall); + if (transportContinuationPrefix) { + const textPart = parts.find(isPlainTextPart); + if (textPart) { + textPart.text = mergeDeliveredPrefix( + transportContinuationPrefix, + textPart.text, + ); + } else { + parts.push({ text: transportContinuationPrefix }); + } + } + if (parts.length > 0) { + this.history.push({ role: 'model', parts }); + this.pendingPartialAssistantTurnIndex = this.history.length - 1; + this.pendingPartialAssistantRecord = { + model, + message: parts.map( + (part) => redactStructuredOutputArgsForRecording(part) ?? part, + ), + tokens: coercedUsage + ? { ...usageMetadata, ...coercedUsage } + : usageMetadata, + contextWindowSize: + this.config.getContentGeneratorConfig()?.contextWindowSize, + ...(goalContext ? { goalContext: { ...goalContext } } : {}), + }; + } + } } + if (abortSignal?.aborted && streamError !== null) { + throw streamError; + } + abortSignal?.throwIfAborted(); + let pendingProtocolChunk: GenerateContentResponse | undefined; if ( streamError === null && pendingProtocolParts.length > 0 && @@ -5607,106 +5789,12 @@ export class LlmChat { syncFunctionCallsField(chunk, parts); hasToolCall ||= parts.some((part) => part.functionCall); allModelParts.push(...parts); - yield chunk; + pendingProtocolChunk = chunk; } } - // A turn can legitimately contain multiple distinct reasoning episodes - // separated by tool calls (Anthropic interleaved thinking, OpenAI - // Responses reasoning items on parallel function calls). Each episode - // must keep its own signature and its own position relative to the - // tool calls it preceded -- merging every thought-flagged part into one - // blob and keeping only the first signature silently discards every - // other episode's replayable payload and destroys the interleaving. - // - // Both wires terminate an episode with a text-less, signature-only - // chunk (anthropicContentGenerator.ts's signature_delta handling; - // responses-converter.ts's output_item.done for a reasoning item), so a - // thought part carrying fresh non-empty text while the open episode - // already has both accumulated text and a signature can only be the - // start of a new episode -- no legitimate continuation of the same - // episode reintroduces text after its signature is set. The - // `openEpisodeText.length > 0` guard additionally protects against a - // non-compliant proxy emitting a signature before any thinking text for - // its episode. Signature fragments are concatenated (not "first seen") - // because a long signature can legitimately arrive split across - // multiple signature_delta events. - // - // Known limitation: two back-to-back thought parts with NO signature at - // all and no intervening non-thought part still merge into one episode - // -- neither boundary condition above can fire without a signature to - // test. This is consistent with both wires' documented invariant that - // every episode ends in a signature-only chunk; it is not reachable via - // Anthropic interleaved thinking or OpenAI Responses reasoning items as - // implemented, but would misattribute text across episodes if a - // non-compliant proxy ever dropped a signature entirely. - // - // Responses emits one complete JSON {id, encrypted_content} payload at - // output_item.done. Close that episode immediately, even without summary - // text; unlike Anthropic signature_delta fragments, it must never be - // concatenated with the next reasoning item's payload. - const consolidatedHistoryParts: Part[] = []; - let openEpisodeText = ''; - let openEpisodeSignature = ''; - let hasOpenEpisode = false; - - const flushThoughtEpisode = () => { - if (!hasOpenEpisode) return; - const text = openEpisodeText.trim(); - // A signature-only episode (no text) is kept, not dropped: it is - // still potentially replayable per Anthropic's spec, and this is - // the ACTIVE (latest) turn's thinking, which must replay byte-exact - // -- unlike converter.ts's dropEmptyTextThinkingBlocks, which drops - // this same empty-text shape but only from non-latest turns, where - // the rationale is that prior-turn thinking is disposable, not that - // an empty-text signed block is inherently invalid. - if (text !== '' || openEpisodeSignature !== '') { - const episodePart: Part = { text, thought: true }; - if (openEpisodeSignature) { - episodePart.thoughtSignature = openEpisodeSignature; - } - consolidatedHistoryParts.push(episodePart); - } - openEpisodeText = ''; - openEpisodeSignature = ''; - hasOpenEpisode = false; - }; - - for (const part of allModelParts) { - if (part.thought) { - const partText = typeof part.text === 'string' ? part.text : ''; - if ( - hasOpenEpisode && - partText !== '' && - openEpisodeText.length > 0 && - openEpisodeSignature !== '' - ) { - flushThoughtEpisode(); - } - hasOpenEpisode = true; - openEpisodeText += partText; - if (part.thoughtSignature) { - openEpisodeSignature += part.thoughtSignature; - if (isCompleteResponsesReasoningSignature(part.thoughtSignature)) { - flushThoughtEpisode(); - } - } - continue; - } - flushThoughtEpisode(); - const lastPart = - consolidatedHistoryParts[consolidatedHistoryParts.length - 1]; - if ( - lastPart?.text && - isValidNonThoughtTextPart(lastPart) && - isValidNonThoughtTextPart(part) - ) { - lastPart.text += part.text; - } else if (isValidContentPart(part)) { - consolidatedHistoryParts.push(part); - } - } - flushThoughtEpisode(); + const consolidatedHistoryParts = + consolidateModelResponseParts(allModelParts); // A thought episode can be flushed while still incomplete if the // stream is cut off before its terminating signature-only chunk @@ -5944,10 +6032,6 @@ export class LlmChat { } } - if (recoveredChunk) { - yield recoveredChunk; - } - // Record assistant turn with raw Content and metadata. Gate matches // the in-memory `this.history.push` decision below so chat-recording // JSONL never carries a partial turn we deliberately dropped from @@ -6144,6 +6228,11 @@ export class LlmChat { role: 'model', parts: acceptedTurnParts, }); + // Persist before these synthetic yields: the consumer may cancel and + // close the generator immediately after receiving a tool call. + if (pendingProtocolChunk) yield pendingProtocolChunk; + if (recoveredChunk) yield recoveredChunk; + abortSignal?.throwIfAborted(); if (deferredFinishReason) { yield { candidates: [{ finishReason: deferredFinishReason }], @@ -6205,6 +6294,9 @@ export class LlmChat { ); // Drop the (userRecovery, modelContinuation) pair. this.history.splice(len - 2, 2); + if (this.pendingPartialAssistantTurnIndex === len - 1) { + this.pendingPartialAssistantTurnIndex = len - 3; + } } } } diff --git a/packages/core/src/services/chatRecordingService.test.ts b/packages/core/src/services/chatRecordingService.test.ts index 939a708b9f9..4eac5ddda4a 100644 --- a/packages/core/src/services/chatRecordingService.test.ts +++ b/packages/core/src/services/chatRecordingService.test.ts @@ -1523,6 +1523,24 @@ describe('ChatRecordingService', () => { ).toBe(false); }); + it.each([ + [undefined, true], + [1_500, true], + [NaN, false], + [Infinity, false], + ['1500', false], + ])('validates cancellation timestamp %s', (cancelledAt, valid) => { + expect( + isTurnResultRecordPayload({ + promptId: 'prompt-1', + state: 'cancelled', + startedAt: 1_000, + cancelledAt, + endedAt: 2_000, + }), + ).toBe(valid); + }); + it('caps promptId, stopReason, and originatorClientId in turn_result payloads', () => { const oversized = 'x'.repeat(TURN_RESULT_IDENTIFIER_MAX_CHARS + 1); expect( diff --git a/packages/core/src/services/chatRecordingService.ts b/packages/core/src/services/chatRecordingService.ts index c50f2421b13..68b15968517 100644 --- a/packages/core/src/services/chatRecordingService.ts +++ b/packages/core/src/services/chatRecordingService.ts @@ -771,6 +771,8 @@ export interface TurnResultRecordPayload { error?: TurnResultErrorPayload; /** Epoch ms the turn started executing (agent clock). */ startedAt?: number; + /** Epoch ms the user-cancel signal was received (agent clock). */ + cancelledAt?: number; /** Epoch ms the turn settled (agent clock). */ endedAt: number; promptText?: string; @@ -812,6 +814,7 @@ export function isTurnResultRecordPayload( if ( !optionalString('stopReason', TURN_RESULT_IDENTIFIER_MAX_CHARS) || !optionalTimestamp('startedAt') || + !optionalTimestamp('cancelledAt') || !optionalString('promptText', TURN_RESULT_TEXT_MAX_CHARS) || !optionalBoolean('promptTextTruncated') || !optionalString('resultText', TURN_RESULT_TEXT_MAX_CHARS) || diff --git a/packages/sdk-typescript/src/daemon/ui/normalizer.ts b/packages/sdk-typescript/src/daemon/ui/normalizer.ts index 627aaf7922e..03f1bf6c9e8 100644 --- a/packages/sdk-typescript/src/daemon/ui/normalizer.ts +++ b/packages/sdk-typescript/src/daemon/ui/normalizer.ts @@ -256,6 +256,14 @@ export function normalizeDaemonEvent( ]; } + case 'turn_complete': + return getString(event.data, 'stopReason') === 'cancelled' + ? normalizePromptCancellation( + isRecord(event.data) ? event.data['promptCancelled'] : undefined, + base, + ) + : []; + case 'followup_suggestion': return normalizeFollowupSuggestion(event, base); @@ -917,6 +925,9 @@ function normalizeSessionUpdate( const parentToolCallId = extractParentToolCallId(update); const meta = extractUpdateMeta(update); const events: DaemonUiEvent[] = []; + if (!parentToolCallId && meta?.['promptCancelled'] !== undefined) { + return normalizePromptCancellation(meta['promptCancelled'], base); + } if (text) { events.push({ ...base, @@ -2168,6 +2179,31 @@ function normalizeAuthDeviceFlowCancelled( return [{ ...base, type: 'auth.device_flow.cancelled', deviceFlowId }]; } +function normalizePromptCancellation( + value: unknown, + base: NormalizedEventBase, +): DaemonUiEvent[] { + const elapsedMs = numberField(value, 'elapsedMs'); + const cancelledAt = numberField(value, 'cancelledAt'); + const promptId = stringField(value, 'promptId') ?? base.promptId; + if ( + !promptId || + elapsedMs === undefined || + elapsedMs < 0 || + cancelledAt === undefined + ) + return []; + return [ + { + ...base, + type: 'prompt.cancelled', + promptId, + elapsedMs, + serverTimestamp: cancelledAt, + }, + ]; +} + function numberField(value: unknown, key: string): number | undefined { if (!isRecord(value)) return undefined; const v = value[key]; diff --git a/packages/sdk-typescript/src/daemon/ui/transcript.ts b/packages/sdk-typescript/src/daemon/ui/transcript.ts index 3dfb61ad557..89d65539010 100644 --- a/packages/sdk-typescript/src/daemon/ui/transcript.ts +++ b/packages/sdk-typescript/src/daemon/ui/transcript.ts @@ -1669,12 +1669,38 @@ function appendPromptCancelledBlock( state: DaemonTranscriptState, event: Extract, ): void { + const existing = event.promptId + ? state.blocks.find( + (block) => + block.kind === 'prompt_cancelled' && + block.promptId === event.promptId, + ) + : undefined; + if (existing) { + if (event.elapsedMs === undefined) return; + const block = getWritableBlockById(state, existing.id); + if (block?.kind !== 'prompt_cancelled') return; + const bytesBefore = estimateBlockBytes(block); + block.elapsedMs = event.elapsedMs; + if (event.serverTimestamp !== undefined) { + block.serverTimestamp = event.serverTimestamp; + } + if (event.sourceRecordIds) block.sourceRecordIds = event.sourceRecordIds; + block.updatedAt = state.now; + state.retainedBytes += estimateBlockBytes(block) - bytesBefore; + return; + } const block: DaemonPromptCancelledTranscriptBlock = { id: allocateBlockId(state, 'prompt_cancelled'), kind: 'prompt_cancelled', clientReceivedAt: state.now, createdAt: state.now, updatedAt: state.now, + ...(event.promptId ? { promptId: event.promptId } : {}), + ...(event.elapsedMs !== undefined ? { elapsedMs: event.elapsedMs } : {}), + ...(event.sourceRecordIds + ? { sourceRecordIds: event.sourceRecordIds } + : {}), ...(event.reason ? { reason: event.reason } : {}), ...(event.eventId !== undefined ? { eventId: event.eventId } : {}), ...(event.serverTimestamp !== undefined diff --git a/packages/sdk-typescript/src/daemon/ui/types.ts b/packages/sdk-typescript/src/daemon/ui/types.ts index 4cb57add876..e61b876148e 100644 --- a/packages/sdk-typescript/src/daemon/ui/types.ts +++ b/packages/sdk-typescript/src/daemon/ui/types.ts @@ -483,6 +483,8 @@ export interface DaemonUiStateResyncRequiredEvent extends DaemonUiEventBase { */ export interface DaemonUiPromptCancelledEvent extends DaemonUiEventBase { type: 'prompt.cancelled'; + /** Execution time before explicit cancellation, excluding cleanup. */ + elapsedMs?: number; /** * Why the turn was cancelled. Absent for a user-initiated cancel; * `'forward_failed'` when the daemon synthesized the cancel because the @@ -1106,6 +1108,7 @@ export interface DaemonPromptCancelledTranscriptBlock extends DaemonTranscriptBlockBase { kind: 'prompt_cancelled'; reason?: string; + elapsedMs?: number; } export type DaemonTranscriptBlock = diff --git a/packages/sdk-typescript/test/unit/daemon-transcript-projection.test.ts b/packages/sdk-typescript/test/unit/daemon-transcript-projection.test.ts index c50e50fef2b..d11da1d1b17 100644 --- a/packages/sdk-typescript/test/unit/daemon-transcript-projection.test.ts +++ b/packages/sdk-typescript/test/unit/daemon-transcript-projection.test.ts @@ -27,6 +27,56 @@ function record( } describe('projectChatRecordsToDaemonTranscript', () => { + it('restores explicit cancellation timing from the persisted turn result', () => { + const projection = projectChatRecordsToDaemonTranscript([ + record('user-1', null), + record('cancel-1', 'user-1', { + type: 'system', + subtype: 'turn_result', + message: undefined, + systemPayload: { + promptId: 'p1', + state: 'cancelled', + startedAt: 1000, + cancelledAt: 11999, + endedAt: 20000, + }, + }), + ]); + expect(projection.complete).toBe(true); + expect(projection.blocks).toHaveLength(2); + expect(projection.blocks[1]).toMatchObject({ + kind: 'prompt_cancelled', + promptId: 'p1', + elapsedMs: 10999, + serverTimestamp: 11999, + sourceRecordIds: ['cancel-1'], + }); + }); + + it.each([ + { state: 'completed', cancelledAt: 2000 }, + { state: 'error', cancelledAt: 2000 }, + { state: 'cancelled' }, + { state: 'cancelled', cancelledAt: '2000' }, + { state: 'cancelled', cancelledAt: 2000, startedAt: '1000' }, + ])('does not invent a user cancellation for %j', (payload) => { + const projection = projectChatRecordsToDaemonTranscript([ + record('cancel-1', null, { + type: 'system', + subtype: 'turn_result', + message: undefined, + systemPayload: { + promptId: 'p1', + startedAt: 1000, + endedAt: 3000, + ...payload, + }, + }), + ]); + expect(projection.blocks).toEqual([]); + }); + it('projects the active branch with deterministic record boundaries', () => { const records = [ record('root', null, { diff --git a/packages/sdk-typescript/test/unit/daemonUi.test.ts b/packages/sdk-typescript/test/unit/daemonUi.test.ts index 81d99eb7da2..f0113cb3319 100644 --- a/packages/sdk-typescript/test/unit/daemonUi.test.ts +++ b/packages/sdk-typescript/test/unit/daemonUi.test.ts @@ -8520,6 +8520,76 @@ describe('R7 review batch — markdown escape + details sanitization', () => { }); describe('cross-client event recognition (prompt_cancelled / replay_complete)', () => { + it('merges authoritative cancellation timing without changing the previous snapshot', () => { + const [provisional] = normalizeDaemonEvent({ + v: 1, + type: 'prompt_cancelled', + promptId: 'p1', + data: {}, + }); + expect(provisional).toMatchObject({ + type: 'prompt.cancelled', + promptId: 'p1', + }); + const before = reduceDaemonTranscriptEvents(createDaemonTranscriptState(), [ + provisional, + ]); + const timing = normalizeDaemonEvent({ + v: 1, + type: 'turn_complete', + promptId: 'p1', + data: { + stopReason: 'cancelled', + promptCancelled: { elapsedMs: 10999, cancelledAt: 12000 }, + }, + }); + let after = reduceDaemonTranscriptEvents(before, timing); + after = reduceDaemonTranscriptEvents(after, [provisional, ...timing]); + expect(before.blocks).toHaveLength(1); + expect(before.blocks[0]).not.toHaveProperty('elapsedMs'); + expect(after.blocks).toEqual([ + expect.objectContaining({ + id: before.blocks[0].id, + kind: 'prompt_cancelled', + promptId: 'p1', + elapsedMs: 10999, + serverTimestamp: 12000, + }), + ]); + const enriched = reduceDaemonTranscriptEvents(after, [ + { type: 'prompt.cancelled', promptId: 'p1', elapsedMs: 11000 }, + ]); + expect(enriched.blocks[0]).toMatchObject({ + serverTimestamp: 12000, + elapsedMs: 11000, + }); + expect(after.blocks[0]).toMatchObject({ elapsedMs: 10999 }); + after = reduceDaemonTranscriptEvents(after, [ + { ...timing[0], promptId: 'p2' }, + ]); + expect(after.blocks).toHaveLength(2); + }); + + it.each([ + { elapsedMs: -1, cancelledAt: 12000 }, + { elapsedMs: NaN, cancelledAt: 12000 }, + { elapsedMs: 1000, cancelledAt: Infinity }, + { elapsedMs: '1000', cancelledAt: 12000 }, + undefined, + ])( + 'ignores malformed or legacy cancellation timing %j', + (promptCancelled) => { + expect( + normalizeDaemonEvent({ + v: 1, + type: 'turn_complete', + promptId: 'p1', + data: { stopReason: 'cancelled', promptCancelled }, + }), + ).toEqual([]); + }, + ); + it('normalizes prompt_cancelled to prompt.cancelled (not debug)', () => { const events = normalizeDaemonEvent({ id: 1, @@ -8538,6 +8608,34 @@ describe('cross-client event recognition (prompt_cancelled / replay_complete)', expect(events[0]).not.toHaveProperty('reason'); }); + it('uses the persisted cancellation identity when replayed during another prompt', () => { + const events = normalizeDaemonEvent({ + v: 1, + type: 'session_update', + promptId: 'current-prompt', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: '' }, + _meta: { + promptCancelled: { + promptId: 'cancelled-prompt', + cancelledAt: 1000, + elapsedMs: 0, + }, + }, + }, + }, + }); + expect(events).toEqual([ + expect.objectContaining({ + type: 'prompt.cancelled', + promptId: 'cancelled-prompt', + elapsedMs: 0, + }), + ]); + }); + it('forwards the prompt_cancelled reason (C3 forward_failed)', () => { const events = normalizeDaemonEvent({ id: 1, diff --git a/packages/web-shell/client/adapters/transcriptToMessages.test.ts b/packages/web-shell/client/adapters/transcriptToMessages.test.ts index 74a4f2f56f1..e496e7ceac5 100644 --- a/packages/web-shell/client/adapters/transcriptToMessages.test.ts +++ b/packages/web-shell/client/adapters/transcriptToMessages.test.ts @@ -4242,6 +4242,20 @@ describe('transcriptBlocksToDaemonMessages', () => { ]); }); + it('preserves cancellation duration for rendering', () => { + const messages = transcriptBlocksToDaemonMessages([ + { + ...promptCancelledBlock('cancel-1', 20), + elapsedMs: 10999, + promptId: 'p1', + }, + ]); + expect(messages[0]).toMatchObject({ + source: 'prompt_cancelled', + data: { elapsedMs: 10999 }, + }); + }); + it('renders localized prompt cancellation messages', () => { const messages = transcriptBlocksToDaemonMessages( [promptCancelledBlock('cancel-1', 20)], diff --git a/packages/web-shell/client/adapters/transcriptToMessages.ts b/packages/web-shell/client/adapters/transcriptToMessages.ts index ea90cac86c4..5574626c20c 100644 --- a/packages/web-shell/client/adapters/transcriptToMessages.ts +++ b/packages/web-shell/client/adapters/transcriptToMessages.ts @@ -1018,6 +1018,9 @@ export function transcriptBlocksToDaemonMessages( content: promptCancelledText, variant: 'info', source: 'prompt_cancelled', + ...(block.elapsedMs !== undefined + ? { data: { elapsedMs: block.elapsedMs } } + : {}), timestamp: blockTime, sourceBlockIds: [block.id], }); diff --git a/packages/web-shell/client/components/MessageList.dom.test.tsx b/packages/web-shell/client/components/MessageList.dom.test.tsx index ae01d8c0ff8..1aa306ac051 100644 --- a/packages/web-shell/client/components/MessageList.dom.test.tsx +++ b/packages/web-shell/client/components/MessageList.dom.test.tsx @@ -4407,6 +4407,31 @@ describe('MessageList — turn collapse (DOM)', () => { expect(parallelAgentsSummary(c)?.hasAttribute('aria-disabled')).toBe(false); }); + it.each([ + [7069, '8s'], + [0, '0s'], + ['1000', '14s'], + [-1, '14s'], + [NaN, '14s'], + [Infinity, '14s'], + ] as const)( + 'uses valid cancellation duration %s for the processed summary', + (elapsedMs, expected) => { + const c = mount([ + { ...userMsg('u1'), timestamp: 1_000 }, + { + id: 't1', + role: 'thinking', + content: 'thinking before cancel', + timestamp: 2_000, + }, + { ...asstMsg('a1'), timestamp: 3_000 }, + { ...systemMsg('c1'), timestamp: 15_000, data: { elapsedMs } }, + ]); + expect(c.textContent).toContain(`Processed ${expected}`); + }, + ); + it('renders collapse metrics in the standalone turn row', () => { const c = mount([ { ...userMsg('u1'), timestamp: 1_000 }, diff --git a/packages/web-shell/client/components/MessageList.tsx b/packages/web-shell/client/components/MessageList.tsx index bfa8f9de111..b49d1ed8195 100644 --- a/packages/web-shell/client/components/MessageList.tsx +++ b/packages/web-shell/client/components/MessageList.tsx @@ -1941,6 +1941,7 @@ export function applyTurnCollapse( const answerIdx = findFinalAnswerIndex(items, start, end); let hiddenCount = 0; let terminalTs: number | undefined; + let cancelledElapsedMs: number | undefined; let assistantTs: number | undefined; let inputTokens = 0; let outputTokens = 0; @@ -1970,6 +1971,24 @@ export function applyTurnCollapse( // Compact mode folds thinking into tool summaries; count it too. thinkingCount += item.message.thoughts.length; } + if ( + item.type === 'message' && + item.message.role === 'system' && + item.message.source === 'prompt_cancelled' + ) { + const data = item.message.data; + const elapsed = + data && typeof data === 'object' && 'elapsedMs' in data + ? data.elapsedMs + : undefined; + if ( + typeof elapsed === 'number' && + Number.isFinite(elapsed) && + elapsed >= 0 + ) { + cancelledElapsedMs = elapsed; + } + } const terminalTimestamp = terminalTurnTimestamp(item); if (terminalTimestamp !== undefined) { terminalTs = @@ -2009,11 +2028,12 @@ export function applyTurnCollapse( : undefined; const lastStepTs = terminalTs ?? assistantTs; const elapsedMs = - promptTs !== undefined && + cancelledElapsedMs ?? + (promptTs !== undefined && lastStepTs !== undefined && lastStepTs >= promptTs ? lastStepTs - promptTs - : undefined; + : undefined); const hasMetrics = hasUsage || elapsedMs !== undefined || liveStartedAt !== undefined; @@ -2250,7 +2270,7 @@ type Translate = ( ) => string; function durationMetricText(elapsedMs: number | undefined): string { - return elapsedMs !== undefined && elapsedMs > 0 + return elapsedMs !== undefined && elapsedMs >= 0 ? formatDuration(elapsedMs) : ''; } diff --git a/packages/web-shell/client/components/messages/SystemMessage.test.tsx b/packages/web-shell/client/components/messages/SystemMessage.test.tsx index 97e731d2836..825218fc36a 100644 --- a/packages/web-shell/client/components/messages/SystemMessage.test.tsx +++ b/packages/web-shell/client/components/messages/SystemMessage.test.tsx @@ -32,6 +32,43 @@ function render(node: ReactNode, language: 'en' | 'zh-CN' = 'en'): HTMLElement { } describe('SystemMessage — prompt_cancelled marker', () => { + it.each([ + ['zh-CN', 10999, '你在 11 秒后取消了请求'], + ['zh-CN', 999, '你在 1 秒后取消了请求'], + ['zh-CN', 0, '你在 0 秒后取消了请求'], + ['zh-CN', 7069, '你在 8 秒后取消了请求'], + ['en', 10999, 'You cancelled this request after 11 seconds'], + ] as const)('renders elapsed seconds in %s', (language, elapsedMs, text) => { + const container = render( + , + language, + ); + expect(container.querySelector('[role="status"]')?.textContent).toBe(text); + }); + + it.each(['1000', -1, NaN, Infinity])( + 'falls back for invalid elapsed time %s', + (elapsedMs) => { + const container = render( + , + 'zh-CN', + ); + expect(container.querySelector('[role="status"]')?.textContent).toBe( + '你已取消请求', + ); + }, + ); + it('renders the user-cancelled marker as a status region', () => { const container = render( , diff --git a/packages/web-shell/client/components/messages/SystemMessage.tsx b/packages/web-shell/client/components/messages/SystemMessage.tsx index d370081b786..8e72d440168 100644 --- a/packages/web-shell/client/components/messages/SystemMessage.tsx +++ b/packages/web-shell/client/components/messages/SystemMessage.tsx @@ -135,9 +135,19 @@ export const SystemMessage = memo(function SystemMessage({ // The user ESC-cancelled a live stream. Render it right-aligned and subtle — // a user-initiated stop reads as belonging to the user side of the transcript. if (source === 'prompt_cancelled') { + const elapsedMs = + data && typeof data === 'object' && 'elapsedMs' in data + ? data.elapsedMs + : undefined; return (
- {t('turn.stopped')} + + {typeof elapsedMs === 'number' && + Number.isFinite(elapsedMs) && + elapsedMs >= 0 + ? t('turn.stoppedAfter', { seconds: Math.ceil(elapsedMs / 1000) }) + : t('turn.stopped')} +
); } diff --git a/packages/web-shell/client/components/sidebar/WebShellSidebar.module.css b/packages/web-shell/client/components/sidebar/WebShellSidebar.module.css index c09b0f0ab64..9bd112a8ef4 100644 --- a/packages/web-shell/client/components/sidebar/WebShellSidebar.module.css +++ b/packages/web-shell/client/components/sidebar/WebShellSidebar.module.css @@ -1042,11 +1042,6 @@ color: var(--success-color, #3fb950); } -.sessionStatusUnknown { - color: var(--muted-foreground); - font-size: 11px; -} - .sessionSourceIcon { width: 28px; height: 28px; diff --git a/packages/web-shell/client/components/sidebar/WebShellSidebar.tsx b/packages/web-shell/client/components/sidebar/WebShellSidebar.tsx index 50cad2f1b9c..540084ffe60 100644 --- a/packages/web-shell/client/components/sidebar/WebShellSidebar.tsx +++ b/packages/web-shell/client/components/sidebar/WebShellSidebar.tsx @@ -4398,8 +4398,6 @@ export function WebShellSidebar({ const isCurrent = standalone?.active ?? isCurrentSession(session); const sessionWorkActive = !session.hasActivePrompt && session.activeWorkState === 'active'; - const activityUnknown = - !session.hasActivePrompt && session.activeWorkState === 'unknown'; // Archiving closes the live session daemon-side, which would end the // running work; keep the action visible but inert while it runs. const running = Boolean(session.hasActivePrompt || sessionWorkActive); @@ -4500,13 +4498,6 @@ export function WebShellSidebar({ data-web-shell-session-active-work aria-hidden="true" /> - ) : activityUnknown && !completedUnread ? ( - - ? - ) : null} {isEditing && showRename ? ( diff --git a/packages/web-shell/client/components/sidebar/WebShellSidebar.workspace-removal.test.tsx b/packages/web-shell/client/components/sidebar/WebShellSidebar.workspace-removal.test.tsx index 83c584c2f63..4c205aadadc 100644 --- a/packages/web-shell/client/components/sidebar/WebShellSidebar.workspace-removal.test.tsx +++ b/packages/web-shell/client/components/sidebar/WebShellSidebar.workspace-removal.test.tsx @@ -4818,7 +4818,9 @@ describe('WebShellSidebar session source switch', () => { expect( row?.querySelector('[data-web-shell-scheduled-task-session]'), ).toBeTruthy(); - expect(row?.querySelector(scenario.selector)).toBeTruthy(); + expect(Boolean(row?.querySelector(scenario.selector))).toBe( + scenario.activeWorkState === 'active', + ); }); it('keeps the scheduled-task marker when a run is grouped by color', async () => { diff --git a/packages/web-shell/client/daemon/session/DaemonSessionProvider.test.tsx b/packages/web-shell/client/daemon/session/DaemonSessionProvider.test.tsx index 03e831e3407..57c2d4f027c 100644 --- a/packages/web-shell/client/daemon/session/DaemonSessionProvider.test.tsx +++ b/packages/web-shell/client/daemon/session/DaemonSessionProvider.test.tsx @@ -7391,6 +7391,51 @@ describe('DaemonSessionProvider', () => { ]); }); + it('updates the cancellation marker from the terminal event', async () => { + sdkMocks.sessions.push( + createMockSession({ + async *events() { + yield { + id: 11, + v: 1, + type: 'prompt_cancelled', + promptId: 'p1', + data: { sessionId: 'session-1' }, + }; + yield { + id: 12, + v: 1, + type: 'turn_complete', + promptId: 'p1', + data: { + sessionId: 'session-1', + promptId: 'p1', + stopReason: 'cancelled', + promptCancelled: { elapsedMs: 10999, cancelledAt: 12000 }, + }, + }; + }, + }), + ); + let blocks: readonly DaemonTranscriptBlock[] = []; + function Harness() { + blocks = useDaemonTranscriptBlocks(); + return null; + } + await renderWithProvider(, { autoConnect: true }); + await act(async () => { + await flushPromises(); + }); + expect(blocks).toEqual([ + expect.objectContaining({ + kind: 'prompt_cancelled', + promptId: 'p1', + elapsedMs: 10999, + serverTimestamp: 12000, + }), + ]); + }); + it('keeps forward-failed prompt cancellations out of blocks', async () => { const session = createMockSession({ events: async function* forwardFailedPromptCancelledEvents( @@ -9916,69 +9961,87 @@ describe('DaemonSessionProvider', () => { expect(promptStatus).toBe('idle'); }); - it('keeps local prompts active when a restored prompt is cancelled', async () => { - const accepted = createDeferred(); - const releaseRestoredCancel = createDeferred(); - const restoredCancelDelivered = createDeferred(); - const releaseLocalComplete = createDeferred(); - const localCompleteDelivered = createDeferred(); - const session = createMockSession({ - hasActivePrompt: true, - submitPrompt: vi.fn(() => accepted.promise), - events: async function* restoredPromptCancelDuringLocalPrompt() { - await releaseRestoredCancel.promise; - yield { - id: 6, - v: 1, - type: 'prompt_cancelled', - originatorClientId: 'client-2', - data: { sessionId: 'session-1', reason: 'user_cancel' }, - } satisfies DaemonEvent; - restoredCancelDelivered.resolve(); - await releaseLocalComplete.promise; - yield { - id: 7, - v: 1, - type: 'turn_complete', - data: { promptId: 'local-prompt', stopReason: 'end_turn' }, - } satisfies DaemonEvent; - localCompleteDelivered.resolve(); - }, - }); - sdkMocks.sessions.push(session); - let actions: DaemonUiSessionActions | undefined; - let promptStatus: ReturnType = 'idle'; + it.each([false, true])( + 'keeps local prompts active when a restored prompt is cancelled (timed terminal: %s)', + async (timedTerminal) => { + const accepted = createDeferred(); + const releaseRestoredCancel = createDeferred(); + const restoredCancelDelivered = createDeferred(); + const releaseLocalComplete = createDeferred(); + const localCompleteDelivered = createDeferred(); + const session = createMockSession({ + hasActivePrompt: true, + submitPrompt: vi.fn(() => accepted.promise), + events: async function* restoredPromptCancelDuringLocalPrompt() { + await releaseRestoredCancel.promise; + yield { + id: 6, + v: 1, + type: 'prompt_cancelled', + promptId: 'restored-prompt', + originatorClientId: 'client-2', + data: { sessionId: 'session-1', reason: 'user_cancel' }, + } satisfies DaemonEvent; + if (timedTerminal) { + yield { + id: 7, + v: 1, + type: 'turn_complete', + promptId: 'restored-prompt', + originatorClientId: 'client-2', + data: { + promptId: 'restored-prompt', + stopReason: 'cancelled', + promptCancelled: { cancelledAt: 12000, elapsedMs: 11000 }, + }, + } satisfies DaemonEvent; + } + restoredCancelDelivered.resolve(); + await releaseLocalComplete.promise; + yield { + id: 8, + v: 1, + type: 'turn_complete', + data: { promptId: 'local-prompt', stopReason: 'end_turn' }, + } satisfies DaemonEvent; + localCompleteDelivered.resolve(); + }, + }); + sdkMocks.sessions.push(session); + let actions: DaemonUiSessionActions | undefined; + let promptStatus: ReturnType = 'idle'; - function Harness() { - actions = useDaemonActions(); - promptStatus = useDaemonPromptStatus(); - return null; - } + function Harness() { + actions = useDaemonActions(); + promptStatus = useDaemonPromptStatus(); + return null; + } - await renderWithProvider(, { - autoConnect: true, - reconnectDelayMs: 1, - maxReconnectDelayMs: 1, - }); - let promptResult: Promise | undefined; - await act(async () => { - promptResult = requireActions(actions).sendPrompt('local prompt'); - accepted.resolve({ promptId: 'local-prompt', lastEventId: 10 }); - await flushPromises(); - releaseRestoredCancel.resolve(); - await restoredCancelDelivered.promise; - await flushPromises(); - }); + await renderWithProvider(, { + autoConnect: true, + reconnectDelayMs: 1, + maxReconnectDelayMs: 1, + }); + let promptResult: Promise | undefined; + await act(async () => { + promptResult = requireActions(actions).sendPrompt('local prompt'); + accepted.resolve({ promptId: 'local-prompt', lastEventId: 10 }); + await flushPromises(); + releaseRestoredCancel.resolve(); + await restoredCancelDelivered.promise; + await flushPromises(); + }); - expect(promptStatus).not.toBe('idle'); + expect(promptStatus).not.toBe('idle'); - await act(async () => { - releaseLocalComplete.resolve(); - await localCompleteDelivered.promise; - await flushPromises(); - }); - await expect(promptResult).resolves.toEqual({ stopReason: 'end_turn' }); - }); + await act(async () => { + releaseLocalComplete.resolve(); + await localCompleteDelivered.promise; + await flushPromises(); + }); + await expect(promptResult).resolves.toEqual({ stopReason: 'end_turn' }); + }, + ); it('does not revive settled restored active prompts after SSE reconnect', async () => { const turnCompleted = createDeferred(); @@ -17900,6 +17963,116 @@ describe('DaemonSessionProvider', () => { }); }); + it.each([true, false])( + 'merges cancellation markers in overlapping history pages (timed: %s)', + async (timed) => { + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['session_transcript_pagination'], + }); + const cancelled = (promptId: string, recordId: string): DaemonEvent => ({ + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: '' }, + _meta: { + 'qwen.session.recordId': recordId, + qwenTranscript: { sourceRecordIds: [recordId] }, + promptCancelled: { + promptId, + cancelledAt: 12000, + elapsedMs: 11000, + }, + }, + }, + }, + }); + const session = createMockSession({ + sessionId: 'session-cancellation-history-dedup', + historyHasMore: true, + replaySnapshot: { + compactedReplay: [ + { + v: 1, + type: 'history_truncated', + data: { recordId: 'newer-anchor', fullTranscriptAvailable: true }, + }, + ], + liveJournal: [ + { + v: 1, + type: timed ? 'turn_complete' : 'prompt_cancelled', + promptId: 'current-cancel', + data: { + promptId: 'current-cancel', + stopReason: 'cancelled', + ...(timed + ? { + promptCancelled: { cancelledAt: 12000, elapsedMs: 11000 }, + } + : {}), + }, + }, + ], + }, + }); + sdkMocks.sessions.push(session); + sdkMocks.getSessionTranscriptPage.mockResolvedValue({ + v: 1, + sessionId: session.sessionId, + events: [ + cancelled('older-cancel', 'older-record'), + cancelled('current-cancel', 'current-record'), + ], + hasMore: false, + }); + let history: ReturnType | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + let retainedBytes = 0; + function Harness() { + history = useDaemonTranscriptHistory(); + blocks = useDaemonTranscriptBlocks(); + retainedBytes = useDaemonTranscriptState().retainedBytes; + return null; + } + await renderWithProvider(, { + autoConnect: true, + historyPageSize: 25, + }); + const live = blocks.find((block) => block.kind === 'prompt_cancelled'); + expect(live).toMatchObject({ + promptId: 'current-cancel', + ...(timed ? { elapsedMs: 11000 } : {}), + }); + if (!timed) expect(live).not.toHaveProperty('elapsedMs'); + await act(async () => { + await history?.loadMore(); + await flushPromises(); + }); + const cancellations = blocks.filter( + (block) => block.kind === 'prompt_cancelled', + ); + expect(cancellations.map((block) => block.promptId)).toEqual([ + 'older-cancel', + 'current-cancel', + ]); + expect(cancellations[1]).toMatchObject({ + id: live?.id, + elapsedMs: 11000, + serverTimestamp: 12000, + sourceRecordIds: ['current-record'], + }); + expect(retainedBytes).toBe( + blocks.reduce( + (total, block) => total + estimateDaemonTranscriptBlockBytes(block), + 0, + ), + ); + }, + ); + it('drops fetched transcript events whose records are already displayed', async () => { // The pagination anchor can sit inside the retained window (e.g. the // daemon's transcript backfill for a live-journal overflow returns the diff --git a/packages/web-shell/client/daemon/session/DaemonSessionProvider.tsx b/packages/web-shell/client/daemon/session/DaemonSessionProvider.tsx index bc90cb578e4..3bf6afde0b3 100644 --- a/packages/web-shell/client/daemon/session/DaemonSessionProvider.tsx +++ b/packages/web-shell/client/daemon/session/DaemonSessionProvider.tsx @@ -200,7 +200,6 @@ interface LiveJournalRepairEpisode { interface TranscriptHistoryMaterialization { blocks: readonly DaemonTranscriptBlock[]; nextOrdinal: number; - retainedBytes: number; toolBlockByCallId: Record; permissionBlockByRequestId: Record; unrecognizedDiagnostics: readonly DaemonUnrecognizedDiagnostic[]; @@ -414,11 +413,16 @@ function materializeTranscriptHistory( break; } } - let pageBytes = 0; - for (const block of pageBlockList) { - pageBytes += estimateDaemonTranscriptBlockBytes(block); - } - const pageBlocks = pageBlockList.length; + const mergedBlocks = mergeTranscriptHistoryBlocks( + current.blocks, + pageBlockList, + ); + const pageBytes = + mergedBlocks.reduce( + (total, block) => total + estimateDaemonTranscriptBlockBytes(block), + 0, + ) - current.retainedBytes; + const pageBlocks = mergedBlocks.length - current.blocks.length; // `impossible` must be evaluated across BOTH dimensions, regardless of // which branch rejects: a page that alone fills the whole block window can // never be admitted (an anchored window always retains at least one block), @@ -457,7 +461,6 @@ function materializeTranscriptHistory( materialization: { blocks: pageBlockList, nextOrdinal: history.nextOrdinal, - retainedBytes: pageBytes, toolBlockByCallId: history.toolBlockByCallId, permissionBlockByRequestId: history.permissionBlockByRequestId, // History pages can carry frames recorded by newer daemon versions, exactly @@ -468,6 +471,44 @@ function materializeTranscriptHistory( }; } +function mergeTranscriptHistoryBlocks( + current: readonly DaemonTranscriptBlock[], + history: readonly DaemonTranscriptBlock[], +): DaemonTranscriptBlock[] { + const cancellations = new Map( + history + .filter((block) => block.kind === 'prompt_cancelled' && block.promptId) + .map((block) => [block.promptId, block]), + ); + const displayedCancelledPromptIds = new Set(); + const updated = current.map((block) => { + if (block.kind !== 'prompt_cancelled' || !block.promptId) return block; + displayedCancelledPromptIds.add(block.promptId); + const persisted = cancellations.get(block.promptId); + if ( + persisted?.kind !== 'prompt_cancelled' || + persisted.elapsedMs === undefined + ) { + return block; + } + return { + ...block, + elapsedMs: persisted.elapsedMs, + serverTimestamp: persisted.serverTimestamp, + sourceRecordIds: persisted.sourceRecordIds, + }; + }); + return [ + ...history.filter( + (block) => + block.kind !== 'prompt_cancelled' || + !block.promptId || + !displayedCancelledPromptIds.has(block.promptId), + ), + ...updated, + ]; +} + function applyTranscriptHistory( current: DaemonTranscriptState, history: TranscriptHistoryMaterialization, @@ -516,10 +557,14 @@ function applyTranscriptHistory( } permissionBlockByRequestId[requestId] = blockId; } + const blocks = mergeTranscriptHistoryBlocks(current.blocks, history.blocks); return { ...current, - blocks: [...history.blocks, ...current.blocks], - retainedBytes: current.retainedBytes + history.retainedBytes, + blocks, + retainedBytes: blocks.reduce( + (total, block) => total + estimateDaemonTranscriptBlockBytes(block), + 0, + ), nextOrdinal: history.nextOrdinal, toolBlockByCallId, trimmedToolNotificationByCallId, @@ -3426,6 +3471,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { enqueueTranscriptEvents(eventsToDispatch); for (const uiEvent of uiEvents) { if ( + event.type === 'prompt_cancelled' && uiEvent.type === 'prompt.cancelled' && (restoredActivePrompt || uiEvent.originatorClientId !== activeSession.clientId) @@ -4968,7 +5014,10 @@ function normalizeAndFilterEvent( }); const goalStatusEvent = normalizeGoalStatusEvent(event); if (isPromptLifecycleTurnEvent(event)) { - return goalStatusEvent ? [goalStatusEvent] : []; + const cancellation = normalized.filter( + (item) => item.type === 'prompt.cancelled', + ); + return goalStatusEvent ? [...cancellation, goalStatusEvent] : cancellation; } return goalStatusEvent ? [...normalized, goalStatusEvent] : normalized; } diff --git a/packages/web-shell/client/i18n.tsx b/packages/web-shell/client/i18n.tsx index 775f3c635a6..9f289274eaf 100644 --- a/packages/web-shell/client/i18n.tsx +++ b/packages/web-shell/client/i18n.tsx @@ -2940,6 +2940,10 @@ const EN: Messages = { return `${n} thought${n === 1 ? '' : 's'}`; }, 'turn.stopped': 'You cancelled this request', + 'turn.stoppedAfter': (v) => { + const seconds = v?.seconds ?? 0; + return `You cancelled this request after ${seconds} second${seconds === 1 ? '' : 's'}`; + }, 'message.renderError': 'This message could not be displayed.', 'tasks.title': 'Background tasks', 'tasks.empty': 'No tasks currently running', @@ -6414,6 +6418,7 @@ const ZH: Messages = { 'turn.toolCalls': (v) => `工具 ${v?.count ?? 0} 次`, 'turn.thinkingCount': (v) => `思考 ${v?.count ?? 0} 次`, 'turn.stopped': '你已取消请求', + 'turn.stoppedAfter': (v) => `你在 ${v?.seconds ?? 0} 秒后取消了请求`, 'message.renderError': '此消息无法显示。', 'tasks.title': '后台任务', 'tasks.empty': '当前没有运行中的任务',