fix(desktop): make the executor's journal, sign-out and terminal stops hold under races

- A journal transition that cannot be made durable now fails: an unrecorded
  claim is not made, and an action whose start could not be recorded is not
  run and is reported as not run. The journal keeps result data within what a
  restart can read back.
- Sign-out fences registrations in flight, and a rotated install id never
  reuses the previous id's client.
- A stopped call's result carries no content, only the acknowledgement.
- Terminal stops: a Stop before the command starts runs nothing; escalation
  signals only the stopped command's own process group, rechecked before each
  signal; a stopped tmux run ends its wait at once.
- A chat's next terminal operation waits for one that outlived its deadline.
- The doorbell bounds its handshake and reconnects at once on wake.
- Approval notifications name neither the chat nor the command.
- A user-local glob that stopped early says so.

Author: waleedlatif1

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

@@ -61,7 +61,7 @@ function harness(options: { enabled?: boolean; focusedChatId?: string | null } =
 }
 
 describe('approval notifications', () => {
-  it('notifies once per waiting call and opens its chat, never naming the command', () => {
+  it('notifies once per waiting call and opens its chat, naming neither chat nor command', () => {
     const { notifier, notifications, openRoute } = harness()
 
     notifier.update([approval('call-1')])
@@ -70,6 +70,7 @@ describe('approval notifications', () => {
     expect(notifications).toHaveLength(1)
     expect(notifications[0]?.show).toHaveBeenCalledOnce()
     expect(notifications[0]?.options.body).not.toContain('rm -rf')
+    expect(notifications[0]?.options.body).not.toContain('Fix CI')
     notifications[0]?.click()
     expect(openRoute).toHaveBeenCalledWith('/workspace/ws-1/chat/chat-b')
   })

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

@@ -1,8 +1,8 @@
 /**
  * Tells the user when a chat they are not looking at waits for their approval. The notification
  * only opens the chat at its approval card; approving happens there, where the user sees what
- * they are approving. It closes itself once the call is decided, and never names the command,
- * since a notification can show on a locked screen.
+ * they are approving. It closes itself once the call is decided, and names neither the chat nor
+ * the command, since a notification can show on a locked screen.
  */
 import type { DesktopApprovalItem } from '@/main/desktop-executor/executor'
 
@@ -25,12 +25,6 @@ export interface ApprovalNotifierDeps {
   }) => ApprovalNotification | null
 }
 
-function approvalBody(item: DesktopApprovalItem): string {
-  return item.chatTitle
-    ? `“${item.chatTitle}” is waiting for your approval.`
-    : 'A chat is waiting for your approval.'
-}
-
 export function createApprovalNotifier(deps: ApprovalNotifierDeps) {
   /** Calls already brought to the user's attention, by notification or by being on screen. */
   const shown = new Map<string, ApprovalNotification | null>()
@@ -54,7 +48,7 @@ export function createApprovalNotifier(deps: ApprovalNotifierDeps) {
         }
         const notification = deps.createNotification({
           title: 'Approval needed',
-          body: approvalBody(item),
+          body: 'A chat is waiting for your approval.',
           silent: !preferences.notificationSounds,
         })
         if (!notification) return

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

@@ -6,25 +6,25 @@ const encoder = new TextEncoder()
 
 /** A stream the test writes SSE text into, closing when told to or when its reader aborts. */
 function controllableStream(signal: AbortSignal) {
-  let controller!: ReadableStreamDefaultController<Uint8Array>
+  let controller: ReadableStreamDefaultController<Uint8Array> | undefined
   const stream = new ReadableStream<Uint8Array>({
     start(c) {
       controller = c
     },
   })
   signal.addEventListener('abort', () => {
     try {
-      controller.error(new Error('aborted'))
+      controller?.error(new Error('aborted'))
     } catch {}
   })
   return {
     stream,
-    write: (text: string) => controller.enqueue(encoder.encode(text)),
-    end: () => controller.close(),
+    write: (text: string) => controller?.enqueue(encoder.encode(text)),
+    end: () => controller?.close(),
   }
 }
 
-function harness(options: { staleAfterMs?: number } = {}) {
+function harness(options: { staleAfterMs?: number; retryBaseMs?: number } = {}) {
   const connections: ReturnType<typeof controllableStream>[] = []
   const failures: DeviceRequestError[] = []
   const onRing = vi.fn()
@@ -106,4 +106,28 @@ describe('InboxDoorbell', () => {
     await vi.waitFor(() => expect(onUnregistered).toHaveBeenCalled())
     doorbell.stop()
   })
+
+  it('reconnects at once on wake, without waiting out a backoff', async () => {
+    const { doorbell, connections, onRing } = harness({ retryBaseMs: 60_000 })
+    doorbell.start()
+    await vi.waitFor(() => expect(connections).toHaveLength(1))
+
+    doorbell.wake()
+
+    await vi.waitFor(() => expect(connections).toHaveLength(2))
+    await vi.waitFor(() => expect(onRing).toHaveBeenCalledTimes(2))
+    doorbell.stop()
+  })
+
+  it('starts on wake after sleep stopped it', async () => {
+    const { doorbell, connections } = harness({ retryBaseMs: 60_000 })
+    doorbell.start()
+    await vi.waitFor(() => expect(connections).toHaveLength(1))
+    doorbell.stop()
+
+    doorbell.wake()
+
+    await vi.waitFor(() => expect(connections).toHaveLength(2))
+    doorbell.stop()
+  })
 })

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

@@ -16,6 +16,7 @@ const logger = createLogger('DesktopExecutorDoorbell')
 /** Sim heartbeats every 30 s; two missed ones mean the connection is gone. */
 const STALE_STREAM_MS = 75_000
 const RECONNECT_MAX_MS = 30_000
+const HANDSHAKE_TIMEOUT_MS = 15_000
 
 interface DoorbellOptions {
   client: Pick<DesktopExecutorClient, 'openInboxStream'>
@@ -54,6 +55,10 @@ export function parseServerSentEvents(buffer: string): {
 
 export class InboxDoorbell {
   private running = false
+  /** Which loop is current; a loop left over from before a stop ends at its next turn. */
+  private loopGeneration = 0
+  /** The connection was dropped on purpose; open the next one without backing off. */
+  private reconnectNow = false
   private connection: AbortController | null = null
   private sleeper: AbortController | null = null
 
@@ -62,30 +67,46 @@ export class InboxDoorbell {
   start(): void {
     if (this.running) return
     this.running = true
-    void this.loop()
+    this.loopGeneration += 1
+    void this.loop(this.loopGeneration)
   }
 
   stop(): void {
     this.running = false
+    this.loopGeneration += 1
     this.connection?.abort()
     this.sleeper?.abort()
   }
 
-  /** Drops the current connection and reconnects now, as after waking or coming back online. */
-  reconnect(): void {
+  /**
+   * After waking or coming back online: a stopped doorbell starts, and a running one drops its
+   * connection, which may be dead without knowing it, and reconnects at once.
+   */
+  wake(): void {
+    if (!this.running) {
+      this.start()
+      return
+    }
+    this.reconnectNow = true
     this.connection?.abort()
     this.sleeper?.abort()
   }
 
-  private async loop(): Promise<void> {
+  private async loop(generation: number): Promise<void> {
     let attempt = 0
-    while (this.running) {
+    const current = () => this.loopGeneration === generation
+    while (current()) {
       const rotated = await this.connectOnce().then(
         (result) => {
           attempt = 0
           return result === 'rotated'
         },
         (error: unknown) => {
+          if (this.reconnectNow) {
+            this.reconnectNow = false
+            attempt = 0
+            return true
+          }
           attempt += 1
           if (error instanceof DeviceRequestError && error.unregistered) {
             this.options.onUnregistered()
@@ -97,7 +118,7 @@ export class InboxDoorbell {
           return false
         }
       )
-      if (!this.running || rotated) continue
+      if (!current() || rotated) continue
       this.sleeper = new AbortController()
       await interruptibleSleep(
         attempt === 0
@@ -123,7 +144,11 @@ export class InboxDoorbell {
       staleTimer = setTimeout(() => connection.abort(), staleAfterMs)
     }
     try {
-      const stream = await this.options.client.openInboxStream(connection.signal)
+      // A stream whose response never starts is as dead as one that went quiet.
+      const handshake = setTimeout(() => connection.abort(), HANDSHAKE_TIMEOUT_MS)
+      const stream = await this.options.client
+        .openInboxStream(connection.signal)
+        .finally(() => clearTimeout(handshake))
       touch()
       this.options.onRing()
       const reader = stream.getReader()

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

@@ -95,10 +95,13 @@ class FakeSim {
 class MemoryJournal implements ExecutorJournal {
   readonly entries = new Map<string, JournalEntry>()
   readonly history: JournalEntry[] = []
+  /** A transition the disk refuses, as a full disk or a failed encryption would. */
+  failOn: JournalEntry['state'] | null = null
   async load() {
     return [...this.entries.values()]
   }
   async put(entry: JournalEntry) {
+    if (entry.state === this.failOn) throw new Error('disk full')
     this.entries.set(entry.toolCallId, entry)
     this.history.push(entry)
   }
@@ -117,16 +120,19 @@ class FakeRunner implements DesktopToolRunner {
   /** When set, every call finishes at once with this completion. */
   immediate: DesktopToolCompletion | null = null
   onStart: ((call: ClaimedDesktopCall) => void) | null = null
+  /** An action that finishes with its full result even after being told to stop. */
+  ignoresAbort = false
 
   async run(call: ClaimedDesktopCall, signal: AbortSignal) {
     this.started.push(call.toolCallId)
     this.onStart?.(call)
     if (this.immediate) return this.immediate
     const result = deferred<DesktopToolCompletion>()
     this.pending.set(call.toolCallId, result)
-    signal.addEventListener('abort', () =>
-      result.resolve({ status: 'error', message: 'This browser action was cancelled.' })
-    )
+    signal.addEventListener('abort', () => {
+      if (!this.ignoresAbort)
+        result.resolve({ status: 'error', message: 'This browser action was cancelled.' })
+    })
     return result.promise
   }
 
@@ -313,7 +319,7 @@ describe('reporting', () => {
     sim.inbox = [callItem('call-1', 'chat-a')]
 
     await executor.reconcile()
-    await vi.waitFor(() => expect(journal.entries.get('call-1')?.state).toBe('result'))
+    await vi.waitFor(() => expect(journal.history.map((entry) => entry.state)).toContain('result'))
 
     await vi.waitFor(() => expect(sim.completions).toHaveLength(1))
     expect(runner.started).toEqual(['call-1'])
@@ -367,6 +373,23 @@ describe('stopping', () => {
     expect(runner.started).toEqual(['a-1'])
   })
 
+  it('sends nothing the stopped action produced, only the acknowledgement', async () => {
+    const { sim, runner, executor } = setup()
+    runner.ignoresAbort = true
+    sim.inbox = [callItem('call-1', 'chat-a', 'read_local_file')]
+    await executor.reconcile()
+    await vi.waitFor(() => expect(runner.started).toEqual(['call-1']))
+
+    sim.inbox = [{ kind: 'cancel', toolCallId: 'call-1' }]
+    await executor.reconcile()
+    await vi.waitFor(() => expect(runner.cancelled).toEqual(['call-1']))
+    runner.finish('call-1', { status: 'success', message: 'read', data: { text: 'secret' } })
+
+    await vi.waitFor(() => expect(sim.completions).toHaveLength(1))
+    expect(sim.completions[0]?.completion.status).toBe('cancelled')
+    expect(JSON.stringify(sim.completions[0])).not.toContain('secret')
+  })
+
   it('ignores a cancel for a call it never held', async () => {
     const { sim, runner, executor } = setup()
     sim.inbox = [{ kind: 'cancel', toolCallId: 'someone-else' }]
@@ -431,4 +454,51 @@ describe('registration', () => {
     expect(journal.entries.size).toBe(0)
     expect(executor.heldCallCount()).toBe(0)
   })
+
+  it('does not keep a claim that Sim answers after sign-out', async () => {
+    const { sim, journal, runner, executor } = setup({ leaseRenewMs: 10 })
+    const answer = deferred<void>()
+    const claim = sim.client.claim
+    sim.client.claim = async (toolCallId) => {
+      await answer.promise
+      return claim(toolCallId)
+    }
+    sim.inbox = [callItem('call-1', 'chat-a')]
+    const reading = executor.reconcile()
+    await vi.waitFor(() => expect(journal.entries.get('call-1')?.state).toBe('claiming'))
+
+    const signingOut = executor.dispose()
+    answer.resolve()
+    await Promise.all([reading, signingOut])
+    await sleep(40)
+
+    expect(executor.heldCallCount()).toBe(0)
+    expect(runner.started).toEqual([])
+    expect(sim.renewals).toEqual([])
+    expect(journal.entries.size).toBe(0)
+  })
+})
+
+describe('recording', () => {
+  it('leaves a call on offer when it cannot record the claim', async () => {
+    const { sim, journal, executor } = setup()
+    journal.failOn = 'claiming'
+    sim.inbox = [callItem('call-1', 'chat-a')]
+
+    await executor.reconcile()
+
+    expect(sim.claims).toEqual([])
+  })
+
+  it('never starts an action it could not record, and reports it as not run', async () => {
+    const { sim, journal, runner, executor } = setup()
+    journal.failOn = 'started'
+    sim.inbox = [callItem('call-1', 'chat-a')]
+
+    await executor.reconcile()
+
+    await vi.waitFor(() => expect(sim.completions).toHaveLength(1))
+    expect(sim.completions[0]?.completion.data).toMatchObject({ notStarted: true })
+    expect(runner.started).toEqual([])
+  })
 })