mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 06:19:43 -10:00
fix(web-shell): persist cancelled responses and elapsed time (#11608)
* fix(web-shell): preserve cancelled responses and elapsed time * fix(core): retain cancelled text during reactive compression --------- Co-authored-by: 钉萁 <[email protected]>
This commit is contained in:
1 parent
d66ad3e3a2
commit
53e6196c6b
28 files changed
+2107
-275
No files matched your search
@@ -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.
|
||||
@@ -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 超时及部分关闭路径映射为用户取消信号。本次保留该
|
||||
分类方式;增加取消原因协议不在范围内。
|
||||
@@ -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',
|
||||
);
|
||||
|
||||
@@ -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<string, unknown>)
|
||||
: 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 } : {}),
|
||||
});
|
||||
|
||||
@@ -981,6 +981,36 @@ class DefaultTranscriptReplayMachine implements TranscriptReplayMachine {
|
||||
emit: (update: SessionUpdate) => TranscriptReplayEmission,
|
||||
meta: UpdateMetaOptions,
|
||||
): Iterable<TranscriptReplayEmission> {
|
||||
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
|
||||
|
||||
@@ -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<void>((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<void>((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' });
|
||||
},
|
||||
);
|
||||
|
||||
|
||||
@@ -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<void> {
|
||||
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)}`,
|
||||
|
||||
@@ -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<AsyncGenerator<GenerateContentResponse>>(
|
||||
(_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(
|
||||
'<invoke name="read_file"><parameter name="file_path">a.ts</parameter></invoke>',
|
||||
'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 =
|
||||
'<invoke name="read_file"><parameter name="file_path">a.ts</parameter></invoke>';
|
||||
|
||||
+209
-117
@@ -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<GenerateContentResponse> {
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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) ||
|
||||
|
||||
@@ -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];
|
||||
|
||||
@@ -1669,12 +1669,38 @@ function appendPromptCancelledBlock(
|
||||
state: DaemonTranscriptState,
|
||||
event: Extract<DaemonUiEvent, { type: 'prompt.cancelled' }>,
|
||||
): 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
|
||||
|
||||
@@ -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 =
|
||||
|
||||
@@ -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, {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)],
|
||||
|
||||
@@ -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],
|
||||
});
|
||||
|
||||
@@ -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 },
|
||||
|
||||
@@ -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)
|
||||
: '';
|
||||
}
|
||||
|
||||
@@ -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(
|
||||
<SystemMessage
|
||||
content=""
|
||||
variant="info"
|
||||
source="prompt_cancelled"
|
||||
data={{ elapsedMs }}
|
||||
/>,
|
||||
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(
|
||||
<SystemMessage
|
||||
content=""
|
||||
variant="info"
|
||||
source="prompt_cancelled"
|
||||
data={{ elapsedMs }}
|
||||
/>,
|
||||
'zh-CN',
|
||||
);
|
||||
expect(container.querySelector('[role="status"]')?.textContent).toBe(
|
||||
'你已取消请求',
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it('renders the user-cancelled marker as a status region', () => {
|
||||
const container = render(
|
||||
<SystemMessage content="" variant="info" source="prompt_cancelled" />,
|
||||
|
||||
@@ -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 (
|
||||
<div className={styles.cancelled} role="status">
|
||||
<span>{t('turn.stopped')}</span>
|
||||
<span>
|
||||
{typeof elapsedMs === 'number' &&
|
||||
Number.isFinite(elapsedMs) &&
|
||||
elapsedMs >= 0
|
||||
? t('turn.stoppedAfter', { seconds: Math.ceil(elapsedMs / 1000) })
|
||||
: t('turn.stopped')}
|
||||
</span>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1042,11 +1042,6 @@
|
||||
color: var(--success-color, #3fb950);
|
||||
}
|
||||
|
||||
.sessionStatusUnknown {
|
||||
color: var(--muted-foreground);
|
||||
font-size: 11px;
|
||||
}
|
||||
|
||||
.sessionSourceIcon {
|
||||
width: 28px;
|
||||
height: 28px;
|
||||
|
||||
@@ -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 ? (
|
||||
<span
|
||||
className={styles.sessionStatusUnknown}
|
||||
aria-label={t('sidebar.activityUnknown')}
|
||||
>
|
||||
?
|
||||
</span>
|
||||
) : null}
|
||||
</span>
|
||||
{isEditing && showRename ? (
|
||||
|
||||
+3
-1
@@ -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 () => {
|
||||
|
||||
@@ -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(<Harness />, { 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<NonBlockingPromptAccepted>();
|
||||
const releaseRestoredCancel = createDeferred<void>();
|
||||
const restoredCancelDelivered = createDeferred<void>();
|
||||
const releaseLocalComplete = createDeferred<void>();
|
||||
const localCompleteDelivered = createDeferred<void>();
|
||||
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<typeof useDaemonPromptStatus> = 'idle';
|
||||
it.each([false, true])(
|
||||
'keeps local prompts active when a restored prompt is cancelled (timed terminal: %s)',
|
||||
async (timedTerminal) => {
|
||||
const accepted = createDeferred<NonBlockingPromptAccepted>();
|
||||
const releaseRestoredCancel = createDeferred<void>();
|
||||
const restoredCancelDelivered = createDeferred<void>();
|
||||
const releaseLocalComplete = createDeferred<void>();
|
||||
const localCompleteDelivered = createDeferred<void>();
|
||||
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<typeof useDaemonPromptStatus> = 'idle';
|
||||
|
||||
function Harness() {
|
||||
actions = useDaemonActions();
|
||||
promptStatus = useDaemonPromptStatus();
|
||||
return null;
|
||||
}
|
||||
function Harness() {
|
||||
actions = useDaemonActions();
|
||||
promptStatus = useDaemonPromptStatus();
|
||||
return null;
|
||||
}
|
||||
|
||||
await renderWithProvider(<Harness />, {
|
||||
autoConnect: true,
|
||||
reconnectDelayMs: 1,
|
||||
maxReconnectDelayMs: 1,
|
||||
});
|
||||
let promptResult: Promise<unknown> | 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(<Harness />, {
|
||||
autoConnect: true,
|
||||
reconnectDelayMs: 1,
|
||||
maxReconnectDelayMs: 1,
|
||||
});
|
||||
let promptResult: Promise<unknown> | 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<void>();
|
||||
@@ -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<typeof useDaemonTranscriptHistory> | undefined;
|
||||
let blocks: readonly DaemonTranscriptBlock[] = [];
|
||||
let retainedBytes = 0;
|
||||
function Harness() {
|
||||
history = useDaemonTranscriptHistory();
|
||||
blocks = useDaemonTranscriptBlocks();
|
||||
retainedBytes = useDaemonTranscriptState().retainedBytes;
|
||||
return null;
|
||||
}
|
||||
await renderWithProvider(<Harness />, {
|
||||
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
|
||||
|
||||
@@ -200,7 +200,6 @@ interface LiveJournalRepairEpisode {
|
||||
interface TranscriptHistoryMaterialization {
|
||||
blocks: readonly DaemonTranscriptBlock[];
|
||||
nextOrdinal: number;
|
||||
retainedBytes: number;
|
||||
toolBlockByCallId: Record<string, string>;
|
||||
permissionBlockByRequestId: Record<string, string>;
|
||||
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<string>();
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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': '当前没有运行中的任务',
|
||||
|
||||
Reference in new issue
Block a user