fix(computer): stop completion delivery at identity reset

Author: Vikhyath Mondreti

apps/sim/lib/mothership/tools/client/completion.ts

@@ -1,6 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { sleep } from '@sim/utils/helpers'
+import { interruptibleSleep } from '@sim/utils/helpers'
 import { isRecordLike } from '@sim/utils/object'
 import { backoffWithJitter } from '@sim/utils/retry'
 import type {
@@ -21,12 +21,16 @@ export class CompletionReportError extends Error {
 }
 
 async function fetchCompletion(input: RequestInfo | URL, init: RequestInit): Promise<Response> {
+  init.signal?.throwIfAborted()
   const controller = new AbortController()
   const timeout = setTimeout(() => {
     controller.abort(new DOMException('Completion report timed out', 'TimeoutError'))
   }, COMPLETION_REPORT_ATTEMPT_TIMEOUT_MS)
   try {
-    return await fetch(input, { ...init, signal: controller.signal })
+    return await fetch(input, {
+      ...init,
+      signal: init.signal ? AbortSignal.any([controller.signal, init.signal]) : controller.signal,
+    })
   } finally {
     clearTimeout(timeout)
   }
@@ -54,7 +58,8 @@ export async function reportClientToolCompletion(
   status: AsyncConfirmationStatus,
   message?: string,
   data?: AsyncCompletionData,
-  executionId?: string
+  executionId?: string,
+  signal?: AbortSignal
 ): Promise<void> {
   const basePayload = {
     toolCallId,
@@ -68,6 +73,7 @@ export async function reportClientToolCompletion(
       method: 'POST',
       headers: { 'Content-Type': 'application/json', ...traceparentHeader() },
       body,
+      signal,
     })
 
   const body = JSON.stringify(basePayload)
@@ -82,6 +88,7 @@ export async function reportClientToolCompletion(
   // from retries are discarded server-side.
   const maxAttempts = 5
   for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+    signal?.throwIfAborted()
     try {
       const response = await send(body)
       if (isSettledDelivery(response, toolCallId)) return
@@ -108,11 +115,12 @@ export async function reportClientToolCompletion(
         lastError = new Error(`Completion failed with status ${response.status}`)
       }
     } catch (error) {
+      if (signal?.aborted) throw error
       lastError = toError(error)
     }
 
     if (attempt < maxAttempts) {
-      await sleep(backoffWithJitter(attempt, null))
+      await interruptibleSleep(backoffWithJitter(attempt, null), signal)
     }
   }
 
@@ -131,7 +139,8 @@ export async function reportClientToolCompletionOnPageExit(
   toolCallId: string,
   status: AsyncConfirmationStatus,
   message: string,
-  data?: AsyncCompletionData
+  data?: AsyncCompletionData,
+  signal?: AbortSignal
 ): Promise<void> {
   // boundary-raw-fetch: keepalive is required so a terminal desktop result survives page unload
   const response = await fetchCompletion(COPILOT_CONFIRM_API_PATH, {
@@ -144,6 +153,7 @@ export async function reportClientToolCompletionOnPageExit(
       ...(data !== undefined ? { data } : {}),
     }),
     keepalive: true,
+    signal,
   })
   if (!isSettledDelivery(response, toolCallId)) {
     throw new CompletionReportError(`Page-exit completion failed with status ${response.status}`)

apps/sim/lib/mothership/tools/client/computer-tool-execution.test.ts

@@ -82,7 +82,8 @@ describe('computer action delivery', () => {
             id,
             completion.status,
             completion.message,
-            completion.data
+            completion.data,
+            expect.objectContaining({ aborted: false })
           )
       } finally {
         report.resolve()
@@ -146,7 +147,9 @@ describe('computer action delivery', () => {
       id,
       'error',
       expect.stringContaining('may already have run'),
-      expect.objectContaining({ doNotRetry: true })
+      expect.objectContaining({ doNotRetry: true }),
+      undefined,
+      expect.objectContaining({ aborted: false })
     )
   })
 
@@ -167,7 +170,9 @@ describe('computer action delivery', () => {
       id,
       'cancelled',
       expect.stringContaining('timed out'),
-      { doNotRetry: true, outcomeUnknown: true }
+      { doNotRetry: true, outcomeUnknown: true },
+      undefined,
+      expect.objectContaining({ aborted: false })
     )
   })
 
@@ -193,7 +198,9 @@ describe('computer action delivery', () => {
         dispatchState: 'not_started',
         doNotRetry: false,
         outcomeUnknown: false,
-      }
+      },
+      undefined,
+      expect.objectContaining({ aborted: false })
     )
     await executeComputerToolOnClient(nextId(), { action: 'list_apps' }, now())
     expect(mocks.execute).toHaveBeenCalledTimes(2)
@@ -213,7 +220,9 @@ describe('computer action delivery', () => {
       id,
       'error',
       expect.any(String),
-      expect.objectContaining({ doNotRetry: true, outcomeUnknown: true })
+      expect.objectContaining({ doNotRetry: true, outcomeUnknown: true }),
+      undefined,
+      expect.objectContaining({ aborted: false })
     )
   })
 
@@ -234,7 +243,9 @@ describe('computer action delivery', () => {
       id,
       'error',
       expect.stringContaining('may already have run'),
-      expect.objectContaining({ doNotRetry: true })
+      expect.objectContaining({ doNotRetry: true }),
+      undefined,
+      expect.objectContaining({ aborted: false })
     )
   })
   it.each([undefined, 'invalid', new Date(Date.now() - 121000).toISOString()])(
@@ -310,7 +321,9 @@ describe('computer action delivery', () => {
       expect.objectContaining({
         sequence: { completedSteps: 1, totalSteps: 2, error: 'Editor focus changed.' },
         observation: expect.objectContaining({ snapshotId: 'after' }),
-      })
+      }),
+      undefined,
+      expect.objectContaining({ aborted: false })
     )
   })
 
