Skip to content

Commit 0b051f5

Browse files
committed
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.
1 parent 0a5f255 commit 0b051f5

11 files changed

Lines changed: 241 additions & 62 deletions

File tree

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,11 @@ export const desktopExecutorErrorPolicy = extendInternalErrorPolicy(
2626

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

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

Lines changed: 50 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -244,8 +244,10 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
244244

245245
afterAll(async () => {
246246
if (chatIds.length) await db.delete(copilotChats).where(inArray(copilotChats.id, chatIds))
247-
if (deviceIds.length)
247+
if (deviceIds.length) {
248+
await db.delete(auditLog).where(inArray(auditLog.resourceId, deviceIds))
248249
await db.delete(desktopDevices).where(inArray(desktopDevices.id, deviceIds))
250+
}
249251
const workspaces = [...workspaceIds.values()]
250252
if (workspaces.length) await db.delete(workspace).where(inArray(workspace.id, workspaces))
251253
if (userIds.length) await db.delete(user).where(inArray(user.id, userIds))
@@ -481,7 +483,7 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
481483
const toolCallId = await pendingCall(
482484
run.runId,
483485
'terminal',
484-
{ operation: 'run', command: 'npm test' },
486+
{ operation: 'run', args: { command: 'npm test' } },
485487
{ awaitingApproval: true }
486488
)
487489
expect((await inbox(desktop)).items).toEqual([
@@ -512,7 +514,7 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
512514
expect((await inbox(desktop)).items).toMatchObject([{ kind: 'call', toolCallId }])
513515
await expect(claim(desktop.principal, desktop.deviceId, toolCallId)).resolves.toMatchObject({
514516
toolName: 'terminal',
515-
args: { operation: 'run', command: 'npm test' },
517+
args: { operation: 'run', args: { command: 'npm test' } },
516518
})
517519
})
518520

@@ -521,7 +523,7 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
521523
const run = await boundRun(desktop)
522524
const toolCallId = await pendingCall(run.runId, 'terminal', {
523525
operation: 'run',
524-
command: 'ls',
526+
args: { command: 'ls' },
525527
})
526528
expect((await inbox(desktop)).items).toMatchObject([{ kind: 'call', toolCallId }])
527529
})
@@ -532,7 +534,7 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
532534
const toolCallId = await pendingCall(
533535
run.runId,
534536
'terminal',
535-
{ operation: 'run', command: 'rm -rf build' },
537+
{ operation: 'run', args: { command: 'rm -rf build' } },
536538
{ awaitingApproval: true }
537539
)
538540
await db
@@ -820,6 +822,49 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
820822
}
821823
})
822824

