fix(web): treat fire stream aborts as transient retries

Author: 0xTomDaniel

typescript/clients/web-ag-ui/apps/web/src/hooks/useAgentConnection.ts

@@ -47,7 +47,7 @@ import { fireAgentRun, logFireCommandDebug } from '../utils/fireAgentRun';
 import { resumeInterruptViaAgent } from '../utils/interruptResolution';
 import { scheduleCycleAfterInterruptResolution } from '../utils/interruptAutoCycle';
 import { canonicalizeChainLabel } from '../utils/iconResolution';
-import { isAgentRunning, isBusyRunError } from '../utils/runConcurrency';
+import { isAbortLikeError, isAgentRunning, isBusyRunError } from '../utils/runConcurrency';
 import { deriveUiState } from '../utils/deriveUiState';
 import {
   isAgentInterrupt,
@@ -494,6 +494,7 @@ export function useAgentConnection(agentId: string): UseAgentConnectionResult {
       runAgent: async (currentAgent) => copilotkit.runAgent({ agent: currentAgent }),
       createId: v7,
       isBusyRunError,
+      isAbortLikeError,
       isAgentRunning,
       onSyncingChange: (isSyncing) => {
         setSyncingState({

typescript/clients/web-ag-ui/apps/web/src/utils/agentCommandScheduler.ts

@@ -38,6 +38,7 @@ export function createAgentCommandScheduler<TAgent extends SchedulableAgent>(par
   runAgent: (agent: TAgent) => Promise<unknown>;
   createId: () => string;
   isBusyRunError: (error: unknown) => boolean;
+  isAbortLikeError?: (error: unknown) => boolean;
   isAgentRunning: (agent: TAgent) => boolean;
   onSyncingChange: (isSyncing: boolean) => void;
   onCommandError?: (command: string, error: unknown) => void;
@@ -122,7 +123,8 @@ export function createAgentCommandScheduler<TAgent extends SchedulableAgent>(par
         syncRunInFlight = false;
 
         const busy = params.isBusyRunError(error) || params.isAgentRunning(agent);
-        if (busy && syncBusyRetries < syncBusyMaxRetries) {
+        const aborted = params.isAbortLikeError?.(error) ?? false;
+        if ((busy || aborted) && syncBusyRetries < syncBusyMaxRetries) {
           syncBusyRetries += 1;
           pendingSyncIntent = true;
           refreshSyncing();
@@ -142,7 +144,8 @@ export function createAgentCommandScheduler<TAgent extends SchedulableAgent>(par
       }
 
       const busy = params.isBusyRunError(error) || params.isAgentRunning(agent);
-      if (busy) {
+      const aborted = params.isAbortLikeError?.(error) ?? false;
+      if (busy || aborted) {
         params.onCommandBusy?.(command, error);
         return;
       }

typescript/clients/web-ag-ui/apps/web/src/utils/agentCommandScheduler.unit.test.ts

@@ -44,6 +44,7 @@ describe('agentCommandScheduler', () => {
       createId: () => 'msg-1',
       isBusyRunError: (error) =>
         error instanceof Error && error.message.includes('already active'),
+      isAbortLikeError: (error) => error instanceof Error && error.message.includes('aborted'),
       isAgentRunning: (value) => value.isRunning === true,
       onSyncingChange,
       onCommandError,
@@ -140,6 +141,31 @@ describe('agentCommandScheduler', () => {
     }
   });
 
+  it('retries sync on abort-like transport failures before surfacing busy state', async () => {
+    vi.useFakeTimers();
+
+    runAgent = vi
+      .fn<() => Promise<void>>()
+      .mockRejectedValueOnce(new Error('BodyStreamBuffer was aborted'))
+      .mockResolvedValueOnce(undefined);
+
+    const scheduler = createScheduler({ syncReplayDelayMs: 10, syncBusyMaxRetries: 2 });
+
+    try {
+      scheduler.dispatch('sync', { allowSyncCoalesce: true });
+
+      await Promise.resolve();
+      await vi.advanceTimersByTimeAsync(20);
+
+      expect(runAgent).toHaveBeenCalledTimes(2);
+      expect(onCommandBusy).not.toHaveBeenCalled();
+      expect(onCommandError).not.toHaveBeenCalled();
+    } finally {
+      scheduler.dispose();
+      vi.useRealTimers();
+    }
+  });
+
   it('rejects non-sync command dispatch while run is in flight', () => {
     runInFlight = true;
     const scheduler = createScheduler();
@@ -189,6 +215,22 @@ describe('agentCommandScheduler', () => {
     scheduler.dispose();
   });
 
+  it('treats abort-like failures for non-sync commands as transient busy responses', async () => {
+    runAgent = vi.fn(async () => {
+      throw new Error('BodyStreamBuffer was aborted');
+    });
+    const scheduler = createScheduler();
+
+    scheduler.dispatch('hire');
+    await Promise.resolve();
+
+    expect(onCommandBusy).toHaveBeenCalledTimes(1);
+    expect(onCommandBusy).toHaveBeenCalledWith('hire', expect.any(Error));
+    expect(onCommandError).not.toHaveBeenCalled();
+
+    scheduler.dispose();
+  });
+
   it('dispatches custom runs through the same scheduler gating path', async () => {
     const scheduler = createScheduler();
     const runCustom = vi.fn(async () => undefined);

typescript/clients/web-ag-ui/apps/web/src/utils/fireAgentRun.ts

@@ -1,5 +1,5 @@
 import type { v7 as uuidv7 } from 'uuid';
-import { isAgentRunning, isBusyRunError } from './runConcurrency';
+import { isAbortLikeError, isAgentRunning, isBusyRunError } from './runConcurrency';
 
 type AgentLike = {
   addMessage: (message: { id: string; role: 'user'; content: string }) => void;
@@ -122,13 +122,14 @@ const startFireRun = async <TAgent extends AgentLike>(
     });
     const maxRetries = options?.maxRetries ?? FIRE_RUN_MAX_RETRIES;
     const retryDelayMs = options?.retryDelayMs ?? FIRE_RUN_RETRY_DELAY_MS;
-    if (isBusyRunError(error) && attempt < maxRetries - 1) {
+    const shouldRetry = isBusyRunError(error) || isAbortLikeError(error);
+    if (shouldRetry && attempt < maxRetries - 1) {
       await sleep(retryDelayMs);
       return startFireRun(runAgent, agent, runInFlightRef, options, attempt + 1);
     }
 
     runInFlightRef.current = false;
-    if (isBusyRunError(error)) {
+    if (isBusyRunError(error) || isAbortLikeError(error)) {
       throw new Error('Agent run is still active. Please retry in a moment.');
     }
     throw error;

typescript/clients/web-ag-ui/apps/web/src/utils/fireAgentRun.unit.test.ts

@@ -284,6 +284,47 @@ describe('fireAgentRun', () => {
     }
   });
 
+  it('retries aborted fire starts and avoids surfacing transient stream abort errors', async () => {
+    vi.useFakeTimers();
+
+    const onError = vi.fn();
+    const agent = {
+      isRunning: false,
+      addMessage: vi.fn(),
+      detachActiveRun: vi.fn(async () => undefined),
+    };
+    const runInFlightRef = { current: false };
+    const runAgent = vi
+      .fn()
+      .mockRejectedValueOnce(new Error('BodyStreamBuffer was aborted'))
+      .mockResolvedValueOnce(undefined);
+
+    try {
+      const ok = await fireAgentRun({
+        agent,
+        runAgent,
+        threadId: 'thread-1',
+        runInFlightRef,
+        createId: () => 'msg-1',
+        onError,
+        busyRunMaxRetries: 4,
+        busyRunRetryDelayMs: 100,
+      });
+
+      expect(ok).toBe(true);
+      expect(runAgent).toHaveBeenCalledTimes(1);
+
+      await vi.advanceTimersByTimeAsync(120);
+
+      expect(runAgent).toHaveBeenCalledTimes(2);
+      expect(onError).not.toHaveBeenCalled();
+      expect(runInFlightRef.current).toBe(true);
+      expect(agent.addMessage).toHaveBeenCalledTimes(1);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it('does nothing when threadId is missing', async () => {
     const agent = { abortRun: vi.fn(), detachActiveRun: vi.fn(), addMessage: vi.fn() };
     const copilotkit = { runAgent: vi.fn() };

typescript/clients/web-ag-ui/apps/web/src/utils/runConcurrency.ts

@@ -2,6 +2,7 @@ type HttpStatus = number | string;
 
 type BusyError = {
   message?: string;
+  name?: string;
   status?: HttpStatus;
   statusCode?: HttpStatus;
   code?: string;
@@ -41,6 +42,27 @@ export function isBusyRunError(error: unknown): boolean {
   );
 }
 
+export function isAbortLikeError(error: unknown): boolean {
+  const maybeError = error as BusyError;
+  const status = toStatusCode(
+    maybeError?.status ??
+      maybeError?.statusCode ??
+      (typeof maybeError?.response === 'object' ? maybeError.response?.status : undefined),
+  );
+  if (status === 499) return true;
+
+  const name = `${maybeError?.name ?? ''}`.toLowerCase();
+  if (name.includes('abort')) return true;
+
+  const message = `${maybeError?.message ?? ''}`.toLowerCase();
+  return (
+    message.includes('aborterror') ||
+    message.includes('aborted') ||
+    message.includes('abort') ||
+    message.includes('bodystreambuffer')
+  );
+}
+
 export function isAgentRunning(agent: RunAwareAgent): boolean {
   const field = agent.isRunning;
   if (typeof field === 'function') return field();

typescript/clients/web-ag-ui/apps/web/src/utils/runConcurrency.unit.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect, it } from 'vitest';
 
-import { isAgentRunning, isBusyRunError } from './runConcurrency';
+import { isAbortLikeError, isAgentRunning, isBusyRunError } from './runConcurrency';
 
 describe('runConcurrency', () => {
   it('recognizes server busy statuses', () => {
@@ -20,6 +20,13 @@ describe('runConcurrency', () => {
     expect(isBusyRunError(new Error('network error'))).toBe(false);
   });
 
+  it('recognizes abort-like transport errors', () => {
+    expect(isAbortLikeError({ name: 'AbortError' })).toBe(true);
+    expect(isAbortLikeError({ message: 'BodyStreamBuffer was aborted' })).toBe(true);
+    expect(isAbortLikeError({ status: 499 })).toBe(true);
+    expect(isAbortLikeError({ message: 'network error' })).toBe(false);
+  });
+
   it('detects running agent flag as boolean or function', () => {
     expect(isAgentRunning({ isRunning: true })).toBe(true);
     expect(isAgentRunning({ isRunning: false })).toBe(false);