feat(desktop): device registry, inbox and leased claims for a background executor

Adds the server half of the desktop background executor's device protocol,
behind the mothership-desktop-background-executor flag (off by default),
built on the existing tool-execution primitives rather than new ones:

- desktop_devices binds an install id to a user and the Better Auth session
  that registered it; copilot_runs.desktop_device_id records which device a
  turn's desktop tools run on (nothing sets it yet).
- Claims, renewals and results reuse claimSimToolExecution,
  renewSimToolExecutionLease and the execution-owner fence, extended for a
  device: only a pending call on a run bound to that device, that the user
  is not still being asked about, with the per-surface claim owner; a result
  is accepted after its lease lapsed until something else settles the call.
- /api/copilot/confirm and the device's complete share one sealing and
  settlement function, so a device result is sealed, settled and published
  exactly like a chat view's.
- POST /api/desktop/devices registers; GET /api/desktop/inbox lists calls,
  approval_needed items (from permission_requested_at) and cancel items;
  GET /api/desktop/inbox/stream is an SSE doorbell over Redis pub/sub.
- Presence is driven by the device: every pull, lease renewal and stream
  open refreshes a 45 s key, and only its TTL removes it.

Author: waleedlatif1

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

@@ -20,18 +20,13 @@ import {
   isWorkflowToolExecutionClaimable,
 } from '@/lib/mothership/async-runs/lifecycle'
 import {
-  completeAsyncToolCall,
-  completeClaimedAsyncToolCall,
-  completePendingAsyncToolCall,
-  detachAsyncToolCall,
   getAsyncToolCall,
   getClaimedWorkflowExecutionId,
   getRunSegment,
 } from '@/lib/mothership/async-runs/repository'
 import { CopilotConfirmOutcome } from '@/lib/mothership/generated/trace-attribute-values-v1'
 import { TraceAttr } from '@/lib/mothership/generated/trace-attributes-v1'
 import { TraceSpan } from '@/lib/mothership/generated/trace-spans-v1'
