fix(mothership): hold desktop leases, the resume wait and first claims to the call deadline

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

@@ -37,6 +37,7 @@ import { createDeferred } from '@sim/testing/helpers/deferred'
 import { generateId } from '@sim/utils/id'
 import { compareStrings } from '@sim/utils/string'
 import { eq, inArray, sql } from 'drizzle-orm'
+import type { PgUpdateSetSource } from 'drizzle-orm/pg-core'
 import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis'
 import type { DbTransaction } from '@/lib/db/types'
 import {
@@ -54,8 +55,10 @@ import {
 } from '@/lib/desktop/executor/errors'
 import { isDesktopPresent } from '@/lib/desktop/executor/presence'
 import { claimDesktopToolCall, requestRunStop } from '@/lib/mothership/async-runs/repository'
+import { TOOL_WATCHDOG_RESUME_GRACE_MS } from '@/lib/mothership/constants'
 import { waitForToolConfirmation } from '@/lib/mothership/persistence/tool-confirm'
 import { unsealClientToolCompletion } from '@/lib/mothership/request/tools/client-completion-seal.server'
+import { settleAbandonedDesktopToolCalls } from '@/lib/mothership/request/tools/desktop-wait'
 
 const CAPABILITIES = { executor: 1, browser: true, terminal: true, localFiles: true }
 
@@ -556,6 +559,7 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
       const toolCallId = await pendingCall(run.runId)
       const { executionToken } = await claim(desktop.principal, desktop.deviceId, toolCallId)
       const input = { deviceId: desktop.deviceId, toolCallId, executionToken }
+      expect((await row(toolCallId)).executionDeadlineAt).toBeNull()
 
       await db
         .update(copilotAsyncToolCalls)
@@ -583,6 +587,41 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
       ).rejects.toBeInstanceOf(DesktopCallRevokedError)
     })
 