@@ -338,7 +351,9 @@ describe('computer action delivery', () => {
       id,
       'cancelled',
       expect.any(String),
-      expect.objectContaining({ doNotRetry: true })
+      expect.objectContaining({ doNotRetry: true }),
+      undefined,
+      expect.objectContaining({ aborted: false })
     )
   })
   it('does not dispatch a tool when Stop already won', async () => {
@@ -402,7 +417,9 @@ describe('computer action delivery', () => {
       rejectedId,
       'error',
       expect.any(String),
-      expect.anything()
+      expect.anything(),
+      undefined,
+      expect.objectContaining({ aborted: false })
     )
   })
 })

apps/sim/lib/mothership/tools/client/computer-tool-execution.ts

@@ -17,6 +17,7 @@ import {
 } from '@/lib/mothership/tools/client/completion'
 import { computerToolResultForModel } from '@/lib/mothership/tools/client/computer-tool-result'
 import { compactCompletionForPageExit } from '@/lib/mothership/tools/client/page-exit-completion'
+import { registerUserDataReset } from '@/stores/user-data-reset-registry'
 
 const logger = createLogger('ComputerToolExecution')
 const MAX_EVENT_AGE_MS = 120_000
@@ -34,13 +35,21 @@ interface Completion {
   data?: unknown
 }
 interface Execution {
+  identitySignal: AbortSignal
   completion?: Completion
   reporting?: Promise<void>
   retryTimer?: ReturnType<typeof setTimeout>
   deliveryAttempts?: number
   release?: () => void
 }
 const executions = new Map<string, Execution>()