-import { publishToolConfirmation } from '@/lib/mothership/persistence/tool-confirm'
 import {
   authenticateCopilotRequestSessionOnly,
   createInternalServerErrorResponse,
@@ -41,9 +36,11 @@ import {
 } from '@/lib/mothership/request/http'
 import { withIncomingGoSpan } from '@/lib/mothership/request/otel'
 import {
-  retainSealedClientToolContext,
-  sealClientToolCompletion,
-} from '@/lib/mothership/request/tools/client-completion-seal.server'
+  type ClientToolSettlementGuard,
+  clientToolCompletionMessage,
+  sealClientToolResult,
+  settleClientToolCall,
+} from '@/lib/mothership/request/tools/client-settlement.server'
 import { isWorkflowToolName } from '@/lib/mothership/tools/client-executed-tools'
 import {
   createStructuralWorkflowToolCompletionData,
@@ -63,20 +60,6 @@ const NATIVE_HANDOFF_INTERRUPTED_MESSAGE =
 
 type ToolCallStatusUpdateOutcome = 'updated' | 'conflict' | 'failed'
 
-interface UpdateToolCallStatusOptions {
-  executionId?: string
-  completionGuard?:
-    | { status: typeof ASYNC_TOOL_STATUS.pending }
-    | { status: typeof ASYNC_TOOL_STATUS.running; claimedBy: string }
-}
-
-function getClientToolCompletionMessage(status: AsyncConfirmationStatus): string {
-  if (status === ASYNC_TOOL_CONFIRMATION_STATUS.success) return 'Tool completed'
-  if (status === ASYNC_TOOL_CONFIRMATION_STATUS.background) return 'Tool is running in background'
-  if (status === ASYNC_TOOL_CONFIRMATION_STATUS.cancelled) return 'Tool cancelled'
-  return 'Tool failed'
-}
-
 function createConfirmationResponse(
   toolCallId: string,
   status: AsyncConfirmationStatus,
@@ -109,53 +92,18 @@ async function updateToolCallStatus(
   status: AsyncConfirmationStatus,
   message?: string,
   data?: AsyncCompletionData,
-  options: UpdateToolCallStatusOptions = {}
+  options: { executionId?: string; guard?: ClientToolSettlementGuard } = {}
 ): Promise<ToolCallStatusUpdateOutcome> {
   const toolCallId = existing.toolCallId
   try {
-    if (status === ASYNC_TOOL_CONFIRMATION_STATUS.background) {
-      const detached = options.executionId
-        ? await detachAsyncToolCall(toolCallId, { preserveClaim: true })
-        : await detachAsyncToolCall(toolCallId)
-      if (!detached) return 'conflict'
-      publishToolConfirmation({
-        toolCallId,
-        status,
-        message: message || undefined,
-        timestamp: new Date().toISOString(),
-        data,
-        ...(options.executionId ? { executionId: options.executionId } : {}),
-      })
-      return 'updated'
-    }
-    const durableStatus =
-      status === ASYNC_TOOL_CONFIRMATION_STATUS.success
-        ? ASYNC_TOOL_STATUS.completed
-        : status === ASYNC_TOOL_CONFIRMATION_STATUS.cancelled
-          ? ASYNC_TOOL_STATUS.cancelled
-          : ASYNC_TOOL_STATUS.failed
-    const completionInput = {
-      toolCallId,
-      status: durableStatus,
-      result: data ?? null,
-      error: status === 'success' ? null : message || status,
-    }
-    const completed =
-      options.completionGuard?.status === ASYNC_TOOL_STATUS.pending
-        ? await completePendingAsyncToolCall(completionInput)
-        : options.completionGuard?.status === ASYNC_TOOL_STATUS.running
-          ? await completeClaimedAsyncToolCall(completionInput, options.completionGuard.claimedBy)
-          : await completeAsyncToolCall(completionInput)
-    if (!completed) return 'conflict'
-    publishToolConfirmation({
+    return await settleClientToolCall({
       toolCallId,
       status,
-      message: message || undefined,
-      timestamp: new Date().toISOString(),
+      message: message ?? '',
       data,
-      ...(options.executionId ? { executionId: options.executionId } : {}),
+      executionId: options.executionId,
+      guard: options.guard ?? { kind: 'open' },
     })
-    return 'updated'
   } catch (error) {
     logger.error('Failed to update tool call status', {
       toolCallId,
@@ -416,28 +364,23 @@ export const POST = withRouteHandler((req: NextRequest) => {
               ),
             }
           : {
-              message: getClientToolCompletionMessage(status),
-              data: {
-                ...retainSealedClientToolContext(existing.result),
-                ...(await sealClientToolCompletion({
-                  toolCallId,
-                  runId: existing.runId,
-                  userId: authenticatedUserId,
-                  ...(isIndeterminateNativeExit
-                    ? {
-                        message: NATIVE_HANDOFF_INTERRUPTED_MESSAGE,
-                        data: {
-                          error: NATIVE_HANDOFF_INTERRUPTED_MESSAGE,
-                          outcomeUnknown: true,
-                          doNotRetry: true,
-                        },
-                      }
-                    : {
-                        ...(message !== undefined ? { message } : {}),
-                        ...(data !== undefined ? { data } : {}),
-                      }),
-                })),
-              },
+              message: clientToolCompletionMessage(status),
+              data: await sealClientToolResult({
+                toolCallId,
+                runId: existing.runId,
+                userId: authenticatedUserId,
+                storedResult: existing.result,
+                ...(isIndeterminateNativeExit
+                  ? {
+                      message: NATIVE_HANDOFF_INTERRUPTED_MESSAGE,
+                      data: {
+                        error: NATIVE_HANDOFF_INTERRUPTED_MESSAGE,
+                        outcomeUnknown: true,
+                        doNotRetry: true,
+                      },
+                    }
+                  : { message, data }),
+              }),
             }
 
         const updateOutcome = await updateToolCallStatus(
@@ -447,9 +390,7 @@ export const POST = withRouteHandler((req: NextRequest) => {
           projected.data,
           {
             ...(isWorkflowTool && executionId ? { executionId } : {}),
-            ...(isPreclaimNativeTerminalOutcome
-              ? { completionGuard: { status: ASYNC_TOOL_STATUS.pending } as const }
-              : {}),
+            ...(isPreclaimNativeTerminalOutcome ? { guard: { kind: 'pending' } as const } : {}),
           }
         )
 
@@ -460,12 +401,7 @@ export const POST = withRouteHandler((req: NextRequest) => {
                 ASYNC_TOOL_CONFIRMATION_STATUS.error,
                 projected.message,
                 projected.data,
-                {
-                  completionGuard: {
-                    status: ASYNC_TOOL_STATUS.running,
-                    claimedBy: nativeClaimOwner,
-                  },
-                }
+                { guard: { kind: 'claimed', claimedBy: nativeClaimOwner } }
               )
             : updateOutcome
 

apps/sim/app/api/desktop/devices/route.ts

@@ -0,0 +1,19 @@
+import { registerDesktopDeviceContract } from '@/lib/api/contracts/desktop-executor'
+import {
+  defineInternalJsonRoute,
+  internalRateLimits,
+  internalSessionAuth,
+} from '@/lib/api/server/routes'
+import { desktopExecutorErrorPolicy } from '@/lib/api/server/routes/desktop-executor'
+import { registerDesktopDevice } from '@/lib/desktop/application/executor'
+
+export const POST = defineInternalJsonRoute({
+  contract: registerDesktopDeviceContract,
+  auth: internalSessionAuth,
+  operation: registerDesktopDevice.operation,
+  rateLimit: internalRateLimits.user({ bucketName: 'desktop-device-register' }),
+  errorPolicy: desktopExecutorErrorPolicy,
+  mapInput: ({ body }) => body,
+  useCase: registerDesktopDevice,
+  staticResponseHeaders: { 'Cache-Control': 'no-store' },
+})

apps/sim/app/api/desktop/inbox/route.ts

@@ -0,0 +1,25 @@
+import { listDesktopInboxContract } from '@/lib/api/contracts/desktop-executor'
+import { defineInternalJsonRoute, internalSessionAuth } from '@/lib/api/server/routes'
+import {
+  desktopExecutorErrorPolicy,
+  desktopExecutorRateLimit,
+} from '@/lib/api/server/routes/desktop-executor'
+import { listDesktopInbox } from '@/lib/desktop/application/executor'
+
+export const dynamic = 'force-dynamic'
+
+export const GET = defineInternalJsonRoute({
+  contract: listDesktopInboxContract,
+  auth: internalSessionAuth,
+  operation: listDesktopInbox.operation,
+  rateLimit: desktopExecutorRateLimit,
+  errorPolicy: desktopExecutorErrorPolicy,
+  mapInput: ({ query }) => ({ deviceId: query.deviceId }),
+  useCase: listDesktopInbox,
+  present: ({ items }) => ({
+    items: items.map((item) =>
+      item.kind === 'call' ? { ...item, createdAt: item.createdAt.toISOString() } : item
+    ),
+  }),
+  staticResponseHeaders: { 'Cache-Control': 'no-store' },
+})

apps/sim/app/api/desktop/inbox/stream/route.ts

@@ -0,0 +1,45 @@
+import { createLogger } from '@sim/logger'
+import type { NextRequest } from 'next/server'
+import { desktopInboxStreamContract } from '@/lib/api/contracts/desktop-executor'
+import { parseRequest } from '@/lib/api/server'
+import { desktopExecutorRateLimit } from '@/lib/api/server/routes/desktop-executor'
+import {
+  InternalUnauthenticatedError,
+  internalSessionAuth,
+} from '@/lib/api/server/routes/internal-json-route'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { openDesktopInboxStream } from '@/lib/desktop/application/executor'
+import { DesktopDeviceUnrecognizedError } from '@/lib/desktop/executor/errors'
+import { createSSEStream } from '@/lib/events/sse-endpoint'
+
+export const dynamic = 'force-dynamic'
+
+const logger = createLogger('DesktopInboxStream')
+
+/**
+ * The background executor's doorbell. A raw route because it streams: authentication, the device
+ * check, and presence all run through the `openDesktopInboxStream` use case.
+ */
+export const GET = withRouteHandler(async (request: NextRequest) => {
+  try {
+    const principal = await internalSessionAuth.authenticate()
+    const limited = await desktopExecutorRateLimit.enforce(request, principal)
+    if (limited) return limited
+    const parsed = await parseRequest(desktopInboxStreamContract, request, {})
+    if (!parsed.success) return parsed.response
+    const { deviceId } = parsed.data.query
+    const inbox = await openDesktopInboxStream.execute({ principal, input: { deviceId } })
+    return createSSEStream(request, {
+      label: 'desktop-inbox',
+      revalidate: inbox.revalidate,
+      subscriptions: [{ subscribe: inbox.subscribe }],
+    })
+  } catch (error) {
+    if (error instanceof InternalUnauthenticatedError)
+      return new Response('Unauthorized', { status: 401 })
+    if (error instanceof DesktopDeviceUnrecognizedError)
+      return new Response(error.message, { status: 401 })
+    logger.error('Failed to open the desktop inbox stream', error)
+    return new Response('Unable to open the desktop inbox', { status: 500 })
+  }
+})

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

@@ -0,0 +1,18 @@
+import { claimDesktopToolContract } from '@/lib/api/contracts/desktop-executor'
+import { defineInternalJsonRoute, internalSessionAuth } from '@/lib/api/server/routes'
+import {
+  desktopExecutorErrorPolicy,
+  desktopExecutorRateLimit,
+} from '@/lib/api/server/routes/desktop-executor'
+import { claimDesktopTool } from '@/lib/desktop/application/executor'
+
+export const POST = defineInternalJsonRoute({
+  contract: claimDesktopToolContract,
+  auth: internalSessionAuth,
+  operation: claimDesktopTool.operation,
+  rateLimit: desktopExecutorRateLimit,
+  errorPolicy: desktopExecutorErrorPolicy,
+  mapInput: ({ body }) => body,
+  useCase: claimDesktopTool,
+  staticResponseHeaders: { 'Cache-Control': 'no-store' },
+})

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

@@ -0,0 +1,18 @@
+import { completeDesktopToolContract } from '@/lib/api/contracts/desktop-executor'
+import { defineInternalJsonRoute, internalSessionAuth } from '@/lib/api/server/routes'
+import {
+  desktopExecutorErrorPolicy,
+  desktopExecutorRateLimit,
+} from '@/lib/api/server/routes/desktop-executor'
+import { completeDesktopTool } from '@/lib/desktop/application/executor'
+
+export const POST = defineInternalJsonRoute({
+  contract: completeDesktopToolContract,
+  auth: internalSessionAuth,
+  operation: completeDesktopTool.operation,
+  rateLimit: desktopExecutorRateLimit,
+  errorPolicy: desktopExecutorErrorPolicy,
+  mapInput: ({ body }) => body,
+  useCase: completeDesktopTool,
+  staticResponseHeaders: { 'Cache-Control': 'no-store' },
+})

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

@@ -0,0 +1,18 @@
+import { renewDesktopToolLeaseContract } from '@/lib/api/contracts/desktop-executor'
+import { defineInternalJsonRoute, internalSessionAuth } from '@/lib/api/server/routes'
+import {
+  desktopExecutorErrorPolicy,
+  desktopExecutorRateLimit,
+} from '@/lib/api/server/routes/desktop-executor'
+import { renewDesktopToolLease } from '@/lib/desktop/application/executor'
+
+export const POST = defineInternalJsonRoute({
+  contract: renewDesktopToolLeaseContract,
+  auth: internalSessionAuth,
+  operation: renewDesktopToolLease.operation,
+  rateLimit: desktopExecutorRateLimit,
+  errorPolicy: desktopExecutorErrorPolicy,
+  mapInput: ({ body }) => body,
+  useCase: renewDesktopToolLease,
+  staticResponseHeaders: { 'Cache-Control': 'no-store' },
+})