fix(desktop): stop a run the model never got the result of (#8749)

* fix(desktop): stop a run the model never got the result of

* fix(desktop): never stop a run the model may hold

* fix(desktop): only a recovered result leaves a duplicate open

* fix(desktop): keep a recovered result's mark while it is parked

Author: waleedlatif1

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

@@ -170,6 +170,8 @@ function setup(
   const busy: boolean[] = []
   /** Calls whose result the executor reported as reaching the model, in order. */
   const delivered: string[] = []
+  /** Calls Sim finished with without their real result reaching the model, in order. */
+  const notDelivered: string[] = []
   const executor = new DesktopExecutor({
     client: sim.client,
     journal,
@@ -180,12 +182,23 @@ function setup(
     onBusyChange: (value) => busy.push(value),
     onApprovals: (items) => approvals.push(items),
     onResultDelivered: (toolCallId) => delivered.push(toolCallId),
+    onResultNotDelivered: (toolCallId) => notDelivered.push(toolCallId),
     ...(options.maxHeldCalls ? { maxHeldCalls: options.maxHeldCalls } : {}),
     ...(options.deliveryAwakeLimitMs !== undefined
       ? { deliveryAwakeLimitMs: options.deliveryAwakeLimitMs }
       : {}),
   })
-  return { sim, journal, runner, executor, onUnregistered, busy, approvals, delivered }
+  return {
+    sim,
+    journal,
+    runner,
+    executor,
+    onUnregistered,
+    busy,
+    approvals,
+    delivered,
+    notDelivered,
+  }
 }
 
 describe('claiming', () => {
@@ -207,13 +220,13 @@ describe('claiming', () => {
   })
 
   it('reports a result as reaching the model only once Sim takes it as the call own', async () => {
-    const { sim, journal, runner, executor, delivered } = setup()
+    const { sim, journal, runner, executor, delivered, notDelivered } = setup()
     runner.immediate = DONE
     // Sim settled this call first: the result never reached the model.
     sim.completionOutcome = 'superseded'
     sim.inbox = [callItem('call-superseded', 'chat-a')]
     await executor.reconcile()
-    await vi.waitFor(() => expect(sim.completions).toHaveLength(1))
+    await vi.waitFor(() => expect(notDelivered).toEqual(['call-superseded']))
     expect(delivered).toEqual([])
 
     // Taken by Sim, even with nothing written locally (no OS encryption, say).
@@ -222,10 +235,57 @@ describe('claiming', () => {
     sim.inbox = [callItem('call-recorded', 'chat-a')]
     await executor.reconcile()
     await vi.waitFor(() => expect(delivered).toEqual(['call-recorded']))
+    expect(notDelivered).toEqual(['call-superseded'])
+  })
+
+  it('reports a result Sim settled first, while a failed send waited to retry, as never reaching the model', async () => {
+    const { sim, runner, executor, delivered, notDelivered } = setup()
+    // The terminal handed back a command still going, and the first send of that result fails.
+    runner.immediate = { status: 'success', message: 'running', data: { status: 'running' } }
+    sim.completeErrors = [new DeviceRequestError(503, 'unavailable')]
+    sim.inbox = [callItem('call-1', 'chat-a', 'terminal')]
+    await executor.reconcile()
+    await vi.waitFor(() => expect(runner.started).toEqual(['call-1']))
+
+    // The user pressed Stop meanwhile, so Sim settled the call before the retry reached it.
+    sim.completionOutcome = 'superseded'
+    sim.inbox = [{ kind: 'cancel', toolCallId: 'call-1' }]
+    await executor.reconcile()
+
+    await vi.waitFor(() => expect(notDelivered).toEqual(['call-1']))
+    expect(delivered).toEqual([])
+  })
+
+  it('reports a stand-in Sim holds as a duplicate as never reaching the model, within one app run', async () => {
+    const { sim, runner, executor, delivered, notDelivered } = setup()
+    // Too large to send, so Sim never took the real result; the stand-in's first answer was lost.
+    runner.immediate = DONE
+    sim.completeErrors = [
+      new DeviceRequestError(413, 'too large'),
+      new DeviceRequestError(0, 'connection reset'),
+    ]
+    sim.completionOutcome = 'duplicate'
+    sim.inbox = [callItem('call-1', 'chat-a', 'terminal')]
+
+    await executor.reconcile()
+
+    await vi.waitFor(() => expect(notDelivered).toEqual(['call-1']))
+    expect(delivered).toEqual([])
+  })
+
+  it('reports a result Sim refused as never reaching the model', async () => {
+    const { sim, runner, executor, notDelivered } = setup()
+    runner.immediate = DONE
+    sim.completeErrors = [new DeviceRequestError(403, 'forbidden')]
+    sim.inbox = [callItem('call-refused', 'chat-a')]
+
+    await executor.reconcile()
+
+    await vi.waitFor(() => expect(notDelivered).toEqual(['call-refused']))
   })
 
   it('reports no result that stands in for what the action produced as reaching the model', async () => {
-    const { sim, runner, executor, delivered } = setup()
+    const { sim, runner, executor, delivered, notDelivered } = setup()
     const standIns: DesktopToolCompletion[] = [
       { status: 'cancelled', message: 'Stopped.' },
       { status: 'error', message: 'Too large.', data: { resultOmitted: true } },
@@ -243,6 +303,7 @@ describe('claiming', () => {
     sim.inbox = [callItem('real', 'chat-a')]
     await executor.reconcile()
     await vi.waitFor(() => expect(delivered).toEqual(['real']))
+    expect(notDelivered).toEqual(['stand-in-0', 'stand-in-1', 'stand-in-2', 'stand-in-3'])
   })
 
   it('claims a whole backlog at once, before any of it runs', async () => {
@@ -492,6 +553,55 @@ describe('restarting', () => {
     expect(runner.started).toEqual([])
     await vi.waitFor(() => expect(journal.entries.size).toBe(0))
   })
+
+  it('leaves open whether the model has a result when Sim already held one for the call', async () => {
+    const { sim, journal, executor, delivered, notDelivered } = setup()
+    // The previous run's send of the real result may have landed just before the app went down.
+    sim.completionOutcome = 'duplicate'
+    await journal.put({ toolCallId: 'started-1', state: 'started', executionToken: 't-started' })
+
+    await executor.recover()
+
+    await vi.waitFor(() => expect(journal.entries.size).toBe(0))
+    expect(sim.completions).toHaveLength(1)
+    expect(delivered).toEqual([])
+    expect(notDelivered).toEqual([])
+  })
+
+  it('still leaves a recovered result open when Sim held one, after parking it', async () => {
+    const { sim, journal, executor, onUnregistered, delivered, notDelivered } = setup()
+    sim.completeErrors = [new DeviceRequestError(401, 'unregistered')]
+    sim.completionOutcome = 'duplicate'
+    await journal.put({ toolCallId: 'started-1', state: 'started', executionToken: 't-started' })
+    await executor.recover()
+    await vi.waitFor(() => expect(onUnregistered).toHaveBeenCalledTimes(1))
+
+    // Registered again.
+    executor.resumeParked()
+
+    await vi.waitFor(() => expect(sim.completions).toHaveLength(1))
+    await vi.waitFor(() => expect(journal.entries.size).toBe(0))
+    expect(delivered).toEqual([])
+    expect(notDelivered).toEqual([])
+  })
+
+  it('reports a recovered result too large to send, then held as a duplicate, as never reaching the model', async () => {
+    const { sim, journal, executor, delivered, notDelivered } = setup()
+    // No run could have sent this result: it is too large every time.
+    sim.completeErrors = [new DeviceRequestError(413, 'too large')]
+    sim.completionOutcome = 'duplicate'
+    await journal.put({
+      toolCallId: 'result-1',
+      state: 'result',
+      executionToken: 't',
+      completion: DONE,
+    })
+
+    await executor.recover()
+
+    await vi.waitFor(() => expect(notDelivered).toEqual(['result-1']))
+    expect(delivered).toEqual([])
+  })
 })
 
 describe('delivery', () => {

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

@@ -87,6 +87,12 @@ export interface DesktopExecutorOptions {
    * pane still running) is in use.
    */
   onResultDelivered?: (toolCallId: string) => void
+  /**
+   * Called once Sim is done with a call without taking its real result: it settled the call
+   * first, refused the result, or recorded one that stands in for it. The model never learns of
+   * anything the action handed back as still going, so nothing will come back to it.
+   */
+  onResultNotDelivered?: (toolCallId: string) => void
   maxHeldCalls?: number
   /** First delivery retry delay; tests shorten it. */
   retryBaseMs?: number
@@ -123,6 +129,11 @@ export class DesktopExecutor {
   private recovering: Promise<void> | null = null
   /** Results a previous app run left, still on their way to Sim. */
   private readonly recoveringIds = new Set<string>()
+  /**
+   * Results a previous app run left, until Sim answers for them, parked ones included: that run's
+   * send of the real result may already have landed.
+   */
+  private readonly recoveredResults = new Set<string>()
   /** Results that have failed to reach Sim for longer than {@link DELIVERY_AWAKE_LIMIT_MS}. */
   private readonly stalledDeliveries = new Set<string>()
   /** Results Sim refused because it no longer recognized this device; sent once it registers again. */
@@ -184,6 +195,7 @@ export class DesktopExecutor {
       })
       // Held awake like a running call: the result exists only on this machine until Sim has it.
       this.recoveringIds.add(entry.toolCallId)
+      this.recoveredResults.add(entry.toolCallId)
       this.updateBusy()
       void this.deliver(entry.toolCallId, entry.executionToken, completion).finally(() => {
         this.recoveringIds.delete(entry.toolCallId)
@@ -413,6 +425,7 @@ export class DesktopExecutor {
     sendingSince: number
   ): Promise<void> {
     let pending = completion
+    let notDelivered = true
     for (let attempt = 1; !this.disposed; attempt++) {
       try {
         const outcome = await this.options.client.complete({
@@ -422,9 +435,13 @@ export class DesktopExecutor {
         })
         logger.info('Desktop call result acknowledged', { toolCallId, outcome })
         // Superseded: Sim settled the call first, so this result never reached the model.
-        if (outcome !== 'superseded' && isDeliveredResult(pending)) {
-          this.options.onResultDelivered?.(toolCallId)
-        }
+        const delivered = outcome !== 'superseded' && isDeliveredResult(pending)
+        if (delivered) this.options.onResultDelivered?.(toolCallId)
+        // A duplicate holds what this app run sent, unless the result was recovered unchanged from
+        // an earlier run, whose send of the real result may have landed before the restart.
+        const takenEarlier =
+          outcome === 'duplicate' && pending === completion && this.recoveredResults.has(toolCallId)
+        notDelivered = !delivered && !takenEarlier
         break
       } catch (error) {
         // Encoding failed on this machine, so nothing was sent; the same data would fail again.
@@ -481,7 +498,10 @@ export class DesktopExecutor {
         )
       }
     }
-    if (!this.disposed) await this.forget(toolCallId)
+    this.recoveredResults.delete(toolCallId)
+    if (this.disposed) return
+    if (notDelivered) this.options.onResultNotDelivered?.(toolCallId)
+    await this.forget(toolCallId)
   }
 
   private async renew(entry: HeldCall): Promise<void> {

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

@@ -64,6 +64,8 @@ export interface DesktopExecutorServiceDeps {
   onBusyChange?: (busy: boolean) => void
   /** Sim has taken a call's real result as the call's own, so the model has it. */
   onResultDelivered?: (toolCallId: string) => void
+  /** Sim is done with a call whose real result the model never got. */
+  onResultNotDelivered?: (toolCallId: string) => void
 }
 
 export interface DesktopExecutorService {
@@ -275,6 +277,7 @@ export function createDesktopExecutorService(
         ...(deps.onApprovals ? { onApprovals: deps.onApprovals } : {}),
         ...(deps.onBusyChange ? { onBusyChange: deps.onBusyChange } : {}),
         ...(deps.onResultDelivered ? { onResultDelivered: deps.onResultDelivered } : {}),
+        ...(deps.onResultNotDelivered ? { onResultNotDelivered: deps.onResultNotDelivered } : {}),
       })
       // Recovery rewrites the journal, so what it held before is read first.
       await pendingResultsSnapshot()

apps/desktop/src/main/index.ts

@@ -610,6 +610,9 @@ function main(): void {
     // A result the model has makes a tmux run it handed back as still going collectable across a
     // restart.
     onResultDelivered: (toolCallId) => terminal.markRunDelivered(toolCallId),
+    // A result the model never got leaves such a run with no one to come back to it, so it is
+    // stopped, as the launch sweep stops a previous process's.
+    onResultNotDelivered: (toolCallId) => void terminal.stopUndeliveredRun(toolCallId),
     runner: createDesktopToolRunner({
       preferences: () => desktopSettings.getPreferences(),
       accountDataAvailable,

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

@@ -938,6 +938,13 @@ export class TerminalService {
     await Promise.allSettled(stops)
   }
 
+  /** Stops the command a plain shell's `run` handed back as still going, if it still is. */
+  async stopAgentCommand(toolCallId: string): Promise<void> {
+    await Promise.allSettled(
+      [...this.sessions.values()].map((session) => this.stopCommand(session, toolCallId))
+    )
+  }
+
   /** Waits for the command a run started to end, up to `ms`. */
   private async commandEnds(
     session: TerminalSession,

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

@@ -119,6 +119,32 @@ describe('stopping recorded tmux runs', () => {
     expect(ledger.list()).toEqual([])
   })
 
+  it("stops only the run of a call whose result never reached the model, this process's or a previous one's", async () => {
+    const dir = ledgerDir()
+    createRunLedger(dir).record(run('previous', '%1'))
+    const ledger = createRunLedger(dir)
+    ledger.record(run('current', '%2'))
+    ledger.record(run('other', '%3'))
+    // Recorded as handed back: the model has its pane.
+    ledger.record(run('handed-back', '%4', 'delivered'))
+    const registry = new TerminalRegistry(undefined, undefined, ledger)
+
+    await registry.stopUndeliveredRun('call-current')
+    await registry.stopUndeliveredRun('call-previous')
+    await registry.stopUndeliveredRun('call-handed-back')
+
+    expect(panes.get('current')).toBe('stopped')
+    expect(panes.get('previous')).toBe('stopped')
+    expect(panes.get('other')).toBeUndefined()
+    expect(panes.get('handed-back')).toBeUndefined()
+    expect(
+      ledger
+        .list()
+        .map((record) => record.runId)
+        .sort()
+    ).toEqual(['handed-back', 'other'])
+  })
+
   it('notes the run of a call whose result reached the model as handed back', () => {
     const dir = ledgerDir()
     const ledger = createRunLedger(dir)

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

@@ -410,23 +410,39 @@ export class TerminalRegistry {
     if (runId) this.runLedger?.advance(runId, 'delivered')
   }
 
+  /**
+   * Stops what a call's `run` handed back as still going once Sim is done with the call without
+   * its result reaching the model: a plain shell's command, or a tmux run, this process's or a
+   * previous one's. A run recorded as handed back is the model's, and is left going.
+   */
+  async stopUndeliveredRun(callId: string): Promise<void> {
+    await Promise.allSettled(
+      [...this.entries.values()].map((entry) => entry.service.stopAgentCommand(callId))
+    )
+    await this.stopRecordedRuns({ callId, keep: (run) => run.state === 'delivered' })
+  }
+
   /**
    * Stops the recorded tmux runs, each only while its pane still carries its tag, and drops the
    * records with nothing left to stop. `excludeLive` skips the runs this process has started;
-   * `keep` names runs to leave going, such as a previous process's runs whose results the model
-   * already has and may come back to.
+   * `callId` limits it to the run that call started; `keep` names runs to leave going, such as a
+   * previous process's runs whose results the model already has and may come back to.
    */
   async stopRecordedRuns(
     options: {
       excludeLive?: boolean
+      callId?: string
       /** Runs to leave going; their records are only dropped once their panes are gone. */
       keep?: (run: RunRecord) => boolean
     } = {}
   ): Promise<void> {
     const ledger = this.runLedger
     if (!ledger) return
+    const runs = ledger
+      .list({ excludeLive: options.excludeLive })
+      .filter((run) => options.callId === undefined || run.callId === options.callId)
     await Promise.allSettled(
-      ledger.list({ excludeLive: options.excludeLive }).map(async (run) => {
+      runs.map(async (run) => {
         let keeping = options.keep?.(run) ?? false
         let state = keeping ? await recordedRunState(run, process.env) : 'ours'
         // A kept run whose command already ended leaves only its dead pane (`remain-on-exit`).