fix(desktop): poll desktop activity only for users with a desktop, keep XDG zsh configs integrated (#8785)

* fix(desktop): poll desktop activity only for users with a desktop, keep XDG zsh configs integrated

* fix(desktop): give user zsh files their own ZDOTDIR, drop stale activity, require a live session

* test(desktop): check delivered events and real requests, not mock calls

Author: waleedlatif1

apps/desktop/e2e/background-executor.spec.ts

@@ -384,7 +384,10 @@ test.describe('background executor', () => {
     const deviceId = await registeredDevice(sim)
     const pulls = () =>
       sim.requests.filter((request) => request.startsWith('GET /api/desktop/inbox')).length
+    const claimAttempts = () =>
+      sim.requests.filter((request) => request === 'POST /api/desktop/tool/claim').length
     const pullsBefore = pulls()
+    const claimAttemptsBefore = claimAttempts()
 
     const gated = sim.issue(
       deviceId,
@@ -399,7 +402,8 @@ test.describe('background executor', () => {
 
     await check('F: the declined call is never claimed, and its command never runs', async () => {
       await sleep(RECONCILE_MS * 2)
-      expect(sim.requireCall(gated).claims).toBe(0)
+      // Attempts, not successful claims: the fixture refuses a declined call's claim itself.
+      expect(claimAttempts() - claimAttemptsBefore).toBe(0)
       expect(sim.requireCall(gated).completions).toEqual([])
       expect(readFileSafe(marker)).toBe('')
     })

apps/desktop/e2e/desktop-tools-live-sim.spec.ts

@@ -48,6 +48,7 @@ const ARRIVAL_MS = 60_000
 /** First requests to a route compile it, which takes minutes on a cold dev app. */
 const COMPILE_MS = 300_000
 const REGISTRATION_PATH = '/api/desktop/devices'
+const DESKTOP_COMPLETION_PATH = '/api/desktop/tool/complete'
 /** Sim's answer to registration on an install that cannot run the background executor. */
 const executorUnavailable = (answer: Record<string, unknown>) => {
   answer.enabled = false
@@ -120,6 +121,7 @@ test.describe('desktop tools against a live Sim', () => {
     await app?.close().catch(() => {})
     app = undefined
     proxy.rewriteAnswer(REGISTRATION_PATH, executorUnavailable)
+    proxy.rewriteAnswer(DESKTOP_COMPLETION_PATH, undefined)
     rmSync(scratch, { recursive: true, force: true })
   })
 
@@ -377,7 +379,7 @@ test.describe('desktop tools against a live Sim', () => {
     entry.method === 'POST' && entry.path === '/api/mothership/chat'
   /** The background executor's report of a call's result. */
   const isDesktopCompletion = (method: string, path: string) =>
-    method === 'POST' && path === '/api/desktop/tool/complete'
+    method === 'POST' && path === DESKTOP_COMPLETION_PATH
   /** A client tool's report of its own result. */
   const isToolReport = (method: string, path: string) =>
     method === 'POST' && path === '/api/copilot/confirm'
@@ -830,14 +832,16 @@ test.describe('desktop tools against a live Sim', () => {
     // Reaches Sim on release although the cut made the app give up on it, as a report already on
     // the wire would: the app cannot know it landed, so it reports again once back online.
     const completion = proxy.hold(isDesktopCompletion, { deliverIfAbandoned: true })
+    const answers = proxy.recordAnswers(DESKTOP_COMPLETION_PATH)
     await send(page, '[network-cut] read my notes')
     await completion.arrival(ARRIVAL_MS, 'The result report')
 
     proxy.cutNetwork()
     await expect.poll(() => completion.isAbandoned, { timeout: 15_000 }).toBe(true)
     completion.release()
     await expect.poll(() => callState(chatId), { timeout: 30_000 }).toMatch(/^completed/)
-    await sleep(5_000)
+    // The late report alone resumed the agent while the app was still offline.
+    await agent.waitForResume(() => Boolean(agent.resultFor(callId)), 60_000)
     const restoredAt = Date.now()
     proxy.restoreNetwork()
 
@@ -852,8 +856,10 @@ test.describe('desktop tools against a live Sim', () => {
     await expect.poll(() => retried().length, { timeout: 60_000 }).toBeGreaterThan(0)
     for (const entry of retried()) expect(entry.status).toBeLessThan(300)
 
-    await agent.waitForResume(() => Boolean(agent.resultFor(callId)), 60_000)
     await proxy.settled(30_000)
+    // Sim had already recorded the late report, so every retry the app heard back from is a no-op.
+    expect(answers.length).toBeGreaterThan(0)
+    expect(answers).toEqual(answers.map(() => expect.objectContaining({ outcome: 'duplicate' })))
     const delivered = agent.resumes.filter((resume) =>
       resume.results.some((entry) => entry.callId === callId)
     )

apps/desktop/e2e/fixtures/live-sim.ts

@@ -259,6 +259,18 @@ export class SimProxy {
     else this.answerRewrites.delete(path)
   }
 
+  /**
+   * Sim's JSON answers to `path` that reach the app from now on, recorded as they pass and left
+   * unchanged. An answer to a request whose client gave up reaches no one, so it is not recorded.
+   */
+  recordAnswers(path: string): Record<string, unknown>[] {
+    const answers: Record<string, unknown>[] = []
+    this.rewriteAnswer(path, (body) => {
+      answers.push(structuredClone(body))
+    })
+    return answers
+  }
+
   /**
    * Cuts the network between the app and Sim: every open connection drops mid-flight, the
    * doorbell stream included, and new ones are reset until `restoreNetwork`.

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

@@ -1420,7 +1420,7 @@ export class TerminalService {
         const screen = (await session.readScrollback(STARTUP_SCREEN_LINES)).output.trim()
         throw new TerminalError(
           'NO_SHELL_INTEGRATION',
-          `The shell has been running its startup files for over ${SHELL_STARTUP_BOUNDS.startingMs / 1000} s without reaching a prompt, so nothing was run. Its screen:\n${screen || '(empty)'}\nIf it is waiting for an answer, ask the user to answer it in that terminal (terminalId ${session.terminalId}), then run the command again.`
+          `The shell began its startup files over ${SHELL_STARTUP_BOUNDS.startingMs / 1000} s ago but never reached a prompt Sim can track, so nothing was run. Its screen:\n${screen || '(empty)'}\nIf a startup file is waiting for an answer, ask the user to answer it in that terminal (terminalId ${session.terminalId}), then run the command again. If the screen shows a prompt, a startup file replaced the shell (such as exec tmux or exec fish), so ask the user to run the command themselves.`
         )
       }
       case 'not-instrumented':

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

@@ -328,7 +328,6 @@ export class TerminalSession {
   private columns: number
   private lines: number
   private shellIntegration = false
-  /** The shell has begun the startup files we generated, so its integration is on the way. */
   /** When the shell began the startup files we generated, or null while it has not. */
   private startupBegunAt: number | null = null
   private readonly spawnedAt = Date.now()

apps/desktop/src/main/terminal/shell-integration.test.ts

@@ -1,5 +1,5 @@
 import { spawnSync } from 'node:child_process'
-import { existsSync, mkdtempSync, writeFileSync } from 'node:fs'
+import { existsSync, mkdirSync, mkdtempSync, writeFileSync } from 'node:fs'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { describe, expect, it } from 'vitest'
@@ -140,6 +140,37 @@ describe('startup marker', () => {
     }
   )
 
+  it.skipIf(!existsSync('/bin/zsh'))(
+    "zsh keeps integrating when the user's .zshenv moves ZDOTDIR, and runs all their files there",
+    () => {
+      const home = mkdtempSync(join(tmpdir(), 'sim-shell-home-'))
+      const userDir = join(home, '.config', 'zsh')
+      mkdirSync(userDir, { recursive: true })
+      writeFileSync(join(home, '.zshenv'), 'export ZDOTDIR="$HOME/.config/zsh"\n')
+      writeFileSync(join(userDir, '.zprofile'), 'USER_PROFILE_RAN=1\n')
+      writeFileSync(join(userDir, 'plugins.zsh'), 'USER_PLUGIN_RAN=1\n')
+      writeFileSync(join(userDir, '.zshrc'), 'USER_RC_RAN=1\nsource "$ZDOTDIR/plugins.zsh"\n')
+      writeFileSync(join(userDir, '.zlogin'), 'echo "login=$USER_RC_RAN"\n')
+      const env = { PATH: process.env.PATH ?? '/usr/bin:/bin', HOME: home }
+      const launch = buildShellLaunch('zsh', mkdtempSync(join(tmpdir(), 'sim-zsh-')), NONCE, env)
+
+      const output = spawnSync(
+        '/bin/zsh',
+        [
+          ...launch.args,
+          '-i',
+          '-c',
+          'echo "profile=$USER_PROFILE_RAN rc=$USER_RC_RAN plugin=$USER_PLUGIN_RAN zdotdir=$ZDOTDIR"; whence -w __sim_precmd',
+        ],
+        { env: { ...env, ...launch.env }, encoding: 'utf8' }
+      ).stdout
+
+      expect(output).toContain(`profile=1 rc=1 plugin=1 zdotdir=${userDir}`)
+      expect(output).toContain('__sim_precmd: function')
+      expect(output).toContain('login=1')
+    }
+  )
+
   it.skipIf(!existsSync('/bin/bash'))("bash sends it before the user's files", () => {
     const home = userHome('.bashrc')
     const env = { PATH: process.env.PATH ?? '/usr/bin:/bin', HOME: home }

apps/desktop/src/main/terminal/shell-integration.ts

@@ -166,8 +166,17 @@ function findTerminator(buffer: string, from: number): { index: number; length:
  * prompt, aliases, and PATH win over ours.
  */
 function writeZshFiles(dir: string, nonce: string, originalZdotdir: string): void {
-  const sourceOriginal = (file: string) =>
-    `[ -f "$SIM_ZDOTDIR_ORIG/${file}" ] && builtin source "$SIM_ZDOTDIR_ORIG/${file}"`
+  // The user's file runs with ZDOTDIR at their own directory, so it finds its siblings and
+  // plugins there. A file that moves ZDOTDIR (an XDG `.zshenv`) moves where the rest of theirs are
+  // read from; ZDOTDIR then points back here, so zsh reads our next file.
+  const sourceOriginal = (file: string) => `if [ -f "$SIM_ZDOTDIR_ORIG/${file}" ]; then
+  __sim_zdotdir="$ZDOTDIR"
+  ZDOTDIR="$SIM_ZDOTDIR_ORIG"
+  builtin source "$SIM_ZDOTDIR_ORIG/${file}"
+  SIM_ZDOTDIR_ORIG="\${ZDOTDIR:-$HOME}"
+  ZDOTDIR="$__sim_zdotdir"
+  builtin unset __sim_zdotdir
+fi`
 
   // `.zshenv` is the first file zsh reads, so the startup marker goes out before any of the user's
   // files run. Only from an interactive shell: a script's output must not carry it.

apps/desktop/src/main/terminal/shell-startup.test.ts

@@ -9,6 +9,7 @@ import { TerminalService } from '@/main/terminal'
  */
 const pty = vi.hoisted(() => ({
   emit: null as ((data: string) => void) | null,
+  exit: null as (() => void) | null,
   writes: [] as string[],
 }))
 
@@ -18,7 +19,9 @@ vi.mock('@lydell/node-pty', () => ({
     onData: (handler: (data: string) => void) => {
       pty.emit = handler
     },
-    onExit: () => {},
+    onExit: (handler: () => void) => {
+      pty.exit = handler
+    },
     write: (data: string) => pty.writes.push(data),
     resize: vi.fn(),
     kill: vi.fn(),
@@ -71,6 +74,7 @@ beforeEach(() => {
 afterEach(() => {
   vi.useRealTimers()
   pty.emit = null
+  pty.exit = null
   pty.writes.length = 0
   vi.unstubAllEnvs()
 })
@@ -122,6 +126,20 @@ describe('a shell that is still starting', () => {
     expect(response).toMatchObject({ ok: false, code: 'NO_SHELL_INTEGRATION' })
     expect(response.error).toContain('[oh-my-zsh] Would you like to update? [Y/n]')
     expect(response.error).toContain('ask the user to answer it')
+    expect(response.error).toContain('a startup file replaced the shell')
+    expect(pty.writes).toEqual([])
+    terminal.dispose()
+  })
+
+  it('reports the session closed when the shell exits while its startup files run', async () => {
+    const terminal = new TerminalService({ loadCwd: () => '/tmp' })
+    const running = terminal.executeTool('call-exit', 'run', { command: 'echo hi' })
+    shell(STARTUP)
+    await vi.advanceTimersByTimeAsync(2_000)
+    pty.exit?.()
+    const response = await running
+
+    expect(response).toMatchObject({ ok: false, code: 'SESSION_CLOSED' })
     expect(pty.writes).toEqual([])
     terminal.dispose()
   })

apps/sim/app/api/mothership/events/route.test.ts

@@ -220,4 +220,25 @@ describe('Mothership owner-scoped event stream', () => {
     expect(authorize).not.toHaveBeenCalled()
     expect(authMockFns.mockGetSession).toHaveBeenCalledTimes(1)
   })
+
+  it('tells each member whether a workspace chat is their own, never whose it is', async () => {
+    const abort = new AbortController()
+    const response = await GET(request('workspaceId=ws-1', abort.signal))
+    if (!response.body) throw new Error('The event stream has no body')
+    const chunks: string[] = []
+    const collected = collect(response.body, chunks)
+    await vi.advanceTimersByTimeAsync(0)
+    emit({ workspaceId: 'ws-1', userId: 'user-1', chatId: 'own-chat', type: 'started' })
+    emit({ workspaceId: 'ws-1', userId: 'teammate-1', chatId: 'teammate-chat', type: 'started' })
+    emit({ workspaceId: 'ws-1', chatId: 'unknown-owner-chat', type: 'started' })
+    abort.abort()
+    await collected
+    const payloads = chunks.map((chunk) => JSON.parse(chunk.split('data: ')[1]))
+    expect(payloads).toEqual([
+      expect.objectContaining({ chatId: 'own-chat', ownChat: true }),
+      expect.objectContaining({ chatId: 'teammate-chat', ownChat: false }),
+      expect.not.objectContaining({ ownChat: expect.anything() }),
+    ])
+    expect(chunks.join('')).not.toMatch(/userId|teammate-1|user-1/)
+  })
 })

apps/sim/app/api/mothership/events/route.ts

@@ -30,14 +30,16 @@ const mothershipEventsHandler = createWorkspaceSSE({
   label: 'mothership-events',
   subscriptions: [
     {
-      subscribe: (workspaceId, send) => {
+      subscribe: (workspaceId, send, viewerUserId) => {
         if (!chatPubSub) return () => {}
         return chatPubSub.onStatusChanged((event) => {
           if (event.workspaceId !== workspaceId) return
           send('task_status', {
             chatId: event.chatId,
             type: event.type,
             ...(event.streamId ? { streamId: event.streamId } : {}),
+            // Whether the chat is the viewer's own, never whose it is; absent when unknown.
+            ...(event.userId ? { ownChat: event.userId === viewerUserId } : {}),
             timestamp: Date.now(),
           })
         })