feat(desktop): bind turns to a desktop and enforce its deadlines from the row (#8650)

* feat(desktop): bind turns to a desktop and enforce its deadlines from the row

A turn sent from a desktop whose background executor is registered to the
same session (and with mothership-desktop-background-executor on) is bound
to that device at admission: copilot_runs.desktop_device_id. Its desktop
calls are persisted pending, offered to the device and claimed through the
executor's own fenced routes; the chat view's authorize and confirm answer
409 for them.

There is no supervisor. The single desktop wait (waitForDesktopToolCall)
branches on the binding: a bound call is offered (pickup_deadline_at, a new
nullable column) and the device's doorbell rung, then the existing durable
wait enforces every deadline from the row on each 5 s check, beside the
lease revocation that already lives there:
- an unclaimed call whose device is offline fails at once as not started
  (reason offline); one still unclaimed past its pickup window fails the same
  way (reason not_responding), as the inverse CAS of the claim;
- a claimed call whose lease lapsed fails as outcome unknown, revoking the
  device's token so its late result is superseded, and rings it to cancel.
Every settlement is sealed like the device's own result. The stale-execution
cron settles, the same way, bound calls whose waiter died with its process.

One not-started builder carries a reason (chat_not_open, offline,
not_responding), and the server-owned failure helper now settles through the
shared client settlement. The device is rung when a bound call needs
approval, when the user answers, and on Stop.

* fix(desktop): refuse a claim past its pickup window and leave executor leases to their own settlement

A device could claim an offered call after its pickup deadline and before the wait's next check settled it, running an action the turn was about to report as not started. The claim and the inbox now treat a closed window as no longer offered. The generic Sim lease sweep no longer settles a desktop executor's lapsed lease with the Sim interrupted result: the bound wait and the stale-execution cron settle it as outcome unknown, so the model is told the action may already have taken effect.

* fix(desktop): settle Stop without the device, survive Redis errors, and only run offered calls

Stop on a call the background executor held never settled the turn's tool
executions unless the device acknowledged: the executor's claim marked a Sim
execution as started, and Stop does not settle that execution. A device that
was asleep, offline or signed out left abortRun unsettled and the next turn's
workbench pending. The executor's claim now takes only its owner token and
lease; no Sim handler runs for it, so Sim-execution quiescence ignores it, as
it already ignores the chat view's desktop claims.

An overdue unclaimed call now settles from its row when presence cannot be
read, and never fails early as offline on a failed read; each call in the
cron's sweep settles in its own try/catch; presence writes are best effort, so
a Redis error no longer fails a pull or a renewal.

The executor takes only a call Sim offered it, within its pickup window, and
the inbox lists unclaimed calls only while offered or waiting for the user. A
call Sim never got to offer gets an implicit deadline one pickup window after
it could first run, so the cron still settles it.

A revoked install id stays revoked on re-registration, a claim racing the
device binding answers "no longer waiting", Stop rings the device only after
its chat is validated, the two device lookups are one, and the desktop inbox
E2E runs in the http-e2e CI job against its own app with Redis.

* test(desktop): assert outcomes instead of mock calls, and drive Stop through abortRun

The desktop tests asserted that mocks were or were not called. They now assert what the caller sees: the 409, the sweep's settled count, and the device's own doorbell, which Stop now rings through the real abortRun against a stand-in worker.

* fix(desktop): never fail an awake device's call as offline because a presence write was lost

Presence writes are best effort, so a pull whose write failed left the key
missing and the next read took the device for offline, failing its pending
call early. A call now fails early as offline only when presence is absent and
the device has not pulled for longer than the presence TTL plus the interval
at which a pull writes last_seen_at; otherwise its pickup window decides.

The audit test no longer depends on the insertion order of audit writes,
which are not awaited.

* fix(desktop): pre-merge review fixes for the bound executor

- Terminal approvals carry their command at args.args.command, so the
  inbox summary read nothing; it now reads the terminal call's real shape,
  and the tests use it.
- A turn budget shorter than the pickup window left a bound call pending; an
  unclaimed call is now settled as never started through the pending-only
  path, as the chat-view wait does.
- Ringing the device is best effort: a publish that throws is logged and can
  no longer break Stop or any other caller.
- Running claimed calls are no longer fetched for the inbox, and offered or
  awaiting calls and cancel items are fetched with separate caps, so neither
  can starve the other.
- No pickup window runs while the user decides: an offer refuses a call still
  awaiting approval, and recording the decision clears any pickup deadline,
  so an allowed call's window starts when it is offered after the answer.
- The per-user rate limit refills every second instead of in one lump a
  minute, so a device that spent its burst can still renew its leases.
- The integration suites delete the audit rows they create.

Author: waleedlatif1

.github/workflows/test-build.yml

@@ -123,7 +123,8 @@ jobs:
   http-e2e:
     name: End-to-end over real HTTP
     runs-on: ${{ (vars.CI_PROVIDER == '' || vars.CI_PROVIDER == 'blacksmith') && 'blacksmith-8vcpu-ubuntu-2404' || 'ubuntu-latest' }}
-    timeout-minutes: 20
+    # Four suites, each starting its own dev app, run in sequence.
+    timeout-minutes: 30
     services:
       postgres:
         image: pgvector/pgvector:pg17
@@ -138,6 +139,16 @@ jobs:
           --health-interval 5s
           --health-timeout 5s
           --health-retries 10
+      # Only the desktop executor's app is given REDIS_URL: its doorbell and presence live there.
+      redis:
+        image: redis:7-alpine
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
     env:
       DATABASE_URL: postgresql://postgres:postgres@127.0.0.1:5432/sim_test
       BETTER_AUTH_SECRET: http-e2e-ci-secret-at-least-32-characters
@@ -298,6 +309,54 @@ jobs:
           STOP_AFTER_E2E_REPORT_PATH="$report_dir/stop-after-http-report.json" \
             bun run test:workflow-stop-after:e2e
 
+      # The desktop background executor's protocol: device registration, the SSE doorbell over
+      # Redis pub/sub, presence, leased claims, Stop and isolation, against its own app with Redis.
+      - name: Verify the desktop background executor's inbox over real HTTP
+        working-directory: apps/sim
+        env:
+          NEXT_PUBLIC_APP_URL: http://127.0.0.1:3019
+          BETTER_AUTH_URL: http://127.0.0.1:3019
+          REDIS_URL: redis://127.0.0.1:6379
+          NEXT_PUBLIC_FORCE_HOSTED: 'false'
+          MSHIP_DESKTOP_BACKGROUND_EXECUTOR: 'true'
+          COPILOT_TOOL_PERMISSIONS_ENABLED: 'true'
+          INTERNAL_API_SECRET: desktop-inbox-http-ci-local-secret-at-least-32-characters
+          DB_TX_TRIPWIRE: throw
+          DISABLE_TELEMETRY: 'true'
+          NEXT_TELEMETRY_DISABLED: '1'
+          READY_TIMEOUT_SECONDS: 300
+        run: |
+          report_dir="$RUNNER_TEMP/e2e"
+          server_log="$report_dir/desktop-inbox-next.log"
+          mkdir -p "$report_dir"
+          node ../../node_modules/next/dist/bin/next dev --hostname 127.0.0.1 --port 3019 > "$server_log" 2>&1 &
+          server_pid=$!
+          finish() {
+            kill "$server_pid" 2>/dev/null || true
+            wait "$server_pid" 2>/dev/null || true
+            awk '/^ (GET|POST|PUT|PATCH|DELETE|HEAD) \/api\// { print }' "$server_log" > "$report_dir/desktop-inbox-http-status.log"
+          }
+          trap finish EXIT
+          fail_startup() {
+            echo "::error::$1"
+            tail -n 200 "$server_log"
+            exit 1
+          }
+          started=$SECONDS
+          until curl --fail --silent --max-time 10 http://127.0.0.1:3019/api/health > /dev/null; do
+            kill -0 "$server_pid" 2>/dev/null || fail_startup 'Local desktop executor app exited during startup.'
+            [ $((SECONDS - started)) -lt "$READY_TIMEOUT_SECONDS" ] ||
+              fail_startup "Local desktop executor app did not become ready within $READY_TIMEOUT_SECONDS seconds."
+            sleep 2
+          done
+          echo "Local desktop executor app ready after $((SECONDS - started))s"
+          DESKTOP_INBOX_E2E_BASE_URL="$NEXT_PUBLIC_APP_URL" \
+          DESKTOP_INBOX_E2E_DATABASE_URL="$DATABASE_URL" \
+          DESKTOP_INBOX_E2E_REDIS_URL="$REDIS_URL" \
+          DESKTOP_INBOX_E2E_AUTH_SECRET="$BETTER_AUTH_SECRET" \
+          DESKTOP_INBOX_E2E_REPORT_PATH="$report_dir/desktop-inbox-http-report.json" \
+            bun run test:desktop-inbox:e2e
+
       - name: Upload end-to-end reports and server logs
         if: failure()
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4

apps/sim/app/api/copilot/confirm/route.test.ts

@@ -131,6 +131,26 @@ describe('Copilot Confirm API Route', () => {
     expect(JSON.stringify(publishToolConfirmation.mock.calls)).not.toContain('resolved-secret')
   })
 
+  it("refuses a chat view's report for a call a desktop's background executor owns", async () => {
+    getAsyncToolCall.mockResolvedValue({
+      ...existingRow,
+      toolName: 'browser_click',
+      status: 'pending',
+      claimedBy: null,
+    })
+    getRunSegment.mockResolvedValue({ id: 'run-1', userId: 'user-1', desktopDeviceId: 'device-1' })
+
+    const response = await POST(
+      createMockPostRequest({
+        toolCallId: 'tool-call-123',
+        status: 'error',
+        message: 'The desktop refused this claim',
+      })
+    )
+
+    expect(response.status).toBe(409)
+  })
+
   it('atomically detaches a live background confirmation', async () => {
     const response = await POST(
       createMockPostRequest({

apps/sim/app/api/copilot/confirm/route.ts

@@ -1,7 +1,7 @@
 import type { Span } from '@opentelemetry/api'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage, toError } from '@sim/utils/errors'
-import { isPlainRecord } from '@sim/utils/object'
+import { isPlainRecord, toRecord } from '@sim/utils/object'
 import { type NextRequest, NextResponse } from 'next/server'
 import { copilotConfirmContract } from '@/lib/api/contracts/copilot'
 import { parseRequest, validationErrorResponse } from '@/lib/api/server'
@@ -40,7 +40,11 @@ import {
   settleClientToolCall,
 } from '@/lib/mothership/request/tools/client-settlement.server'
 import { isWorkflowToolName } from '@/lib/mothership/tools/client-executed-tools'
-import { getDesktopToolClaimOwner, isNativeDesktopTool } from '@/lib/mothership/tools/desktop-tools'
+import {
+  getDesktopToolClaimOwner,
+  isDesktopToolCall,
+  isNativeDesktopTool,
+} from '@/lib/mothership/tools/desktop-tools'
 import {
   createStructuralWorkflowToolCompletionData,
   getWorkflowToolCompletionExecutionId,
@@ -206,6 +210,17 @@ export const POST = withRouteHandler((req: NextRequest) => {
           return NextResponse.json({ error: 'Forbidden' }, { status: 403 })
         }
 
+        if (run.desktopDeviceId && isDesktopToolCall(existing.toolName, toRecord(existing.args))) {
+          span.setAttribute(TraceAttr.CopilotConfirmOutcome, CopilotConfirmOutcome.Forbidden)
+          return NextResponse.json(
+            {
+              error:
+                "This chat's desktop actions report through the desktop app's background executor",
+            },
+            { status: 409 }
+          )
+        }
+
         const isWorkflowTool = isWorkflowToolName(existing.toolName || '')
         const workflowId = isWorkflowTool
           ? resolveWorkflowToolTargetId(existing.args, run.workflowId)

apps/sim/app/api/copilot/tool-permission/route.ts

@@ -5,6 +5,7 @@ import { copilotToolPermissionContract } from '@/lib/api/contracts/copilot'
 import { parseRequest, validationErrorResponse } from '@/lib/api/server'
 import { isCopilotToolPermissionsEnabled } from '@/lib/core/config/env-flags'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { ringDesktopInbox } from '@/lib/desktop/executor/doorbell'
 import {
   getAsyncToolCall,
   getRunSegment,
@@ -136,6 +137,8 @@ async function applyDecision(
     toolName: claimed.toolName,
     decidedAt: claimed.permissionDecidedAt?.toISOString(),
   })
+  // A bound device lists the call for approval; the answer turns it into a call or drops it.
+  if (run.desktopDeviceId) ringDesktopInbox(run.desktopDeviceId, 'approval')
 
   return { toolCallId, decision, applied: true }
 }

apps/sim/app/api/desktop/tool/authorize/route.test.ts

@@ -72,6 +72,26 @@ describe('desktop tool authorization', () => {
     })
   })
 
+  it("refuses a chat view's claim on a run bound to a desktop's background executor", async () => {
+    getAsyncToolCall.mockResolvedValueOnce({
+      toolCallId: 'bound-click',
+      runId: 'run-1',
+      status: 'pending',
+      toolName: 'browser_click',
+      args: { ref: 'e1' },
+    })
+    getRunSegment.mockResolvedValueOnce({
+      id: 'run-1',
+      chatId: 'chat-1',
+      userId: 'user-1',
+      status: 'active',
+      desktopDeviceId: 'device-1',
+    })
+
+    const response = await POST(request('bound-click'))
+    expect(response.status).toBe(409)
+  })
+
   it('rejects retired browser tools retained only for history', async () => {
     getAsyncToolCall.mockResolvedValueOnce({
       toolCallId: 'retired-browser-tool',

apps/sim/app/api/desktop/tool/authorize/route.ts

@@ -74,6 +74,12 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
   if (run.status === 'complete' || run.status === 'error' || run.status === 'cancelled') {
     return createNotFoundResponse('Pending client tool call not found')
   }
+  // The device's background executor claims a bound run's calls through its own fenced route.
+  if (run.desktopDeviceId)
+    return NextResponse.json(
+      { error: "This chat's desktop actions run in the desktop app's background executor" },
+      { status: 409 }
+    )
 
   const args = isRecordLike(toolCall.args) ? (toolCall.args as Record<string, unknown>) : {}
   if (!isDesktopToolCall(toolCall.toolName, args)) {

apps/sim/background/cleanup-stale-executions.ts

@@ -31,6 +31,7 @@ import {
   type StaleSweepableExecutionStatus,
 } from '@/lib/logs/types'
 import { sweepOrphanedRuns } from '@/lib/mothership/async-runs/orphaned-runs'
+import { settleAbandonedDesktopToolCalls } from '@/lib/mothership/request/tools/desktop-wait'
 import { cancelStaleDispatches } from '@/lib/table/dispatcher'
 import { deleteFile } from '@/lib/uploads/core/storage-service'
 import {
@@ -743,6 +744,19 @@ export async function runCleanupStaleExecutions() {
     })
   }
 
+  /**
+   * Settle desktop calls on device-bound runs whose waiter died with its process: an offered call
+   * nobody claimed, or a claimed one whose device stopped renewing its lease.
+   */
+  let abandonedDesktopCallsSettled = 0
+  try {
+    abandonedDesktopCallsSettled = await settleAbandonedDesktopToolCalls()
+  } catch (error) {
+    logger.error('Failed to settle abandoned desktop tool calls:', {
+      error: toError(error).message,
+    })
+  }
+
   return {
     executions: {
       found: staleExecutionsFound,
@@ -774,6 +788,7 @@ export async function runCleanupStaleExecutions() {
     },
     chatRuns: {
       orphanedSettled: orphanedRunsSettled,
+      abandonedDesktopCallsSettled,
     },
   }
 }

apps/sim/lib/api/server/routes/desktop-executor.ts

@@ -26,9 +26,11 @@ export const desktopExecutorErrorPolicy = extendInternalErrorPolicy(
 
 /**
  * One busy device renews a lease per running call every 20 s and pulls its inbox on every
- * doorbell, so the bucket allows a sustained 10 requests a second per user.
+ * doorbell, so the bucket allows a sustained 10 requests a second per user. It refills every
+ * second rather than once a minute: a device that spent its burst must still renew its leases
+ * well before they lapse.
  */
 export const desktopExecutorRateLimit = internalRateLimits.user({
   bucketName: 'desktop-executor',
-  config: { maxTokens: 600, refillRate: 600, refillIntervalMs: 60_000 },
+  config: { maxTokens: 600, refillRate: 10, refillIntervalMs: 1_000 },
 })