From 3b0f77b15811aa96b04684f9d319c7bd0bdd8566 Mon Sep 17 00:00:00 2001 From: Is14w Date: Sat, 22 Aug 2026 11:54:39 +0800 Subject: [PATCH 01/15] fix(chat): recover from stopped and stalled runs --- .../src/lib/cancellation/abortRace.ts | 43 +++ .../src/lib/chat/compaction/controller.ts | 70 +++- .../conversation/run/gatewayBridgeEvents.ts | 7 +- .../src/lib/providers/runtime/streamRetry.ts | 242 ++++++++++++- .../agent-gui/src/lib/trajectory/recorder.ts | 83 ++++- .../src/lib/trajectory/recorderRegistry.ts | 29 +- crates/agent-gui/src/pages/ChatPage.tsx | 2 + .../history/useConversationHistoryActions.ts | 17 + .../pages/chat/runtime/chatRunFinalization.ts | 47 +++ .../pages/chat/runtime/useManualCompaction.ts | 59 ++- .../src/pages/chat/runtime/useSendChatTurn.ts | 339 ++++++++++++++---- .../test/chat/chat-stop-timing.test.mjs | 90 +++++ .../test/chat/compaction-controller.test.mjs | 76 ++++ .../test/providers/stream-retry.test.mjs | 330 +++++++++++++++++ .../test/trajectory/desktop-live.test.mjs | 35 ++ .../test/trajectory/recorder.test.mjs | 23 +- .../src/pages/chat/ChatComposerBar.tsx | 16 + 17 files changed, 1386 insertions(+), 122 deletions(-) create mode 100644 crates/agent-gui/src/lib/cancellation/abortRace.ts diff --git a/crates/agent-gui/src/lib/cancellation/abortRace.ts b/crates/agent-gui/src/lib/cancellation/abortRace.ts new file mode 100644 index 000000000..f4720a1c5 --- /dev/null +++ b/crates/agent-gui/src/lib/cancellation/abortRace.ts @@ -0,0 +1,43 @@ +function abortReason(signal: AbortSignal): unknown { + return signal.reason ?? new DOMException("Aborted", "AbortError"); +} + +/** + * Resolves/rejects with `operation` unless `signal` aborts first. The source + * promise remains observed after an abort, so a late rejection never becomes + * an unhandled rejection. + */ +export function raceWithAbort(operation: PromiseLike | T, signal?: AbortSignal): Promise { + const source = Promise.resolve(operation); + if (!signal) return source; + if (signal.aborted) { + // The caller may already have started an IPC/network promise before it + // checks cancellation. Keep that promise observed even though this race + // has already been decided, otherwise a late rejection becomes unhandled. + void source.catch(() => undefined); + return Promise.reject(abortReason(signal)); + } + + return new Promise((resolve, reject) => { + let settled = false; + const finish = (callback: (value: T) => void, value: T) => { + if (settled) return; + settled = true; + signal.removeEventListener("abort", onAbort); + callback(value); + }; + const fail = (error: unknown) => { + if (settled) return; + settled = true; + signal.removeEventListener("abort", onAbort); + reject(error); + }; + const onAbort = () => fail(abortReason(signal)); + + signal.addEventListener("abort", onAbort, { once: true }); + source.then( + (value) => finish(resolve, value), + (error) => fail(error), + ); + }); +} diff --git a/crates/agent-gui/src/lib/chat/compaction/controller.ts b/crates/agent-gui/src/lib/chat/compaction/controller.ts index 1c050675f..546dab1f9 100644 --- a/crates/agent-gui/src/lib/chat/compaction/controller.ts +++ b/crates/agent-gui/src/lib/chat/compaction/controller.ts @@ -87,6 +87,8 @@ export type CompactionTurnBinding = { cancellation: TurnCancellation; debugLogger?: StreamDebugLogger; complete?: CompleteAssistantFn; + /** Observer owned by this turn; prevents a late old turn from reporting into a replacement. */ + observer?: CompactionObserver; sinks: CompactionSinks; buildPreparedContext: ( state: ConversationViewState, @@ -186,6 +188,11 @@ export class CompactionController { private pressure = createCompactionPressure(); private readonly ledger = new TokenLedger(); private binding: CompactionTurnBinding | null = null; + // A conversation can start a replacement turn after force-stop while an old + // provider task is still unwinding. The lease keeps that old task from + // clearing or rolling back the replacement turn's binding. + private bindingGeneration = 0; + private activeBindingGeneration: number | null = null; private rollbackSnapshot: RollbackSnapshot | null = null; private inFlight = false; private statusPhase: CompactionStatus["phase"] = "idle"; @@ -213,18 +220,30 @@ export class CompactionController { bindTurn(binding: CompactionTurnBinding) { // A defensive rebind must not strand the previous observer interval. this.settleAbortedIfRunning(); + const generation = ++this.bindingGeneration; this.binding = binding; + this.activeBindingGeneration = generation; this.rollbackSnapshot = null; this.inFlight = false; + return generation; } - unbindTurn() { + isTurnBound(generation: number) { + return this.binding !== null && this.activeBindingGeneration === generation; + } + + unbindTurn(expectedGeneration?: number) { + if (expectedGeneration !== undefined && !this.isTurnBound(expectedGeneration)) { + return false; + } // Every published start receives exactly one terminal notification, even when a caller // tears down the turn without first reaching the ordinary completion path. this.settleAbortedIfRunning(); this.binding = null; + this.activeBindingGeneration = null; this.rollbackSnapshot = null; this.inFlight = false; + return true; } get stats() { @@ -324,8 +343,10 @@ export class CompactionController { includeUploadedFilesMetadata?: boolean; }): Promise { const binding = this.binding; + const bindingGeneration = this.activeBindingGeneration; const presend = binding?.presend; - if (!binding || !presend) return false; + if (!binding || !presend || bindingGeneration === null) return false; + const ownsBinding = () => this.isTurnBound(bindingGeneration); if (binding.cancellation.userStop.signal.aborted) { throw createCompactionAbortError(); } @@ -420,6 +441,9 @@ export class CompactionController { ); return true; } catch (error) { + if (!ownsBinding()) { + throw createCompactionAbortError(); + } if (this.isAbortOutcome(scope.controller.signal, error)) { throw error; } @@ -441,8 +465,10 @@ export class CompactionController { return false; } finally { scope.release(); - this.inFlight = false; - this.binding?.sinks.setBridgeToolStatus?.(null); + if (ownsBinding()) { + this.inFlight = false; + this.binding?.sinks.setBridgeToolStatus?.(null); + } } } @@ -458,9 +484,11 @@ export class CompactionController { manualContextUsage?: ManualContextUsageSnapshot; }): Promise { const binding = this.binding; - if (!binding) { + const bindingGeneration = this.activeBindingGeneration; + if (!binding || bindingGeneration === null) { return { context: null, shouldDisableProtection: false, outcome: "skipped" }; } + const ownsBinding = () => this.isTurnBound(bindingGeneration); // 覆盖"mid-stream abort 后、summarizer 启动前"用户恰好点停止的间隙。 if (binding.cancellation.userStop.signal.aborted) { throw createCompactionAbortError(); @@ -594,6 +622,9 @@ export class CompactionController { ); return { context: resumeContext, shouldDisableProtection: false, outcome: "compacted" }; } catch (error) { + if (!ownsBinding()) { + throw createCompactionAbortError(); + } if (this.isAbortOutcome(scope.controller.signal, error)) { throw error; } @@ -630,8 +661,10 @@ export class CompactionController { : { context: null, shouldDisableProtection: false, outcome: "failed" }; } finally { scope.release(); - this.inFlight = false; - this.binding?.sinks.setBridgeToolStatus?.(null); + if (ownsBinding()) { + this.inFlight = false; + this.binding?.sinks.setBridgeToolStatus?.(null); + } } } @@ -653,7 +686,7 @@ export class CompactionController { }, ): Promise { if (this.binding || this.inFlight) return { status: "busy" }; - this.bindTurn(binding); + const bindingGeneration = this.bindTurn(binding); try { const probe = this.probeManualDecision(binding, state, contextUsage, options?.tools); if (!probe.shouldCompact) { @@ -681,12 +714,12 @@ export class CompactionController { } } catch { // 中止或意外异常：走统一善后（回滚快照 / running 态复位 idle）。 - await this.handleTurnAbort(); + await this.handleTurnAbort(bindingGeneration); return binding.cancellation.userStop.signal.aborted ? { status: "failed", aborted: true } : { status: "failed" }; } finally { - this.unbindTurn(); + this.unbindTurn(bindingGeneration); } } @@ -721,7 +754,10 @@ export class CompactionController { } // 用户中止后的统一善后：有快照则回滚（恢复状态/输入框/可选持久化）并返回 true。 - async handleTurnAbort(): Promise { + async handleTurnAbort(expectedGeneration?: number): Promise { + if (expectedGeneration !== undefined && !this.isTurnBound(expectedGeneration)) { + return false; + } const binding = this.binding; const snapshot = this.rollbackSnapshot; this.rollbackSnapshot = null; @@ -808,6 +844,10 @@ export class CompactionController { this.binding?.sinks.publishStatus?.(status); } + private activeObserver() { + return this.binding?.observer ?? this.observer; + } + private publishRunning( trigger: CompactionTrigger, sourceSegmentIndex: number, @@ -818,7 +858,7 @@ export class CompactionController { this.observedTrigger = trigger; this.observedTokensBefore = decision.totalTokens; this.notifyObserver(() => - this.observer?.onStart({ trigger, tokensBefore: decision.totalTokens }), + this.activeObserver()?.onStart({ trigger, tokensBefore: decision.totalTokens }), ); this.publishStatus({ phase: "running", @@ -842,7 +882,7 @@ export class CompactionController { // was unwinding. Late completion is then operationally stale and must not emit a second end. if (this.observedTrigger !== trigger || this.observedOperationId !== operationId) return; this.notifyObserver(() => - this.observer?.onEnd({ + this.activeObserver()?.onEnd({ trigger, status: "complete", ...(this.observedTokensBefore === undefined @@ -866,7 +906,7 @@ export class CompactionController { private settleFailed(trigger: CompactionTrigger, message: string, operationId: number) { if (this.observedTrigger !== trigger || this.observedOperationId !== operationId) return; this.notifyObserver(() => - this.observer?.onEnd({ + this.activeObserver()?.onEnd({ trigger, status: "error", ...(this.observedTokensBefore === undefined @@ -886,7 +926,7 @@ export class CompactionController { return false; } this.notifyObserver(() => - this.observer?.onEnd({ + this.activeObserver()?.onEnd({ trigger, status: "aborted", ...(this.observedTokensBefore === undefined diff --git a/crates/agent-gui/src/lib/chat/conversation/run/gatewayBridgeEvents.ts b/crates/agent-gui/src/lib/chat/conversation/run/gatewayBridgeEvents.ts index 2fc3b35fc..a60d4b7db 100644 --- a/crates/agent-gui/src/lib/chat/conversation/run/gatewayBridgeEvents.ts +++ b/crates/agent-gui/src/lib/chat/conversation/run/gatewayBridgeEvents.ts @@ -218,12 +218,17 @@ export function createGatewayBridgeEventController( } }, emitError(message: string, conversationIdOverride?: string) { - queueEvent({ + const sendResult = queueEvent({ type: "error", message, conversation_id: conversationIdOverride ?? params.resolveErrorConversationId?.() ?? params.conversationId, }); + if (sendResult && typeof (sendResult as Promise).then === "function") { + (sendResult as Promise).catch((error) => { + console.warn("error event failed", error); + }); + } }, close() { streamClosed = true; diff --git a/crates/agent-gui/src/lib/providers/runtime/streamRetry.ts b/crates/agent-gui/src/lib/providers/runtime/streamRetry.ts index aef4299d6..171596054 100644 --- a/crates/agent-gui/src/lib/providers/runtime/streamRetry.ts +++ b/crates/agent-gui/src/lib/providers/runtime/streamRetry.ts @@ -5,6 +5,7 @@ import { createAssistantMessageEventStream, isRetryableAssistantError, } from "@earendil-works/pi-ai"; +import { raceWithAbort } from "../../cancellation/abortRace"; export type { RetryAttemptRecord } from "@liveagent/ui/lib/chat/retryAttempts"; @@ -13,10 +14,13 @@ export const DEFAULT_STREAM_RETRY_MAX_ATTEMPTS = 6; const STREAM_RETRY_BASE_DELAY_MS = 200; const STREAM_RETRY_BACKOFF_FACTOR = 2; +const DEFAULT_STREAM_RETRY_IDLE_TIMEOUT_MS = 30_000; export type StreamRetryConfig = { maxAttempts?: number; disabled?: boolean; + /** Maximum time a provider attempt may wait for its next event or result. */ + idleTimeoutMs?: number; /** * Retry ordinal (1..maxRetries) about to be attempted, invoked before the * backoff sleep. `errorMessage` is the failure that triggered this retry. @@ -46,6 +50,12 @@ function terminalMessage(event: TerminalEvent) { return event.type === "done" ? event.message : event.error; } +function terminalAssistantMessage( + terminal: TerminalEvent | undefined, +): AssistantMessage | undefined { + return terminal ? (terminalMessage(terminal) as AssistantMessage) : undefined; +} + /** Codex-style backoff: base * factor^(attempt-1) * uniform(0.9, 1.1), uncapped. */ export function computeStreamRetryBackoffMs(attempt: number): number { const base = STREAM_RETRY_BASE_DELAY_MS * STREAM_RETRY_BACKOFF_FACTOR ** (attempt - 1); @@ -67,6 +77,59 @@ function buildAbortedAssistantMessage(previous: AssistantMessage | undefined): A } as AssistantMessage; } +function errorText(error: unknown): string { + if (error instanceof Error && error.message.trim()) return error.message; + if (typeof error === "string" && error.trim()) return error; + if (error && typeof error === "object") { + const candidate = error as { errorMessage?: unknown; message?: unknown; error?: unknown }; + if (typeof candidate.errorMessage === "string" && candidate.errorMessage.trim()) { + return candidate.errorMessage; + } + if (typeof candidate.message === "string" && candidate.message.trim()) { + return candidate.message; + } + if (typeof candidate.error === "string" && candidate.error.trim()) return candidate.error; + } + return String(error) || "Provider stream failed"; +} + +function buildTransportErrorMessage( + error: unknown, + previous?: AssistantMessage, +): AssistantMessage { + return { + ...(previous ?? {}), + role: "assistant", + content: previous?.content ?? [], + stopReason: "error", + errorMessage: errorText(error), + } as AssistantMessage; +} + +function isRetryableTransportFailure(error: unknown): boolean { + const assistantError = + error && typeof error === "object" && "role" in error + ? (error as AssistantMessage) + : undefined; + if (assistantError && isRetryableAssistantError(assistantError)) return true; + const message = errorText(error); + if (/\b(?:abort|aborted|cancel|cancelled|canceled)\b/i.test(message)) return false; + return /(?:fetch failed|network|timed?\s*out|timeout|econn(?:reset|refused|aborted)|connection|socket|stream|\b(?:408|425|429|500|502|503|504|522|524)\b|temporarily unavailable|service unavailable|overloaded)/i.test( + message, + ); +} + +function createSyntheticErrorStream( + error: unknown, + previous?: AssistantMessage, +): AssistantMessageEventStream { + const stream = createAssistantMessageEventStream(); + const failed = buildTransportErrorMessage(error, previous); + stream.push({ type: "error", reason: "error", error: failed }); + stream.end(failed); + return stream; +} + function sleepWithAbort(ms: number, signal: AbortSignal | undefined): Promise { if (signal?.aborted) return Promise.reject(signal.reason ?? new Error("Aborted")); if (ms <= 0) return Promise.resolve(); @@ -83,6 +146,58 @@ function sleepWithAbort(ms: number, signal: AbortSignal | undefined): Promise( + operation: PromiseLike | T, + timeoutMs: number, + signal?: AbortSignal, +): Promise { + if (timeoutMs <= 0) return raceWithAbort(operation, signal); + let timer: ReturnType | undefined; + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error("Provider stream idle timeout")), timeoutMs); + }); + return Promise.race([raceWithAbort(operation, signal), timeout]).finally(() => { + if (timer !== undefined) clearTimeout(timer); + }); +} + +function readTerminalEventAfterAbort( + iterator: AsyncIterator, +): Promise> { + // A provider may have synchronously buffered its terminal error before the + // user pressed Stop. Preserve that terminal for retry accounting, but never + // forward a queued text/thinking/tool event after cancellation. + return new Promise((resolve) => { + let settled = false; + const finish = (value: IteratorResult) => { + if (settled) return; + settled = true; + clearTimeout(timer); + resolve(value); + }; + const timer = setTimeout(() => finish({ done: true, value: undefined }), 0); + Promise.resolve() + .then(() => iterator.next()) + .then( + (next) => { + if (next.done || isTerminalEvent(next.value)) finish(next); + else finish({ done: true, value: undefined }); + }, + () => finish({ done: true, value: undefined }), + ); + }); +} + +function abandonIterator(iterator: AsyncIterator) { + try { + // A hung provider iterator can keep its return() promise pending too. + // Request cleanup without making cancellation wait on provider code. + void Promise.resolve(iterator.return?.()).catch(() => undefined); + } catch { + // Some minimal provider/test iterators do not support return(). + } +} + /** * Wraps a fresh-stream factory with attempt-scoped retry for transient * provider/transport failures. @@ -111,10 +226,25 @@ export function withStreamRetry( ): AssistantMessageEventStream { const maxAttempts = Math.max(1, options?.maxAttempts ?? DEFAULT_STREAM_RETRY_MAX_ATTEMPTS); const disabled = options?.disabled ?? false; + const idleTimeoutMs = Math.max( + 0, + options?.idleTimeoutMs ?? DEFAULT_STREAM_RETRY_IDLE_TIMEOUT_MS, + ); const signal = options?.signal; const output = createAssistantMessageEventStream(); - const firstSource = factory(); + let firstSource: AssistantMessageEventStream; + try { + firstSource = factory(); + } catch (error) { + firstSource = createSyntheticErrorStream(error); + } + + const endAsAborted = (previous?: AssistantMessage) => { + const aborted = buildAbortedAssistantMessage(previous); + output.push({ type: "error", reason: "aborted", error: aborted }); + output.end(aborted); + }; void (async () => { let attempt = 1; @@ -125,8 +255,32 @@ export function withStreamRetry( let committed = false; const buffered: AssistantMessageEvent[] = []; let terminal: TerminalEvent | undefined; + let iteratorFailure = false; - for await (const event of source) { + const iterator = source[Symbol.asyncIterator](); + while (true) { + let next: IteratorResult; + try { + next = signal?.aborted + ? await readTerminalEventAfterAbort(iterator) + : await raceWithTimeout(iterator.next(), idleTimeoutMs, signal); + } catch (error) { + if (signal?.aborted) { + abandonIterator(iterator); + endAsAborted(terminalAssistantMessage(terminal)); + return; + } + abandonIterator(iterator); + iteratorFailure = true; + const failed = buildTransportErrorMessage(error, terminalAssistantMessage(terminal)); + terminal = { type: "error", reason: "error", error: failed }; + const failedEvent = terminal; + if (committed) output.push(failedEvent); + else buffered.push(failedEvent); + break; + } + if (next.done) break; + const event = next.value; if (!committed && COMMITTING_EVENT_TYPES.has(event.type)) { committed = true; for (const bufferedEvent of buffered.splice(0)) output.push(bufferedEvent); @@ -140,19 +294,62 @@ export function withStreamRetry( } else { buffered.push(event); } - if (isTerminalEvent(event)) terminal = event; + if (isTerminalEvent(event)) { + terminal = event; + abandonIterator(iterator); + break; + } + } + + // The abort-aware terminal drain may stop waiting before the provider iterator + // acknowledges cancellation. Ask it to release its transport without blocking the UI. + if (signal?.aborted) abandonIterator(iterator); + + let result: AssistantMessage | undefined; + if (!iteratorFailure && terminal === undefined) { + if (signal?.aborted) { + if (terminal === undefined) { + endAsAborted(); + return; + } + result = terminalMessage(terminal) as AssistantMessage; + } else { + try { + result = await raceWithTimeout(source.result(), idleTimeoutMs, signal); + } catch (error) { + if (signal?.aborted) { + endAsAborted(terminalAssistantMessage(terminal)); + return; + } + if (terminal === undefined) { + const failed = buildTransportErrorMessage(error); + terminal = { type: "error", reason: "error", error: failed }; + if (committed) output.push(terminal); + else buffered.push(terminal); + } + } + } + } + + // A few provider adapters finish iteration without emitting a terminal + // event and expose the failure only through result(). Treat that shape + // exactly like an in-stream error so an uncommitted network failure can + // still be retried. + if (terminal === undefined && result?.stopReason === "error") { + terminal = { type: "error", reason: "error", error: result }; + if (committed) output.push(terminal); + else buffered.push(terminal); } if (terminal?.type === "error" && !committed && !disabled && attempt < maxAttempts) { - if (isRetryableAssistantError(terminalMessage(terminal))) { + const terminalError = terminalMessage(terminal); + if (isRetryableAssistantError(terminalError) || isRetryableTransportFailure(terminalError)) { const errorMessage = terminalMessage(terminal)?.errorMessage || "Unknown error"; attempt += 1; options?.onRetry?.(attempt - 1, maxAttempts - 1, errorMessage); hasRetried = true; try { await sleepWithAbort(computeStreamRetryBackoffMs(attempt - 1), signal); - source = factory(); - continue; } catch { // Stopped mid-backoff: the terminal must say "aborted", not replay // the prior attempt's transport error. Handing the consumer that @@ -160,16 +357,25 @@ export function withStreamRetry( // abort branches upstream never fire, so nothing records the // cancellation and the status row falls back to a spinner. if (signal?.aborted) { - const aborted = buildAbortedAssistantMessage( - terminalMessage(terminal) as AssistantMessage | undefined, - ); - output.push({ type: "error", reason: "aborted", error: aborted }); - output.end(aborted); + endAsAborted(terminalAssistantMessage(terminal)); return; } // The next attempt failed to start — surface the prior attempt's // real failure below instead of hanging the consumer on a retry // that will never happen. + break; + } + try { + source = factory(); + continue; + } catch (error) { + // A synchronous provider construction failure is still a + // transport attempt. Feed it through the same bounded retry loop. + source = createSyntheticErrorStream( + error, + terminalAssistantMessage(terminal), + ); + continue; } } } @@ -181,10 +387,20 @@ export function withStreamRetry( // done/error event through iteration and only expose the final message // via result(). output.end() is idempotent once a terminal event has // already been pushed above, so this also safety-nets that case. - output.end(await source.result()); + output.end(result ?? terminalAssistantMessage(terminal) ?? buildTransportErrorMessage("Provider stream ended without a result")); return; } - })(); + })().catch((error) => { + if (signal?.aborted) { + endAsAborted(); + return; + } + const failed = buildAbortedAssistantMessage(undefined); + failed.stopReason = "error"; + failed.errorMessage = error instanceof Error ? error.message : String(error); + output.push({ type: "error", reason: "error", error: failed }); + output.end(failed); + }); return output; } diff --git a/crates/agent-gui/src/lib/trajectory/recorder.ts b/crates/agent-gui/src/lib/trajectory/recorder.ts index e77399b4c..076c95eaf 100644 --- a/crates/agent-gui/src/lib/trajectory/recorder.ts +++ b/crates/agent-gui/src/lib/trajectory/recorder.ts @@ -50,6 +50,8 @@ export type TrajectoryStepEndInfo = { }; export type TrajectoryRecorder = { + /** Internal hook used by conversation runs that share one recorder. */ + selectTurn?: (turn: number) => void; /** * 一轮开始（用户消息落定）。同时把 `turn` 记为当前轮，后续调用不再重复传—— * 让每个埋点点自己传 turn 号，多一个参数就多一处传错的机会。 @@ -134,6 +136,76 @@ export const NOOP_TRAJECTORY_RECORDER: TrajectoryRecorder = { discard: () => {}, }; +/** + * Bind a recorder to one conversation turn. A conversation can briefly have + * two runs alive while a force-stop finalizer is unwinding; every synchronous + * recorder call must select its own turn before emitting an event. + */ +export function scopeTrajectoryRecorder( + recorder: TrajectoryRecorder, + turn: number, + onSelect?: () => void, +): TrajectoryRecorder { + const select = () => { + onSelect?.(); + recorder.selectTurn?.(turn); + }; + return { + selectTurn: recorder.selectTurn, + beginTurn: (info) => { + select(); + recorder.beginTurn({ ...info, turn }); + }, + noteContext: (info) => { + select(); + recorder.noteContext(info); + }, + captureHeader: (input) => { + select(); + return recorder.captureHeader(input); + }, + stepStart: (step, headerId) => { + select(); + recorder.stepStart(step, headerId); + }, + firstToken: (step) => { + select(); + recorder.firstToken(step); + }, + stepEnd: (step, info) => { + select(); + recorder.stepEnd(step, info); + }, + noteRetry: (step, info) => { + select(); + recorder.noteRetry(step, info); + }, + toolStart: (step, toolCall) => { + select(); + recorder.toolStart(step, toolCall); + }, + toolEnd: (callId, info) => { + select(); + recorder.toolEnd(callId, info); + }, + compactionStart: (options) => { + select(); + recorder.compactionStart(options); + }, + compactionEnd: (info) => { + select(); + recorder.compactionEnd(info); + }, + endTurn: (info) => { + select(); + recorder.endTurn(info); + }, + flush: () => recorder.flush(), + dispose: () => recorder.dispose(), + discard: () => recorder.discard(), + }; +} + /** * 创建会话级 recorder。 * @@ -152,7 +224,7 @@ export function createTrajectoryRecorder(params: { let lastHeader: { headerId: string; refs: TrajectorySectionRefs } | undefined; // Events before beginTurn are retained as turn 1 rather than discarded. let currentTurn = 1; - let turnOpen = false; + const openTurns = new Set(); let timer: ReturnType | null = null; let disposed = false; const firstTokenSeen = new Set(); @@ -225,9 +297,12 @@ export function createTrajectoryRecorder(params: { }; return { + selectTurn: (turn) => { + if (Number.isFinite(turn)) currentTurn = Math.max(1, Math.trunc(turn)); + }, beginTurn: ({ turn, messageIndex, messageId, text }) => { currentTurn = turn; - turnOpen = true; + openTurns.add(turn); emit({ k: "user", t: turn, @@ -347,8 +422,8 @@ export function createTrajectoryRecorder(params: { }); }, endTurn: (info) => { - if (!turnOpen) return; - turnOpen = false; + if (!openTurns.has(currentTurn)) return; + openTurns.delete(currentTurn); const unfinishedSteps = [...openSteps] .map((key) => { const [turnText, stepText] = key.split(" "); diff --git a/crates/agent-gui/src/lib/trajectory/recorderRegistry.ts b/crates/agent-gui/src/lib/trajectory/recorderRegistry.ts index 76d3a5a62..8b7d26bc0 100644 --- a/crates/agent-gui/src/lib/trajectory/recorderRegistry.ts +++ b/crates/agent-gui/src/lib/trajectory/recorderRegistry.ts @@ -13,7 +13,11 @@ import { type PreparedSystemPromptSlots, } from "../../pages/chat/runtime/conversationContextBuilders"; import { appendDesktopLiveTrajectory, clearDesktopLiveTrajectory } from "./liveTrajectory"; -import { createTrajectoryRecorder, type TrajectoryRecorder } from "./recorder"; +import { + createTrajectoryRecorder, + scopeTrajectoryRecorder, + type TrajectoryRecorder, +} from "./recorder"; import { createTauriTrajectoryPorts, resolvePersistedTrajectoryTurnNumber, @@ -47,12 +51,23 @@ export function acquireTrajectoryRecorder( conversationId: string, segmentIndex: number, publish?: TrajectoryPublish, + turn?: number, ): { recorder: TrajectoryRecorder; readSlots: () => PreparedSystemPromptSlots } { const existing = entries.get(conversationId); if (existing !== undefined) { existing.segmentIndex = segmentIndex; existing.publish = publish; - return { recorder: existing.recorder, readSlots: existing.slots.read }; + return { + recorder: + turn === undefined + ? existing.recorder + : scopeTrajectoryRecorder(existing.recorder, turn, () => { + // Recorder calls are synchronous. Select the matching bridge immediately before + // each event so an old run cannot publish late events through a replacement run. + existing.publish = publish; + }), + readSlots: existing.slots.read, + }; } const slots = createPreparedSystemPromptSlotHolder(); const entry: Entry = { @@ -69,7 +84,15 @@ export function acquireTrajectoryRecorder( }), }; entries.set(conversationId, entry); - return { recorder: entry.recorder, readSlots: slots.read }; + return { + recorder: + turn === undefined + ? entry.recorder + : scopeTrajectoryRecorder(entry.recorder, turn, () => { + entry.publish = publish; + }), + readSlots: slots.read, + }; } /** 供上下文构建器写入分段原文。 */ diff --git a/crates/agent-gui/src/pages/ChatPage.tsx b/crates/agent-gui/src/pages/ChatPage.tsx index b55119fb0..4d1ffeb80 100644 --- a/crates/agent-gui/src/pages/ChatPage.tsx +++ b/crates/agent-gui/src/pages/ChatPage.tsx @@ -1461,6 +1461,7 @@ export function ChatPage(props: ChatPageProps) { buildRuntimeEntryFromVisibleState, updateConversationRuntimeEntry, setConversationAbortController, + getConversationAbortController, getConversationStopRequestVersion, isConversationStopRequested, consumeConversationStop, @@ -1543,6 +1544,7 @@ export function ChatPage(props: ChatPageProps) { t, currentConversationIdRef, isConversationRunning, + getConversationAbortController, setConversationRunningState, setConversationAbortController, setConversationStopHandler, diff --git a/crates/agent-gui/src/pages/chat/history/useConversationHistoryActions.ts b/crates/agent-gui/src/pages/chat/history/useConversationHistoryActions.ts index 004cc8042..838e416a6 100644 --- a/crates/agent-gui/src/pages/chat/history/useConversationHistoryActions.ts +++ b/crates/agent-gui/src/pages/chat/history/useConversationHistoryActions.ts @@ -50,6 +50,8 @@ export type PersistConversationParams = { createdAt: number; titlePromise: Promise | null; titleLookahead?: boolean; + /** Reject a stale run after title lookahead but before touching durable history. */ + shouldPersist?: () => boolean; }; // 成功返回盖好 revision 的持久化状态（revision 是 replace/分页的 CAS 令牌， @@ -497,6 +499,7 @@ export function useConversationHistoryActions(params: UseConversationHistoryActi createdAt, titlePromise, titleLookahead = true, + shouldPersist, } = params; const pendingConversationTitle = t("chat.pendingTitle"); @@ -512,6 +515,10 @@ export function useConversationHistoryActions(params: UseConversationHistoryActi } } + // A force-stopped run may finish its title lookahead after a replacement run has + // installed a new controller. Do not let that stale snapshot enter the history queue. + if (shouldPersist && !shouldPersist()) return null; + const updatedAt = Date.now(); markLocalHistorySnapshotSynced(conversationId, updatedAt); const selectedModelToPersist = resolvePersistedConversationModelSelection({ @@ -537,6 +544,9 @@ export function useConversationHistoryActions(params: UseConversationHistoryActi commitPersistenceCursor: (cursor) => conversationPersistenceCursorRef.current.set(conversationId, cursor), }); + // The replacement run may have taken ownership while the write was in flight. + // The durable write is already ordered, but its stale summary must not become visible. + if (shouldPersist && !shouldPersist()) return null; markLocalHistorySnapshotSynced(conversationId, summary.updatedAt); // The write landed, so the durable row now matches `state` exactly — // stamp the CAS revision the backend will derive for it. Callers that @@ -573,6 +583,7 @@ export function useConversationHistoryActions(params: UseConversationHistoryActi })); sidebarStore.upsertLocal({ ...summary, isPending: undefined }); } catch (err) { + if (shouldPersist && !shouldPersist()) return null; markLocalHistorySnapshotSynced(conversationId, -1); const msg = err instanceof Error ? err.message : String(err); const persistFailedMessage = t("chat.history.persistFailed").replace( @@ -592,6 +603,7 @@ export function useConversationHistoryActions(params: UseConversationHistoryActi void titlePromise .then(async (resolvedTitle) => { if (!resolvedTitle || resolvedTitle === initialStoredTitle) return; + if (shouldPersist && !shouldPersist()) return; const currentItem = sidebarStore.peek(conversationId); if (!currentItem || currentItem.title !== initialStoredTitle) return; @@ -605,8 +617,13 @@ export function useConversationHistoryActions(params: UseConversationHistoryActi return; } + if (shouldPersist && !shouldPersist()) return; markLocalHistorySnapshotSynced(conversationId, Number.MAX_SAFE_INTEGER); const summary = await renameChatHistory(conversationId, resolvedTitle); + // The replacement run may have taken ownership while the rename IPC + // was in flight. Do not publish the stale title into the new run's + // sidebar state. + if (shouldPersist && !shouldPersist()) return; markLocalHistorySnapshotSynced(summary.id, summary.updatedAt); sidebarStore.upsertLocal({ ...summary, isPending: undefined }); }) diff --git a/crates/agent-gui/src/pages/chat/runtime/chatRunFinalization.ts b/crates/agent-gui/src/pages/chat/runtime/chatRunFinalization.ts index 0ff26fb39..eb336ec15 100644 --- a/crates/agent-gui/src/pages/chat/runtime/chatRunFinalization.ts +++ b/crates/agent-gui/src/pages/chat/runtime/chatRunFinalization.ts @@ -124,6 +124,53 @@ export async function trackTerminalHistoryPersist( } } +/** + * Keep a terminal write attached to the run that produced it. A force-stopped + * run can finish after its replacement has installed a new controller; it + * must not enqueue its old snapshot or mark the replacement as persist-failed. + */ +export async function persistOwnedTerminalHistory(params: { + input: T; + ownsRun: () => boolean; + persist: (input: T & { shouldPersist: () => boolean }) => Promise; + markFailed: () => void; + options?: { + maxAttempts?: number; + retryDelayMs?: number; + sleep?: (ms: number) => Promise; + onRetry?: (error: unknown, attempt: number, maxAttempts: number) => void; + }; +}): Promise { + const { input, ownsRun, persist, markFailed, options } = params; + if (!ownsRun()) return false; + + let ownershipLost = false; + const persistWhileOwned = async () => { + if (!ownsRun()) { + ownershipLost = true; + return true; + } + const persisted = await persist({ ...input, shouldPersist: ownsRun }); + if (!ownsRun()) { + ownershipLost = true; + return true; + } + return persisted; + }; + + try { + const persisted = await persistTerminalHistoryWithRetry(persistWhileOwned, options); + if (!persisted && !ownershipLost) { + markFailed(); + } + return ownershipLost ? false : persisted; + } catch (error) { + if (!ownsRun()) return false; + markFailed(); + throw error; + } +} + /** * Ordered chat-run finalization: history persistence must land before the * gateway stream close / terminal runtime snapshot become observable remotely diff --git a/crates/agent-gui/src/pages/chat/runtime/useManualCompaction.ts b/crates/agent-gui/src/pages/chat/runtime/useManualCompaction.ts index 2acb88664..3654e7208 100644 --- a/crates/agent-gui/src/pages/chat/runtime/useManualCompaction.ts +++ b/crates/agent-gui/src/pages/chat/runtime/useManualCompaction.ts @@ -5,6 +5,7 @@ import { useCallback } from "react"; import { readMessageContextUsage } from "../../../lib/chat/compaction/contextUsageMetadata"; import type { CompactionController, + CompactionObserver, CompactionSinks, ManualCompactionOutcome, ManualContextUsageSnapshot, @@ -97,6 +98,7 @@ export function useManualCompaction(params: { t: (key: string) => string; currentConversationIdRef: MutableRefObject; isConversationRunning: (conversationId: string) => boolean; + getConversationAbortController: (conversationId: string) => AbortController | null; setConversationRunningState: (conversationId: string, value: boolean) => void; setConversationAbortController: ( conversationId: string, @@ -141,6 +143,7 @@ export function useManualCompaction(params: { t, currentConversationIdRef, isConversationRunning, + getConversationAbortController, setConversationRunningState, setConversationAbortController, setConversationStopHandler, @@ -198,6 +201,9 @@ export function useManualCompaction(params: { let stopHandlerRegistered = false; let stopRequestVersion: number | null = null; let flushTrajectory: (() => Promise) | null = null; + const ownsManualRun = () => + !stopHandlerRegistered || + getConversationAbortController(conversationId) === cancellation.userStop; // 停止处理器与发送链路 handleConversationStop 同款：记录版本号供 finally // 消费 stop intent；abort 使 compactManually 中止（controller 返回 aborted）。 const handleStop: ConversationStopHandler = (options) => { @@ -322,13 +328,17 @@ export function useManualCompaction(params: { let compactionFailureMessage = ""; const sinks: CompactionSinks = { - applyState: (state) => - updateConversationRuntimeEntry(conversationId, (prev) => ({ ...prev, state })), + applyState: (state) => { + if (!ownsManualRun()) return; + updateConversationRuntimeEntry(conversationId, (prev) => ({ ...prev, state })); + }, applyStateMidRun: (state) => { + if (!ownsManualRun()) return; updateConversationRuntimeEntry(conversationId, (prev) => ({ ...prev, state })); resetLiveTranscript(transcriptStore); }, publishStatus: (status) => { + if (!ownsManualRun()) return; if (status.phase === "failed") compactionFailureMessage = status.message; updateConversationRuntimeEntry(conversationId, (prev) => ({ ...prev, @@ -336,13 +346,17 @@ export function useManualCompaction(params: { })); }, setBridgeToolStatus: (status, isCompaction = false) => { + if (!ownsManualRun()) return; gatewayBridgeEvents.queueToolStatus(status, isCompaction); updateToolStatus(status, transcriptStore); }, - queueCheckpoint: (state, contextUsageTokens) => - gatewayBridgeEvents.queueCheckpoint(state, contextUsageTokens), - persist: (state) => - persistConversation({ + queueCheckpoint: (state, contextUsageTokens) => { + if (!ownsManualRun()) return; + gatewayBridgeEvents.queueCheckpoint(state, contextUsageTokens); + }, + persist: (state) => { + if (!ownsManualRun()) return Promise.resolve(false); + return persistConversation({ conversationId, sessionId: runtimeEntry.sessionId, providerId, @@ -353,10 +367,14 @@ export function useManualCompaction(params: { fallbackTitle: t("chat.pendingTitle"), createdAt: runtimeEntry.createdAt, titlePromise: null, - }), + shouldPersist: ownsManualRun, + }); + }, // 压缩把携带 memory 增量块的 user 消息移出 active segment;丢弃注入 // 状态后,下一轮发送的 getSystemText 回退到现读快照并重新冻结。 - onCompacted: () => memoryTurnInjection.invalidate(conversationId), + onCompacted: () => { + if (ownsManualRun()) memoryTurnInjection.invalidate(conversationId); + }, }; const compactionController = getCompactionController(conversationId); @@ -373,9 +391,10 @@ export function useManualCompaction(params: { }); } }, + 1, ); flushTrajectory = trajectoryRecording.recorder.flush; - compactionController.setObserver({ + const compactionObserver: CompactionObserver = { onStart: ({ trigger }) => { trajectoryRecording.recorder.compactionStart({ standalone: trigger === "manual" }); }, @@ -391,13 +410,14 @@ export function useManualCompaction(params: { updateTrajectoryRecorderSegment(conversationId, newSegmentIndex); } }, - }); + }; const outcome = await compactionController.compactManually( { providerId, model, runtime, cancellation, + observer: compactionObserver, sinks, buildPreparedContext: (state, tools, options) => buildPreparedConversationContext({ @@ -430,6 +450,7 @@ export function useManualCompaction(params: { { tools: runtimeEntry.state.meta.tools, onProceed: () => { + if (!ownsManualRun()) return; proceeded = true; if (hasRemoteGatewayTarget) { // 与 useSendChatTurn 同款注册镜像：userMessage 取最近一条用户消息 @@ -471,27 +492,36 @@ export function useManualCompaction(params: { try { result = await run(); - if (result.status === "failed" && result.message && isCurrentConversation()) { + if ( + ownsManualRun() && + result.status === "failed" && + result.message && + isCurrentConversation() + ) { setErrorMessage(result.message); } return result; } catch (error) { const message = error instanceof Error ? error.message : String(error); - if (isCurrentConversation()) { + if (ownsManualRun() && isCurrentConversation()) { setErrorMessage(message); } result = { status: "failed", message }; return result; } finally { + const ownsRunOnFinalization = + stopHandlerRegistered && getConversationAbortController(conversationId) === cancellation.userStop; const flushRecordedTrajectory = flushTrajectory as (() => Promise) | null; if (flushRecordedTrajectory !== null) { await flushRecordedTrajectory(); } if (stopHandlerRegistered) { clearConversationStopHandler(conversationId, handleStop); - setConversationAbortController(conversationId, null); + if (ownsRunOnFinalization) { + setConversationAbortController(conversationId, null); + } } - if (runningStateClaimed) { + if (runningStateClaimed && ownsRunOnFinalization) { setConversationRunningState(conversationId, false); } // 停止意图必须消费，否则残留会吞掉该会话的下一条消息。版本号不匹配 @@ -546,6 +576,7 @@ export function useManualCompaction(params: { finishGatewayRunMirror, flushGatewayBridgeEventsForRequest, getCompactionController, + getConversationAbortController, getConversationLiveTranscriptStore, isConversationRunning, persistConversation, diff --git a/crates/agent-gui/src/pages/chat/runtime/useSendChatTurn.ts b/crates/agent-gui/src/pages/chat/runtime/useSendChatTurn.ts index 1170a3d53..18718e457 100644 --- a/crates/agent-gui/src/pages/chat/runtime/useSendChatTurn.ts +++ b/crates/agent-gui/src/pages/chat/runtime/useSendChatTurn.ts @@ -23,6 +23,8 @@ import { invoke } from "@tauri-apps/api/core"; import type { Dispatch, MutableRefObject, SetStateAction } from "react"; import { useCallback } from "react"; import { createHookRunScope } from "../../../lib/automation/hookRunner"; +import { raceWithAbort } from "../../../lib/cancellation/abortRace"; +import type { CompactionObserver } from "../../../lib/chat/compaction/controller"; import { buildPersistableMessagesFromSnapshot, type SuppressedToolTraceSnapshot, @@ -37,6 +39,7 @@ import { type HistoryMessageRef, setTaskListState, } from "../../../lib/chat/conversation/conversationState"; +import type { LiveTranscriptStore } from "../../../lib/chat/conversation/liveTranscriptStore"; import { createConversationHookLifecycle, createGatewayBridgeEventController, @@ -120,9 +123,9 @@ import { } from "./chatPageRuntime"; import { finalizeChatRunInOrder, + persistOwnedTerminalHistory, releaseChatRunUi, settleChatRunFinalization, - trackTerminalHistoryPersist, } from "./chatRunFinalization"; import { buildPreparedContext as buildPreparedConversationContext, @@ -174,6 +177,7 @@ type UseSendChatTurnParams = { buildRuntimeEntryFromVisibleState: ChatPageRuntimeStore["buildRuntimeEntryFromVisibleState"]; updateConversationRuntimeEntry: ChatPageRuntimeStore["updateConversationRuntimeEntry"]; setConversationAbortController: ChatPageRuntimeStore["setConversationAbortController"]; + getConversationAbortController: ChatPageRuntimeStore["getConversationAbortController"]; getConversationStopRequestVersion: ChatPageRuntimeStore["getConversationStopRequestVersion"]; isConversationStopRequested: ChatPageRuntimeStore["isConversationStopRequested"]; consumeConversationStop: ChatPageRuntimeStore["consumeConversationStop"]; @@ -250,6 +254,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { buildRuntimeEntryFromVisibleState, updateConversationRuntimeEntry, setConversationAbortController, + getConversationAbortController, getConversationStopRequestVersion, isConversationStopRequested, consumeConversationStop, @@ -424,12 +429,14 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { gatewayBridgeRequest?.conversationId ?? currentConversationIdRef.current, }); const updateGatewayBridgeToolStatus = (status: string | null, isCompaction = false) => { + if (!ownsConversationRun()) return; gatewayBridgeEvents.queueToolStatus(status, isCompaction); updateToolStatus(status, transcriptStore); }; // Mirrors the live retry-attempt list to remote WebUI clients alongside // the local live-transcript update. const updateGatewayBridgeRetryAttempts: typeof updateRetryAttempts = (attempts, store) => { + if (!ownsConversationRun()) return; gatewayBridgeEvents.queueRetryAttempts(attempts); updateRetryAttempts(attempts, store); }; @@ -497,6 +504,10 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { gatewayBridgeRequest?.runtimeControlsOverride ?? overrides?.runtimeControlsOverride ?? settings.chatRuntimeControls; + // Runtime callbacks can outlive a force-stopped turn. Keep the controller + // identity available before constructing the failover callbacks so late + // callbacks cannot mutate the replacement turn's runtime entry. + let activeTurnController: AbortController | null = null; const providerConfig = createProviderRuntimeConfig(provider, model, runtimeControls); // cc-switch style auto-failover plan for this turn (shared by the agent // and text runtimes). The switch callback makes the winning fallback the @@ -513,6 +524,12 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { round: number; errorMessage: string; }) => { + if ( + activeTurnController === null || + getConversationAbortController(conversationId) !== activeTurnController + ) { + return; + } const nextSelectedModel = event.target?.selectedModel ?? failoverPlan.primary.selectedModel; updateConversationRuntimeEntry(conversationId, (prev) => @@ -538,6 +555,12 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { : undefined; const handleMemoryExtractionModelFailure = memoryExtractionModel ? (failedModel: { selectedModel?: SelectedModel }) => { + if ( + activeTurnController === null || + getConversationAbortController(conversationId) !== activeTurnController + ) { + return; + } const failedSelectedModel = failedModel.selectedModel; setSettings((prev) => { if (!selectedModelsMatch(prev.memory.summaryModel, failedSelectedModel)) { @@ -606,7 +629,10 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { setIsImportingPastedText(false); } } - if (isConversationStopRequested(conversationId)) { + const stopRequestedForActiveRun = + isConversationStopRequested(conversationId) && + getConversationAbortController(conversationId) !== null; + if (stopRequestedForActiveRun) { const stopRequestVersion = getConversationStopRequestVersion(conversationId); if (gatewayBridgeRequest) { void invoke("gateway_chat_cancel_request", { @@ -621,6 +647,12 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { void settleChatRunFinalization(gatewayBridgeEvents.close()); return false; } + // Force-stop clears the old controller before its provider finally + // unwinds. A new manual message after that point is a new run, so the old + // stop intent must not silently discard it. + if (isConversationStopRequested(conversationId)) { + consumeConversationStop(conversationId, getConversationStopRequestVersion(conversationId)); + } const userMessage = createUserMessageWithUploads(text, uploadedFiles, Date.now()); if (!userMessage) { @@ -651,6 +683,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { // 轮次级取消：会话 abort controller 只注册 userStop 一次；每个 LLM 请求 // （主请求/压缩摘要/标题任务）各自派生子 scope，杜绝 abort 换代丢停止的窗口。 const cancellation = createTurnCancellation(); + activeTurnController = cancellation.userStop; const conversationDebugLogger = createStreamDebugLogger({ enabled: effectiveIsAgentDevExecutionMode, conversationId, @@ -761,6 +794,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { let terminalHistoryPersistPromise: Promise | null = null; let runCleanupPromise: Promise = Promise.resolve(); let compactionBound = false; + let compactionBindingGeneration: number | null = null; let runStopRequestVersion: number | null = null; function registerGatewayRuntimeRun(state: GatewayRuntimeSnapshotState) { @@ -808,12 +842,14 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { async function persistTerminalConversation( input: Parameters[0], ) { - return trackTerminalHistoryPersist( - () => persistConversationWithHistorySync(input), - () => { + return persistOwnedTerminalHistory({ + input, + ownsRun: ownsConversationRun, + persist: persistConversationWithHistorySync, + markFailed: () => { terminalHistoryPersistFailed = true; }, - ); + }); } function acknowledgeGatewayRunStarted() { @@ -838,9 +874,9 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { return; } conversationRunStarted = true; + setConversationAbortController(conversationId, cancellation.userStop); applyConversationState(nextConversationState); resetLiveTranscript(transcriptStore); - setConversationAbortController(conversationId, cancellation.userStop); if (isConversationStopRequested(conversationId)) { cancellation.userStop.abort(); } @@ -854,9 +890,40 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { } } + function ownsConversationRun() { + return ( + conversationRunStarted && + getConversationAbortController(conversationId) === cancellation.userStop + ); + } + + // The provider runtime may finish callbacks after Stop has released this run's + // controller. Keep every live-transcript sink scoped to the controller that + // created it so a late old callback cannot overwrite a replacement run. + const runResetLiveTranscript = (store: LiveTranscriptStore) => { + if (ownsConversationRun()) resetLiveTranscript(store); + }; + const runSettleLiveTranscript = (store: LiveTranscriptStore) => { + if (ownsConversationRun()) settleLiveTranscript(store); + }; + const runAppendDraftAssistantText = (delta: string, store: LiveTranscriptStore) => { + if (ownsConversationRun()) appendDraftAssistantText(delta, store); + }; + const runBatchLiveRoundsUpdate = ( + updater: Parameters[0], + store: LiveTranscriptStore, + ) => { + if (ownsConversationRun()) batchLiveRoundsUpdate(updater, store); + }; + const runUpdateToolStatus = (status: string | null, store: LiveTranscriptStore) => { + if (ownsConversationRun()) updateToolStatus(status, store); + }; function releaseConversationRunUi() { if (!conversationRunStarted || conversationUiReleased) return; conversationUiReleased = true; + // A force-stopped run can finish after a new turn has installed its own + // controller. Only the current controller owner may clear shared UI. + if (!ownsConversationRun()) return; releaseChatRunUi({ clearAbortController: () => setConversationAbortController(conversationId, null), clearSendingState: () => setConversationSendingState(conversationId, false), @@ -864,6 +931,16 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { }); } + function releaseCompactionTurn() { + if (!compactionBound) return; + if (compactionBindingGeneration === null) { + compaction.unbindTurn(); + } else { + compaction.unbindTurn(compactionBindingGeneration); + } + compactionBound = false; + } + function requestRemoteGatewayCancellation() { if (remoteGatewayCancelRequested) return; remoteGatewayCancelRequested = true; @@ -965,12 +1042,12 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { cancellation.userStop.abort(); requestRemoteGatewayCancellation(); gatewayBridgeEvents.emitError("Cancelled", conversationId); + const ownsRunOnStop = ownsConversationRun(); releaseConversationRunUi(); - if (compactionBound) { - compaction.unbindTurn(); - compactionBound = false; + releaseCompactionTurn(); + if (ownsRunOnStop) { + clearAbortSnapshot(transcriptStore); } - clearAbortSnapshot(transcriptStore); await finalizeConversationRun("cancelled"); clearConversationStopHandler(conversationId, handleConversationStop); consumeConversationStop(conversationId, runStopRequestVersion); @@ -978,6 +1055,20 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { return true; } + async function awaitBeforeRuntime(operation: PromiseLike | T) { + try { + return { + cancelled: false as const, + value: await raceWithAbort(operation, cancellation.userStop.signal), + }; + } catch (error) { + if (cancellation.userStop.signal.aborted) { + return { cancelled: true as const }; + } + throw error; + } + } + async function markLocalGatewayRunStarted() { if (!mirrorsLocalRunToGateway || localGatewayRunStarted) { return; @@ -1084,7 +1175,11 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { }; if (mirrorsLocalRunToGateway) { try { - await markLocalGatewayRunStarted(); + const result = await awaitBeforeRuntime(markLocalGatewayRunStarted()); + if (result.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } } catch (error) { console.warn("gateway_chat_mark_local_started failed", error); } @@ -1094,7 +1189,11 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { } if (overrides?.beforeRuntimeStart) { try { - await overrides.beforeRuntimeStart(); + const result = await awaitBeforeRuntime(overrides.beforeRuntimeStart()); + if (result.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } if (await finishRequestedStopBeforeRuntime()) { return true; } @@ -1114,11 +1213,18 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { } if (!initialUserTurnPersisted) { - trajectoryTurn = await resolveTrajectoryTurnNumber({ - conversationId, - currentUserPersisted: false, - fallbackTurn: nextConversationState.meta.totalMessageCount, - }); + const result = await awaitBeforeRuntime( + resolveTrajectoryTurnNumber({ + conversationId, + currentUserPersisted: false, + fallbackTurn: nextConversationState.meta.totalMessageCount, + }), + ); + if (result.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } + trajectoryTurn = result.value; if (await finishRequestedStopBeforeRuntime()) { return true; } @@ -1140,10 +1246,16 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { createdAt, titlePromise, titleLookahead: true, + shouldPersist: ownsConversationRun, }); const initialPersist = initialPersistPromise; if (overrides?.afterInitialHistoryPersist && !overrides.beforeRuntimeStart) { - const persisted = await initialPersist; + const initialPersistResult = await awaitBeforeRuntime(initialPersist); + if (initialPersistResult.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } + const persisted = initialPersistResult.value; if (await finishRequestedStopBeforeRuntime()) { return true; } @@ -1160,7 +1272,11 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { return true; } try { - await overrides.afterInitialHistoryPersist(); + const result = await awaitBeforeRuntime(overrides.afterInitialHistoryPersist()); + if (result.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } if (await finishRequestedStopBeforeRuntime()) { return true; } @@ -1200,10 +1316,17 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { void initialPersistConfirmation; } if (gatewayBridgeRequest || hasRemoteGatewayTarget) { - const persisted = await initialPersist.catch((error) => { - console.warn("initial conversation history persist before gateway stream failed", error); - return false; - }); + const initialPersistResult = await awaitBeforeRuntime( + initialPersist.catch((error) => { + console.warn("initial conversation history persist before gateway stream failed", error); + return false; + }), + ); + if (initialPersistResult.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } + const persisted = initialPersistResult.value; if (!persisted) { console.warn("gateway stream started before initial user turn was persisted"); } @@ -1211,20 +1334,32 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { return true; } } - await gatewayBridgeEvents.queueUserMessage(text, uploadedFiles, { - messageId: pendingUserMessage.id, - baseMessageRef: overrides?.editResendBaseMessageRef, - // The new message's own stable identity: lets remote transcripts bind - // their user bubble's messageRef immediately, so a follow-up edit of - // this message can anchor its rebase without a history round-trip. - messageRef: findHistoryMessageRefByMessageId(nextConversationState, pendingUserMessage.id), - }); + const queueUserMessageResult = await awaitBeforeRuntime( + gatewayBridgeEvents.queueUserMessage(text, uploadedFiles, { + messageId: pendingUserMessage.id, + baseMessageRef: overrides?.editResendBaseMessageRef, + // The new message's own stable identity: lets remote transcripts bind + // their user bubble's messageRef immediately, so a follow-up edit of + // this message can anchor its rebase without a history round-trip. + messageRef: findHistoryMessageRefByMessageId(nextConversationState, pendingUserMessage.id), + }), + ); + if (queueUserMessageResult.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } if (effectiveIsAgentMode) { try { - await invoke("checkpoint_begin_turn", { - conversation_id: conversationId, - turn_id: pendingUserMessage.id, - }); + const checkpointResult = await awaitBeforeRuntime( + invoke("checkpoint_begin_turn", { + conversation_id: conversationId, + turn_id: pendingUserMessage.id, + }), + ); + if (checkpointResult.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } } catch (error) { console.warn("checkpoint turn boundary failed", error); } @@ -1233,10 +1368,17 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { return true; } acknowledgeGatewayRunStarted(); - const [{ memoryTurnInjection }, { buildMemoryOverviewSection }] = await Promise.all([ - import("../../../lib/chat/memory/injectionController"), - import("../../../lib/memory/prompts/injection"), - ]); + const promptModulesResult = await awaitBeforeRuntime( + Promise.all([ + import("../../../lib/chat/memory/injectionController"), + import("../../../lib/memory/prompts/injection"), + ]), + ); + if (promptModulesResult.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } + const [{ memoryTurnInjection }, { buildMemoryOverviewSection }] = promptModulesResult.value; let skillsPrompt = ""; let memoryPrompt = ""; /** 本轮 `/skill-name` 显式提及块;没有提及时恒为空串,不会挂出任何内容。 */ @@ -1268,10 +1410,11 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { }); } }, + trajectoryTurn, ); // 压缩有四条触发路径，逐个调用点埋点必漏；订阅控制器生命周期一次覆盖全部。 // manual 发生在两轮之间，不属于任何 turn。 - compaction.setObserver({ + const compactionObserver: CompactionObserver = { onStart: ({ trigger }) => { trajectoryRecording.recorder.compactionStart({ standalone: trigger === "manual" }); }, @@ -1287,7 +1430,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { updateTrajectoryRecorderSegment(conversationId, newSegmentIndex); } }, - }); + }; function buildPreparedContext( state: ConversationViewState, @@ -1345,12 +1488,13 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { }); } - compaction.bindTurn({ + compactionBindingGeneration = compaction.bindTurn({ providerId, model, runtime: providerConfig, cancellation, debugLogger: compactionDebugLogger, + observer: compactionObserver, buildPreparedContext, buildResumeContext, presend: { @@ -1363,16 +1507,24 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { sinks: { applyState: applyConversationState, applyStateMidRun: rebaseConversationStateDuringRun, - publishStatus: (status) => + publishStatus: (status) => { + if (!ownsConversationRun()) return; updateConversationRuntimeEntry(conversationId, (prev) => ({ ...prev, compactionStatus: status, - })), - setBridgeToolStatus: updateGatewayBridgeToolStatus, - queueCheckpoint: (state, contextUsageTokens) => - gatewayBridgeEvents.queueCheckpoint(state, contextUsageTokens), - persist: (state) => - persistConversation({ + })); + }, + setBridgeToolStatus: (status, isCompaction) => { + if (!ownsConversationRun()) return; + updateGatewayBridgeToolStatus(status, isCompaction); + }, + queueCheckpoint: (state, contextUsageTokens) => { + if (!ownsConversationRun()) return; + gatewayBridgeEvents.queueCheckpoint(state, contextUsageTokens); + }, + persist: async (state) => { + if (!ownsConversationRun()) return false; + return persistConversation({ conversationId, sessionId, providerId, @@ -1383,8 +1535,11 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { fallbackTitle, createdAt, titlePromise, - }), + shouldPersist: ownsConversationRun, + }); + }, restoreComposer: (composerText, restoredUploads) => { + if (!ownsConversationRun()) return; if (isConversationVisible() && typeof composerText === "string") { composerRef.current?.setText(composerText); composerRef.current?.focus(); @@ -1392,8 +1547,9 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { setPendingUploadsForConversation(conversationId, restoredUploads); }, persistRollback: async (state) => { + if (!ownsConversationRun()) return false; abortedConversationCommitted = true; - await persistConversationWithHistorySync({ + return persistConversationWithHistorySync({ conversationId, sessionId, providerId, @@ -1404,12 +1560,17 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { fallbackTitle, createdAt, titlePromise, + shouldPersist: ownsConversationRun, }); }, // 压缩把携带 memory 增量块的 user 消息移出 active segment,增量对模型 // 永久不可见;丢弃注入状态,下一轮把 fresh 快照重冻结进 system 段 —— // 压缩本来就要重建前缀,这次重冻结免费。 - onCompacted: () => memoryTurnInjection.invalidate(conversationId), + onCompacted: () => { + if (ownsConversationRun()) { + memoryTurnInjection.invalidate(conversationId); + } + }, }, }); compactionBound = true; @@ -1423,7 +1584,12 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { let byName = new Map(skillsList.map((s) => [s.name, s])); let missing = selectedSkillNames.filter((n) => !byName.has(n)); if (missing.length > 0 && workspaceResources.mode !== "custom") { - const fresh = await refreshSkills(); + const freshResult = await awaitBeforeRuntime(refreshSkills()); + if (freshResult.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } + const fresh = freshResult.value; if (await finishRequestedStopBeforeRuntime()) { return true; } @@ -1492,7 +1658,14 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { // 消息上的那个断点,不额外占用 Anthropic 的 4 个 cache_control 名额。 let memoryOverview: string | null = null; try { - memoryOverview = await buildMemoryOverviewSection(effectiveWorkdir); + const memoryOverviewResult = await awaitBeforeRuntime( + buildMemoryOverviewSection(effectiveWorkdir), + ); + if (memoryOverviewResult.cancelled) { + await finishRequestedStopBeforeRuntime(); + return true; + } + memoryOverview = memoryOverviewResult.value; } catch (error) { console.warn("Failed to build memory overview prompt", error); // null 表示这轮没读到,基线维持原样;空串是「一条记忆都没有」,属于正常内容。 @@ -1524,6 +1697,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { conversationId, workdir: effectiveWorkdir, onWarning: (warning) => { + if (!ownsConversationRun()) return; updateConversationRuntimeEntry(conversationId, (prev) => ({ ...prev, hookWarning: formatHookWarningMessage(settings.locale, t, warning), @@ -1544,6 +1718,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { suppressedToolTrace: [], }; const commitVisibleAbortedConversation = () => { + if (!ownsConversationRun()) return false; if (abortedConversationCommitted) return true; const snapshot = getAbortSnapshot(transcriptStore); @@ -1579,6 +1754,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { }; const commitErroredConversation = (rawMessage: string) => { + if (!ownsConversationRun()) return; const snapshot = getAbortSnapshot(transcriptStore); const partialMessages = buildPersistableMessagesFromSnapshot({ executionMode: effectiveExecutionMode, @@ -1621,6 +1797,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { function applyConversationState(nextState: ConversationViewState) { nextConversationState = nextState; + if (!ownsConversationRun()) return; updateConversationRuntimeEntry(conversationId, (prev) => ({ ...prev, state: nextState, @@ -1631,6 +1808,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { // Once a compaction/prune result is committed into visible history, the // corresponding live transcript becomes stale and must be cleared. applyConversationState(nextState); + if (!ownsConversationRun()) return; resetLiveTranscript(transcriptStore); } @@ -1642,6 +1820,9 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { runId: gatewayBridgeRequestId, getState: () => nextConversationState.meta.taskList, commitState: async (taskList) => { + if (!ownsConversationRun()) { + throw new Error("Stale conversation run cannot persist task state."); + } const persisted = await persistConversationWithHistorySync({ conversationId, sessionId, @@ -1653,10 +1834,14 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { fallbackTitle, createdAt, titlePromise, + shouldPersist: ownsConversationRun, }).catch(() => false); if (!persisted) { throw new Error("Failed to persist task state."); } + if (!ownsConversationRun()) { + throw new Error("Stale conversation run cannot apply task state."); + } applyConversationState(setTaskListState(nextConversationState, taskList)); }, }; @@ -1682,6 +1867,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { skillsRootDir: skillsRootDirForTools, skillAccessPolicy: skillAccessPolicyForTools, onManagedSkillsChanged: (change) => { + if (!ownsConversationRun()) return; if (change.action !== "delete") { enableManagedSkills(change.names); return; @@ -1701,6 +1887,7 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { commandSafetyMode: effectiveCommandSafetyMode, planModeEnabled: effectivePlanModeEnabled, applyMcpOps: (ops) => { + if (!ownsConversationRun()) return; const removedIds = ops.filter((op) => op.kind === "remove").map((op) => op.serverId); setSettings((prev) => removeWorkspaceResourceReferences(applyMcpOpsToAppSettings(prev, ops), { @@ -1715,11 +1902,13 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { sshManagerRemoteAllowed: !gatewayBridgeRequest || settings.remote.enableWebSshTerminal === true, onSshSessionsChanged: (change) => { + if (!ownsConversationRun()) return; if (change.action === "create") { ensureSshTunnelToolTab(change.projectPathKey); } }, onTunnelsChanged: (change) => { + if (!ownsConversationRun()) return; if (change.action === "create") { ensureTunnelToolTab(change.projectPathKey); } @@ -1742,10 +1931,10 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { buildPreparedContext, compaction, cancellation, - resetLiveTranscript, - settleLiveTranscript, - batchLiveRoundsUpdate, - updateToolStatus, + resetLiveTranscript: runResetLiveTranscript, + settleLiveTranscript: runSettleLiveTranscript, + batchLiveRoundsUpdate: runBatchLiveRoundsUpdate, + updateToolStatus: runUpdateToolStatus, updateRetryAttempts: updateGatewayBridgeRetryAttempts, updatePersistableAgentProgress: (progress) => { persistableAgentProgress = progress; @@ -1790,10 +1979,10 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { buildPreparedContext, compaction, cancellation, - resetLiveTranscript, - settleLiveTranscript, - appendDraftAssistantText, - batchLiveRoundsUpdate, + resetLiveTranscript: runResetLiveTranscript, + settleLiveTranscript: runSettleLiveTranscript, + appendDraftAssistantText: runAppendDraftAssistantText, + batchLiveRoundsUpdate: runBatchLiveRoundsUpdate, updateGatewayBridgeToolStatus, updateRetryAttempts: updateGatewayBridgeRetryAttempts, commitVisibleAbortedConversation, @@ -1819,11 +2008,19 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { hookScope.cancel(); requestRemoteGatewayCancellation(); runCleanupPromise = (async () => { - const rolledBack = await compaction.handleTurnAbort(); + const rolledBack = + compactionBindingGeneration !== null && + compaction.isTurnBound(compactionBindingGeneration) + ? await compaction.handleTurnAbort(compactionBindingGeneration) + : false; if (!rolledBack) { commitVisibleAbortedConversation(); } - if (shouldCreatePendingHistoryItem && !abortedConversationCommitted) { + if ( + ownsConversationRun() && + shouldCreatePendingHistoryItem && + !abortedConversationCommitted + ) { sidebarStore.removeLocal(conversationId); } })(); @@ -1832,18 +2029,18 @@ export function useSendChatTurn(params: UseSendChatTurnParams) { commitErroredConversation(msg || "Request failed"); } gatewayBridgeEvents.emitError(remoteErrorMessage, conversationId); - if (titleJobRef.current?.conversationId === conversationId) { + if (ownsConversationRun() && titleJobRef.current?.conversationId === conversationId) { titleJobRef.current = null; } } finally { + const ownsRunOnFinalization = ownsConversationRun(); releaseConversationRunUi(); - if (compactionBound) { - compaction.unbindTurn(); - compactionBound = false; - } + releaseCompactionTurn(); hookLifecycle.endAgent(); hookScope.close(); - clearAbortSnapshot(transcriptStore); + if (ownsRunOnFinalization) { + clearAbortSnapshot(transcriptStore); + } const stopped = runStopRequestVersion !== null || cancellation.userStop.signal.aborted; if (stopped) { gatewayRuntimeFinalState = "cancelled"; diff --git a/crates/agent-gui/test/chat/chat-stop-timing.test.mjs b/crates/agent-gui/test/chat/chat-stop-timing.test.mjs index 9b8364e79..64a2c31a6 100644 --- a/crates/agent-gui/test/chat/chat-stop-timing.test.mjs +++ b/crates/agent-gui/test/chat/chat-stop-timing.test.mjs @@ -1,8 +1,14 @@ import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; import test from "node:test"; import { createTsModuleLoader } from "../helpers/load-ts-module.mjs"; +const chatComposerBarSource = readFileSync( + new URL("../../../agent-ui/src/pages/chat/ChatComposerBar.tsx", import.meta.url), + "utf8", +); + function createHookHarness() { const refs = []; const states = []; @@ -80,6 +86,43 @@ async function flushPromises() { await new Promise((resolve) => setImmediate(resolve)); } +test("an aborted pre-runtime wait releases without waiting for local work", async () => { + const loader = createTsModuleLoader(); + const { raceWithAbort } = loader.loadModule("src/lib/cancellation/abortRace.ts"); + const localWork = deferred(); + const controller = new AbortController(); + const waiting = raceWithAbort(localWork.promise, controller.signal); + + controller.abort(new Error("cancelled by user")); + + await assert.rejects(waiting, /cancelled by user/); + // Finish the underlying local work after cancellation. Its settlement must + // remain observed and must not turn into an unhandled rejection. + localWork.resolve(); + await flushPromises(); +}); + +test("a pre-aborted wait still observes local work that rejects later", async () => { + const loader = createTsModuleLoader(); + const { raceWithAbort } = loader.loadModule("src/lib/cancellation/abortRace.ts"); + const localWork = deferred(); + const controller = new AbortController(); + controller.abort(new Error("already cancelled")); + + await assert.rejects(raceWithAbort(localWork.promise, controller.signal), /already cancelled/); + localWork.reject(new Error("late local failure")); + await flushPromises(); +}); + +test("a queued draft keeps a direct Stop control available", () => { + assert.match( + chatComposerBarSource, + /\{canQueueDraftWhileSending \? \(\s*/); +}); + /** * Per-conversation live transcript stores, matching * useLiveTranscriptController: every conversation owns its own store, so a @@ -695,6 +738,53 @@ test("terminal history persistence marks both false results and thrown errors", assert.equal(failures, 2); }); +test("a stale terminal run cannot enqueue or publish a history snapshot", async () => { + const loader = createTsModuleLoader(); + const { persistOwnedTerminalHistory } = loader.loadModule( + "src/pages/chat/runtime/chatRunFinalization.ts", + ); + let ownsRun = false; + let persistCalls = 0; + let markedFailed = 0; + + const skipped = await persistOwnedTerminalHistory({ + input: { state: "old terminal snapshot" }, + ownsRun: () => ownsRun, + persist: async () => { + persistCalls += 1; + return true; + }, + markFailed: () => { + markedFailed += 1; + }, + }); + + assert.equal(skipped, false); + assert.equal(persistCalls, 0); + assert.equal(markedFailed, 0); + + ownsRun = true; + let persistenceGuard; + const persisted = await persistOwnedTerminalHistory({ + input: { state: "current terminal snapshot" }, + ownsRun: () => ownsRun, + persist: async (input) => { + persistCalls += 1; + persistenceGuard = input.shouldPersist; + return true; + }, + markFailed: () => { + markedFailed += 1; + }, + }); + + assert.equal(persisted, true); + assert.equal(persistCalls, 1); + assert.equal(typeof persistenceGuard, "function"); + assert.equal(persistenceGuard(), true); + assert.equal(markedFailed, 0); +}); + test("terminal history persistence retries transient failures before succeeding", async () => { const loader = createTsModuleLoader(); const { persistTerminalHistoryWithRetry, trackTerminalHistoryPersist } = loader.loadModule( diff --git a/crates/agent-gui/test/chat/compaction-controller.test.mjs b/crates/agent-gui/test/chat/compaction-controller.test.mjs index 3ffdd4b82..1d8ae996d 100644 --- a/crates/agent-gui/test/chat/compaction-controller.test.mjs +++ b/crates/agent-gui/test/chat/compaction-controller.test.mjs @@ -140,6 +140,17 @@ function bindController(controller, overrides = {}) { return { cancellation, recorder }; } +test("a stale turn lease cannot unbind a replacement compaction binding", () => { + const controller = new CompactionController(); + const oldLease = controller.bindTurn({ sinks: {} }); + const replacementLease = controller.bindTurn({ sinks: {} }); + + assert.equal(controller.unbindTurn(oldLease), false); + assert.equal(controller.isTurnBound(replacementLease), true); + assert.equal(controller.unbindTurn(replacementLease), true); + assert.equal(controller.isTurnBound(replacementLease), false); +}); + test("pre-send compaction: checkpoint, persist, re-appended user message, paired status", async () => { const controller = new CompactionController(); const baseState = bigState(); @@ -362,6 +373,9 @@ test("a late result cannot settle a newer compaction with the same trigger", asy releaseOld(); await assert.rejects(oldPending, /abort/i); assert.equal(oldBinding.recorder.byKind("persist").length, 0); + // The old task has now unwound. Its finally block must not clear the + // replacement compaction's in-flight guard. + assert.equal(controller.shouldProtectMidStream(1_000_000), false); assert.equal(observed.at(-1)[0], "start"); releaseNew(); @@ -379,6 +393,68 @@ test("a late result cannot settle a newer compaction with the same trigger", asy ); }); +test("a replacement binding keeps a late old terminal on the old observer", async () => { + const controller = new CompactionController(); + const oldObserved = []; + const newObserved = []; + let releaseOld; + const oldGate = new Promise((resolve) => { + releaseOld = resolve; + }); + + const oldBinding = bindController(controller, { + observer: { + onStart: (info) => oldObserved.push(["start", info]), + onEnd: (info) => oldObserved.push(["end", info]), + }, + complete: async () => { + await oldGate; + return summaryResponse(); + }, + }); + const oldPending = controller.compactDuringRun({ trigger: "post-tool", state: bigState() }); + await new Promise((resolve) => setImmediate(resolve)); + + controller.unbindTurn(); + let releaseNew; + const newGate = new Promise((resolve) => { + releaseNew = resolve; + }); + const newBinding = bindController(controller, { + observer: { + onStart: (info) => newObserved.push(["start", info]), + onEnd: (info) => newObserved.push(["end", info]), + }, + complete: async () => { + await newGate; + return summaryResponse(); + }, + }); + const newPending = controller.compactDuringRun({ trigger: "post-tool", state: bigState() }); + await new Promise((resolve) => setImmediate(resolve)); + + releaseOld(); + await assert.rejects(oldPending, /abort/i); + assert.deepEqual( + oldObserved.map(([kind, info]) => [kind, info.status ?? info.trigger]), + [ + ["start", "post-tool"], + ["end", "aborted"], + ], + ); + assert.deepEqual( + newObserved.map(([kind, info]) => [kind, info.status ?? info.trigger]), + [["start", "post-tool"]], + ); + + releaseNew(); + const result = await newPending; + assert.equal(result.outcome, "compacted"); + assert.equal(newObserved.at(-1)[1].status, "complete"); + assert.equal(oldObserved.at(-1)[1].status, "aborted"); + assert.notEqual(oldBinding.cancellation, newBinding.cancellation); +}); + test("summarizer failure degrades to prune and still returns a usable context", async () => { const controller = new CompactionController(); // 大工具输出（200k 字符 ≈ 50k tokens > 40k 保护额度）必须在"最近 2 个用户轮次"之前才可被裁剪。 diff --git a/crates/agent-gui/test/providers/stream-retry.test.mjs b/crates/agent-gui/test/providers/stream-retry.test.mjs index bab095560..62cf2c584 100644 --- a/crates/agent-gui/test/providers/stream-retry.test.mjs +++ b/crates/agent-gui/test/providers/stream-retry.test.mjs @@ -42,6 +42,18 @@ function createErrorStream(errorMessage) { }; } +function createErrorStreamWithHangingResult(errorMessage) { + return { + async *[Symbol.asyncIterator]() { + yield { type: "error", error: createAssistant(undefined, "error", { errorMessage }) }; + await new Promise(() => {}); + }, + async result() { + return await new Promise(() => {}); + }, + }; +} + function createSuccessStream(text) { const assistant = createAssistant(text, "stop"); return { @@ -92,12 +104,140 @@ function createAbortedDoneStream() { }; } +function createNeverYieldingStream() { + return { + async *[Symbol.asyncIterator]() { + await new Promise(() => {}); + }, + async result() { + return await new Promise(() => {}); + }, + }; +} + +function createCommittedIdleStream(text) { + const assistant = createAssistant(text, "stop"); + return { + async *[Symbol.asyncIterator]() { + yield { type: "start", partial: { ...assistant, content: [] } }; + yield { + type: "text_delta", + contentIndex: 0, + delta: text, + partial: { ...assistant, content: [{ type: "text", text }] }, + }; + await new Promise(() => {}); + }, + async result() { + return await new Promise(() => {}); + }, + }; +} + +function createNeverResolvingResultStream() { + return { + async *[Symbol.asyncIterator]() { + return; + }, + async result() { + return await new Promise(() => {}); + }, + }; +} + +function createRejectingIteratorStream(errorMessage) { + return { + [Symbol.asyncIterator]() { + return { + async next() { + throw new Error(errorMessage); + }, + async return() { + return { done: true }; + }, + }; + }, + async result() { + throw new Error(errorMessage); + }, + }; +} + +function createRejectingIteratorWithHangingResultStream(errorMessage) { + return { + [Symbol.asyncIterator]() { + return { + async next() { + throw new Error(errorMessage); + }, + async return() { + return { done: true }; + }, + }; + }, + async result() { + return await new Promise(() => {}); + }, + }; +} + +function createRejectingResultStream(errorMessage) { + return { + async *[Symbol.asyncIterator]() {}, + async result() { + throw new Error(errorMessage); + }, + }; +} + +function createResultOnlyErrorStream(errorMessage) { + return { + async *[Symbol.asyncIterator]() {}, + async result() { + return createAssistant(undefined, "error", { errorMessage }); + }, + }; +} + +function createBufferedEventStream(events) { + return { + [Symbol.asyncIterator]() { + let index = 0; + return { + async next() { + return index < events.length + ? { value: events[index++], done: false } + : { done: true }; + }, + async return() { + return { done: true }; + }, + }; + }, + async result() { + return createAssistant(undefined, "aborted"); + }, + }; +} + async function collectEvents(eventStream) { const events = []; for await (const event of eventStream) events.push(event); return events; } +async function resolveWithin(promise, timeoutMs = 250) { + let timeoutId; + const timeout = new Promise((_, reject) => { + timeoutId = setTimeout(() => reject(new Error(`timed out after ${timeoutMs}ms`)), timeoutMs); + }); + try { + return await Promise.race([promise, timeout]); + } finally { + clearTimeout(timeoutId); + } +} + test("withStreamRetry succeeds after N retryable errors without leaking failed-attempt events", async () => { let calls = 0; const wrapped = withStreamRetry( @@ -120,6 +260,140 @@ test("withStreamRetry succeeds after N retryable errors without leaking failed-a assert.equal(final.content[0].text, "final answer"); }); +test("withStreamRetry retries a provider iterator rejection before content commits", async () => { + let calls = 0; + const wrapped = withStreamRetry( + () => { + calls += 1; + return calls === 1 ? createRejectingIteratorStream("fetch failed") : createSuccessStream("recovered"); + }, + { maxAttempts: 3 }, + ); + + const events = await collectEvents(wrapped); + assert.equal(calls, 2); + assert.deepEqual(events.map((event) => event.type), ["start", "text_delta", "done"]); + assert.equal((await wrapped.result()).content[0].text, "recovered"); +}); + +test("withStreamRetry does not await result after an iterator rejection", async () => { + let calls = 0; + const wrapped = withStreamRetry( + () => { + calls += 1; + return calls === 1 + ? createRejectingIteratorWithHangingResultStream("network connection reset") + : createSuccessStream("recovered"); + }, + { maxAttempts: 2, idleTimeoutMs: 5_000 }, + ); + + const events = await resolveWithin(collectEvents(wrapped), 1_500); + assert.equal(calls, 2); + assert.deepEqual(events.map((event) => event.type), ["start", "text_delta", "done"]); + assert.equal((await wrapped.result()).content[0].text, "recovered"); +}); + +test("withStreamRetry retries an explicit terminal error without awaiting a hanging result", async () => { + let calls = 0; + const wrapped = withStreamRetry( + () => { + calls += 1; + return calls === 1 + ? createErrorStreamWithHangingResult("network connection reset") + : createSuccessStream("recovered"); + }, + { maxAttempts: 2, idleTimeoutMs: 5_000 }, + ); + + const events = await resolveWithin(collectEvents(wrapped), 1_500); + assert.equal(calls, 2); + assert.deepEqual(events.map((event) => event.type), ["start", "text_delta", "done"]); + assert.equal((await wrapped.result()).content[0].text, "recovered"); +}); + +test("withStreamRetry retries a synchronous provider construction failure", async () => { + let calls = 0; + const wrapped = withStreamRetry( + () => { + calls += 1; + if (calls === 1) throw new Error("network unavailable"); + return createSuccessStream("recovered"); + }, + { maxAttempts: 3 }, + ); + + await collectEvents(wrapped); + assert.equal(calls, 2); + assert.equal((await wrapped.result()).content[0].text, "recovered"); +}); + +test("withStreamRetry retries a provider result rejection before content commits", async () => { + let calls = 0; + const wrapped = withStreamRetry( + () => { + calls += 1; + return calls === 1 ? createRejectingResultStream("network timeout") : createSuccessStream("recovered"); + }, + { maxAttempts: 3 }, + ); + + await collectEvents(wrapped); + assert.equal(calls, 2); + assert.equal((await wrapped.result()).content[0].text, "recovered"); +}); + +test("withStreamRetry retries a result-only provider error before content commits", async () => { + let calls = 0; + const wrapped = withStreamRetry( + () => { + calls += 1; + return calls === 1 + ? createResultOnlyErrorStream("503 service unavailable") + : createSuccessStream("recovered"); + }, + { maxAttempts: 3 }, + ); + + const events = await collectEvents(wrapped); + assert.equal(calls, 2); + assert.deepEqual(events.map((event) => event.type), ["start", "text_delta", "done"]); + assert.equal((await wrapped.result()).content[0].text, "recovered"); +}); + +test("withStreamRetry retries an uncommitted idle provider stream", async () => { + let calls = 0; + const wrapped = withStreamRetry( + () => { + calls += 1; + return calls === 1 ? createNeverYieldingStream() : createSuccessStream("recovered"); + }, + { maxAttempts: 2, idleTimeoutMs: 5 }, + ); + + await collectEvents(wrapped); + assert.equal(calls, 2); + assert.equal((await wrapped.result()).content[0].text, "recovered"); +}); + +test("withStreamRetry drops buffered ordinary events after an already-aborted stop", async () => { + const controller = new AbortController(); + controller.abort(new Error("cancelled")); + const assistant = createAssistant("late", "stop"); + const wrapped = withStreamRetry( + () => + createBufferedEventStream([ + { type: "text_delta", contentIndex: 0, delta: "late", partial: assistant }, + { type: "done", message: assistant }, + ]), + { signal: controller.signal }, + ); + + const events = await collectEvents(wrapped); + assert.deepEqual(events.map((event) => event.type), ["error"]); + assert.equal(events[0].reason, "aborted"); +}); + test("withStreamRetry invokes onRetry per attempt and onRetryRecovered once content commits", async () => { let calls = 0; const retryCalls = []; @@ -198,6 +472,24 @@ test("withStreamRetry does not retry once content has been committed", async () assert.equal(final.stopReason, "error"); }); +test("withStreamRetry ends a committed stream that becomes idle without retrying it", async () => { + let calls = 0; + const wrapped = withStreamRetry( + () => { + calls += 1; + return createCommittedIdleStream("partial"); + }, + { maxAttempts: 2, idleTimeoutMs: 5 }, + ); + + const events = await resolveWithin(collectEvents(wrapped), 1_500); + assert.equal(calls, 1); + assert.deepEqual(events.map((event) => event.type), ["start", "text_delta", "error"]); + const final = await wrapped.result(); + assert.equal(final.stopReason, "error"); + assert.match(final.errorMessage, /idle timeout/i); +}); + test("withStreamRetry never retries an aborted stream", async () => { let calls = 0; const wrapped = withStreamRetry(() => { @@ -215,6 +507,44 @@ test("withStreamRetry never retries an aborted stream", async () => { assert.equal(final.stopReason, "aborted"); }); +test("withStreamRetry aborts a provider iterator that never yields", async () => { + const controller = new AbortController(); + let calls = 0; + const wrapped = withStreamRetry( + () => { + calls += 1; + return createNeverYieldingStream(); + }, + { signal: controller.signal }, + ); + const eventsPromise = collectEvents(wrapped); + + controller.abort(new Error("cancelled by user")); + + const events = await resolveWithin(eventsPromise); + assert.equal(calls, 1); + assert.deepEqual(events.map((event) => event.type), ["error"]); + assert.equal(events[0].reason, "aborted"); + assert.equal((await resolveWithin(wrapped.result())).stopReason, "aborted"); +}); + +test("withStreamRetry aborts a provider result that never resolves", async () => { + const controller = new AbortController(); + const wrapped = withStreamRetry(() => createNeverResolvingResultStream(), { + signal: controller.signal, + }); + const eventsPromise = collectEvents(wrapped); + + // Let the eager pump reach source.result() before simulating Stop. + await Promise.resolve(); + controller.abort(new Error("cancelled by user")); + + const events = await resolveWithin(eventsPromise); + assert.deepEqual(events.map((event) => event.type), ["error"]); + assert.equal(events[0].reason, "aborted"); + assert.equal((await resolveWithin(wrapped.result())).stopReason, "aborted"); +}); + test("withStreamRetry respects maxAttempts and surfaces the last failure", async () => { let calls = 0; const wrapped = withStreamRetry( diff --git a/crates/agent-gui/test/trajectory/desktop-live.test.mjs b/crates/agent-gui/test/trajectory/desktop-live.test.mjs index f70577f8b..04509e41c 100644 --- a/crates/agent-gui/test/trajectory/desktop-live.test.mjs +++ b/crates/agent-gui/test/trajectory/desktop-live.test.mjs @@ -90,3 +90,38 @@ test("the recorder registry exclusively owns desktop live trajectory writes", () assert.doesNotMatch(source, /appendDesktopLiveTrajectory/); } }); + +test("late scoped trajectory events publish through their owning run", () => { + const conversationId = "desktop-live-interleaved-runs"; + const publishedByRun = { old: [], replacement: [] }; + clearDesktopLiveTrajectory(conversationId); + + const oldRun = acquireTrajectoryRecorder( + conversationId, + 0, + (events) => publishedByRun.old.push(...events), + 7, + ); + const replacementRun = acquireTrajectoryRecorder( + conversationId, + 0, + (events) => publishedByRun.replacement.push(...events), + 8, + ); + + try { + oldRun.recorder.beginTurn({ turn: 7, messageIndex: 0, text: "old" }); + replacementRun.recorder.beginTurn({ turn: 8, messageIndex: 1, text: "replacement" }); + oldRun.recorder.stepStart(1); + replacementRun.recorder.stepStart(1); + oldRun.recorder.endTurn({ status: "aborted" }); + replacementRun.recorder.endTurn({ status: "complete" }); + + assert.equal(publishedByRun.old.every((event) => event.t === 7), true); + assert.equal(publishedByRun.replacement.every((event) => event.t === 8), true); + assert.equal(publishedByRun.old.some((event) => event.k === "turn_end"), true); + assert.equal(publishedByRun.replacement.some((event) => event.k === "turn_end"), true); + } finally { + discardTrajectoryRecorder(conversationId); + } +}); diff --git a/crates/agent-gui/test/trajectory/recorder.test.mjs b/crates/agent-gui/test/trajectory/recorder.test.mjs index d42f65464..4382586e6 100644 --- a/crates/agent-gui/test/trajectory/recorder.test.mjs +++ b/crates/agent-gui/test/trajectory/recorder.test.mjs @@ -4,7 +4,7 @@ import test from "node:test"; import { createTsModuleLoader } from "../helpers/load-ts-module.mjs"; const loader = createTsModuleLoader(); -const { createTrajectoryRecorder, NOOP_TRAJECTORY_RECORDER } = loader.loadModule( +const { createTrajectoryRecorder, scopeTrajectoryRecorder, NOOP_TRAJECTORY_RECORDER } = loader.loadModule( "src/lib/trajectory/recorder.ts", ); @@ -366,6 +366,27 @@ test("turn end is idempotent across happy-path and finalizer calls", () => { assert.equal(published.find((event) => event.k === "turn_end").st, "complete"); }); +test("interleaved scoped runs close only their own trajectory turn", () => { + const { recorder, published } = harness(); + const first = scopeTrajectoryRecorder(recorder, 1); + const second = scopeTrajectoryRecorder(recorder, 2); + + first.beginTurn({ turn: 1 }); + first.stepStart(1); + second.beginTurn({ turn: 2 }); + second.stepStart(1); + first.endTurn({ status: "aborted" }); + second.endTurn({ status: "complete" }); + + assert.deepEqual( + published.filter((event) => event.k === "turn_end").map((event) => [event.t, event.st]), + [ + [1, "aborted"], + [2, "complete"], + ], + ); +}); + test("new request headers declare the seven-slot runtime layout version", () => { const { recorder, published } = harness(); const headerId = recorder.captureHeader({ diff --git a/crates/agent-ui/src/pages/chat/ChatComposerBar.tsx b/crates/agent-ui/src/pages/chat/ChatComposerBar.tsx index 9449c9a34..e6f5a9ed3 100644 --- a/crates/agent-ui/src/pages/chat/ChatComposerBar.tsx +++ b/crates/agent-ui/src/pages/chat/ChatComposerBar.tsx @@ -1185,6 +1185,22 @@ export const ChatComposerBar = memo(function ChatComposerBar(props: ChatComposer
+ {canQueueDraftWhileSending ? ( + + ) : null}