825+
it('lists new work however many calls the device is already running', async () => {
826+
const desktop = await signedInDesktop()
827+
const run = await boundRun(desktop)
828+
await db.insert(copilotAsyncToolCalls).values(
829+
Array.from({ length: 501 }, () => ({
830+
runId: run.runId,
831+
toolCallId: generateId(),
832+
toolName: 'browser_click',
833+
args: { ref: 'e1' },
834+
status: 'running' as const,
835+
claimedBy: 'desktop-browser',
836+
executionOwnerToken: generateId(),
837+
executionLeaseExpiresAt: sql`now() + interval '1 minute'`,
838+
createdAt: new Date(Date.now() - 60_000),
839+
}))
840+
)
841+
/** A call Sim settled without the device's result, still to be acknowledged. */
842+
const stopped = generateId()
843+
await db.insert(copilotAsyncToolCalls).values({
844+
runId: run.runId,
845+
toolCallId: stopped,
846+
toolName: 'browser_click',
847+
args: { ref: 'e2' },
848+
status: 'cancelled',
849+
executionOwnerToken: generateId(),
850+
executionRevokedAt: new Date(),
851+
createdAt: new Date(Date.now() - 30_000),
852+
})
853+
const offered = await pendingCall(run.runId)
854+
const gated = await pendingCall(
855+
run.runId,
856+
'terminal',
857+
{ operation: 'run', args: { command: 'make' } },
858+
{ awaitingApproval: true }
859+
)
860+
861+
expect((await inbox(desktop)).items).toMatchObject([
862+
{ kind: 'cancel', toolCallId: stopped },
863+
{ kind: 'call', toolCallId: offered },
864+
{ kind: 'approval_needed', toolCallId: gated },
865+
])
866+
})
867+
823868
it('lists pending calls in persistence order even when their doorbell was never heard', async () => {
824869
const desktop = await signedInDesktop()
825870
const run = await boundRun(desktop)

‎apps/sim/lib/desktop/executor/bound-turn.integration.ts‎

Lines changed: 49 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ vi.mock('@/lib/core/config/feature-flags', () => featureFlagsMock)
3939
import type { SessionPrincipal } from '@sim/auth/principal'
4040
import { db } from '@sim/db'
4141
import {
42+
auditLog,
4243
copilotAsyncToolCalls,
4344
copilotChats,
4445
copilotRuns,
@@ -363,8 +364,10 @@ describe.runIf(Boolean(redisUrl))("a turn bound to a desktop's background execut
363364

364365
afterAll(async () => {
365366
if (chatIds.length > 0) await db.delete(copilotChats).where(inArray(copilotChats.id, chatIds))
366-
if (deviceIds.length > 0)
367+
if (deviceIds.length > 0) {
368+
await db.delete(auditLog).where(inArray(auditLog.resourceId, deviceIds))
367369
await db.delete(desktopDevices).where(inArray(desktopDevices.id, deviceIds))
370+
}
368371
await db.delete(workspace).where(eq(workspace.id, workspaceId))
369372
await db.delete(user).where(eq(user.id, userId))
370373
})
@@ -545,7 +548,9 @@ describe.runIf(Boolean(redisUrl))("a turn bound to a desktop's background execut
545548
(rings) => rings.length > 0
546549
)
547550
expect(desktop.rings).toEqual(['approval'])
548-
expect((await desktop.pull()).items).toMatchObject([{ kind: 'approval_needed', toolCallId }])
551+
expect((await desktop.pull()).items).toMatchObject([
552+
{ kind: 'approval_needed', toolCallId, summary: 'rm -rf build' },
553+
])
549554

550555
const decided = await post(toolPermissionPOST, '/api/copilot/tool-permission', {
551556
decisions: [{ toolCallId, decision: 'allow' }],
@@ -566,6 +571,48 @@ describe.runIf(Boolean(redisUrl))("a turn bound to a desktop's background execut
566571
TURN_WAIT_MS
567572
)
568573

574+
it(
575+
'runs no pickup window while the user decides, and offers the call afresh once allowed',
576+
async () => {
577+
const desktop = await signedInDesktop()
578+
const run = await boundRun(desktop)
579+
await desktop.pull()
580+
581+
const { toolCallId, answer, context } = await agentCalls(
582+
run,
583+
'terminal',
584+
{ operation: 'run', args: { command: 'make release' } },
585+
{ gated: true }
586+
)
587+
/** The user takes longer than a pickup window to answer, with a deadline already on the row. */
588+
await db
589+
.update(copilotAsyncToolCalls)
590+
.set({
591+
createdAt: new Date(Date.now() - 600_000),
592+
pickupDeadlineAt: sql`now() - interval '5 minutes'`,
593+
})
594+
.where(eq(copilotAsyncToolCalls.toolCallId, toolCallId))
595+
await runCleanupStaleExecutions()
596+
expect((await storedCall(toolCallId)).status).toBe('pending')
597+
expect((await desktop.pull()).items).toMatchObject([{ kind: 'approval_needed', toolCallId }])
598+
599+
const decided = await post(toolPermissionPOST, '/api/copilot/tool-permission', {
600+
decisions: [{ toolCallId, decision: 'allow' }],
601+
})
602+
expect(decided.status).toBe(200)
603+
await until(
604+
() => storedCall(toolCallId),
605+
(row) => row.pickupDeadlineAt !== null && row.pickupDeadlineAt.getTime() > Date.now()
606+
)
607+
const { executionToken } = await desktop.claim(toolCallId)
608+
await desktop.complete(toolCallId, executionToken, { output: 'released' })
609+
await answer
610+
611+
expect(resultOf(context, toolCallId)).toMatchObject({ success: true })
612+
},
613+
TURN_WAIT_MS
614+
)
615+
569616
it(
570617
'ends a running call on Stop without the device, and tells the device to cancel it',
571618
async () => {
Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
import { describe, expect, it, vi } from 'vitest'
2+
3+
vi.mock('@/lib/events/pubsub', () => ({
4+
createPubSubChannel: () => ({
5+
publish: () => {
6+
throw new Error('Redis connection closed')
7+
},
8+
subscribe: vi.fn(),
9+
dispose: vi.fn(),
10+
}),
11+
}))
12+
13+
import { ringDesktopInbox } from '@/lib/desktop/executor/doorbell'
14+
15+
describe('desktop inbox doorbell', () => {
16+
it('never fails its caller when the ring cannot be published', () => {
17+
expect(() => ringDesktopInbox('device-1', 'cancel')).not.toThrow()
18+
})
19+
})

‎apps/sim/lib/desktop/executor/doorbell.ts‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
1+
import { createLogger } from '@sim/logger'
2+
import { toError } from '@sim/utils/errors'
13
import { createPubSubChannel, type PubSubChannel } from '@/lib/events/pubsub'
24

5+
const logger = createLogger('DesktopInboxDoorbell')
6+
37
/** Why a device should re-read its inbox. The event is only a hint; the inbox is the record. */
48
export type DesktopInboxChangeReason = 'call' | 'approval' | 'cancel'
59

@@ -22,9 +26,21 @@ function channel(): PubSubChannel<DesktopInboxDoorbell> {
2226
return scope._desktopInboxDoorbell
2327
}
2428

25-
/** Tells every pod serving this device's inbox stream that the inbox changed. */
29+
/**
30+
* Tells every pod serving this device's inbox stream that the inbox changed. Best effort: the
31+
* ring only hurries the device's next pull, and its periodic reconcile reads the same inbox, so a
32+
* failed ring is logged and never fails the caller (Stop, a decision, an offer).
33+
*/
2634
export function ringDesktopInbox(deviceId: string, reason: DesktopInboxChangeReason): void {
27-
channel().publish({ deviceId, reason })
35+
try {
36+
channel().publish({ deviceId, reason })
37+
} catch (error) {
38+
logger.warn('Could not ring the desktop inbox', {
39+
deviceId,
40+
reason,
41+
error: toError(error).message,
42+
})
43+
}
2844
}
2945

3046
/** Subscribes to one device's doorbell; returns the unsubscribe. */

‎apps/sim/lib/desktop/executor/inbox.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,9 +27,11 @@ export type DesktopInboxEntry =
2727
}
2828
| { kind: 'cancel'; toolCallId: string }
2929

30+
/** The command a gated terminal run would execute: `{ operation: 'run', args: { command } }`. */
3031
function summarize(args: Record<string, unknown>): string | null {
31-
return typeof args.command === 'string' && args.command.trim()
32-
? truncate(args.command.trim(), SUMMARY_MAX_LENGTH)
32+
const command = isPlainRecord(args.args) ? args.args.command : undefined
33+
return typeof command === 'string' && command.trim()
34+
? truncate(command.trim(), SUMMARY_MAX_LENGTH)
3335
: null
3436
}
3537

‎apps/sim/lib/desktop/executor/repository.ts‎

Lines changed: 59 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import {
1313
inArray,
1414
isNotNull,
1515
isNull,
16+
ne,
1617
notInArray,
1718
or,
1819
type SQL,
@@ -152,55 +153,64 @@ export async function touchDesktopDevice(deviceId: string): Promise<void> {
152153
}
153154

154155
/**
155-
* The rows a device's inbox is built from: every unclaimed pending desktop call on its recent open
156-
* runs, plus every call it claimed and has not yet acknowledged. Pending calls on stopped or ended
157-
* runs are left out here, so they cannot crowd actionable rows past the limit. Ordered by
158-
* persistence time, the order the device claims in.
156+
* The rows a device's inbox is built from, in two sets with their own limits so neither can crowd
157+
* out the other: unclaimed calls on its recent open runs that are offered or waiting for the
158+
* user's decision, and calls it claimed that Sim settled without its result and it has not yet
159+
* acknowledged (cancel items). A call the device is still running is never listed: it already
160+
* holds it. Ordered by persistence time, the order the device claims in.
159161
*/
160162
export async function listDesktopInboxRows(identity: Omit<DesktopDeviceIdentity, 'sessionId'>) {
161-
return db
162-
.select({
163-
toolCallId: copilotAsyncToolCalls.toolCallId,
164-
toolName: copilotAsyncToolCalls.toolName,
165-
args: copilotAsyncToolCalls.args,
166-
status: copilotAsyncToolCalls.status,
167-
permissionRequestedAt: copilotAsyncToolCalls.permissionRequestedAt,
168-
permissionDecision: copilotAsyncToolCalls.permissionDecision,
169-
claimed: sql<boolean>`${copilotAsyncToolCalls.executionOwnerToken} IS NOT NULL`,
170-
createdAt: copilotAsyncToolCalls.createdAt,
171-
chatId: copilotRuns.chatId,
172-
chatTitle: copilotChats.title,
173-
workspaceId: copilotRuns.workspaceId,
174-
})
175-
.from(copilotRuns)
176-
.innerJoin(copilotAsyncToolCalls, eq(copilotAsyncToolCalls.runId, copilotRuns.id))
177-
.innerJoin(copilotChats, eq(copilotChats.id, copilotRuns.chatId))
178-
.where(
179-
and(
180-
eq(copilotRuns.desktopDeviceId, identity.deviceId),
181-
eq(copilotRuns.userId, identity.userId),
182-
sql`${copilotRuns.startedAt} > now() - make_interval(hours => ${DESKTOP_INBOX_HORIZON_HOURS})`,
183-
inArray(copilotAsyncToolCalls.toolName, [...DESKTOP_TOOL_CALL_NAMES]),
184-
or(
185-
and(
186-
eq(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.pending),
187-
isNull(copilotAsyncToolCalls.executionOwnerToken),
188-
or(
189-
sql`${copilotAsyncToolCalls.pickupDeadlineAt} > clock_timestamp()`,
190-
awaitingPermission
191-
),
192-
inArray(copilotRuns.status, LIVE_RUN_STATUSES),
193-
isNull(copilotRuns.toolAdmissionClosedAt)
194-
),
195-
and(
196-
isNotNull(copilotAsyncToolCalls.executionOwnerToken),
197-
isNull(copilotAsyncToolCalls.executionSettledAt)
198-
)
163+
const rowsWhere = (state: SQL | undefined) =>
164+
db
165+
.select({
166+
toolCallId: copilotAsyncToolCalls.toolCallId,
167+
toolName: copilotAsyncToolCalls.toolName,
168+
args: copilotAsyncToolCalls.args,
169+
status: copilotAsyncToolCalls.status,
170+
permissionRequestedAt: copilotAsyncToolCalls.permissionRequestedAt,
171+
permissionDecision: copilotAsyncToolCalls.permissionDecision,
172+
claimed: sql<boolean>`${copilotAsyncToolCalls.executionOwnerToken} IS NOT NULL`,
173+
createdAt: copilotAsyncToolCalls.createdAt,
174+
chatId: copilotRuns.chatId,
175+
chatTitle: copilotChats.title,
176+
workspaceId: copilotRuns.workspaceId,
177+
})
178+
.from(copilotRuns)
179+
.innerJoin(copilotAsyncToolCalls, eq(copilotAsyncToolCalls.runId, copilotRuns.id))
180+
.innerJoin(copilotChats, eq(copilotChats.id, copilotRuns.chatId))
181+
.where(
182+
and(
183+
eq(copilotRuns.desktopDeviceId, identity.deviceId),
184+
eq(copilotRuns.userId, identity.userId),
185+
sql`${copilotRuns.startedAt} > now() - make_interval(hours => ${DESKTOP_INBOX_HORIZON_HOURS})`,
186+
inArray(copilotAsyncToolCalls.toolName, [...DESKTOP_TOOL_CALL_NAMES]),
187+
state
199188
)
200189
)
201-
)
202-
.orderBy(asc(copilotAsyncToolCalls.createdAt), asc(copilotAsyncToolCalls.toolCallId))
203-
.limit(INBOX_ROW_LIMIT)
190+
.orderBy(asc(copilotAsyncToolCalls.createdAt), asc(copilotAsyncToolCalls.toolCallId))
191+
.limit(INBOX_ROW_LIMIT)
192+
const [waiting, cancelled] = await Promise.all([
193+
rowsWhere(
194+
and(
195+
eq(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.pending),
196+
isNull(copilotAsyncToolCalls.executionOwnerToken),
197+
or(sql`${copilotAsyncToolCalls.pickupDeadlineAt} > clock_timestamp()`, awaitingPermission),
198+
inArray(copilotRuns.status, LIVE_RUN_STATUSES),
199+
isNull(copilotRuns.toolAdmissionClosedAt)
200+
)
201+
),
202+
rowsWhere(
203+
and(
204+
isNotNull(copilotAsyncToolCalls.executionOwnerToken),
205+
isNull(copilotAsyncToolCalls.executionSettledAt),
206+
ne(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.running)
207+
)
208+
),
209+
])
210+
return [...waiting, ...cancelled].sort(
211+
(a, b) =>
212+
a.createdAt.getTime() - b.createdAt.getTime() || a.toolCallId.localeCompare(b.toolCallId)
213+
)
204214
}
205215

206216
export type DesktopInboxRow = Awaited<ReturnType<typeof listDesktopInboxRows>>[number]
@@ -295,7 +305,9 @@ export async function getRunDesktopDeviceId(runId: string): Promise<string | nul
295305

296306
/**
297307
* Offers a pending call on a bound run to its device by opening its pickup window. Only an
298-
* unclaimed call can be offered, and only once, so a re-dispatched call keeps its first deadline.
308+
* unclaimed call the user is not still deciding on can be offered, and only once, so a
309+
* re-dispatched call keeps its first deadline; recording the user's decision clears it, so an
310+
* allowed call's window starts when it is offered after the answer.
299311
*/
300312
export async function offerDesktopToolCall(input: {
301313
toolCallId: string
@@ -313,6 +325,7 @@ export async function offerDesktopToolCall(input: {
313325
eq(copilotAsyncToolCalls.runId, input.runId),
314326
eq(copilotAsyncToolCalls.status, ASYNC_TOOL_STATUS.pending),
315327
isNull(copilotAsyncToolCalls.pickupDeadlineAt),
328+
sql`NOT (${awaitingPermission})`,
316329
sql`EXISTS (SELECT 1 FROM ${copilotRuns} r WHERE r.id = ${copilotAsyncToolCalls.runId} AND r.desktop_device_id IS NOT NULL)`
317330
)
318331
)

0 commit comments

Comments
 (0)