+let identity = new AbortController()
+
+registerUserDataReset('computer-tool-execution', () => {
+  identity.abort()
+  identity = new AbortController()
+  for (const [toolCallId, execution] of executions) releaseExecution(toolCallId, execution)
+})
 
 function releaseExecution(toolCallId: string, execution: Execution): void {
   if (executions.get(toolCallId) === execution) executions.delete(toolCallId)
@@ -60,10 +69,13 @@ async function deliver(toolCallId: string, execution: Execution): Promise<void>
     toolCallId,
     completion.status,
     completion.message,
-    completion.data
+    completion.data,
+    undefined,
+    execution.identitySignal
   )
     .then(() => releaseExecution(toolCallId, execution))
     .catch((error) => {
+      if (execution.identitySignal.aborted) return
       logger.warn('Computer action result delivery failed; retained for redelivery', {
         toolCallId,
         error: getErrorMessage(error),
@@ -93,7 +105,7 @@ export async function executeComputerToolOnClient(
 ): Promise<void> {
   const existing = executions.get(toolCallId)
   if (existing) return deliver(toolCallId, existing)
-  const execution: Execution = {}
+  const execution: Execution = { identitySignal: identity.signal }
   const claim = replayLedger.claim(toolCallId)
   const reject = async (message: string, data?: Record<string, unknown>) => {
     const dataWithError = { error: message, ...data }
@@ -111,7 +123,9 @@ export async function executeComputerToolOnClient(
       toolCallId,
       ASYNC_TOOL_CONFIRMATION_STATUS.error,
       message,
-      dataWithError
+      dataWithError,
+      undefined,
+      execution.identitySignal
     ).catch((error) =>
       logger.warn('Could not report computer action rejection', {
         toolCallId,
@@ -145,8 +159,11 @@ export async function executeComputerToolOnClient(
     )
   executions.set(toolCallId, execution)
   const actionController = new AbortController()
+  const actionSignal = signal
+    ? AbortSignal.any([signal, execution.identitySignal])
+    : execution.identitySignal
   let dispatched = false
-  let cancelled = signal?.aborted === true
+  let cancelled = actionSignal.aborted
   const cancel = () => {
     cancelled = true
     actionController.abort()
@@ -173,7 +190,8 @@ export async function executeComputerToolOnClient(
         toolCallId,
         completion.status,
         completion.message,
-        completion.data
+        completion.data,
+        execution.identitySignal
       )
         .then(() => releaseExecution(toolCallId, execution))
         .catch((error) =>
@@ -196,7 +214,7 @@ export async function executeComputerToolOnClient(
     }
   }
   execution.release = () => window.removeEventListener('pagehide', onPageHide)
-  signal?.addEventListener('abort', cancel, { once: true })
+  actionSignal.addEventListener('abort', cancel, { once: true })
   window.addEventListener('pagehide', onPageHide)
   let timer: ReturnType<typeof setTimeout> | undefined
   try {
@@ -252,7 +270,7 @@ export async function executeComputerToolOnClient(
     }
   } finally {
     if (timer) clearTimeout(timer)
-    signal?.removeEventListener('abort', cancel)
+    actionSignal.removeEventListener('abort', cancel)
   }
   await deliver(toolCallId, execution)
 }

apps/sim/lib/mothership/tools/client/computer-tool-unload.test.ts

@@ -7,6 +7,7 @@ import { createDeferred } from '@sim/testing/helpers/deferred'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { copilotConfirmBodySchema } from '@/lib/api/contracts/copilot'
 import { executeComputerToolOnClient } from '@/lib/mothership/tools/client/computer-tool-execution'
+import { clearUserData } from '@/stores'
 
 const native = vi.hoisted(() => ({ execute: vi.fn(), cancel: vi.fn() }))
 vi.mock('@/lib/computer-use/transport', () => ({
@@ -345,4 +346,80 @@ describe('native computer completion during page unload', () => {
       await execution
     }
   })
+
+  it.each(['retained', 'backoff', 'request', 'native'] as const)(
+    'stops old-identity delivery during %s and admits the next identity',
+    async (phase) => {
+      const nativeResult = createDeferred<ComputerUseResult>()
+      const pendingResponse = createDeferred<Response>()
+      const pendingSignals: AbortSignal[] = []
+      let signedOut = false
+      let connected = false
+      const fetcher = vi.fn<typeof fetch>((_url, init) => {
+        if (connected) return Promise.resolve(new Response(null, { status: 200 }))
+        if (phase === 'request') {
+          const signal = init?.signal
+          if (!signal) throw new Error('Missing completion abort signal')
+          pendingSignals.push(signal)
+          return new Promise<Response>((resolve, reject) => {
+            signal.addEventListener('abort', () => reject(signal.reason), { once: true })
+            pendingResponse.promise.then(resolve, reject)
+          })
+        }
+        return Promise.resolve(new Response(null, { status: signedOut ? 401 : 503 }))
+      })
+      vi.stubGlobal('fetch', fetcher)
+      const beacon = vi.fn(() => false)
+      Object.defineProperty(navigator, 'sendBeacon', { configurable: true, value: beacon })
+      if (phase === 'native') native.execute.mockReturnValueOnce(nativeResult.promise)
+      else native.execute.mockResolvedValueOnce({ kind: 'apps', apps: [] })
+      const id = nextId()
+      const timestamp = new Date().toISOString()
+      const execution = executeComputerToolOnClient(id, { action: 'list_apps' }, timestamp)
+      await flushMicrotasks()
+      if (phase === 'retained') {
+        await vi.advanceTimersByTimeAsync(15_000)
+        await execution
+        expect(fetcher.mock.calls.length).toBeGreaterThanOrEqual(5)
+      }
+      if (phase === 'request') {
+        window.dispatchEvent(new Event('pagehide'))
+        await flushMicrotasks()
+        expect(pendingSignals).toHaveLength(2)
+      }
+      const callsBeforeSignOut = fetcher.mock.calls.length
+      try {
+        signedOut = true
+        expect(await clearUserData()).toBe(true)
+        nativeResult.resolve({ kind: 'apps', apps: [] })
+        await flushMicrotasks()
+        expect(vi.getTimerCount()).toBe(0)
+        if (phase === 'request') expect(pendingSignals.every((signal) => signal.aborted)).toBe(true)
+        if (phase === 'native') expect(native.cancel).toHaveBeenCalledExactlyOnceWith(id)
+        beacon.mockClear()
+        window.dispatchEvent(new Event('pagehide'))
+        await vi.advanceTimersByTimeAsync(120_000)
+        expect(fetcher).toHaveBeenCalledTimes(callsBeforeSignOut)
+        expect(beacon).not.toHaveBeenCalled()
+        expect(native.execute).toHaveBeenCalledOnce()
+        connected = true
+        native.execute.mockResolvedValueOnce({ kind: 'apps', apps: [] })
+        const next = nextId()
+        await executeComputerToolOnClient(next, { action: 'list_apps' }, new Date().toISOString())
+        expect(
+          copilotConfirmBodySchema.parse(JSON.parse(String(fetcher.mock.lastCall?.[1]?.body)))
+        ).toMatchObject({ toolCallId: next, status: 'success' })
+        expect(native.execute).toHaveBeenCalledTimes(2)
+      } finally {
+        connected = true
+        nativeResult.resolve({ kind: 'apps', apps: [] })
+        pendingResponse.resolve(new Response(null, { status: 200 }))
+        await vi.advanceTimersByTimeAsync(120_000)
+        await execution
+        const cleanup = executeComputerToolOnClient(id, { action: 'list_apps' }, timestamp)
+        await vi.advanceTimersByTimeAsync(15_000)
+        await cleanup
+      }
+    }
+  )
 })