+    it('stops renewing a call once the resume gate gave up on its deadline, then settles it as outcome unknown', async () => {
+      const desktop = await signedInDesktop()
+      const run = await boundRun(desktop)
+      const toolCallId = await pendingCall(run.runId)
+      const setRow = (fields: PgUpdateSetSource<typeof copilotAsyncToolCalls>) =>
+        db
+          .update(copilotAsyncToolCalls)
+          .set(fields)
+          .where(eq(copilotAsyncToolCalls.toolCallId, toolCallId))
+      const deadlinePassedBy = (seconds: number) =>
+        setRow({ executionDeadlineAt: sql`clock_timestamp() - ${seconds} * interval '1 second'` })
+      await setRow({ executionDeadlineAt: sql`clock_timestamp() + interval '10 minutes'` })
+      const { executionToken } = await claim(desktop.principal, desktop.deviceId, toolCallId)
+      const input = { deviceId: desktop.deviceId, toolCallId, executionToken }
+
+      // Inside the gate's grace past the deadline, the device keeps the call.
+      await deadlinePassedBy(TOOL_WATCHDOG_RESUME_GRACE_MS / 1000 - 10)
+      await expect(
+        renewDesktopToolLease.execute({ principal: desktop.principal, input })
+      ).resolves.toEqual({ renewed: true })
+
+      await deadlinePassedBy(TOOL_WATCHDOG_RESUME_GRACE_MS / 1000 + 1)
+      await expect(
+        renewDesktopToolLease.execute({ principal: desktop.principal, input })
+      ).rejects.toBeInstanceOf(DesktopCallRevokedError)
+
+      // Unrenewed, the lease lapses, and the cleanup settles the call as it would any lapse.
+      await setRow({ executionLeaseExpiresAt: sql`clock_timestamp() - interval '1 second'` })
+      await settleAbandonedDesktopToolCalls(0)
+      expect(await row(toolCallId)).toMatchObject({
+        status: 'failed',
+        error: expect.stringContaining('It may already have taken effect'),
+      })
+    })
+
     it('after Stop, refuses the stopped calls, revokes the running one and lists it to cancel', async () => {
       const desktop = await signedInDesktop()
       const run = await boundRun(desktop)

apps/sim/lib/mothership/async-runs/orphaned-runs.integration.ts

@@ -262,6 +262,56 @@ describe.runIf(Boolean(redisUrl))('Chat runs no controller owns', () => {
     return { ...run, marker: chat?.conversationId ?? null }
   }
 
+  describe('a Sim claim past its start window', () => {
+    async function lateTool(
+      runId: string,
+      fields: Partial<{ status: 'running'; claimedBy: string }> = {}
+    ) {
+      const tool = await dispatchedTool(runId)
+      await db
+        .update(copilotAsyncToolCalls)
+        .set({ executionDeadlineAt: sql`clock_timestamp() + interval '30 seconds'`, ...fields })
+        .where(eq(copilotAsyncToolCalls.toolCallId, tool.toolCallId))
+      return tool
+    }
+
+    async function storedTool(toolCallId: string) {
+      const [row] = await db
+        .select()
+        .from(copilotAsyncToolCalls)
+        .where(eq(copilotAsyncToolCalls.toolCallId, toolCallId))
+      return row
+    }
+
+    it('leaves an admitted pending call unclaimed for its caller to settle as never run', async () => {
+      const run = await admittedRun()
+      const tool = await lateTool(run.runId)
+
+      expect(await claimToolExecution(tool)).toEqual({ outcome: 'too_late' })
+      expect(await storedTool(tool.toolCallId)).toMatchObject({
+        status: 'pending',
+        executionStartedAt: null,
+        executionOwnerToken: null,
+      })
+    })
+
+    it('reports a stopped run as closed before the start window', async () => {
+      const run = await admittedRun({ stopped: true })
+      const tool = await lateTool(run.runId)
+
+      expect(await claimToolExecution(tool)).toEqual({ outcome: 'closed' })
+    })
+
+    it('still claims a call a client already claimed, and one without a deadline', async () => {
+      const run = await admittedRun()
+      const clientClaimed = await lateTool(run.runId, { status: 'running', claimedBy: 'client' })
+      const undated = await dispatchedTool(run.runId)
+
+      expect(await claimToolExecution(clientClaimed)).toEqual({ outcome: 'claimed' })
+      expect(await claimToolExecution(undated)).toEqual({ outcome: 'claimed' })
+    })
+  })
+
   it('settles a run whose controller died once its recovery window has passed', async () => {
     const orphan = await admittedRun({ idleMinutes: 90, status: 'paused_waiting_for_tool' })
 

apps/sim/lib/mothership/async-runs/repository.ts

@@ -55,6 +55,7 @@ import {
   isAwaitingToolPermission,
   SIM_TOOL_EXECUTION_VERSION,
 } from '@/lib/mothership/async-runs/lifecycle'
+import { TOOL_WATCHDOG_RESUME_GRACE_MS } from '@/lib/mothership/constants'
 import { TraceAttr } from '@/lib/mothership/generated/trace-attributes-v1'
 import { TraceSpan } from '@/lib/mothership/generated/trace-spans-v1'
 import {
@@ -867,6 +868,12 @@ export type ToolExecutionClaim =
   /** Already claimed or settled. */
   | { outcome: 'existing' }
 
+/** A Sim handler's claim, which also reports an unclaimed call too late to start. */
+export type SimToolExecutionClaim =
+  | ToolExecutionClaim
+  /** Admitted but unclaimed, with too little of its deadline left to start: nothing ran it. */
+  | { outcome: 'too_late' }
+
 /**
  * Every condition at once, typed as a condition: drizzle's `and` may return undefined, which a
  * negation would render as `NOT (NULL)` and match nothing.
@@ -897,6 +904,19 @@ const unfinishedSimExecution = allOf(
  */
 const tooLateToStart = sql`(${copilotAsyncToolCalls.executionDeadlineAt} IS NOT NULL AND ${copilotAsyncToolCalls.executionDeadlineAt} < clock_timestamp() + ${SIM_TOOL_MIN_START_BUDGET_MS} * interval '1 millisecond')`
 
+/**
+ * A call past its deadline plus the resume grace: the resume wait's ceiling for it (`run.ts`), past
+ * which nothing still waits for its result. A call without a deadline never matches.
+ */
+const pastResumeGiveUp = sql`(${copilotAsyncToolCalls.executionDeadlineAt} IS NOT NULL AND ${copilotAsyncToolCalls.executionDeadlineAt} + ${TOOL_WATCHDOG_RESUME_GRACE_MS} * interval '1 millisecond' <= clock_timestamp())`
+
+/** A call nothing claimed or started whose deadline leaves too little to start it. */
+const pendingTooLateToStart = allOf(
+  eq(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.pending),
+  isNull(copilotAsyncToolCalls.executionStartedAt),
+  tooLateToStart
+)
+
 /**
  * A Sim execution whose owner died before recording a result: its lease lapsed, which a live
  * owner's heartbeat never lets happen. A takeover runs it again until it has started
@@ -925,11 +945,12 @@ const rerunnableSimExecution = allOf(
  * Sim claims a call it is about to execute under an execution lease; a terminal result never
  * releases it. The claim takes a call's first execution, or re-runs one whose owner died
  * mid-execution: the new owner token fences the dead owner out, since its renewals and its
- * result both require the token and an unexpired lease.
+ * result both require the token and an unexpired lease. A pending call too late to start is left
+ * unclaimed, under the same admission check, for its caller to settle as never run.
  */
 export async function claimToolExecution(
   owner: SimToolExecutionOwner
-): Promise<ToolExecutionClaim> {
+): Promise<SimToolExecutionClaim> {
   return await claimUnderRunAdmission(owner, 'sim-stream', true, async (tx, thisCall) => {
     const startedAt = new Date()
     const [claimed] = await traceMothershipQuery('UPDATE', 'copilot_async_tool_calls', () =>
@@ -954,7 +975,8 @@ export async function claimToolExecution(
                 inArray(copilotAsyncToolCalls.status, [
                   ASYNC_TOOL_STATUS.pending,
                   ASYNC_TOOL_STATUS.running,
-                ])
+                ]),
+                not(pendingTooLateToStart)
               ),
               interruptedSimExecution
             )
@@ -964,10 +986,13 @@ export async function claimToolExecution(
     )
     if (claimed) return { outcome: 'claimed' } as const
     const [record] = await traceMothershipQuery('SELECT', 'copilot_async_tool_calls', () =>
-      tx.select({ id: copilotAsyncToolCalls.id }).from(copilotAsyncToolCalls).where(thisCall)
+      tx
+        .select({ tooLate: sql<boolean>`${pendingTooLateToStart}` })
+        .from(copilotAsyncToolCalls)
+        .where(thisCall)
     )
     if (!record) throw new Error('Tool execution record is unavailable')
-    return { outcome: 'existing' } as const
+    return record.tooLate ? ({ outcome: 'too_late' } as const) : ({ outcome: 'existing' } as const)
   })
 }
 
@@ -1069,7 +1094,8 @@ export async function claimDesktopToolCall(
 /**
  * Expired ownership cannot be renewed, even before a follower has observed the expiry. With
  * `desktop`, the call must still be running on a run bound to that device, so a call Stop settled
- * cannot be renewed either.
+ * cannot be renewed either, nor one past its deadline plus the resume grace: its lease then lapses
+ * and the call settles as outcome unknown.
  */
 export async function renewSimToolExecutionLease(
   owner: SimToolExecutionOwner,
@@ -1092,6 +1118,7 @@ export async function renewSimToolExecutionLease(
         desktop
           ? and(
               eq(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.running),
+              not(pastResumeGiveUp),
               'deviceId' in desktop
                 ? sql`EXISTS (SELECT 1 FROM ${copilotRuns} r WHERE r.id = ${copilotAsyncToolCalls.runId} AND r.desktop_device_id = ${desktop.deviceId})`
                 : and(

apps/sim/lib/mothership/request/lifecycle/run.test.ts

@@ -3672,6 +3672,41 @@ describe('runCopilotLifecycle', () => {
     }
   })
 
+  it('fails a call whose deadline already passed at that deadline plus grace, not a grace from now', async () => {
+    vi.useFakeTimers()
+    try {
+      mothershipAsyncRunsMockFns.mockGetChatViewDesktopLeaseRemainingMs.mockResolvedValue(50_000)
+      const turn = runImportTurn(Date.now() - 10_000)
+      await vi.advanceTimersByTimeAsync(19_000)
+      expect(turn.bodies).toHaveLength(1)
+      await vi.advanceTimersByTimeAsync(2_000)
+      expect((await turn.lifecycle).success).toBe(true)
+      expect(turn.resumedWithLostImport()).toBe(true)
+      expect(
+        mothershipAsyncRunsMockFns.mockGetChatViewDesktopLeaseRemainingMs
+      ).not.toHaveBeenCalled()
+    } finally {
+      mothershipAsyncRunsMockFns.mockGetChatViewDesktopLeaseRemainingMs.mockReset()
+      vi.useRealTimers()
+    }
+  })
+
+  it('a call without a deadline keeps its full budget and grace', async () => {
+    vi.useFakeTimers()
+    try {
+      mothershipAsyncRunsMockFns.mockGetChatViewDesktopLeaseRemainingMs.mockResolvedValue(null)
+      const turn = runImportTurn()
+      await vi.advanceTimersByTimeAsync(89_000)
+      expect(turn.bodies).toHaveLength(1)
+      await vi.advanceTimersByTimeAsync(2_000)
+      expect((await turn.lifecycle).success).toBe(true)
+      expect(turn.resumedWithLostImport()).toBe(true)
+    } finally {
+      mothershipAsyncRunsMockFns.mockGetChatViewDesktopLeaseRemainingMs.mockReset()
+      vi.useRealTimers()
+    }
+  })
+
   it('a lease lookup that fails is checked again instead of failing a renewed import', async () => {
     vi.useFakeTimers()
     try {

apps/sim/lib/mothership/request/lifecycle/run.ts

@@ -1450,21 +1450,22 @@ async function runCheckpointLoop(
           const waitBudgetMs =
             pendingToolWaitBudgetMs(toolCall, context.desktopDeviceId) +
             TOOL_WATCHDOG_RESUME_GRACE_MS
-          const ceilingAt = now + Math.max(waitBudgetMs, CLIENT_TOOL_RESULT_TIMEOUT_MS)
+          const clientCapAt = now + Math.max(waitBudgetMs, CLIENT_TOOL_RESULT_TIMEOUT_MS)
+          // Neither the wait nor a lease carries a call past the deadline the worker named for it.
+          const ceilingAt =
+            toolCall?.deadlineAt === undefined
+              ? clientCapAt
+              : Math.min(clientCapAt, toolCall.deadlineAt + TOOL_WATCHDOG_RESUME_GRACE_MS)
           pendingWatchdogs.set(toolCallId, {
             promise,
             settlement: promise.then(
               () => ({ toolCallId, promise }),
               () => ({ toolCallId, promise })
             ),
-            deadlineAt: now + waitBudgetMs,
+            deadlineAt: Math.min(now + waitBudgetMs, ceilingAt),
             waitBudgetMs,
             startedAt: now,
-            // A lease never carries a call past the deadline the worker named for it.
-            ceilingAt:
-              toolCall?.deadlineAt === undefined
-                ? ceilingAt
-                : Math.min(ceilingAt, toolCall.deadlineAt + TOOL_WATCHDOG_RESUME_GRACE_MS),
+            ceilingAt,
           })
           maximumWaitBudgetMs = Math.max(maximumWaitBudgetMs, waitBudgetMs)
         }

apps/sim/lib/mothership/request/tools/execution-lifetime.ts

@@ -15,9 +15,9 @@ import {
   prepareWorkbenchAccess,
   recordSimSandboxProcess,
   renewSimToolExecutionLease,
+  type SimToolExecutionClaim,
   settleSimSandboxProcess,
   settleSimToolExecution,
-  type ToolExecutionClaim,
 } from '@/lib/mothership/async-runs/repository'
 import { TraceAttr } from '@/lib/mothership/generated/trace-attributes-v1'
 import { TraceSpan } from '@/lib/mothership/generated/trace-spans-v1'
@@ -30,7 +30,7 @@ export interface ToolExecutionLifetime {
   readonly signal: AbortSignal
   readonly owner?: SimToolExecutionOwner
   complete(input: CompleteAsyncToolCallInput): Promise<void>
-  claim(runId: string, userId: string): Promise<ToolExecutionClaim>
+  claim(runId: string, userId: string): Promise<SimToolExecutionClaim>
   hold<T>(work: Promise<T>): Promise<T>
 }
 

apps/sim/lib/mothership/request/tools/executor.test.ts

@@ -369,8 +369,10 @@ describe('executeToolAndReport provenance isolation', () => {
   beforeEach(() => {
     executeTool.mockReset()
     completeAsyncToolCall.mockResolvedValue(null)
+    completePendingAsyncToolCall.mockReset()
     markAsyncToolRunning.mockResolvedValue(null)
     upsertAsyncToolCall.mockResolvedValue(null)
+    claimToolExecution.mockReset()
     claimToolExecution.mockResolvedValue({ outcome: 'claimed' })
     settleSimToolExecution.mockResolvedValue(undefined)
     waitForToolConfirmation.mockReset()
@@ -698,6 +700,70 @@ describe('executeToolAndReport provenance isolation', () => {
     expect(tool.error).toBe(TOOL_DEADLINE_BEFORE_START_MESSAGE)
   })
 
+  it('settles a call its claim found too late to start as never run, without claiming it', async () => {
+    claimToolExecution.mockResolvedValueOnce({ outcome: 'too_late' })
+    completePendingAsyncToolCall.mockImplementationOnce(async (input) => ({ ...input }))
+    const tool: ToolCallState = {
+      ...buildPendingToolCall(),
+      name: 'generate_audio',
+      deadlineAt: Date.now() + 30_000,
+    }
+    const completion = await executeToolAndReport(tool.id, buildStreamingContext(tool), {
+      userId: 'user-1',
+      workflowId: 'workflow-1',
+      resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry(),
+    })
+
+    expect(executeTool).not.toHaveBeenCalled()
+    expect(completePendingAsyncToolCall).toHaveBeenCalledExactlyOnceWith(
+      expect.objectContaining({
+        toolCallId: tool.id,
+        status: 'failed',
+        error: TOOL_DEADLINE_BEFORE_START_MESSAGE,
+      })
+    )
+    expect(completion.status).toBe('error')
+    expect(tool.result?.output).toMatchObject({
+      error: TOOL_DEADLINE_BEFORE_START_MESSAGE,
+      notStarted: true,
+    })
+  })
+
+  it('settles the call of a stopped run as stopped even past its start window', async () => {
+    claimToolExecution.mockResolvedValueOnce({ outcome: 'closed' })
+    completePendingAsyncToolCall.mockImplementation(async (input) => ({ ...input }))
+    const tool: ToolCallState = {
+      ...buildPendingToolCall(),
+      name: 'generate_audio',
+      deadlineAt: Date.now() + 30_000,
+    }
+    const completion = await executeToolAndReport(tool.id, buildStreamingContext(tool), {
+      userId: 'user-1',
+      workflowId: 'workflow-1',
+      resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry(),
+    })
+
+    expect(completePendingAsyncToolCall).not.toHaveBeenCalled()
+    expect(completion).toMatchObject({
+      status: 'cancelled',
+      message: 'Run stopped before tool admission',
+    })
+  })
+
+  it('claims a call without a deadline without trying to settle it first', async () => {
+    executeTool.mockResolvedValue({ success: true, output: {} })
+    const tool: ToolCallState = { ...buildPendingToolCall(), name: 'generate_audio' }
+    await executeToolAndReport(tool.id, buildStreamingContext(tool), {
+      userId: 'user-1',
+      workflowId: 'workflow-1',
+      resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry(),
+    })
+
+    expect(completePendingAsyncToolCall).not.toHaveBeenCalled()
+    expect(claimToolExecution).toHaveBeenCalledOnce()
+    expect(executeTool).toHaveBeenCalledOnce()
+  })
+
   it('runs a call again only until the deadline recorded when it was first handed', async () => {
     vi.useFakeTimers()
     let signal: AbortSignal | undefined