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):

- 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).
- POST /api/desktop/devices registers the device and returns the timing
  contract; GET /api/desktop/inbox is the reconciling pull (calls offered to
  the device, calls awaiting approval, and claimed calls to cancel);
  GET /api/desktop/inbox/stream is an SSE doorbell over Redis pub/sub that
  keeps a 45 s presence key while it is open.
- POST /api/desktop/tool/claim, /lease and /complete claim an offered call
  with an execution token and a 60 s lease, renew it (410 once stopped or
  lapsed), and record the sealed result idempotently: a retry is a
  duplicate, a result for a call Sim settled first is superseded.

Every operation requires the device to be bound to the caller's user and
session, the run to be bound to that device, and the run's tool admission
to be open.

Author: waleedlatif1

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,26 @@
+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, hasActiveRun }) => ({
+    hasActiveRun,
+    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,42 @@
+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 {
+  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 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,19 @@
+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,
+  present: (claim) => ({ ...claim, leaseExpiresAt: claim.leaseExpiresAt.toISOString() }),
+  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,19 @@
+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,
+  present: ({ leaseExpiresAt }) => ({ leaseExpiresAt: leaseExpiresAt.toISOString() }),
+  staticResponseHeaders: { 'Cache-Control': 'no-store' },
+})

apps/sim/lib/api/contracts/desktop-executor.ts

@@ -0,0 +1,183 @@
+import { z } from 'zod'
+import { desktopToolCallIdSchema } from '@/lib/api/contracts/desktop-tool-authorization'
+import { defineRouteContract } from '@/lib/api/contracts/types'
+
+/**
+ * The desktop background executor's wire protocol. The Electron main process calls every route
+ * here with the cookie of the Better Auth session it registered under; Sim refuses any other
+ * session for that device.
+ */
+
+/** The install id the desktop app generates once and keeps in its user data. */
+const desktopDeviceIdSchema = z.string().uuid('Device ID must be a UUID')
+
+const desktopDeviceCapabilitiesSchema = z.object({
+  executor: z.number().int().min(1).max(1000),
+  browser: z.boolean(),
+  terminal: z.boolean(),
+  localFiles: z.boolean(),
+})
+
+const registerDesktopDeviceBodySchema = z.object({
+  deviceId: desktopDeviceIdSchema,
+  name: z.string().trim().min(1, 'Device name is required').max(128, 'Device name is too long'),
+  appVersion: z.string().trim().min(1, 'App version is required').max(64),
+  platform: z.string().trim().min(1, 'Platform is required').max(64),
+  capabilities: desktopDeviceCapabilitiesSchema,
+})
+export type RegisterDesktopDeviceBody = z.input<typeof registerDesktopDeviceBodySchema>
+
+/**
+ * `enabled: false` is the kill switch: the device keeps finishing calls on runs already bound to
+ * it, but no new turn binds to it.
+ */
+export const registerDesktopDeviceResponseSchema = z.object({
+  enabled: z.boolean(),
+  protocolVersion: z.number().int().min(1),
+  leaseMs: z.number().int().positive(),
+  leaseRenewMs: z.number().int().positive(),
+  pickupGraceMs: z.number().int().positive(),
+  activeReconcileMs: z.number().int().positive(),
+  idleReconcileMs: z.number().int().positive(),
+})
+
+export const registerDesktopDeviceContract = defineRouteContract({
+  method: 'POST',
+  path: '/api/desktop/devices',
+  body: registerDesktopDeviceBodySchema,
+  response: { mode: 'json', schema: registerDesktopDeviceResponseSchema },
+  error: z.object({ error: z.string() }),
+})
+
+const desktopInboxQuerySchema = z.object({ deviceId: desktopDeviceIdSchema })
+
+const desktopInboxCallSchema = z.object({
+  kind: z.literal('call'),
+  toolCallId: z.string().min(1),
+  toolName: z.string().min(1),
+  chatId: z.string().min(1),
+  workspaceId: z.string().nullable(),
+  createdAt: z.string(),
+})
+
+const desktopInboxApprovalSchema = z.object({
+  kind: z.literal('approval_needed'),
+  toolCallId: z.string().min(1),
+  toolName: z.string().min(1),
+  chatId: z.string().min(1),
+  chatTitle: z.string().nullable(),
+  workspaceId: z.string().nullable(),
+  /** A short description of what the call would do, such as the command to run. */
+  summary: z.string().nullable(),
+})
+
+const desktopInboxCancelSchema = z.object({
+  kind: z.literal('cancel'),
+  toolCallId: z.string().min(1),
+})
+
+/** Ordered by when each call was persisted, which is the order the executor claims them in. */
+const desktopInboxItemSchema = z.discriminatedUnion('kind', [
+  desktopInboxCallSchema,
+  desktopInboxApprovalSchema,
+  desktopInboxCancelSchema,
+])
+export type DesktopInboxItem = z.output<typeof desktopInboxItemSchema>
+
+export const desktopInboxResponseSchema = z.object({
+  items: z.array(desktopInboxItemSchema),
+  /** True while any run bound to this device is active, so the device pulls at the active cadence. */
+  hasActiveRun: z.boolean(),
+})
+
+export const listDesktopInboxContract = defineRouteContract({
+  method: 'GET',
+  path: '/api/desktop/inbox',
+  query: desktopInboxQuerySchema,
+  response: { mode: 'json', schema: desktopInboxResponseSchema },
+  error: z.object({ error: z.string() }),
+})
+
+/**
+ * Server-sent events: `inbox_changed` with `{ reason: 'call' | 'approval' | 'cancel' }`, plus
+ * heartbeat comments and a `rotate` event before the server closes a long-lived stream. Events
+ * are hints; the device re-reads `GET /api/desktop/inbox` on each one. While the stream is open
+ * the device counts as online.
+ */
+export const desktopInboxStreamContract = defineRouteContract({
+  method: 'GET',
+  path: '/api/desktop/inbox/stream',
+  query: desktopInboxQuerySchema,
+  response: { mode: 'stream' },
+})
+
+const claimDesktopToolBodySchema = z.object({
+  deviceId: desktopDeviceIdSchema,
+  toolCallId: desktopToolCallIdSchema,
+})
+export type ClaimDesktopToolBody = z.input<typeof claimDesktopToolBodySchema>
+
+export const claimDesktopToolResponseSchema = z.object({
+  toolName: z.string().min(1),
+  args: z.record(z.string(), z.unknown()),
+  chatId: z.string().min(1),
+  workspaceId: z.string().nullable(),
+  /** Presented on every lease renewal and on completion; it fences out stale owners. */
+  executionToken: z.string().min(1),
+  leaseExpiresAt: z.string(),
+})
+
+export const claimDesktopToolContract = defineRouteContract({
+  method: 'POST',
+  path: '/api/desktop/tool/claim',
+  body: claimDesktopToolBodySchema,
+  response: { mode: 'json', schema: claimDesktopToolResponseSchema },
+  error: z.object({ error: z.string() }),
+})
+
+const renewDesktopToolLeaseBodySchema = z.object({
+  deviceId: desktopDeviceIdSchema,
+  toolCallId: desktopToolCallIdSchema,
+  executionToken: z.string().min(1).max(128),
+})
+export type RenewDesktopToolLeaseBody = z.input<typeof renewDesktopToolLeaseBodySchema>
+
+export const renewDesktopToolLeaseResponseSchema = z.object({ leaseExpiresAt: z.string() })
+
+/** A 410 means the call was stopped, settled, or its lease lapsed: cancel the local action. */
+export const renewDesktopToolLeaseContract = defineRouteContract({
+  method: 'POST',
+  path: '/api/desktop/tool/lease',
+  body: renewDesktopToolLeaseBodySchema,
+  response: { mode: 'json', schema: renewDesktopToolLeaseResponseSchema },
+  error: z.object({ error: z.string() }),
+})
+
+const completeDesktopToolBodySchema = z.object({
+  deviceId: desktopDeviceIdSchema,
+  toolCallId: desktopToolCallIdSchema,
+  executionToken: z.string().min(1).max(128),
+  status: z.enum(['success', 'error', 'cancelled']),
+  message: z.string().max(10_000).optional(),
+  data: z.unknown().optional(),
+})
+export type CompleteDesktopToolBody = z.input<typeof completeDesktopToolBodySchema>
+
+/**
+ * Every 200 is an acknowledgement the device's outbox can drop the result on. `recorded`: this
+ * request settled the call. `duplicate`: an earlier request with the same token already did.
+ * `superseded`: Sim settled the call first (Stop, or a lapsed lease), and `status` is what the
+ * model received.
+ */
+export const completeDesktopToolResponseSchema = z.object({
+  outcome: z.enum(['recorded', 'duplicate', 'superseded']),
+  status: z.enum(['completed', 'failed', 'cancelled']),
+})
+
+export const completeDesktopToolContract = defineRouteContract({
+  method: 'POST',
+  path: '/api/desktop/tool/complete',
+  body: completeDesktopToolBodySchema,
+  response: { mode: 'json', schema: completeDesktopToolResponseSchema },
+  error: z.object({ error: z.string() }),
+})

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

@@ -0,0 +1,34 @@
+import {
+  extendInternalErrorPolicy,
+  internalErrorResponse,
+  internalOrchestrationErrorPolicy,
+  internalRateLimits,
+} from '@/lib/api/server/routes/internal-json-route'
+import {
+  DesktopCallRevokedError,
+  DesktopDeviceUnrecognizedError,
+} from '@/lib/desktop/executor/errors'
+
+/**
+ * A device's unrecognized binding is a 401, so it registers again; a call taken away from its
+ * token is a 410, so it stops the local action.
+ */
+export const desktopExecutorErrorPolicy = extendInternalErrorPolicy(
+  internalOrchestrationErrorPolicy,
+  (error) => {
+    if (error instanceof DesktopDeviceUnrecognizedError)
+      return internalErrorResponse(401, { error: error.message })
+    if (error instanceof DesktopCallRevokedError)
+      return internalErrorResponse(410, { error: error.message })
+    return null
+  }
+)
+
+/**
+ * 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.
+ */
+export const desktopExecutorRateLimit = internalRateLimits.user({
+  bucketName: 'desktop-executor',
+  config: { maxTokens: 600, refillRate: 600, refillIntervalMs: 60_000 },
+})

apps/sim/lib/core/config/env.ts

@@ -601,6 +601,7 @@ export const env = createEnv({
     DASHBOARDS: z.boolean().optional(),
     PROJECT_API_ENABLED:                   z.boolean().optional(),                 // Fallback for the `projects` feature flag off AppConfig
     MSHIP_MODEL_SELECTOR: z.boolean().optional(),
+    MSHIP_DESKTOP_BACKGROUND_EXECUTOR:     z.boolean().optional(),                 // Fallback for the `mothership-desktop-background-executor` feature flag off AppConfig
     INBOX_ENABLED:                         z.boolean().optional(),                 // Enable inbox (Sim Mailer) on self-hosted (bypasses hosted requirements)
     SANDBOXES_ENABLED:                     z.boolean().optional(),                 // Enable custom sandboxes on self-hosted (bypasses hosted requirements)
 

apps/sim/lib/core/config/feature-flags.test.ts

@@ -44,6 +44,7 @@ setEnv({
   TABLE_ROW_TTL: undefined,
   MSHIP_MODEL_SELECTOR: undefined,
   MSHIP_PLAN_MODE: undefined,
+  MSHIP_DESKTOP_BACKGROUND_EXECUTOR: undefined,
   AGENT_MEMORY_HISTORY: undefined,
   CREDENTIAL_GROUPS: undefined,
   KNOWLEDGE_MEMBER_ACCESS: undefined,