fix(desktop): leave a stopped run's desktop calls to Stop

The supervisor's not-started and outcome-unknown settlements now hold the
run row against Stop, like the claim and a device result, and refuse once
admission is closed; a supervisor whose run was stopped exits. An unreadable
presence falls back to the pickup window instead of failing the call as
offline.

Author: waleedlatif1

apps/sim/lib/desktop/executor/repository.ts

@@ -8,7 +8,8 @@ import {
   desktopDevices,
 } from '@sim/db/schema'
 import { sanitizeValueForJsonb } from '@sim/utils/string'
-import { and, asc, eq, inArray, isNotNull, isNull, ne, or, sql } from 'drizzle-orm'
+import { and, asc, eq, inArray, isNotNull, isNull, ne, notInArray, or, sql } from 'drizzle-orm'
+import type { DbTransaction } from '@/lib/db/types'
 import {
   DESKTOP_CALL_LEASE_SECONDS,
   DESKTOP_INBOX_HORIZON_HOURS,
@@ -485,8 +486,10 @@ export async function getDesktopCallState(toolCallId: string) {
       msUntilLeaseEnd: sql<
         number | null
       >`(extract(epoch from (${copilotAsyncToolCalls.executionLeaseExpiresAt} - clock_timestamp())) * 1000)::float8`,
+      runOpen: sql<boolean>`${copilotRuns.toolAdmissionClosedAt} IS NULL AND ${copilotRuns.status} NOT IN ('complete', 'error', 'cancelled')`,
     })
     .from(copilotAsyncToolCalls)
+    .innerJoin(copilotRuns, eq(copilotRuns.id, copilotAsyncToolCalls.runId))
     .where(eq(copilotAsyncToolCalls.toolCallId, toolCallId))
     .limit(1)
   return row ?? null
@@ -499,35 +502,61 @@ interface DesktopCallFailure {
   error: string
 }
 
