fix(chat): recover reliably after Stop when a turn stalls

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<T>(operation: PromiseLike<T> | T, signal?: AbortSignal): Promise<T> {
+  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<T>((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),
+    );
+  });
+}

crates/agent-gui/src/lib/chat/compaction/controller.ts

@@ -5,6 +5,7 @@ import {
   positiveTokenCount,
 } from "@liveagent/ui/lib/chat/contextUsage";
 import type { PendingUploadedFile } from "@liveagent/ui/lib/chat/uploadedFiles";
+import { raceWithAbort } from "../../cancellation/abortRace";
 import type { StreamDebugLogger } from "../../debug/agentDebug";
 import type { ProviderId } from "../../settings";
 import { type ConversationViewState, getActiveSegment } from "../conversation/conversationState";
@@ -87,6 +88,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 +189,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 +221,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() {
@@ -235,7 +255,14 @@ export class CompactionController {
     binding: CompactionTurnBinding,
     state: ConversationViewState,
   ): Promise<ConversationViewState> {
-    const persisted = await binding.sinks.persist?.(state);
+    // Checkpoint durability matters, but an already-started write must not
+    // keep a Stop request from releasing the run. raceWithAbort continues to
+    // observe the underlying write after cancellation, so a late failure is
+    // not left as an unhandled rejection.
+    const persisted = await raceWithAbort(
+      binding.sinks.persist?.(state),
+      binding.cancellation.userStop.signal,
+    );
     if (persisted === false || persisted === null) {
       throw new Error("compaction checkpoint persistence failed");
     }
@@ -324,8 +351,10 @@ export class CompactionController {
     includeUploadedFilesMetadata?: boolean;
   }): Promise<boolean> {
     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 +449,9 @@ export class CompactionController {
       );
       return true;
     } catch (error) {
+      if (!ownsBinding()) {
+        throw createCompactionAbortError();
+      }
       if (this.isAbortOutcome(scope.controller.signal, error)) {
         throw error;
       }
@@ -441,8 +473,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 +492,11 @@ export class CompactionController {
     manualContextUsage?: ManualContextUsageSnapshot;
   }): Promise<CompactionDuringRunResult> {
     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 +630,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 +669,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 +694,7 @@ export class CompactionController {
     },
   ): Promise<ManualCompactionOutcome> {
     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 +722,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 +762,10 @@ export class CompactionController {
   }
 
   // 用户中止后的统一善后：有快照则回滚（恢复状态/输入框/可选持久化）并返回 true。
-  async handleTurnAbort(): Promise<boolean> {
+  async handleTurnAbort(expectedGeneration?: number): Promise<boolean> {
+    if (expectedGeneration !== undefined && !this.isTurnBound(expectedGeneration)) {
+      return false;
+    }
     const binding = this.binding;
     const snapshot = this.rollbackSnapshot;
     this.rollbackSnapshot = null;
@@ -808,6 +852,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 +866,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 +890,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 +914,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 +934,7 @@ export class CompactionController {
       return false;
     }
     this.notifyObserver(() =>
-      this.observer?.onEnd({
+      this.activeObserver()?.onEnd({
         trigger,
         status: "aborted",
         ...(this.observedTokensBefore === undefined

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<void>).then === "function") {
+        (sendResult as Promise<void>).catch((error) => {
+          console.warn("error event failed", error);
+        });
+      }
     },
     close() {
       streamClosed = true;

crates/agent-gui/src/lib/chat/runner/agentRunner.ts

@@ -14,7 +14,12 @@ import {
   mergeHostedSearchBlocks,
 } from "@liveagent/ui/lib/chat/hostedSearch";
 import type { PreparedProxyRequest } from "@liveagent/ui/lib/providers/proxy";
-import { buildStreamRequestDebugPayload, type StreamDebugLogger } from "../../debug/agentDebug";
+import { raceWithAbort } from "../../cancellation/abortRace";
+import {
+  buildStreamRequestDebugPayload,
+  flushDebugLoggerInBackground,
+  type StreamDebugLogger,
+} from "../../debug/agentDebug";
 import { capturePrefixShape, comparePrefixShape } from "../../debug/prefixCacheShape";
 import { readPreviousPrefixShape, recordPrefixShape } from "../../debug/prefixShapeStore";
 import {
@@ -855,7 +860,10 @@ export async function runAssistantWithTools(params: {
         // Await the round's probe finalization (message_end already queued this
         // exact promise) so the coverage decision reads the complete in-band
         // search metadata instead of racing the response-clone parser.
-        const blocks = await finishHostedSearchRound(currentRound, "completed");
+        const blocks = await raceWithAbort(
+          finishHostedSearchRound(currentRound, "completed"),
+          params.signal,
+        );
         return blocks.some((block) => block.status === "completed" && block.sources.length > 0);
       }
       // web_fetch bridges never add new information; once the model has
@@ -1038,7 +1046,7 @@ export async function runAssistantWithTools(params: {
     ) {
       const finalization = finishHostedSearchRound(round, mode)
         .then((hostedSearchBlocks) => {
-          if (!assistantRef) return;
+          if (!assistantRef || params.signal?.aborted) return;
           const nextAssistant = applyHostedSearchBlocksToAssistant(
             assistantRef.current,
             round,
@@ -1064,7 +1072,15 @@ export async function runAssistantWithTools(params: {
 
     async function waitForHostedSearchFinalizations() {
       while (hostedSearchFinalizations.size > 0) {
-        await Promise.allSettled([...hostedSearchFinalizations]);
+        const pending = Promise.allSettled([...hostedSearchFinalizations]);
+        try {
+          await raceWithAbort(pending, params.signal);
+        } catch (error) {
+          // Probe finalization unregisters before it waits for a response
+          // clone. A cancelled turn must not wait for that clone to close.
+          if (params.signal?.aborted) return;
+          throw error;
+        }
       }
     }
 
@@ -1955,7 +1971,7 @@ export async function runAssistantWithTools(params: {
         throw new Error(normalizeErrorMessage(assistant.errorMessage, "Cancelled"));
       }
 
-      await params.debugLogger?.flush();
+      flushDebugLoggerInBackground(params.debugLogger, "agent runner");
       return {
         messages,
         assistant,
@@ -1967,7 +1983,7 @@ export async function runAssistantWithTools(params: {
       nativeWebSearchStatusController.finish();
       params.onToolStatus?.(null);
       params.debugLogger?.logError(error);
-      await params.debugLogger?.flush();
+      flushDebugLoggerInBackground(params.debugLogger, "agent runner");
       throw error;
     } finally {
       queueAllHostedSearchFinalizations("dispose");

crates/agent-gui/src/lib/debug/agentDebug.ts

@@ -26,6 +26,25 @@ export type StreamDebugLogger = {
   flush: () => Promise<void>;
 };
 
+/**
+ * Agent dev logging is diagnostic only. Keep its queued IPC writes out of
+ * request completion and cancellation paths, while still observing a late
+ * failure so it cannot become an unhandled rejection.
+ */
+export function flushDebugLoggerInBackground(
+  logger: StreamDebugLogger | undefined,
+  context: string,
+): void {
+  if (!logger) return;
+  try {
+    void Promise.resolve(logger.flush()).catch((error) => {
+      console.warn(`Agent dev debug ${context} flush failed`, error);
+    });
+  } catch (error) {
+    console.warn(`Agent dev debug ${context} flush failed`, error);
+  }
+}
+
 const writeQueues = new Map<string, Promise<void>>();
 const REDACTED_DEBUG_CREDENTIAL = "[redacted credential]";