fix(desktop): address review on the background executor protocol

- Presence is a per-connection set, so overlapping streams never erase
  each other's presence.
- A result posted after Stop settles the call as stopped (superseded)
  instead of recording a success for a stopped turn.
- Nothing reads as awaiting approval while tool permissions are off.
- One reconcile cadence (10 s), shorter than the pickup grace, so a lost
  doorbell cannot fail the first call of a turn the device has not seen.
- The inbox query skips unclaimed calls on stopped runs and non-desktop
  tools, so they cannot crowd actionable rows past its limit.
- The doorbell stream applies the executor rate limit; inbox ids use the
  claim route's id schema; the index build recovers from an interrupted
  concurrent build; the E2E waits for the cancel doorbell specifically.

Author: waleedlatif1

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

@@ -16,8 +16,7 @@ export const GET = defineInternalJsonRoute({
   errorPolicy: desktopExecutorErrorPolicy,
   mapInput: ({ query }) => ({ deviceId: query.deviceId }),
   useCase: listDesktopInbox,
-  present: ({ items, hasActiveRun }) => ({
-    hasActiveRun,
+  present: ({ items }) => ({
     items: items.map((item) =>
       item.kind === 'call' ? { ...item, createdAt: item.createdAt.toISOString() } : item
     ),

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

@@ -2,6 +2,7 @@ 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,
@@ -22,6 +23,8 @@ const logger = createLogger('DesktopInboxStream')
 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

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

@@ -37,8 +37,7 @@ export const registerDesktopDeviceResponseSchema = z.object({
   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(),
+  reconcileMs: z.number().int().positive(),
 })
 
 export const registerDesktopDeviceContract = defineRouteContract({
@@ -53,7 +52,7 @@ const desktopInboxQuerySchema = z.object({ deviceId: desktopDeviceIdSchema })
 
 const desktopInboxCallSchema = z.object({
   kind: z.literal('call'),
-  toolCallId: z.string().min(1),
+  toolCallId: desktopToolCallIdSchema,
   toolName: z.string().min(1),
   chatId: z.string().min(1),
   workspaceId: z.string().nullable(),
@@ -62,7 +61,7 @@ const desktopInboxCallSchema = z.object({
 
 const desktopInboxApprovalSchema = z.object({
   kind: z.literal('approval_needed'),
-  toolCallId: z.string().min(1),
+  toolCallId: desktopToolCallIdSchema,
   toolName: z.string().min(1),
   chatId: z.string().min(1),
   chatTitle: z.string().nullable(),
@@ -73,7 +72,7 @@ const desktopInboxApprovalSchema = z.object({
 
 const desktopInboxCancelSchema = z.object({
   kind: z.literal('cancel'),
-  toolCallId: z.string().min(1),
+  toolCallId: desktopToolCallIdSchema,
 })
 
 /** Ordered by when each call was persisted, which is the order the executor claims them in. */
@@ -86,8 +85,6 @@ 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({

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

@@ -9,8 +9,9 @@ import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 const { redisUrl } = await vi.hoisted(async () => {
   const { readTestRedisUrl } = await import('@sim/db/testing/test-infrastructure')
   const url = readTestRedisUrl()
-  /** The real Redis module reads this at import. */
+  /** The real Redis module and the tool-permission switch read these at import. */
   if (url) process.env.REDIS_URL = url
+  process.env.COPILOT_TOOL_PERMISSIONS_ENABLED = 'true'
   return { redisUrl: url }
 })
 
@@ -28,6 +29,7 @@ import {
   workspace,
 } from '@sim/db/schema'
 import { featureFlagsMock, featureFlagsMockFns } from '@sim/testing/mocks/feature-flags.mock'
+import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { eq, inArray, sql } from 'drizzle-orm'
 import { closeRedisConnection } from '@/lib/core/config/redis'
@@ -278,7 +280,7 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
       ).rejects.toBeInstanceOf(DesktopDeviceUnrecognizedError)
       await expect(
         listDesktopInbox.execute({ principal: newer, input: { deviceId: desktop.deviceId } })
-      ).resolves.toEqual({ items: [], hasActiveRun: false })
+      ).resolves.toEqual({ items: [] })
     })
 
     it('disconnects the device when its session is signed out', async () => {
@@ -519,18 +521,20 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
         input: { deviceId: desktop.deviceId },
       })
       expect(inbox.items).toEqual([{ kind: 'cancel', toolCallId: running }])
-      expect(inbox.hasActiveRun).toBe(false)
 
-      /** The device's cancelled result acknowledges the cancellation. */
-      await completeDesktopTool.execute({
-        principal: desktop.principal,
-        input: {
-          deviceId: desktop.deviceId,
-          toolCallId: running,
-          executionToken,
-          status: 'cancelled',
-        },
-      })
+      /** Stop already answered the turn: even a success after it settles the call as stopped. */
+      await expect(
+        completeDesktopTool.execute({
+          principal: desktop.principal,
+          input: {
+            deviceId: desktop.deviceId,
+            toolCallId: running,
+            executionToken,
+            status: 'success',
+          },
+        })
+      ).resolves.toEqual({ outcome: 'superseded', status: 'cancelled' })
+      expect((await row(running)).status).toBe('cancelled')
       const after = await listDesktopInbox.execute({
         principal: desktop.principal,
         input: { deviceId: desktop.deviceId },
@@ -654,7 +658,6 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
         principal: desktop.principal,
         input: { deviceId: desktop.deviceId },
       })
-      expect(inbox.hasActiveRun).toBe(true)
       expect(inbox.items.map((item) => item.kind === 'call' && item.toolCallId)).toEqual([
         first,
         second,
@@ -682,6 +685,28 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
       await expect.poll(() => isDesktopPresent(desktop.deviceId)).toBe(false)
     })
 
+    it('keeps the device online while a replacement stream overlaps the one it replaces', async () => {
+      const desktop = await signedInDesktop()
+      const open = () =>
+        openDesktopInboxStream.execute({
+          principal: desktop.principal,
+          input: { deviceId: desktop.deviceId },
+        })
+      /** The stream that wrote presence last is the one that closes, while the other stays open. */
+      const closeStaying = (await open()).subscribe(() => {})
+      await expect.poll(() => isDesktopPresent(desktop.deviceId)).toBe(true)
+      const closeLeaving = (await open()).subscribe(() => {})
+      try {
+        await sleep(200)
+        closeLeaving()
+        await sleep(200)
+        expect(await isDesktopPresent(desktop.deviceId)).toBe(true)
+      } finally {
+        closeStaying()
+      }
+      await expect.poll(() => isDesktopPresent(desktop.deviceId)).toBe(false)
+    })
+
     it('ends a stream whose session was signed out at its next revalidation', async () => {
       const desktop = await signedInDesktop()
       const stream = await openDesktopInboxStream.execute({

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

@@ -12,8 +12,7 @@ import {
   DESKTOP_CALL_LEASE_SECONDS,
   DESKTOP_CALL_PICKUP_GRACE_MS,
   DESKTOP_EXECUTOR_PROTOCOL_VERSION,
-  DESKTOP_INBOX_ACTIVE_RECONCILE_MS,
-  DESKTOP_INBOX_IDLE_RECONCILE_MS,
+  DESKTOP_INBOX_RECONCILE_MS,
   DESKTOP_PRESENCE_REFRESH_MS,
 } from '@/lib/desktop/executor/constants'
 import {
@@ -35,7 +34,6 @@ import {
   claimOfferedDesktopCall,
   getBoundDesktopCall,
   getBoundDesktopDevice,
-  hasActiveDesktopRun,
   listDesktopInboxRows,
   recordDesktopCallResult,
   renewDesktopCallLease,
@@ -126,8 +124,7 @@ export const registerDesktopDevice = defineAuthorizedCredentialUserUseCase({
       leaseMs: DESKTOP_CALL_LEASE_SECONDS * 1000,
       leaseRenewMs: DESKTOP_CALL_LEASE_RENEW_MS,
       pickupGraceMs: DESKTOP_CALL_PICKUP_GRACE_MS,
-      activeReconcileMs: DESKTOP_INBOX_ACTIVE_RECONCILE_MS,
-      idleReconcileMs: DESKTOP_INBOX_IDLE_RECONCILE_MS,
+      reconcileMs: DESKTOP_INBOX_RECONCILE_MS,
     }
   },
 })
@@ -146,15 +143,14 @@ export const listDesktopInbox = defineAuthorizedCredentialUserUseCase({
   }: {
     principal: SessionPrincipal
     input: DeviceInput
-  }): Promise<{ items: DesktopInboxEntry[]; hasActiveRun: boolean }> {
+  }): Promise<{ items: DesktopInboxEntry[] }> {
     await requireBoundDevice(principal, input.deviceId)
     const identity = { deviceId: input.deviceId, userId: principal.userId }
-    const [rows, hasActiveRun] = await Promise.all([
+    const [rows] = await Promise.all([
       listDesktopInboxRows(identity),
-      hasActiveDesktopRun(identity),
       touchDesktopDevice(input.deviceId),
     ])
-    return { items: classifyDesktopInbox(rows), hasActiveRun }
+    return { items: classifyDesktopInbox(rows) }
   },
 })
 
@@ -320,6 +316,9 @@ export interface CompleteDesktopToolInput extends DesktopCallTokenInput {
   data?: unknown
 }
 
+const STOPPED_WHILE_RUNNING_MESSAGE =
+  'Stopped by the user while the Sim desktop app was running this action. It may already have taken effect; inspect the current state before repeating it.'
+
 const COMPLETION_STATUS = {
   success: { durable: ASYNC_TOOL_STATUS.completed, message: 'Tool completed' },
   error: { durable: ASYNC_TOOL_STATUS.failed, message: 'Tool failed' },
@@ -396,6 +395,7 @@ export const completeDesktopTool = defineAuthorizedCredentialUserUseCase({
       toolCallId: call.toolCallId,
       runId: call.runId,
       ownerToken: input.executionToken,
+      stoppedMessage: STOPPED_WHILE_RUNNING_MESSAGE,
     })
     if (acknowledged.outcome === 'unknown')
       throw new OrchestrationError('not_found', 'Desktop tool call not found')

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

@@ -16,13 +16,10 @@ export const DESKTOP_CALL_LEASE_RENEW_MS = 20_000
 export const DESKTOP_CALL_PICKUP_GRACE_MS = 15_000
 
 /**
- * The device's safety-net inbox pull while it has an active bound run. Shorter than the pickup
- * grace, so a lost doorbell still lets the device claim a call before it fails.
+ * The device's safety-net inbox pull. Shorter than the pickup grace, so a lost doorbell still lets
+ * the device claim a call before it fails, including the first call of a turn it has not seen.
  */
-export const DESKTOP_INBOX_ACTIVE_RECONCILE_MS = 10_000
-
-/** The device's safety-net inbox pull while none of its runs are active. */
-export const DESKTOP_INBOX_IDLE_RECONCILE_MS = 60_000
+export const DESKTOP_INBOX_RECONCILE_MS = 10_000
 
 /** Presence outlives two missed refreshes, so a rolling deploy does not read as offline. */
 export const DESKTOP_PRESENCE_TTL_SECONDS = 45

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

@@ -2,17 +2,10 @@ import { getRedisClient } from '@/lib/core/config/redis'
 import { DESKTOP_PRESENCE_TTL_SECONDS } from '@/lib/desktop/executor/constants'
 
 /**
- * A device is present while one of its inbox streams is open. Each stream writes its own
- * connection id, so a stream that closes after its replacement opened cannot erase the newer
- * stream's presence.
+ * A device is present while one of its inbox streams is open. Each stream holds its own entry in
+ * the device's set, scored by when that entry expires, so streams that overlap during a reconnect
+ * or rotation never erase each other's presence. The key itself expires with its newest entry.
  */
-const RELEASE_OWN_PRESENCE = `
-if redis.call('GET', KEYS[1]) == ARGV[1] then
-  return redis.call('DEL', KEYS[1])
-end
-return 0
-`
-
 function presenceKey(deviceId: string): string {
   return `desktop:presence:${deviceId}`
 }
@@ -26,22 +19,28 @@ export function isDesktopPresenceAvailable(): boolean {
 export async function markDesktopPresent(deviceId: string, connectionId: string): Promise<void> {
   const redis = getRedisClient()
   if (!redis) throw new Error('Desktop presence requires Redis')
-  await redis.set(presenceKey(deviceId), connectionId, 'EX', DESKTOP_PRESENCE_TTL_SECONDS)
+  const key = presenceKey(deviceId)
+  await redis
+    .multi()
+    .zremrangebyscore(key, '-inf', Date.now())
+    .zadd(key, Date.now() + DESKTOP_PRESENCE_TTL_SECONDS * 1000, connectionId)
+    .expire(key, DESKTOP_PRESENCE_TTL_SECONDS)
+    .exec()
 }
 
-/** Clears presence only when this connection is still the one recorded. */
+/** Removes only this connection's entry; any other open stream keeps the device present. */
 export async function releaseDesktopPresence(
   deviceId: string,
   connectionId: string
 ): Promise<void> {
   const redis = getRedisClient()
   if (!redis) return
-  await redis.eval(RELEASE_OWN_PRESENCE, 1, presenceKey(deviceId), connectionId)
+  await redis.zrem(presenceKey(deviceId), connectionId)
 }
 
-/** A device with no open inbox stream cannot pick up a call; a missing Redis reads as absent. */
+/** A device with no unexpired stream entry cannot pick up a call; a missing Redis reads as absent. */
 export async function isDesktopPresent(deviceId: string): Promise<boolean> {
   const redis = getRedisClient()
   if (!redis) return false
-  return (await redis.exists(presenceKey(deviceId))) === 1
+  return (await redis.zcount(presenceKey(deviceId), Date.now(), '+inf')) > 0
 }