+/**
+ * Runs a settlement while holding the run row against Stop, the same serialization the claim and a
+ * device result use. Nothing is settled on a stopped or ended run: Stop answers for its calls.
+ */
+async function settleOnOpenRun(
+  runId: string,
+  settle: (tx: DbTransaction) => Promise<boolean>
+): Promise<boolean> {
+  return db.transaction(async (tx) => {
+    const [run] = await tx
+      .select({ id: copilotRuns.id })
+      .from(copilotRuns)
+      .where(
+        and(
+          eq(copilotRuns.id, runId),
+          isNull(copilotRuns.toolAdmissionClosedAt),
+          notInArray(copilotRuns.status, TERMINAL_RUN_STATUSES)
+        )
+      )
+      .for('share')
+    return run ? settle(tx) : false
+  })
+}
+
 /**
  * Fails a call nobody claimed: the inverse CAS of the claim, so exactly one of the two wins.
  * `deadlinePassed` limits it to a call whose pickup window has closed.
  */
 export async function failUnclaimedDesktopCall(
   input: DesktopCallFailure & { deadlinePassed: boolean }
 ): Promise<boolean> {
-  const [row] = await db
-    .update(copilotAsyncToolCalls)
-    .set({
-      status: ASYNC_TOOL_STATUS.failed,
-      result: sanitizeValueForJsonb(input.result),
-      error: input.error,
-      completedAt: sql`now()`,
-      updatedAt: sql`now()`,
-    })
-    .where(
-      and(
-        eq(copilotAsyncToolCalls.toolCallId, input.toolCallId),
-        eq(copilotAsyncToolCalls.runId, input.runId),
-        eq(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.pending),
-        isNull(copilotAsyncToolCalls.executionOwnerToken),
-        input.deadlinePassed
-          ? sql`${copilotAsyncToolCalls.executionLeaseExpiresAt} <= clock_timestamp()`
-          : undefined
+  return settleOnOpenRun(input.runId, async (tx) => {
+    const [row] = await tx
+      .update(copilotAsyncToolCalls)
+      .set({
+        status: ASYNC_TOOL_STATUS.failed,
+        result: sanitizeValueForJsonb(input.result),
+        error: input.error,
+        completedAt: sql`now()`,
+        updatedAt: sql`now()`,
+      })
+      .where(
+        and(
+          eq(copilotAsyncToolCalls.toolCallId, input.toolCallId),
+          eq(copilotAsyncToolCalls.runId, input.runId),
+          eq(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.pending),
+          isNull(copilotAsyncToolCalls.executionOwnerToken),
+          input.deadlinePassed
+            ? sql`${copilotAsyncToolCalls.executionLeaseExpiresAt} <= clock_timestamp()`
+            : undefined
+        )
       )
-    )
-    .returning({ toolCallId: copilotAsyncToolCalls.toolCallId })
-  return Boolean(row)
+      .returning({ toolCallId: copilotAsyncToolCalls.toolCallId })
+    return Boolean(row)
+  })
 }
 
 /**
@@ -537,27 +566,29 @@ export async function failUnclaimedDesktopCall(
 export async function failLapsedDesktopCall(
   input: DesktopCallFailure & { ownerToken: string }
 ): Promise<boolean> {
-  const [row] = await db
-    .update(copilotAsyncToolCalls)
-    .set({
-      status: ASYNC_TOOL_STATUS.failed,
-      result: sanitizeValueForJsonb(input.result),
-      error: input.error,
-      claimedBy: null,
-      claimedAt: null,
-      completedAt: sql`now()`,
-      updatedAt: sql`now()`,
-    })
-    .where(
-      and(
-        eq(copilotAsyncToolCalls.toolCallId, input.toolCallId),
-        eq(copilotAsyncToolCalls.runId, input.runId),
-        eq(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.running),
-        eq(copilotAsyncToolCalls.executionOwnerToken, input.ownerToken),
-        isNull(copilotAsyncToolCalls.executionRevokedAt),
-        sql`${copilotAsyncToolCalls.executionLeaseExpiresAt} <= clock_timestamp()`
+  return settleOnOpenRun(input.runId, async (tx) => {
+    const [row] = await tx
+      .update(copilotAsyncToolCalls)
+      .set({
+        status: ASYNC_TOOL_STATUS.failed,
+        result: sanitizeValueForJsonb(input.result),
+        error: input.error,
+        claimedBy: null,
+        claimedAt: null,
+        completedAt: sql`now()`,
+        updatedAt: sql`now()`,
+      })
+      .where(
+        and(
+          eq(copilotAsyncToolCalls.toolCallId, input.toolCallId),
+          eq(copilotAsyncToolCalls.runId, input.runId),
+          eq(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.running),
+          eq(copilotAsyncToolCalls.executionOwnerToken, input.ownerToken),
+          isNull(copilotAsyncToolCalls.executionRevokedAt),
+          sql`${copilotAsyncToolCalls.executionLeaseExpiresAt} <= clock_timestamp()`
+        )
       )
-    )
-    .returning({ toolCallId: copilotAsyncToolCalls.toolCallId })
-  return Boolean(row)
+      .returning({ toolCallId: copilotAsyncToolCalls.toolCallId })
+    return Boolean(row)
+  })
 }

apps/sim/lib/desktop/executor/supervisor.integration.ts

@@ -20,11 +20,13 @@ import { db } from '@sim/db'
 import {
   copilotAsyncToolCalls,
   copilotChats,
+  copilotRuns,
   desktopDevices,
   session,
   user,
   workspace,
 } from '@sim/db/schema'
+import { createDeferred } from '@sim/testing/helpers/deferred'
 import { featureFlagsMock, featureFlagsMockFns } from '@sim/testing/mocks/feature-flags.mock'
 import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
@@ -40,7 +42,7 @@ import {
   resolveTurnDesktopDevice,
 } from '@/lib/desktop/application/executor'
 import { isDesktopPresent } from '@/lib/desktop/executor/presence'
-import { failLapsedDesktopCall } from '@/lib/desktop/executor/repository'
+import { failLapsedDesktopCall, failUnclaimedDesktopCall } from '@/lib/desktop/executor/repository'
 import {
   DESKTOP_LEASE_LOST_MESSAGE,
   DESKTOP_NOT_RESPONDING_MESSAGE,
@@ -521,6 +523,71 @@ describe.runIf(Boolean(redisUrl))('desktop call supervision and turn binding', (
       })
     })
 
+    it('serializes a not-started settlement with a concurrent Stop', async () => {
+      const desktop = await signedInDesktop()
+      const run = await boundRun(desktop.userId, desktop.deviceId)
+      const toolCallId = await pendingCall(run.runId)
+
+      /** Stop's run-row update, held open until the settlement is queued behind it. */
+      const stopHeld = createDeferred<number>()
+      const releaseStop = createDeferred<void>()
+      const stopping = db.transaction(async (tx) => {
+        await tx
+          .update(copilotRuns)
+          .set({ toolAdmissionClosedAt: sql`now()` })
+          .where(eq(copilotRuns.id, run.runId))
+        const [backend] = await tx.execute<{ pid: number }>(sql`SELECT pg_backend_pid() AS pid`)
+        stopHeld.resolve(backend.pid)
+        await releaseStop.promise
+      })
+      const stopPid = await stopHeld.promise
+      const settling = failUnclaimedDesktopCall({
+        toolCallId,
+        runId: run.runId,
+        result: {},
+        error: 'offline',
+        deadlinePassed: false,
+      })
+      await expect
+        .poll(async () => {
+          const [waiter] = await db.execute<{ pid: number }>(sql`
+            SELECT pid FROM pg_stat_activity WHERE wait_event_type = 'Lock'
+              AND ${stopPid}::int = ANY(pg_blocking_pids(pid)) LIMIT 1`)
+          return Boolean(waiter)
+        })
+        .toBe(true)
+      releaseStop.resolve()
+      await stopping
+
+      await expect(settling).resolves.toBe(false)
+      expect((await row(toolCallId)).status).toBe('pending')
+    })
+
+    it('leaves a call on a stopped run to Stop rather than failing it as not responding', async () => {
+      const desktop = await signedInDesktop()
+      const run = await boundRun(desktop.userId, desktop.deviceId)
+      const toolCallId = await pendingCall(run.runId)
+
+      await online(desktop, async () => {
+        const supervision = superviseDesktopCall({
+          toolCallId,
+          runId: run.runId,
+          userId: desktop.userId,
+          deviceId: desktop.deviceId,
+          pickupGraceMs: 400,
+        })
+        await expect
+          .poll(async () => (await row(toolCallId)).executionLeaseExpiresAt)
+          .not.toBeNull()
+        await db
+          .update(copilotRuns)
+          .set({ toolAdmissionClosedAt: sql`now()` })
+          .where(eq(copilotRuns.id, run.runId))
+        await expect(supervision).resolves.toBe('settled')
+      })
+      expect((await row(toolCallId)).status).toBe('pending')
+    })
+
     it('never offers a call on a run the chat view serves', async () => {
       const desktop = await signedInDesktop()
       const run = await boundRun(desktop.userId, null)

apps/sim/lib/desktop/executor/supervisor.ts

@@ -39,7 +39,7 @@ export type DesktopCallSupervision =
   | 'offline'
   | 'not_responding'
   | 'lease_lost'
-  /** The device's result, Stop, or another settlement ended it. */
+  /** The device's result, Stop, or another settlement ended it, or its run was stopped. */
   | 'settled'
   | 'aborted'
 
@@ -153,7 +153,15 @@ export async function superviseDesktopCall(
   })
   if (offered) {
     ringDesktopInbox(input.deviceId, 'call')
-    if (!(await isDesktopPresent(input.deviceId).catch(() => false))) {
+    /** Only a confirmed absence fails fast; an unreadable presence falls back to the pickup window. */
+    const present = await isDesktopPresent(input.deviceId).catch((error) => {
+      logger.warn('Desktop presence could not be read; waiting out the pickup window', {
+        deviceId: input.deviceId,
+        error: getErrorMessage(error),
+      })
+      return null
+    })
+    if (present === false) {
       if (await failUnclaimed(input, 'offline', DESKTOP_OFFLINE_MESSAGE, false)) return 'offline'
     }
   } else if (!isOfferedOrClaimed(await getDesktopCallState(input.toolCallId))) {
@@ -163,7 +171,8 @@ export async function superviseDesktopCall(
   for (;;) {
     if (input.signal?.aborted) return 'aborted'
     const state = await getDesktopCallState(input.toolCallId)
-    if (!state || isTerminalAsyncStatus(state.status)) return 'settled'
+    /** Settled, held by a legacy claim, or its run was stopped: Stop answers for its calls. */
+    if (!state || isTerminalAsyncStatus(state.status) || !state.runOpen) return 'settled'
     if (!isOfferedOrClaimed(state)) return 'settled'
     const remainingMs = state.msUntilLeaseEnd ?? 0
     if (remainingMs <= 0) {