fix(files): keep watcher streams across SSH reconnect races (#24722)

* fix(files): preserve watcher streams across SSH reconnect races

* fix(types): prune the checked SSH watcher from suppression baseline

* fix(files): retain established SSH watches through connection loss

* fix(files): fence initial SSH watcher callbacks across reconnects

* test(files): verify guarded SSH watch errors and cancellation

config/ts-nocheck-baseline.txt

@@ -163,7 +163,6 @@ src/main/runtime/runtime-file-commands-revoke-terminal-file-grants-for-client.ts
 src/main/runtime/runtime-file-commands-search-local-runtime-files.ts
 src/main/runtime/runtime-file-commands-search-remote-quick-open-file-paths.ts
 src/main/runtime/runtime-file-commands-search-runtime-files.ts
-src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.ts
 src/main/runtime/runtime-file-commands-terminal-artifact-access.ts
 src/main/runtime/runtime-file-commands-terminal-file-paths.ts
 src/main/runtime/runtime-file-commands-write-file-explorer-file.ts

src/main/runtime/orca-runtime-files-ssh-rearm.test.ts

@@ -1,5 +1,7 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import type { FsChangeEvent } from '../../shared/filesystem-entry-types'
+import { createSshDisposalError } from '../ssh/ssh-channel-multiplexer'
+import { createSshFileExplorerWatchTestConnection } from './ssh-file-explorer-watch-test-connection'
 
 const {
   resolveAuthorizedPathMock,
@@ -47,6 +49,8 @@ import {
 const ROOT_PATH = '/home/me/repo'
 const CONNECTION_ID = 'conn-1'
 const OVERFLOW_EVENTS: FsChangeEvent[] = [{ kind: 'overflow', absolutePath: ROOT_PATH }]
+const connections: ReturnType<typeof createSshFileExplorerWatchTestConnection>[] = []
+const releases: (() => Promise<void>)[] = []
 
 /** Drive the provider-registration hook the way a relay reconnect would. */
 function emitProviderRegistered(connectionId: string): void {
@@ -80,29 +84,168 @@ describe('remote file-explorer watch re-arm', () => {
   })
 
   afterEach(async () => {
+    for (const { mux } of connections) {
+      mux.dispose()
+    }
+    for (const release of releases.splice(0)) {
+      await release()
+    }
     await awaitRuntimeFileWatcherUnsubscribes()
     _resetRuntimeFileWatcherLeasesForTests()
+    for (const { provider } of connections.splice(0)) {
+      provider.dispose()
+    }
   })
 
+  function connect() {
+    const connection = createSshFileExplorerWatchTestConnection(ROOT_PATH, CONNECTION_ID)
+    connections.push(connection)
+    return connection
+  }
+
+  async function startInitialWatch() {
+    const first = connect()
+    getSshFilesystemProviderMock.mockReturnValue(first.provider)
+    const watch = vi.spyOn(first.provider, 'watch')
+    const commands = createRuntimeFileCommands()
+    const onEvents = vi.fn()
+    const controller = new AbortController()
+    const onTerminalError = vi.fn(() => controller.abort())
+    const setup = commands.watchFileExplorer(
+      'id:wt-1',
+      onEvents,
+      onTerminalError,
+      controller.signal
+    )
+    await vi.waitFor(() => expect(first.countRequests('fs.watch')).toBe(1))
+    const finish = async () => {
+      first.settleWatch()
+      const release = await setup
+      releases.push(release)
+      return release
+    }
+    return { first, watch, onEvents, onTerminalError, controller, setup, finish }
+  }
+
+  it.each(['event', 'terminal error'] as const)(
+    'fences the initial provider %s while replacement setup is pending',
+    async (stage) => {
+      const { first, onEvents, onTerminalError, controller, finish } = await startInitialWatch()
+      const release = await finish()
+      const next = connect()
+      getSshFilesystemProviderMock.mockReturnValue(next.provider)
+      emitProviderRegistered(CONNECTION_ID)
+      await vi.waitFor(() => expect(next.countRequests('fs.watch')).toBe(1))
+      if (stage === 'event') {
+        first.emitChange()
+      } else {
+        first.failWatch('late initial failure')
+      }
+      expect(onEvents).not.toHaveBeenCalled()
+      expect(onTerminalError).not.toHaveBeenCalled()
+      expect(controller.signal.aborted).toBe(false)
+      next.settleWatch()
+      await vi.waitFor(() => expect(onEvents).toHaveBeenCalledExactlyOnceWith(OVERFLOW_EVENTS))
+      next.emitChange()
+      expect(onEvents).toHaveBeenCalledTimes(2)
+      await release()
+      expect(next.countRequests('fs.unwatch')).toBe(1)
+    }
+  )
+
+  it('fences initial callbacks after same-provider replacement and unsubscribe', async () => {
+    const { first, watch, onEvents, onTerminalError, finish } = await startInitialWatch()
+    const release = await finish()
+    const initialEvent = watch.mock.calls[0]?.[1]
+    const initialError = watch.mock.calls[0]?.[2]?.onTerminalError
+    emitProviderRegistered(CONNECTION_ID)
+    await vi.waitFor(() => expect(onEvents).toHaveBeenCalledExactlyOnceWith(OVERFLOW_EVENTS))
+    initialEvent?.([{ kind: 'update', absolutePath: `${ROOT_PATH}/stale.ts` }])
+    initialError?.(new Error('old generation failed'))
+    expect(onEvents).toHaveBeenCalledOnce()
+    expect(onTerminalError).not.toHaveBeenCalled()
+    first.emitChange()
+    expect(onEvents).toHaveBeenCalledTimes(2)
+    await release()
+    initialEvent?.([{ kind: 'update', absolutePath: `${ROOT_PATH}/closed.ts` }])
+    initialError?.(new Error('closed initial watch failed'))
+    expect(onEvents).toHaveBeenCalledTimes(2)
+    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(first.countRequests('fs.unwatch')).toBe(1)
+  })
+
+  it('delivers current initial events during setup and genuine current failures after setup', async () => {
+    const { first, onEvents, onTerminalError, controller, finish } = await startInitialWatch()
+    first.emitChange()
+    expect(onEvents).toHaveBeenCalledOnce()
+    await finish()
+    first.emitChange()
+    expect(onEvents).toHaveBeenCalledTimes(2)
+    first.failWatch('current initial failure')
+    expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(
+      expect.objectContaining({ message: 'current initial failure' })
+    )
+    expect(controller.signal.aborted).toBe(true)
+  })
+
+  it('keeps a completed initial watch armed after typed connection loss', async () => {
+    const { watch, onEvents, onTerminalError, controller, finish } = await startInitialWatch()
+    await finish()
+    watch.mock.calls[0]?.[2]?.onTerminalError?.(createSshDisposalError('connection_lost'))
+    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(controller.signal.aborted).toBe(false)
+    const next = connect()
+    getSshFilesystemProviderMock.mockReturnValue(next.provider)
+    emitProviderRegistered(CONNECTION_ID)
+    await vi.waitFor(() => expect(next.countRequests('fs.watch')).toBe(1))
+    next.settleWatch()
+    await vi.waitFor(() => expect(onEvents).toHaveBeenCalledExactlyOnceWith(OVERFLOW_EVENTS))
+  })
+
+  it.each(['connection loss', 'cancellation'] as const)(
+    'rejects initial setup on %s and fences callbacks after setup ends',
+    async (reason) => {
+      const { first, watch, onEvents, onTerminalError, controller, setup } =
+        await startInitialWatch()
+      if (reason === 'connection loss') {
+        first.mux.dispose('connection_lost')
+        await expect(setup).rejects.toMatchObject({ code: 'CONNECTION_LOST' })
+      } else {
+        controller.abort()
+        await expect(setup).rejects.toMatchObject({ name: 'AbortError' })
+      }
+      watch.mock.calls[0]?.[1]([{ kind: 'update', absolutePath: `${ROOT_PATH}/late.ts` }])
+      watch.mock.calls[0]?.[2]?.onTerminalError?.(new Error('late setup failure'))
+      expect(onEvents).not.toHaveBeenCalled()
+      expect(onTerminalError).not.toHaveBeenCalled()
+      expect(providerRegistrationListeners.size).toBe(0)
+    }
+  )
+
   it('reinstalls and resyncs when the connection re-registers its provider', async () => {
     // Why: dispose() on transport loss stops the registration without firing onTerminalError, so
     // nothing else tells this watch it died.
     const firstUnwatch = vi.fn()
     const secondUnwatch = vi.fn()
-    const watch = vi.fn().mockResolvedValueOnce(firstUnwatch).mockResolvedValueOnce(secondUnwatch)
-    getSshFilesystemProviderMock.mockReturnValue({ watch })
+    const firstProvider = { watch: vi.fn().mockResolvedValue(firstUnwatch) }
+    const secondProvider = { watch: vi.fn().mockResolvedValue(secondUnwatch) }
+    getSshFilesystemProviderMock.mockReturnValue(firstProvider)
     const commands = createRuntimeFileCommands()
     const onEvents = vi.fn()
 
-    await commands.watchFileExplorer('id:wt-1', onEvents)
-    expect(watch).toHaveBeenCalledTimes(1)
+    const unsubscribe = await commands.watchFileExplorer('id:wt-1', onEvents)
+    expect(firstProvider.watch).toHaveBeenCalledOnce()
 
+    getSshFilesystemProviderMock.mockReturnValue(secondProvider)
     emitProviderRegistered(CONNECTION_ID)
-    await vi.waitFor(() => expect(watch).toHaveBeenCalledTimes(2))
+    await vi.waitFor(() => expect(secondProvider.watch).toHaveBeenCalledOnce())
 
     expect(onEvents).toHaveBeenCalledWith(OVERFLOW_EVENTS)
     // The dead transport's handle must not be closed against the fresh registration.
     expect(firstUnwatch).not.toHaveBeenCalled()
+    await unsubscribe()
+    expect(secondUnwatch).toHaveBeenCalledOnce()
+    expect(firstUnwatch).not.toHaveBeenCalled()
   })
 
   it('ignores registrations for other connections', async () => {

src/main/runtime/orca-runtime-files-watch-host-scope.test.ts

@@ -1,6 +1,8 @@
 import { describe, expect, it, vi } from 'vitest'
 import { EventEmitter } from 'node:events'
+import type { FsChangeEvent } from '../../shared/filesystem-entry-types'
 import {
+  getSshFilesystemProviderMock,
   resolveAuthorizedPathMock,
   statMock,
   watchInWatcherProcessMock,
@@ -102,7 +104,9 @@ describe('RuntimeFileCommands', () => {
 
   it('keeps SSH runtime watches on the remote filesystem provider', async () => {
     const remoteDispose = vi.fn()
-    const providerWatch = vi.fn(() => remoteDispose)
+    const providerWatch = vi.fn(
+      (_rootPath: string, _callback: (events: FsChangeEvent[]) => void) => remoteDispose
+    )
     vi.mocked(getSshFilesystemProvider).mockReturnValue({ watch: providerWatch } as never)
     const { commands, store } = createRuntimeFileCommands({ path: '/remote/repo' })
     store.getRepo.mockReturnValue({ connectionId: 'ssh-1' })
@@ -110,10 +114,15 @@ describe('RuntimeFileCommands', () => {
 
     const unsubscribe = await commands.watchFileExplorer('id:wt-1', onEvents)
 
-    expect(providerWatch).toHaveBeenCalledWith('/remote/repo', onEvents, {
+    expect(providerWatch).toHaveBeenCalledWith('/remote/repo', expect.any(Function), {
       signal: undefined,
       onTerminalError: expect.any(Function)
     })
+    const callback = providerWatch.mock.calls[0]?.[1]
+    callback?.([{ kind: 'update', absolutePath: '/remote/repo/current.ts' }])
+    expect(onEvents).toHaveBeenCalledWith([
+      { kind: 'update', absolutePath: '/remote/repo/current.ts' }
+    ])
     expect(watchInWatcherProcessMock).not.toHaveBeenCalled()
     await unsubscribe()
     expect(remoteDispose).toHaveBeenCalledTimes(1)
@@ -149,11 +158,10 @@ describe('RuntimeFileCommands', () => {
   it('scopes same-path runtime watcher teardown to its SSH execution host', async () => {
     const firstDispose = vi.fn()
     const secondDispose = vi.fn()
-    vi.mocked(getSshFilesystemProvider).mockImplementation(
-      (connectionId) =>
-        ({
-          watch: vi.fn(() => (connectionId === 'ssh-1' ? firstDispose : secondDispose))
-        }) as never
+    const firstProvider = { watch: vi.fn(() => firstDispose) }
+    const secondProvider = { watch: vi.fn(() => secondDispose) }
+    getSshFilesystemProviderMock.mockImplementation((connectionId) =>
+      connectionId === 'ssh-1' ? firstProvider : secondProvider
     )
     const first = createRuntimeFileCommands({ path: '/same/repo' })
     const second = createRuntimeFileCommands({ path: '/same/repo' })

src/main/runtime/orca-runtime-files-watch.test.ts

@@ -4,6 +4,7 @@ import type * as Fs from 'node:fs'
 import type * as FsPromises from 'node:fs/promises'
 import type * as FilesystemAuth from '../ipc/filesystem-auth'
 import type { FsChangeEvent } from '../../shared/filesystem-entry-types'
+import type { IFilesystemProvider } from '../providers/types'
 import { WatcherProcessFailure } from '../ipc/parcel-watcher-process-failure'
 import { acquireWatcherRemovalGate } from '../ipc/watcher-removal-gate'
 
@@ -543,38 +544,62 @@ describe('RuntimeFileCommands file watching', () => {
     expect(drained).toBe(true)
   })
 
-  it('forwards the abort signal into SSH-backed file explorer watches', async () => {
-    const watch = vi.fn(async () => () => {})
-    getSshFilesystemProviderMock.mockReturnValue({ watch })
-    const store = { getRepo: vi.fn(() => ({ connectionId: 'ssh-1' })) }
-    const commands = new RuntimeFileCommands({
-      getRuntimeId: () => 'runtime-1',
-      requireStore: () => store,
-      resolveWorktreeSelector: vi.fn(async () => ({
-        id: 'wt-1',
-        repoId: 'repo-1',
-        path: '/remote/repo'
-      })),
-      resolveRuntimeFileTarget: vi.fn(async () => ({
-        worktree: {
+  it.each(['abort', 'unsubscribe'] as const)(
+    'forwards the SSH watch signal and fences terminal errors after %s',
+    async (stop) => {
+      const dispose = vi.fn()
+      const watch = vi.fn<IFilesystemProvider['watch']>().mockResolvedValue(dispose)
+      getSshFilesystemProviderMock.mockReturnValue({ watch })
+      const store = { getRepo: vi.fn(() => ({ connectionId: 'ssh-1' })) }
+      const commands = new RuntimeFileCommands({
+        getRuntimeId: () => 'runtime-1',
+        requireStore: () => store,
+        resolveWorktreeSelector: vi.fn(async () => ({
           id: 'wt-1',
           repoId: 'repo-1',
           path: '/remote/repo'
-        },
-        executionHostId: 'ssh:ssh-1'
-      })),
-      resolveRuntimeGitTarget: vi.fn(),
-      openFile: vi.fn()
-    } as never)
-    const controller = new AbortController()
-    const onTerminalError = vi.fn()
+        })),
+        resolveRuntimeFileTarget: vi.fn(async () => ({
+          worktree: {
+            id: 'wt-1',
+            repoId: 'repo-1',
+            path: '/remote/repo'
+          },
+          executionHostId: 'ssh:ssh-1'
+        })),
+        resolveRuntimeGitTarget: vi.fn(),
+        openFile: vi.fn()
+      } as never)
+      const controller = new AbortController()
+      const onTerminalError = vi.fn()
 
-    await commands.watchFileExplorer('id:wt-1', vi.fn(), onTerminalError, controller.signal)
+      const unsubscribe = await commands.watchFileExplorer(
+        'id:wt-1',
+        vi.fn(),
+        onTerminalError,
+        controller.signal
+      )
 
-    expect(watch).toHaveBeenCalledWith('/remote/repo', expect.any(Function), {
-      signal: controller.signal,
-      onTerminalError
-    })
-    expect(watchInWatcherProcessMock).not.toHaveBeenCalled()
-  })
+      expect(watch).toHaveBeenCalledWith('/remote/repo', expect.any(Function), {
+        signal: controller.signal,
+        onTerminalError: expect.any(Function)
+      })
+      const options = watch.mock.calls[0]?.[2]
+      expect(options?.signal).toBe(controller.signal)
+      const currentError = new Error('current remote watch failed')
+      options?.onTerminalError?.(currentError)
+      expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(currentError)
+      if (stop === 'abort') {
+        controller.abort()
+      } else {
+        await unsubscribe()
+        expect(controller.signal.aborted).toBe(false)
+      }
+      options?.onTerminalError?.(new Error('late remote watch failure'))
+      expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(currentError)
+      await unsubscribe()
+      expect(dispose).toHaveBeenCalledOnce()
+      expect(watchInWatcherProcessMock).not.toHaveBeenCalled()
+    }
+  )
 })

src/main/runtime/runtime-file-commands-assert-remote-terminal-file-grant-path-still-canonical.ts

@@ -100,21 +100,47 @@ export class RuntimeFileCommandsWithAssertRemoteTerminalFileGrantPathStillCanoni
           if (!route.provider) {
             throw new Error(SSH_FILESYSTEM_PROVIDER_UNAVAILABLE_MESSAGE)
           }
-          // Why: the RPC layer already threads AbortSignal for local watches; SSH must cancel the remote fs.watch, not wait it out.
-          const close = await route.provider.watch(target.path, callback, {
-            signal,
-            onTerminalError
-          })
-          const rearm = armSshFileExplorerWatchRearm({
-            runtimeId: this.host.getRuntimeId(),
-            connectionId: route.connectionId,
-            rootPath: target.path,
-            callback,
-            onTerminalError,
-            signal,
-            initialUnwatch: close
-          })
-          return { unsubscribe: rearm.unsubscribe, rootPaths: [target.path] }
+          const provider = route.provider
+          const isCurrentInitialProvider = (): boolean =>
+            !signal?.aborted && provider === getSshFilesystemProvider(route.connectionId)
+          let initialCallbacks = {
+            callback: (events: FsChangeEvent[]) => {
+              if (isCurrentInitialProvider()) {
+                callback(events)
+              }
+            },
+            onTerminalError: (error: Error) => {
+              if (isCurrentInitialProvider()) {
+                onTerminalError(error)
+              }
+            }
+          }
+          try {
+            // Initial callbacks join the rearm generation after setup succeeds.
+            const close = await provider.watch(
+              target.path,
+              (events) => initialCallbacks.callback(events),
+              { signal, onTerminalError: (error) => initialCallbacks.onTerminalError(error) }
+            )
+            const rearm = armSshFileExplorerWatchRearm({
+              runtimeId: this.host.getRuntimeId(),
+              connectionId: route.connectionId,
+              rootPath: target.path,
+              callback,
+              onTerminalError,
+              signal,
+              initialUnwatch: close,
+              initialProvider: provider
+            })
+            initialCallbacks = rearm.initialCallbacks
+            return { unsubscribe: rearm.unsubscribe, rootPaths: [target.path] }
+          } catch (error) {
+            initialCallbacks = {
+              callback: () => undefined,
+              onTerminalError: () => undefined
+            }
+            throw error
+          }
         }
 
         const rootPath = await resolveAuthorizedPath(target.path, this.host.requireStore())

src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.test.ts

@@ -0,0 +1,421 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type { IFilesystemProvider } from '../providers/types'
+import {
+  createSshDisposalError,
+  SSH_MUX_REQUEST_TIMEOUT_CODE
+} from '../ssh/ssh-channel-multiplexer'
+import { createSshFileExplorerWatchTestConnection as createWatchConnection } from './ssh-file-explorer-watch-test-connection'
+import { armSshFileExplorerWatchRearm } from './runtime-file-commands-ssh-file-watcher-rearm'
+
+type WatchProvider = Pick<IFilesystemProvider, 'watch'>
+
+const { getProvider, registrationListeners, rearms } = vi.hoisted(() => ({
+  getProvider: vi.fn<(connectionId: string) => WatchProvider | undefined>(),
+  registrationListeners: new Set<(connectionId: string) => void>(),
+  rearms: new Map<string, Set<() => void>>()
+}))
+
+vi.mock('../providers/ssh-filesystem-dispatch', () => ({
+  getSshFilesystemProvider: getProvider,
+  onSshFilesystemProviderRegistered: (listener: (connectionId: string) => void) => {
+    registrationListeners.add(listener)
+    return () => registrationListeners.delete(listener)
+  }
+}))
+
+vi.mock('./runtime-file-commands-mobile-file-list-limit', () => ({
+  runtimeWatcherReleaseKey: (runtimeId: string, connectionId: string, rootPath: string) =>
+    `${runtimeId}:${connectionId}:${rootPath}`,
+  sshFileExplorerWatchRearms: rearms
+}))
+
+function registerProvider(provider: WatchProvider): void {
+  getProvider.mockReturnValue(provider)
+  for (const listener of registrationListeners) {
+    listener('ssh-1')
+  }
+}
+
+function pendingWatch() {
+  let resolve: (unwatch: () => void) => void = () => undefined
+  let reject: (error: Error) => void = () => undefined
+  const promise = new Promise<() => void>((resolvePromise, rejectPromise) => {
+    resolve = resolvePromise
+    reject = rejectPromise
+  })
+  return { watch: vi.fn<IFilesystemProvider['watch']>(() => promise), resolve, reject }
+}
+
+describe('SSH file explorer watcher rearm', () => {
+  let unsubscribe: (() => Promise<void>) | undefined
+  const connections: ReturnType<typeof createWatchConnection>[] = []
+
+  beforeEach(() => {
+    getProvider.mockReset()
+    registrationListeners.clear()
+    rearms.clear()
+  })
+
+  afterEach(async () => {
+    for (const { mux } of connections) {
+      mux.dispose()
+    }
+    await unsubscribe?.()
+    unsubscribe = undefined
+    for (const { provider } of connections.splice(0)) {
+      provider.dispose()
+    }
+    expect(registrationListeners.size).toBe(0)
+    expect(rearms.size).toBe(0)
+  })
+
+  function install(
+    initialProvider: WatchProvider = { watch: vi.fn() },
+    initialUnwatch: () => void = vi.fn()
+  ) {
+    if (!getProvider.getMockImplementation()) {
+      getProvider.mockReturnValue(initialProvider)
+    }
+    const onEvents = vi.fn()
+    const controller = new AbortController()
+    const onTerminalError = vi.fn((_error: Error) => controller.abort())
+    const rearm = armSshFileExplorerWatchRearm({
+      runtimeId: 'runtime-1',
+      connectionId: 'ssh-1',
+      rootPath: '/remote/repo',
+      callback: onEvents,
+      onTerminalError,
+      signal: controller.signal,
+      initialUnwatch,
+      initialProvider
+    })
+    unsubscribe = rearm.unsubscribe
+    return { onEvents, onTerminalError, initialUnwatch, controller }
+  }
+
+  it('ignores a superseded setup failure and installs the current provider', async () => {
+    const { onEvents, onTerminalError, controller } = install()
+    const old = pendingWatch()
+    registerProvider(old)
+    await vi.waitFor(() => expect(old.watch).toHaveBeenCalledOnce())
+
+    const currentUnwatch = vi.fn()
+    const current = { watch: vi.fn(async () => currentUnwatch) }
+    registerProvider(current)
+    old.reject(new Error('previous transport disconnected'))
+
+    await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce())
+    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(controller.signal.aborted).toBe(false)
+    expect(onEvents).toHaveBeenCalledExactlyOnceWith([
+      { kind: 'overflow', absolutePath: '/remote/repo' }
+    ])
+    await unsubscribe?.()
+    expect(currentUnwatch).toHaveBeenCalledOnce()
+  })
+
+  it('replaces an initial watch whose provider changed before rearm was installed', async () => {
+    const original = pendingWatch()
+    const initialUnwatch = vi.fn()
+    const initialSetup = original.watch('/remote/repo', vi.fn())
+    const currentUnwatch = vi.fn()
+    const current = { watch: vi.fn(async () => currentUnwatch) }
+    registerProvider(current)
+    original.resolve(initialUnwatch)
+    await initialSetup
+    const { onEvents, onTerminalError } = install(original, initialUnwatch)
+
+    await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce())
+    expect(current.watch).toHaveBeenCalledOnce()
+    expect(initialUnwatch).not.toHaveBeenCalled()
+    expect(onTerminalError).not.toHaveBeenCalled()
+    await unsubscribe?.()
+    expect(currentUnwatch).toHaveBeenCalledOnce()
+    expect(initialUnwatch).not.toHaveBeenCalled()
+  })
+
+  it('coalesces registrations that arrive before replacement setup starts', async () => {
+    const { onEvents } = install()
+    const old = { watch: vi.fn(async () => vi.fn()) }
+    const current = { watch: vi.fn(async () => vi.fn()) }
+    registerProvider(old)
+    registerProvider(current)
+
+    await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce())
+    expect(old.watch).not.toHaveBeenCalled()
+    expect(current.watch).toHaveBeenCalledOnce()
+  })
+
+  it('closes a superseded successful setup without publishing its refresh', async () => {
+    const { onEvents, onTerminalError } = install()
+    const old = pendingWatch()
+    registerProvider(old)
+    await vi.waitFor(() => expect(old.watch).toHaveBeenCalledOnce())
+    const currentUnwatch = vi.fn()
+    const current = { watch: vi.fn(async () => currentUnwatch) }
+    registerProvider(current)
+    const obsoleteUnwatch = vi.fn()
+    old.resolve(obsoleteUnwatch)
+
+    await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce())
+    expect(obsoleteUnwatch).toHaveBeenCalledOnce()
+    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(onEvents).toHaveBeenCalledOnce()
+    await unsubscribe?.()
+    expect(currentUnwatch).toHaveBeenCalledOnce()
+  })
+
+  it('reports a genuine current-provider setup failure', async () => {
+    const { onTerminalError, onEvents } = install()
+    const error = new Error('current host refused watch')
+    registerProvider({
+      watch: vi.fn(async () => {
+        throw error
+      })
+    })
+
+    await vi.waitFor(() => expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(error))
+    expect(onEvents).not.toHaveBeenCalled()
+  })
+
+  it.each(['setup', 'terminal callback'] as const)(
+    'waits for registration after a current-provider %s loses its connection',
+    async (stage) => {
+      const { onTerminalError, onEvents, controller } = install()
+      const lost = pendingWatch()
+      registerProvider(lost)
+      await vi.waitFor(() => expect(lost.watch).toHaveBeenCalledOnce())
+      const error = createSshDisposalError('connection_lost')
+      if (stage === 'setup') {
+        lost.reject(error)
+      } else {
+        lost.resolve(vi.fn())
+        await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce())
+        lost.watch.mock.calls[0]?.[2]?.onTerminalError?.(error)
+      }
+      await new Promise<void>((resolve) => setImmediate(resolve))
+      expect(onTerminalError).not.toHaveBeenCalled()
+      expect(controller.signal.aborted).toBe(false)
+
+      const currentUnwatch = vi.fn()
+      const current = { watch: vi.fn(async () => currentUnwatch) }
+      registerProvider(current)
+      await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce())
+      expect(onEvents).toHaveBeenLastCalledWith([
+        { kind: 'overflow', absolutePath: '/remote/repo' }
+      ])
+      await unsubscribe?.()
+      expect(currentUnwatch).toHaveBeenCalledOnce()
+    }
+  )
+
+  it.each([
+    ['shutdown', createSshDisposalError('shutdown')],
+    [
+      'timeout',
+      Object.assign(new Error('watch timed out'), { code: SSH_MUX_REQUEST_TIMEOUT_CODE })
+    ],
+    ['uncoded connection message', new Error('CONNECTION_LOST')]
+  ])('reports a current-provider %s setup failure', async (_label, error) => {
+    const { onTerminalError } = install()
+    registerProvider({
+      watch: vi.fn(async () => {
+        throw error
+      })
+    })
+    await vi.waitFor(() => expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(error))
+  })
+
+  it.each(['active', 'canceled'] as const)(
+    'keeps an %s owner correct through actual multiplexer loss and later registration',
+    async (owner) => {
+      const connect = () => {
+        const connection = createWatchConnection()
+        connections.push(connection)
+        return connection
+      }
+      const first = connect()
+      getProvider.mockReturnValue(first.provider)
+      const initialSetup = first.provider.watch('/remote/repo', vi.fn())
+      first.settleWatch()
+      const { onEvents, onTerminalError, controller } = install(first.provider, await initialSetup)
+      first.mux.dispose('connection_lost')
+      first.provider.dispose()
+      const next = connect()
+      registerProvider(next.provider)
+      await vi.waitFor(() => expect(next.countRequests('fs.watch')).toBe(1))
+      next.mux.dispose('connection_lost')
+      await new Promise<void>((resolve) => setImmediate(resolve))
+      expect(getProvider('ssh-1')).toBe(next.provider)
+      expect(onTerminalError).not.toHaveBeenCalled()
+      expect(controller.signal.aborted).toBe(false)
+      expect(onEvents).not.toHaveBeenCalled()
+      if (owner === 'canceled') {
+        await unsubscribe?.()
+      }
+
+      next.provider.dispose()
+      const current = connect()
+      registerProvider(current.provider)
+      if (owner === 'canceled') {
+        await new Promise<void>((resolve) => setImmediate(resolve))
+        expect(current.countRequests('fs.watch')).toBe(0)
+        return
+      }
+      await vi.waitFor(() => expect(current.countRequests('fs.watch')).toBe(1))
+      current.settleWatch()
+      await vi.waitFor(() =>
+        expect(onEvents).toHaveBeenCalledExactlyOnceWith([
+          { kind: 'overflow', absolutePath: '/remote/repo' }
+        ])
+      )
+      current.emitChange()
+      expect(onEvents).toHaveBeenLastCalledWith([
+        { kind: 'update', absolutePath: '/remote/repo/current.ts' }
+      ])
+      await unsubscribe?.()
+      expect(current.countRequests('fs.unwatch')).toBe(1)
+      current.emitChange()
+      expect(onEvents).toHaveBeenCalledTimes(2)
+    }
+  )
+
+  it.each(['unregistered', 'replaced'] as const)(
+    'ignores a setup failure when its provider is %s between rejection handlers',
+    async (change) => {
+      const { onTerminalError, onEvents, controller } = install()
+      const old = pendingWatch()
+      registerProvider(old)
+      await vi.waitFor(() => expect(old.watch).toHaveBeenCalledOnce())
+      const current = { watch: vi.fn(async () => vi.fn()) }
+      const checkedWhileCurrent = vi.fn(() => {
+        queueMicrotask(() => {
+          if (change === 'replaced') {
+            registerProvider(current)
+          } else {
+            getProvider.mockReturnValue(undefined)
+          }
+        })
+        return old
+      })
+      getProvider.mockImplementationOnce(checkedWhileCurrent)
+      old.reject(new Error('previous setup failed'))
+      await new Promise<void>((resolve) => setImmediate(resolve))
+
+      expect(checkedWhileCurrent).toHaveBeenCalledOnce()
+      expect(onTerminalError).not.toHaveBeenCalled()
+      expect(controller.signal.aborted).toBe(false)
+      if (change === 'replaced') {
+        expect(current.watch).toHaveBeenCalledOnce()
+        expect(onEvents).toHaveBeenCalledExactlyOnceWith([
+          { kind: 'overflow', absolutePath: '/remote/repo' }
+        ])
+      } else {
+        expect(onEvents).not.toHaveBeenCalled()
+      }
+    }
+  )
+
+  it('ignores late terminal callbacks from a superseded provider', async () => {
+    const { onTerminalError, controller } = install()
+    const old = pendingWatch()
+    registerProvider(old)
+    await vi.waitFor(() => expect(old.watch).toHaveBeenCalledOnce())
+    const current = { watch: vi.fn(async () => vi.fn()) }
+    registerProvider(current)
+    old.watch.mock.calls[0]?.[2]?.onTerminalError?.(new Error('previous watcher stopped'))
+    old.resolve(vi.fn())
+
+    await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce())
+    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(controller.signal.aborted).toBe(false)
+  })
+
+  it('reports a terminal callback from the current provider', async () => {
+    const { onTerminalError, onEvents } = install()
+    const current = pendingWatch()
+    registerProvider(current)
+    await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce())
+    current.resolve(vi.fn())
+    await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce())
+    const error = new Error('current watcher stopped')
+    current.watch.mock.calls[0]?.[2]?.onTerminalError?.(error)
+
+    expect(onTerminalError).toHaveBeenCalledExactlyOnceWith(error)
+  })
+
+  it('ignores late events from a superseded provider and delivers current events', async () => {
+    const { onEvents } = install()
+    const old = pendingWatch()
+    registerProvider(old)
+    await vi.waitFor(() => expect(old.watch).toHaveBeenCalledOnce())
+    const current = pendingWatch()
+    registerProvider(current)
+    old.watch.mock.calls[0]?.[1]([{ kind: 'update', absolutePath: '/remote/repo/old.ts' }])
+    old.resolve(vi.fn())
+    await vi.waitFor(() => expect(current.watch).toHaveBeenCalledOnce())
+    current.resolve(vi.fn())
+    await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce())
+    const currentEvents = [{ kind: 'update' as const, absolutePath: '/remote/repo/current.ts' }]
+    current.watch.mock.calls[0]?.[1](currentEvents)
+
+    expect(onEvents).toHaveBeenCalledTimes(2)
+    expect(onEvents).toHaveBeenLastCalledWith(currentEvents)
+  })
+
+  it('releases the previous subscriber when the same provider is registered again', async () => {
+    const { onEvents } = install()
+    const callbacks = new Set<Parameters<IFilesystemProvider['watch']>[1]>()
+    const provider = {
+      watch: vi.fn<IFilesystemProvider['watch']>(async (_root, callback) => {
+        callbacks.add(callback)
+        return () => {
+          callbacks.delete(callback)
+        }
+      })
+    }
+    registerProvider(provider)
+    await vi.waitFor(() => expect(onEvents).toHaveBeenCalledOnce())
+    registerProvider(provider)
+    await vi.waitFor(() => expect(onEvents).toHaveBeenCalledTimes(2))
+
+    expect(callbacks.size).toBe(1)
+    const events = [{ kind: 'update' as const, absolutePath: '/remote/repo/current.ts' }]
+    for (const callback of callbacks) {
+      callback(events)
+    }
+    expect(onEvents).toHaveBeenCalledTimes(3)
+    expect(onEvents).toHaveBeenLastCalledWith(events)
+    await unsubscribe?.()
+    expect(callbacks.size).toBe(0)
+  })
+
+  it('does not report a setup failure after unsubscribe starts', async () => {
+    const { onTerminalError, onEvents, initialUnwatch } = install()
+    const pending = pendingWatch()
+    registerProvider(pending)
+    await vi.waitFor(() => expect(pending.watch).toHaveBeenCalledOnce())
+    const closed = unsubscribe?.()
+    pending.reject(new Error('setup canceled during shutdown'))
+
+    await closed
+    expect(initialUnwatch).toHaveBeenCalledOnce()
+    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(onEvents).not.toHaveBeenCalled()
+  })
+
+  it('closes a replacement that finishes after unsubscribe starts', async () => {
+    const { onTerminalError, onEvents } = install()
+    const pending = pendingWatch()
+    registerProvider(pending)
+    await vi.waitFor(() => expect(pending.watch).toHaveBeenCalledOnce())
+    const closed = unsubscribe?.()
+    const lateUnwatch = vi.fn()
+    pending.resolve(lateUnwatch)
+
+    await closed
+    expect(lateUnwatch).toHaveBeenCalledOnce()
+    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(onEvents).not.toHaveBeenCalled()
+  })
+})

src/main/runtime/runtime-file-commands-ssh-file-watcher-rearm.ts

@@ -1,5 +1,6 @@
-// @ts-nocheck -- mechanically split declarations.
 import type { FsChangeEvent } from '../../shared/filesystem-entry-types'
+import type { IFilesystemProvider } from '../providers/types'
+import { toRemoteRuntimeClientErrorLike } from '../../shared/remote-runtime-client-error-classification'
 import {
   runtimeWatcherReleaseKey,
   sshFileExplorerWatchRearms
@@ -17,40 +18,99 @@ export function armSshFileExplorerWatchRearm(args: {
   onTerminalError: (error: Error) => void
   signal?: AbortSignal
   initialUnwatch: () => void
-}): { unsubscribe: () => Promise<void> } {
+  initialProvider: Pick<IFilesystemProvider, 'watch'>
+}): {
+  initialCallbacks: {
+    callback: (events: FsChangeEvent[]) => void
+    onTerminalError: (error: Error) => void
+  }
+  unsubscribe: () => Promise<void>
+} {
   const key = runtimeWatcherReleaseKey(args.runtimeId, args.connectionId, args.rootPath)
   let currentUnwatch = args.initialUnwatch
+  let currentProvider = args.initialProvider
   let stopped = false
   let reinstalling: Promise<void> | null = null
+  let providerGeneration = 0
 
-  const reinstall = async (): Promise<void> => {
-    const provider = getSshFilesystemProvider(args.connectionId)
-    if (stopped || !provider) {
+  const reportTerminalError = (error: unknown): void => {
+    // Connection loss leaves the established stream armed for the next provider.
+    if (toRemoteRuntimeClientErrorLike(error).code === 'CONNECTION_LOST') {
       return
     }
-    // Why: the old handle is scoped to the dead transport; closing it here would only risk
-    // unwatching the root we just re-registered on the new one.
-    const nextUnwatch = await provider.watch(args.rootPath, args.callback, {
-      signal: args.signal,
-      onTerminalError: args.onTerminalError
-    })
-    if (stopped) {
+    args.onTerminalError(error instanceof Error ? error : new Error(String(error)))
+  }
+
+  const reinstall = async (
+    generation: number,
+    provider: Pick<IFilesystemProvider, 'watch'> | undefined
+  ): Promise<void> => {
+    if (stopped || generation !== providerGeneration || !provider) {
+      return
+    }
+    const isCurrent = (): boolean =>
+      !stopped &&
+      !args.signal?.aborted &&
+      generation === providerGeneration &&
+      provider === getSshFilesystemProvider(args.connectionId)
+    let nextUnwatch: () => void
+    try {
+      nextUnwatch = await provider.watch(
+        args.rootPath,
+        (events) => {
+          if (isCurrent()) {
+            args.callback(events)
+          }
+        },
+        {
+          signal: args.signal,
+          onTerminalError: (error) => {
+            if (isCurrent()) {
+              reportTerminalError(error)
+            }
+          }
+        }
+      )
+    } catch (error) {
+      if (!isCurrent()) {
+        return
+      }
+      throw error
+    }
+    if (!isCurrent()) {
       nextUnwatch()
       return
     }
+    // Why: a dead transport's handle must never unwatch the replacement transport's root.
+    if (currentProvider === provider) {
+      currentUnwatch()
+    }
     currentUnwatch = nextUnwatch
+    currentProvider = provider
     args.callback([{ kind: 'overflow', absolutePath: args.rootPath }])
   }
 
-  const unsubscribeRearm = onSshFilesystemProviderRegistered((registeredId) => {
-    if (registeredId !== args.connectionId || stopped) {
+  const scheduleReinstall = (): void => {
+    if (stopped) {
       return
     }
-    // Why: reconnect storms can register repeatedly; chain so a second one can't double-install.
+    const generation = ++providerGeneration
+    let attemptProvider: Pick<IFilesystemProvider, 'watch'> | undefined
+    // Why: obsolete reconnect attempts cannot terminate the stream or publish a stale refresh.
     const attempt = (reinstalling ?? Promise.resolve())
-      .then(reinstall)
+      .then(() => {
+        attemptProvider = getSshFilesystemProvider(args.connectionId)
+        return reinstall(generation, attemptProvider)
+      })
       .catch((error: unknown) => {
-        args.onTerminalError(error instanceof Error ? error : new Error(String(error)))
+        if (
+          !stopped &&
+          !args.signal?.aborted &&
+          generation === providerGeneration &&
+          attemptProvider === getSshFilesystemProvider(args.connectionId)
+        ) {
+          reportTerminalError(error)
+        }
       })
       .finally(() => {
         if (reinstalling === attempt) {
@@ -58,6 +118,11 @@ export function armSshFileExplorerWatchRearm(args: {
         }
       })
     reinstalling = attempt
+  }
+  const unsubscribeRearm = onSshFilesystemProviderRegistered((registeredId) => {
+    if (registeredId === args.connectionId) {
+      scheduleReinstall()
+    }
   })
 
   const stop = (): void => {
@@ -72,8 +137,28 @@ export function armSshFileExplorerWatchRearm(args: {
   const rearms = sshFileExplorerWatchRearms.get(key) ?? new Set<() => void>()
   rearms.add(stop)
   sshFileExplorerWatchRearms.set(key, rearms)
+  if (getSshFilesystemProvider(args.connectionId) !== args.initialProvider) {
+    scheduleReinstall()
+  }
+  const isInitialCurrent = (): boolean =>
+    !stopped &&
+    !args.signal?.aborted &&
+    providerGeneration === 0 &&
+    args.initialProvider === getSshFilesystemProvider(args.connectionId)
 
   return {
+    initialCallbacks: {
+      callback: (events) => {
+        if (isInitialCurrent()) {
+          args.callback(events)
+        }
+      },
+      onTerminalError: (error) => {
+        if (isInitialCurrent()) {
+          reportTerminalError(error)
+        }
+      }
+    },
     unsubscribe: () => {
       stop()
       const close = async (): Promise<void> => currentUnwatch()

src/main/runtime/ssh-file-explorer-watch-test-connection.ts

@@ -0,0 +1,65 @@
+import { SshFilesystemProvider } from '../providers/ssh-filesystem-provider'
+import { SshChannelMultiplexer, type MultiplexerTransport } from '../ssh/ssh-channel-multiplexer'
+import {
+  encodeJsonRpcFrame,
+  HEADER_LENGTH,
+  MessageType,
+  parseJsonRpcMessage,
+  type JsonRpcMessage
+} from '../ssh/relay-protocol'
+
+export function createSshFileExplorerWatchTestConnection(
+  rootPath = '/remote/repo',
+  connectionId = 'ssh-1'
+) {
+  const written: Buffer[] = []
+  let receive: (data: Buffer) => void = () => undefined
+  let sequence = 1
+  const transport: MultiplexerTransport = {
+    write: (data) => {
+      written.push(data)
+    },
+    onData: (callback) => {
+      receive = callback
+    },
+    onClose: () => undefined
+  }
+  const mux = new SshChannelMultiplexer(transport)
+  const provider = new SshFilesystemProvider(connectionId, mux)
+  const messages = (): JsonRpcMessage[] =>
+    written
+      .filter((frame) => frame[0] === MessageType.Regular)
+      .map((frame) => parseJsonRpcMessage(frame.subarray(HEADER_LENGTH)))
+  const send = (message: JsonRpcMessage): void => {
+    receive(encodeJsonRpcFrame(message, sequence++, 0))
+  }
+  const watchRequest = () => {
+    const request = messages().find(
+      (message) => 'method' in message && message.method === 'fs.watch'
+    )
+    if (!request || !('id' in request) || !('method' in request)) {
+      throw new Error('No pending fs.watch request')
+    }
+    return request
+  }
+  return {
+    mux,
+    provider,
+    countRequests: (method: string) =>
+      messages().filter((message) => 'method' in message && message.method === method).length,
+    settleWatch: () => send({ jsonrpc: '2.0', id: watchRequest().id, result: null }),
+    emitChange: (absolutePath = `${rootPath}/current.ts`) =>
+      send({
+        jsonrpc: '2.0',
+        method: 'fs.changed',
+        params: { events: [{ kind: 'update', absolutePath }] }
+      }),
+    failWatch: (message: string) => {
+      const watchId = watchRequest().params?.watchId
+      if (typeof watchId !== 'number') {
+        throw new Error('No remote watch id')
+      }
+      send({ jsonrpc: '2.0', method: 'fs.watchFailed', params: { rootPath, watchId, message } })
+    }
+  }
+}