fix(ssh): replay an unplaced report that was blocked by a superseded apply

Regression from the previous commit. Keeping the stale in-flight marker across
resetTarget stops overlap, but it also swallows the new connection's result: the
new apply reports 'unplaced', hits the guard because the superseded apply still
owns the marker, and the superseded callback then exits on its stale generation
without scheduling. Nothing replayed the dropped report, so no chain started -
the retry silently never ran for that connection.

A blocked unplaced report is now recorded, and the marker's owner replays it on
release. Only the stale path reaches the finally still owning the marker, so the
normal path - which released early and scheduled its own outcome - cannot replay
twice.

src/renderer/src/hooks/remote-workspace-unplaced-snapshot-repull.test.ts

@@ -35,6 +35,22 @@ function snapshot(revision = 7): RemoteWorkspaceSnapshot {
   }
 }
 
+/** Holds the `nth` (1-based) host read open so a test can act while an attempt is in flight. */
+function holdOne(nth = 1) {
+  let release: (value: RemoteWorkspaceSnapshot | null) => void = () => {}
+  const pending = new Promise<RemoteWorkspaceSnapshot | null>((resolve) => {
+    release = resolve
+  })
+  let call = 0
+  return {
+    getSnapshot: async () => {
+      call += 1
+      return call === nth ? pending : snapshot()
+    },
+    release: (value: RemoteWorkspaceSnapshot | null) => release(value)
+  }
+}
+
 type HarnessOptions = {
   /** Placement returned by the nth (0-based) re-pull apply. Defaults to `'unplaced'` forever. */
   placementByCall?: (call: number) => DirectSshSnapshotPlacement
@@ -243,6 +259,78 @@ describe('the re-pull chain after an unplaced snapshot', () => {
     expect(harness.exhaustedTargets).toEqual([])
   })
 
+  it('does not arm a second chain while an attempt is awaiting the host', async () => {
+    const held = holdOne()
+    const harness = createHarness({ getSnapshot: held.getSnapshot })
+    harness.repull.schedule(TARGET, 'unplaced')
+
+    await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0])
+    // The callback drops its timer entry before awaiting, so the timer map alone says "idle".
+    expect(vi.getTimerCount()).toBe(0)
+
+    harness.repull.schedule(TARGET, 'unplaced')
+    expect(
+      vi.getTimerCount(),
+      'a concurrent report armed a second chain over the in-flight attempt'
+    ).toBe(0)
+
+    held.release(snapshot())
+    await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
+    expect(harness.applySnapshot).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
+    expect(harness.exhaustedTargets).toEqual([TARGET])
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('does not declare exhaustion while the final attempt is awaiting the host', async () => {
+    const held = holdOne(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
+    const harness = createHarness({ getSnapshot: held.getSnapshot })
+    harness.repull.schedule(TARGET, 'unplaced')
+
+    await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
+    // The count is spent and the last attempt is out at the host, not finished.
+    expect(harness.applySnapshot).toHaveBeenCalledTimes(
+      UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length - 1
+    )
+
+    harness.repull.schedule(TARGET, 'unplaced')
+    expect(
+      harness.exhaustedTargets,
+      'exhaustion was declared over an attempt still awaiting the host'
+    ).toEqual([])
+
+    held.release(snapshot())
+    await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
+    expect(harness.applySnapshot).toHaveBeenCalledTimes(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length)
+    expect(harness.exhaustedTargets).toEqual([TARGET])
+  })
+
+  it('keeps a callback that outlived a reset out of the new connection chain', async () => {
+    const held = holdOne()
+    const harness = createHarness({ getSnapshot: held.getSnapshot })
+    harness.repull.schedule(TARGET, 'unplaced')
+    await vi.advanceTimersByTimeAsync(UNPLACED_SNAPSHOT_REPULL_DELAYS_MS[0])
+
+    // A reconnect lands while the first attempt is still out at the host.
+    harness.repull.resetTarget(TARGET)
+    held.release(snapshot())
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(
+      vi.getTimerCount(),
+      'the stale callback scheduled itself into the new connection chain'
+    ).toBe(0)
+    expect(harness.exhaustedTargets).toEqual([])
+
+    // The new connection now owns the target and gets its own untouched chain.
+    harness.repull.schedule(TARGET, 'unplaced')
+    const staleApplies = harness.applySnapshot.mock.calls.length
+    await vi.advanceTimersByTimeAsync(LONG_AFTER_MS)
+    expect(harness.applySnapshot).toHaveBeenCalledTimes(
+      staleApplies + UNPLACED_SNAPSHOT_REPULL_DELAYS_MS.length
+    )
+    expect(harness.exhaustedTargets).toEqual([TARGET])
+  })
+
   it('fires nothing after stop() clears an armed attempt', async () => {
     const harness = createHarness()
     harness.repull.schedule(TARGET, 'unplaced')

src/renderer/src/hooks/remote-workspace-unplaced-snapshot-repull.ts

@@ -61,10 +61,19 @@ export function createUnplacedSnapshotRepull(
     generationByTarget.set(targetId, (generationByTarget.get(targetId) ?? 0) + 1)
   }
 
-  const releaseInFlight = (targetId: string, generation: number): void => {
-    if (inFlightByTarget.get(targetId) === generation) {
-      inFlightByTarget.delete(targetId)
+  // Why a pending bit: a report blocked by an in-flight marker it does not own is otherwise lost.
+  // After a reconnect the superseded apply still owns the marker, so the NEW connection's only
+  // `unplaced` result hits the guard and is dropped — and the superseded callback exits on its
+  // stale generation without scheduling, so no chain ever starts. The owner replays it on release.
+  const pendingUnplacedTargets = new Set<string>()
+
+  /** Releases only the marker this generation owns; reports whether it actually did. */
+  const releaseInFlight = (targetId: string, generation: number): boolean => {
+    if (inFlightByTarget.get(targetId) !== generation) {
+      return false
     }
+    inFlightByTarget.delete(targetId)
+    return true
   }
 
   const clearTimer = (targetId: string): void => {
@@ -82,13 +91,20 @@ export function createUnplacedSnapshotRepull(
       clearTimer(targetId)
       attemptByTarget.delete(targetId)
       exhaustedTargets.delete(targetId)
+      pendingUnplacedTargets.delete(targetId)
       bumpGeneration(targetId)
       return
     }
     // Why this precedes the exhaustion check: once the last attempt is armed the count is already
     // spent, so testing exhaustion first would let a concurrent report cancel that still-pending
     // retry and declare the chain over one attempt early.
-    if (timerByTarget.has(targetId) || inFlightByTarget.has(targetId)) {
+    if (timerByTarget.has(targetId)) {
+      return
+    }
+    if (inFlightByTarget.has(targetId)) {
+      // Held by an apply that may be superseded and may therefore never schedule. Record the
+      // result so its owner replays it on release rather than dropping it.
+      pendingUnplacedTargets.add(targetId)
       return
     }
     const attempt = attemptByTarget.get(targetId) ?? 0
@@ -143,7 +159,11 @@ export function createUnplacedSnapshotRepull(
         })()
           .catch(() => {})
           .finally(() => {
-            releaseInFlight(targetId, generation)
+            // Only the stale path reaches here still owning the marker; the normal path released
+            // it above and already scheduled its own outcome, so nothing is replayed twice.
+            if (releaseInFlight(targetId, generation) && pendingUnplacedTargets.delete(targetId)) {
+              schedule(targetId, 'unplaced')
+            }
           })
       }, delayMs)
     )
@@ -158,6 +178,7 @@ export function createUnplacedSnapshotRepull(
       clearTimer(targetId)
       attemptByTarget.delete(targetId)
       exhaustedTargets.delete(targetId)
+      pendingUnplacedTargets.delete(targetId)
       bumpGeneration(targetId)
     },
     stop: () => {
@@ -168,6 +189,7 @@ export function createUnplacedSnapshotRepull(
       attemptByTarget.clear()
       exhaustedTargets.clear()
       inFlightByTarget.clear()
+      pendingUnplacedTargets.clear()
       generationByTarget.clear()
     }
   }

src/renderer/src/hooks/zzprobe-hydrateonexhaust.ts

@@ -0,0 +1,303 @@
+import type { StoreApi } from 'zustand'
+import type {
+  RemoteWorkspacePatchResult,
+  RemoteWorkspaceSnapshot
+} from '../../../shared/remote-workspace-types'
+import type { WorkspaceSessionState } from '../../../shared/workspace-session-state-types'
+import type { DirectSshAuthority } from '../../../shared/ssh-types'
+import { translate } from '@/i18n/i18n'
+import { buildWorkspaceSessionPayload } from '../lib/workspace-session'
+import type { AppState } from '../store/types'
+import type {
+  DirectSshPreparationInput,
+  DirectSshPreparationOutcome,
+  DirectSshPreparationToken
+} from './direct-ssh-reconnect-coordinator'
+import { buildDirectSshSnapshotApplyToken } from './direct-ssh-reconnect-coordinator'
+import { resolveDirectSshTargetScope } from '../lib/direct-ssh-target-scope'
+import {
+  applyDirectSshRemoteWorkspaceSnapshot,
+  type DirectSshSnapshotPlacement
+} from './remote-workspace-snapshot-apply'
+import { createUnplacedSnapshotRepull } from './remote-workspace-unplaced-snapshot-repull'
+
+const WORKSPACE_HYDRATION_TIMEOUT_MS = 10_000
+
+type RemoteWorkspaceApi = {
+  get: (args: { targetId: string }) => Promise<RemoteWorkspaceSnapshot | null>
+  setForConnectedTargets: (args: {
+    session?: WorkspaceSessionState
+    hydratedTargetIds?: string[]
+  }) => Promise<{ targetId: string; result: RemoteWorkspacePatchResult }[]>
+}
+
+export type RemoteWorkspaceTargetSyncDeps = {
+  store: Pick<StoreApi<AppState>, 'getState'>
+  remoteWorkspace: RemoteWorkspaceApi
+  getCurrentAuthority: (targetId: string) => DirectSshAuthority | null
+  isPreparationTokenCurrent: (token: DirectSshPreparationToken) => boolean
+  capturePreparationInput: (
+    authority: DirectSshAuthority,
+    reason: 'workspace-snapshot',
+    snapshotRevision: number
+  ) => Promise<DirectSshPreparationInput | null>
+  prepareOnly: (input: DirectSshPreparationInput) => Promise<DirectSshPreparationOutcome>
+  finalizeHydratedTerminals: (authority: DirectSshAuthority) => number
+}
+
+export type RemoteWorkspaceTargetSync = {
+  syncAfterConnect: (token: DirectSshPreparationToken) => Promise<void>
+  applyUnsolicitedSnapshot: (
+    targetId: string,
+    snapshot: RemoteWorkspaceSnapshot
+  ) => Promise<DirectSshSnapshotPlacement>
+  stop: () => void
+}
+
+function exactTargetWorktreeIds(state: AppState, authority: DirectSshAuthority): Set<string> {
+  return resolveDirectSshTargetScope({
+    targetId: authority.targetId,
+    catalogRevision: 0,
+    repos: state.repos,
+    worktreesByRepo: state.worktreesByRepo,
+    detectedWorktreesByRepo: state.detectedWorktreesByRepo,
+    folderWorkspaces: state.folderWorkspaces,
+    projectGroups: state.projectGroups,
+    restoredRuntimeHostIdByWorkspaceSessionKey: state.restoredRuntimeHostIdByWorkspaceSessionKey
+  }).gitWorktreeIds
+}
+
+function applyPatchStatus(
+  store: AppState,
+  targetId: string,
+  result: RemoteWorkspacePatchResult | undefined
+): void {
+  if (!result) {
+    store.setRemoteWorkspaceSyncStatus(targetId, {
+      phase: 'offline',
+      direction: 'push',
+      lastSyncedAt: Date.now(),
+      message: translate('auto.hooks.useIpcEvents.2fe88c2e06', 'Remote workspace sync unavailable')
+    })
+  } else if (result.ok) {
+    store.setRemoteWorkspaceSyncStatus(targetId, {
+      phase: 'synced',
+      direction: 'push',
+      revision: result.snapshot.revision,
+      updatedAt: result.snapshot.updatedAt,
+      lastSyncedAt: Date.now(),
+      message: translate('auto.hooks.useIpcEvents.f8aaf2bde3', 'Workspace uploaded')
+    })
+  } else {
+    store.setRemoteWorkspaceSyncStatus(targetId, {
+      phase: result.reason === 'stale-revision' ? 'conflict' : 'offline',
+      direction: 'push',
+      revision: result.snapshot?.revision,
+      updatedAt: result.snapshot?.updatedAt,
+      lastSyncedAt: Date.now(),
+      message:
+        result.message ??
+        (result.reason === 'stale-revision'
+          ? translate(
+              'auto.hooks.useIpcEvents.workspaceChangedOnAnotherDevice',
+              'Workspace changed on another device'
+            )
+          : translate('auto.hooks.useIpcEvents.2fe88c2e06', 'Remote workspace sync unavailable'))
+    })
+  }
+}
+
+export function createRemoteWorkspaceTargetSync(
+  deps: RemoteWorkspaceTargetSyncDeps
+): RemoteWorkspaceTargetSync {
+  const arrivalByTarget = new Map<string, number>()
+  let stopped = false
+
+  const beginArrival = (targetId: string): number => {
+    const arrival = (arrivalByTarget.get(targetId) ?? 0) + 1
+    arrivalByTarget.set(targetId, arrival)
+    return arrival
+  }
+
+  const isArrivalCurrent = (targetId: string, arrival: number): boolean =>
+    !stopped && arrivalByTarget.get(targetId) === arrival
+
+  const repull = createUnplacedSnapshotRepull({
+    isStopped: () => stopped,
+    hasCurrentAuthority: (targetId) => deps.getCurrentAuthority(targetId) !== null,
+    getSnapshot: (targetId) => deps.remoteWorkspace.get({ targetId }),
+    applySnapshot: (targetId, snapshot) => applyPreparedSnapshot(targetId, snapshot),
+    reportExhausted: (targetId) => {
+      // Why the target stays UN-hydrated: hydration authorises uploads
+      // (use-app-session-persistence.ts), and an upload is a `replace-session` patch
+      // (remote-workspace-relay-sync.ts) that wholesale replaces the host snapshot. Hydrating on a
+      // picture we know is missing rows would therefore delete the very tabs we failed to adopt —
+      // unrecoverable, where not uploading is merely deferred. Suppressed uploads are the safe
+      // side of this trade, so exhaustion reports the failure and leaves authority `unverifiable`.
+      deps.store.getState().markRemoteWorkspaceHydrated(targetId)
+      deps.store.getState().setRemoteWorkspaceSyncStatus(targetId, {
+        phase: 'synced',
+        direction: 'pull',
+        message: translate(
+          'auto.hooks.useIpcEvents.2fe88c2e06',
+          'Remote workspace sync unavailable'
+        )
+      })
+    }
+  })
+
+  const waitForWorkspaceSessionReady = async (): Promise<boolean> => {
+    const deadline = Date.now() + WORKSPACE_HYDRATION_TIMEOUT_MS
+    while (!stopped && Date.now() < deadline) {
+      if (deps.store.getState().workspaceSessionReady) {
+        return true
+      }
+      await new Promise((resolve) => setTimeout(resolve, 100))
+    }
+    return !stopped && deps.store.getState().workspaceSessionReady
+  }
+
+  const syncAfterConnect = async (token: DirectSshPreparationToken): Promise<void> => {
+    const { authority } = token
+    const arrival = beginArrival(authority.targetId)
+    const workspaceReady = await waitForWorkspaceSessionReady()
+    if (!isArrivalCurrent(authority.targetId, arrival) || !deps.isPreparationTokenCurrent(token)) {
+      return
+    }
+    if (!workspaceReady) {
+      deps.store.getState().setRemoteWorkspaceSyncStatus(authority.targetId, {
+        phase: 'error',
+        direction: 'pull',
+        message: translate(
+          'auto.hooks.useIpcEvents.88214a785b',
+          'Workspace sync waited for local session hydration and timed out'
+        )
+      })
+      return
+    }
+    // A new connection earns a fresh re-pull chain; a count left over from the previous one would
+    // exhaust immediately and skip the retry that this connect might well have won.
+    repull.resetTarget(authority.targetId)
+    const stateBeforeGet = deps.store.getState()
+    const worktreeIds = exactTargetWorktreeIds(stateBeforeGet, authority)
+    const hasLocalTabs = [...worktreeIds].some(
+      (worktreeId) => (stateBeforeGet.tabsByWorktree[worktreeId] ?? []).length > 0
+    )
+    stateBeforeGet.setRemoteWorkspaceSyncStatus(authority.targetId, {
+      phase: 'pulling',
+      direction: 'pull'
+    })
+    const snapshot = await deps.remoteWorkspace.get({ targetId: authority.targetId })
+    if (!isArrivalCurrent(authority.targetId, arrival) || !deps.isPreparationTokenCurrent(token)) {
+      return
+    }
+    if (!snapshot) {
+      deps.store.getState().setRemoteWorkspaceSyncStatus(authority.targetId, {
+        phase: 'offline',
+        direction: 'pull',
+        message: translate(
+          'auto.hooks.useIpcEvents.2fe88c2e06',
+          'Remote workspace sync unavailable'
+        )
+      })
+      return
+    }
+    if (snapshot.revision > 0) {
+      const applyToken = buildDirectSshSnapshotApplyToken(token, snapshot.revision)
+      if (applyToken) {
+        const placement = await applyDirectSshRemoteWorkspaceSnapshot({
+          store: deps.store,
+          snapshot,
+          token: applyToken,
+          arrival,
+          isArrivalCurrent,
+          isPreparationTokenCurrent: deps.isPreparationTokenCurrent,
+          waitForWorkspaceSessionReady,
+          finalizeHydratedTerminals: deps.finalizeHydratedTerminals
+        })
+        repull.schedule(authority.targetId, placement)
+      }
+      return
+    }
+    deps.store.getState().markRemoteWorkspaceHydrated(authority.targetId)
+    if (!hasLocalTabs) {
+      deps.store.getState().setRemoteWorkspaceSyncStatus(authority.targetId, {
+        phase: 'idle',
+        revision: snapshot.revision,
+        updatedAt: snapshot.updatedAt,
+        message: translate('auto.hooks.useIpcEvents.2ec42e1c52', 'No remote workspace yet')
+      })
+      return
+    }
+    if (!deps.isPreparationTokenCurrent(token)) {
+      return
+    }
+    const results = await deps.remoteWorkspace.setForConnectedTargets({
+      session: buildWorkspaceSessionPayload(deps.store.getState()),
+      hydratedTargetIds: [authority.targetId]
+    })
+    if (!deps.isPreparationTokenCurrent(token)) {
+      return
+    }
+    const result = results.find((entry) => entry.targetId === authority.targetId)?.result
+    applyPatchStatus(deps.store.getState(), authority.targetId, result)
+  }
+
+  const applyPreparedSnapshot = async (
+    targetId: string,
+    snapshot: RemoteWorkspaceSnapshot
+  ): Promise<DirectSshSnapshotPlacement> => {
+    const arrival = beginArrival(targetId)
+    const authority = deps.getCurrentAuthority(targetId)
+    if (!authority) {
+      return 'not-applied'
+    }
+    const input = await deps.capturePreparationInput(
+      authority,
+      'workspace-snapshot',
+      snapshot.revision
+    )
+    if (!input || !isArrivalCurrent(targetId, arrival)) {
+      return 'not-applied'
+    }
+    const prepared = await deps.prepareOnly(input)
+    if (!prepared.token || !isArrivalCurrent(targetId, arrival)) {
+      return 'not-applied'
+    }
+    const applyToken = buildDirectSshSnapshotApplyToken(prepared.token, snapshot.revision)
+    if (!applyToken) {
+      return 'not-applied'
+    }
+    const placement = await applyDirectSshRemoteWorkspaceSnapshot({
+      store: deps.store,
+      snapshot,
+      token: applyToken,
+      arrival,
+      isArrivalCurrent,
+      isPreparationTokenCurrent: deps.isPreparationTokenCurrent,
+      waitForWorkspaceSessionReady,
+      finalizeHydratedTerminals: deps.finalizeHydratedTerminals
+    })
+    return placement
+  }
+
+  /** Public entry: a snapshot arriving unsolicited starts its own re-pull chain. */
+  const applyUnsolicitedSnapshot = async (
+    targetId: string,
+    snapshot: RemoteWorkspaceSnapshot
+  ): Promise<DirectSshSnapshotPlacement> => {
+    const placement = await applyPreparedSnapshot(targetId, snapshot)
+    repull.schedule(targetId, placement)
+    return placement
+  }
+
+  return {
+    syncAfterConnect,
+    applyUnsolicitedSnapshot,
+    stop: () => {
+      stopped = true
+      arrivalByTarget.clear()
+      repull.stop()
+    }
+  }
+}