Skip to content

Commit 6e5104c

Browse files
committed
fix(desktop): count a tmux run as handed back only once its result is journaled, and check a pane's tag in the same tmux command that acts on it
- A run is marked handed back only once the executor has journaled the result that hands it back; a chat-view run cannot prove its result reached the model, so after a restart it is treated as orphaned and stopped. - A run that a stop for everything (sign-out, Terminal off) could not confirm stays marked to stop, so a later launch for the same user never keeps it. - Every tag-guarded action (Ctrl-C, closing the pane, for live and recorded runs) is one `if-shell -F` command that checks the tag and acts, so a tmux restart between a check and an action cannot hand it to another pane.
1 parent 10674d0 commit 6e5104c

11 files changed

Lines changed: 181 additions & 27 deletions

File tree

‎apps/desktop/src/main/desktop-executor/executor.test.ts‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -168,6 +168,8 @@ function setup(
168168
const runner = new FakeRunner()
169169
const onUnregistered = vi.fn()
170170
const busy: boolean[] = []
171+
/** Calls whose result the executor reported as journaled, in order. */
172+
const journaled: string[] = []
171173
const executor = new DesktopExecutor({
172174
client: sim.client,
173175
journal,
@@ -177,12 +179,13 @@ function setup(
177179
onUnregistered,
178180
onBusyChange: (value) => busy.push(value),
179181
onApprovals: (items) => approvals.push(items),
182+
onResultRecorded: (toolCallId) => journaled.push(toolCallId),
180183
...(options.maxHeldCalls ? { maxHeldCalls: options.maxHeldCalls } : {}),
181184
...(options.deliveryAwakeLimitMs !== undefined
182185
? { deliveryAwakeLimitMs: options.deliveryAwakeLimitMs }
183186
: {}),
184187
})
185-
return { sim, journal, runner, executor, onUnregistered, busy, approvals }
188+
return { sim, journal, runner, executor, onUnregistered, busy, approvals, journaled }
186189
}
187190

188191
describe('claiming', () => {
@@ -203,6 +206,22 @@ describe('claiming', () => {
203206
expect(executor.heldCallCount()).toBe(0)
204207
})
205208

209+
it('reports a result as durable only once the journal holds it', async () => {
210+
const { sim, journal, runner, executor, journaled } = setup()
211+
runner.immediate = DONE
212+
journal.failOn = 'result'
213+
sim.inbox = [callItem('call-unjournaled', 'chat-a')]
214+
await executor.reconcile()
215+
await vi.waitFor(() => expect(sim.completions).toHaveLength(1))
216+
expect(journaled).toEqual([])
217+
218+
journal.failOn = null
219+
sim.inbox = [callItem('call-journaled', 'chat-a')]
220+
await executor.reconcile()
221+
await vi.waitFor(() => expect(sim.completions).toHaveLength(2))
222+
expect(journaled).toEqual(['call-journaled'])
223+
})
224+
206225
it('claims a whole backlog at once, before any of it runs', async () => {
207226
const { sim, runner, executor } = setup()
208227
sim.inbox = [callItem('a-1', 'chat-a'), callItem('a-2', 'chat-a'), callItem('a-3', 'chat-a')]

‎apps/desktop/src/main/desktop-executor/executor.ts‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,11 @@ export interface DesktopExecutorOptions {
6767
onApprovals?: (items: DesktopApprovalItem[]) => void
6868
/** Called whenever the number of held calls changes between zero and more. */
6969
onBusyChange?: (busy: boolean) => void
70+
/**
71+
* Called once a call's result is in the journal: from then on it reaches Sim, now or after a
72+
* restart, so anything it hands back (a pane still running) can be counted on.
73+
*/
74+
onResultRecorded?: (toolCallId: string, completion: DesktopToolCompletion) => void
7075
maxHeldCalls?: number
7176
/** First delivery retry delay; tests shorten it. */
7277
retryBaseMs?: number
@@ -375,7 +380,9 @@ export class DesktopExecutor {
375380
): Promise<void> {
376381
if (this.disposed) return
377382
// Best effort: unrecorded, a crash reports the call from its `started` entry as outcome unknown.
378-
await this.record({ toolCallId, state: 'result', executionToken, completion })
383+
if (await this.record({ toolCallId, state: 'result', executionToken, completion })) {
384+
this.options.onResultRecorded?.(toolCallId, completion)
385+
}
379386
const sendingSince = Date.now()
380387
try {
381388
await this.sendResult(toolCallId, executionToken, completion, sendingSince)

‎apps/desktop/src/main/desktop-executor/service.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import { hostname } from 'node:os'
88
import { join } from 'node:path'
99
import type { DesktopExecutorDevice } from '@sim/desktop-bridge'
10+
import type { DesktopToolCompletion } from '@sim/desktop-bridge/tool-results'
1011
import { createLogger } from '@sim/logger'
1112
import { getErrorMessage } from '@sim/utils/errors'
1213
import { generateId } from '@sim/utils/id'
@@ -61,6 +62,8 @@ export interface DesktopExecutorServiceDeps {
6162
onApprovals?: (items: DesktopApprovalItem[]) => void
6263
/** Whether any chat has desktop work claimed on this machine changed. */
6364
onBusyChange?: (busy: boolean) => void
65+
/** A call's result is in the journal, so it reaches Sim even across a restart. */
66+
onResultRecorded?: (toolCallId: string, completion: DesktopToolCompletion) => void
6467
}
6568

6669
export interface DesktopExecutorService {
@@ -245,6 +248,7 @@ export function createDesktopExecutorService(
245248
onUnregistered: handleUnrecognized,
246249
...(deps.onApprovals ? { onApprovals: deps.onApprovals } : {}),
247250
...(deps.onBusyChange ? { onBusyChange: deps.onBusyChange } : {}),
251+
...(deps.onResultRecorded ? { onResultRecorded: deps.onResultRecorded } : {}),
248252
})
249253
await executor.recover()
250254
// Signed out while recovering: sign-out already disposed this executor.

‎apps/desktop/src/main/index.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -607,6 +607,13 @@ function main(): void {
607607
accountDataAvailable,
608608
onApprovals: (items) => approvalNotifier.update(items),
609609
onBusyChange: (busy) => sleepBlocker.setBusy(busy),
610+
// A result that will reach the model (not one reported as not started or outcome unknown)
611+
// makes a tmux run it handed back as still going collectable across a restart.
612+
onResultRecorded: (toolCallId, completion) => {
613+
if (completion.data?.outcomeUnknown !== true && completion.data?.notStarted !== true) {
614+
terminal.markRunDelivered(toolCallId)
615+
}
616+
},
610617
runner: createDesktopToolRunner({
611618
preferences: () => desktopSettings.getPreferences(),
612619
accountDataAvailable,

‎apps/desktop/src/main/terminal/index.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1351,8 +1351,8 @@ export class TerminalService {
13511351
handle.dispose()
13521352
}
13531353
// Still going, it stays tracked, and nothing polls the status file again: `read` captures
1354-
// the pane instead. Its result now points the model at that pane, so a restart must not end it.
1355-
if (!outcome.done && handle.runId) ledger?.markDelivered(handle.runId)
1354+
// the pane instead. Its record is marked handed back only once that result is durable
1355+
// (the executor's journal); see `TerminalRegistry.markRunDelivered`.
13561356

13571357
const { text, truncated } = elideOutput(outcome.output)
13581358
return {

‎apps/desktop/src/main/terminal/registry-run-ledger.test.ts‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,35 @@ describe('stopping recorded tmux runs', () => {
8888
expect(ledger.list().map((record) => record.runId)).toEqual(['handed-back'])
8989
})
9090

91+
it('keeps meaning to stop a run a stop for everything could not confirm', async () => {
92+
const dir = ledgerDir()
93+
createRunLedger(dir).record(run('unconfirmed', '%1', true))
94+
panes.set('unconfirmed', 'unknown')
95+
const ledger = createRunLedger(dir)
96+
// Sign-out could not confirm the run ended.
97+
await new TerminalRegistry(undefined, undefined, ledger).stopAgentCommands()
98+
panes.set('unconfirmed', 'running')
99+
100+
// A later launch for the same user, with the journal cleared, still stops it.
101+
await new TerminalRegistry(undefined, undefined, ledger).stopUncollectableRuns(new Set())
102+
103+
expect(panes.get('unconfirmed')).toBe('stopped')
104+
expect(ledger.list()).toEqual([])
105+
})
106+
107+
it('notes a run as handed back once its call result is durable', () => {
108+
const dir = ledgerDir()
109+
const ledger = createRunLedger(dir)
110+
ledger.record(run('watched', '%1'))
111+
ledger.record(run('other', '%2'))
112+
113+
new TerminalRegistry(undefined, undefined, ledger).markRunDelivered('call-watched')
114+
115+
expect(
116+
Object.fromEntries(ledger.list().map((record) => [record.runId, record.delivered]))
117+
).toEqual({ watched: true, other: false })
118+
})
119+
91120
it("at launch, stops the previous process's runs and none this one has started", async () => {
92121
const dir = ledgerDir()
93122
createRunLedger(dir).record(run('previous', '%1'))

‎apps/desktop/src/main/terminal/registry.ts‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -396,10 +396,22 @@ export class TerminalRegistry {
396396
stopUncollectableRuns(unresolvedCalls: ReadonlySet<string> | null): Promise<void> {
397397
return this.stopRecordedRuns({
398398
excludeLive: true,
399-
keep: (run) => run.delivered && !unresolvedCalls?.has(run.callId),
399+
keep: (run) => run.delivered && !run.mustStop && !unresolvedCalls?.has(run.callId),
400400
})
401401
}
402402

403+
/**
404+
* Notes that a call's result is durable (in the executor's journal), so a tmux run it handed
405+
* back as still going may be left to the model across a restart.
406+
*/
407+
markRunDelivered(callId: string): void {
408+
const ledger = this.runLedger
409+
if (!ledger) return
410+
for (const run of ledger.list()) {
411+
if (run.callId === callId) ledger.markDelivered(run.runId)
412+
}
413+
}
414+
403415
/**
404416
* Stops the recorded tmux runs, each only while its pane still carries its tag, and drops the
405417
* records with nothing left to stop. `excludeLive` skips the runs this process has started;
@@ -417,10 +429,14 @@ export class TerminalRegistry {
417429
if (!ledger) return
418430
await Promise.allSettled(
419431
ledger.list({ excludeLive: options.excludeLive }).map(async (run) => {
420-
const state = options.keep?.(run)
432+
const keeping = options.keep?.(run) ?? false
433+
const state = keeping
421434
? await recordedRunState(run, process.env)
422435
: await stopRecordedRun(run, process.env, RECORDED_RUN_GRACE_MS)
423436
if (state === 'gone') ledger.forget(run.runId)
437+
// A run this sweep meant to stop but could not confirm stays meant to stop, so no later
438+
// sweep keeps it.
439+
else if (!keeping) ledger.markMustStop(run.runId)
424440
})
425441
)
426442
}

‎apps/desktop/src/main/terminal/run-ledger.ts‎

Lines changed: 28 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -27,13 +27,17 @@ export interface RunRecord extends RecordedRun {
2727
* pane): from then on the model can come back to the pane, so a restart must leave it be.
2828
*/
2929
delivered: boolean
30+
/** A stop for everything (sign-out, Terminal off) could not confirm this run ended. */
31+
mustStop?: boolean
3032
}
3133

3234
export interface RunLedger {
3335
/** Saves a run's record; false when it could not be saved, so the run must not start. */
3436
record(run: RunRecord): boolean
3537
/** Notes that the run's call has handed back its result, with the run still going. */
3638
markDelivered(runId: string): void
39+
/** Notes that the run must be stopped, whatever a later launch would otherwise decide. */
40+
markMustStop(runId: string): void
3741
forget(runId: string): void
3842
/** Every recorded run; `excludeLive` leaves out runs this process recorded. */
3943
list(options?: { excludeLive?: boolean }): RunRecord[]
@@ -61,6 +65,7 @@ function parseRecord(text: string): RunRecord | null {
6165
socket: parsed.socket,
6266
callId: parsed.callId,
6367
delivered: parsed.delivered,
68+
...(parsed.mustStop === true ? { mustStop: true } : {}),
6469
}
6570
}
6671
} catch {
@@ -83,6 +88,24 @@ export function createRunLedger(dir: string): RunLedger {
8388
const live = new Set<string>()
8489
const pathFor = (runId: string) => join(dir, `${runId}.json`)
8590

91+
/** Rewrites a saved record; `change` returns null to leave it as it is. */
92+
const update = (runId: string, change: (record: RunRecord) => RunRecord | null): void => {
93+
if (!RUN_ID.test(runId)) return
94+
let record: RunRecord | null = null
95+
try {
96+
record = parseRecord(readFileSync(pathFor(runId), 'utf8'))
97+
} catch {
98+
record = null
99+
}
100+
const changed = record ? change(record) : null
101+
if (!changed) return
102+
try {
103+
writeJsonFileAtomicallySync(pathFor(runId), changed)
104+
} catch (error) {
105+
logger.warn('Could not update a tmux run record', { error: getErrorMessage(error) })
106+
}
107+
}
108+
86109
return {
87110
record(run) {
88111
if (!RUN_ID.test(run.runId)) return false
@@ -96,20 +119,11 @@ export function createRunLedger(dir: string): RunLedger {
96119
}
97120
},
98121
markDelivered(runId) {
99-
if (!RUN_ID.test(runId)) return
100-
let record: RunRecord | null = null
101-
try {
102-
record = parseRecord(readFileSync(pathFor(runId), 'utf8'))
103-
} catch {
104-
record = null
105-
}
106-
if (!record || record.delivered) return
107-
try {
108-
writeJsonFileAtomicallySync(pathFor(runId), { ...record, delivered: true })
109-
} catch (error) {
110-
// Left undelivered, a restart stops the run: the conservative side.
111-
logger.warn('Could not note a tmux run as handed back', { error: getErrorMessage(error) })
112-
}
122+
// Left undelivered, a restart stops the run: the conservative side.
123+
update(runId, (record) => (record.delivered ? null : { ...record, delivered: true }))
124+
},
125+
markMustStop(runId) {
126+
update(runId, (record) => (record.mustStop ? null : { ...record, mustStop: true }))
113127
},
114128
forget(runId) {
115129
live.delete(runId)

‎apps/desktop/src/main/terminal/service.test.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -662,8 +662,8 @@ describe('agent commands in tmux', () => {
662662
pane,
663663
socket: '/tmp/tmux-fake/default',
664664
callId: 'call-long',
665-
// Handed back as still running, with its pane: a restart must leave it be.
666-
delivered: true,
665+
// Handed back, but only the executor's journal can make that durable.
666+
delivered: false,
667667
},
668668
])
669669

‎apps/desktop/src/main/terminal/tmux.test.ts‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -136,6 +136,8 @@ interface FakeTmuxState {
136136
retagAtCheck?: number
137137
/** display-message calls so far. */
138138
checks?: number
139+
/** tmux restarts and the user's pane takes the id just before the next guarded action. */
140+
retagBeforeAction?: boolean
139141
/** Attached clients, as `list-clients` reports them. */
140142
clients?: Array<{ pid: string; tty: string; session: string }>
141143
/** Commands the fake holds until the file named here exists, like a busy tmux server. */
@@ -174,6 +176,18 @@ if (state.hold && state.hold[args[0]]) {
174176
}
175177
state.held = state.held.filter((command) => command !== args[0])
176178
}
179+
// \`if-shell -F -t pane '#{==:#{option},value}' command\`: the check and the action in one command.
180+
if (args[0] === 'if-shell') {
181+
const pane = state.panes[target()]
182+
if (pane && state.retagBeforeAction) {
183+
pane.options['@sim-run-id'] = 'someone-else'
184+
state.retagBeforeAction = false
185+
save()
186+
}
187+
const check = /^#\\{==:#\\{([^}]+)\\},(.*)\\}$/.exec(args[args.length - 2])
188+
if (!pane || !check || (pane.options[check[1]] ?? '') !== check[2]) process.exit(0)
189+
args = args[args.length - 1].split(' ')
190+
}
177191
if (state.fail && state.fail[args[0]]) fail(state.fail[args[0]])
178192
switch (args[0]) {
179193
case 'new-window': {
@@ -406,6 +420,17 @@ describe('stopping a run another process started, from its record', () => {
406420
expect([...saved]).toEqual([])
407421
})
408422

423+
it('acts on no pane that took the recorded id between the last check and the action', async () => {
424+
const tmux = fakeTmux()
425+
const { record } = await recorded(tmux)
426+
tmux.write({ ...tmux.read(), retagBeforeAction: true })
427+
428+
await stopRecordedRun(record, tmux.env, 0)
429+
430+
expect(tmux.read().log).toEqual([])
431+
expect(Object.keys(tmux.read().panes)).toEqual([record.pane])
432+
})
433+
409434
it('keeps the record when tmux stops answering part-way through the stop', async () => {
410435
const tmux = fakeTmux()
411436
const { record } = await recorded(tmux)
@@ -485,6 +510,17 @@ describe('stopping a tmux run touches only its own pane', () => {
485510
expect(Object.keys(tmux.read().panes)).toEqual([users])
486511
})
487512

513+
it('sends a live run nothing once its pane is retagged between the check and the action', async () => {
514+
const tmux = fakeTmux()
515+
const run = await started(tmux)
516+
tmux.write({ ...tmux.read(), retagBeforeAction: true })
517+
518+
await stopRun(run, tmux.env, 0)
519+
520+
expect(tmux.read().log).toEqual([])
521+
expect(Object.keys(tmux.read().panes)).toEqual([run.pane])
522+
})
523+
488524
it('sends nothing to a pane that reused the run pane id after tmux restarted', async () => {
489525
const tmux = fakeTmux()
490526
const run = await started(tmux)

0 commit comments

Comments
 (0)