merge: PR #19572 (ssh host partition adoption) into integration stack

Conflict: config/reliability-gates.jsonc - both sides appended a new gate entry at the same array position; kept both.
2026-09-22 00:02:31 +00:00 · 2026-09-10 15:03:11 -07:00
parent eb2f2d52ae 4b0d984193
commit 1aa426aa48
21 changed files with 1690 additions and 209 deletions
@@ -104,6 +104,127 @@
      "demotionRule": "Keep experimental until soak and platform evidence support promotion; do not relax value or identity assertions to hide failures."
    },

+    {
+      "id": "workspace-session.ssh-host-partition-round-trip",
+      "title": "An SSH workspace round-trips through its own partition without losing tabs, editor files or browser state",
+      "maturity": "experimental",
+      "protection": "partial",
+      "owner": "workspace-session-persistence",
+      "layer": "persistence-integration",
+      "surfaces": [
+        "workspace session partitions",
+        "direct SSH remote workspace sync",
+        "boot session hydration"
+      ],
+      "platforms": ["macos", "linux", "windows"],
+      "providers": ["local", "ssh"],
+      "coveredPlatforms": ["macos"],
+      "coveredProviders": ["local", "ssh"],
+      "coverageNotes": "Drives the real Store against a temp profile, the real remoteWorkspace:setForConnectedTargets handler, the real session projection and the real pull-side merge. The relay transport is faked at the multiplexer boundary, so no live SSH host or relay is exercised; the partition routing and projection code under test is platform-independent. Runtime (orca environment) partitions, PTY lifecycle and mobile rendering are unaffected.",
+      "motivatingLinks": [
+        "https://github.com/stablyai/orca/issues/12721",
+        "https://github.com/stablyai/orca/issues/18173",
+        "https://github.com/stablyai/orca/blob/main/src/shared/workspace-session-partition-owner.ts"
+      ],
+      "invariant": "Every workspace the `ssh:<targetId>` partition names is hydrated at boot and published to the host, whatever kind of state it holds - terminal tabs, open editor files with unsaved hot-exit drafts, browser workspaces, tab groups, or host-qualified visit recency. An empty tab row is read as a gap, never as evidence the tabs were closed, so a replace-session upload can never delete a populated host list. A workspace the local partition already holds terminal tabs for is left untouched, and every workspace routes back to the partition that owns it. Adoption never destroys what it was not told about: a host row the partition holds nothing for cannot replace a populated base row, and a bare id the read found contested is gap-filled rather than replaced, because that is exactly the id where `local` and `ssh:<targetId>` are not one workspace written twice. Tab-, pane- and file-keyed rows are recovered through the same indexes the split routed them by. An explicit empty tab row written by this build survives the round trip as a present empty row and is declined by the shared seeding predicate.",
+      "oracle": "Seed a real Store the way shipping builds leave it: the local blob holds the worktree key with an empty list while `ssh:<targetId>` holds the real one, with a second populated SSH partition present. Publish through the IPC handler with no session argument (the path the debounced writer takes) and assert the host snapshot carries the runtime-authored tabs rather than []. Separately drive the shipping split (buildWorkspaceSessionHostSnapshots) and feed its own output back through the real boot read, pinning the write and read halves to each other rather than to a hand-built fixture: an SSH workspace with open editor files, an unsaved dirtyDraftContent and no terminal tabs must come back intact, as must one with no tabsByWorktree key at all. Export and re-import through the real projection and merge through mergeDirectSshRemoteWorkspaceSession, asserting tabs survive a publish, the next pull, and an older client publishing an empty list for them. Assert the reunited workspace routes to `ssh:<targetId>`, that a workspace the base holds tabs for is not modified, and that a contested id claimed by an SSH and a runtime host does not send the SSH rows into the rotating runtime partition.",
+      "commands": [
+        "pnpm exec vitest run --config config/vitest.config.ts src/main/ipc/ssh-host-partition-session-export.test.ts src/renderer/src/lib/workspace-session-ssh-partition-round-trip.test.ts src/renderer/src/lib/workspace-session-host-contention.test.ts src/shared/workspace-session-partition-owner.test.ts"
+      ],
+      "testFiles": [
+        "src/main/ipc/ssh-host-partition-session-export.test.ts",
+        "src/renderer/src/lib/workspace-session-ssh-partition-round-trip.test.ts",
+        "src/renderer/src/lib/workspace-session-host-contention.test.ts",
+        "src/shared/workspace-session-partition-owner.test.ts"
+      ],
+      "assertionRefs": [
+        {
+          "file": "src/main/ipc/ssh-host-partition-session-export.test.ts",
+          "assertions": [
+            "publishes tabs the runtime persisted into the target ssh partition",
+            "never replaces the host snapshot with an empty list for a worktree that has tabs"
+          ]
+        },
+        {
+          "file": "src/renderer/src/lib/workspace-session-ssh-partition-round-trip.test.ts",
+          "assertions": [
+            "hydrates tabs the runtime persisted into the ssh partition",
+            "adopts the stranded workspace rows alongside its tabs",
+            "adopts a hibernated agent record the host partition alone holds",
+            "leaves a workspace the local partition already holds tabs for untouched",
+            "routes the reunited workspace back to the partition that owns it",
+            "does not delete the worktree tabs across a publish and the next pull",
+            "publishes the stranded tabs rather than an empty list",
+            "does not let an empty host row destroy an unsaved draft the base alone holds",
+            "still adopts a populated host row over the base leftovers",
+            "adopts the layout of a tab the host slice names only in unifiedTabs",
+            "does not overwrite a contested workspace's own rows with the ssh workspace's",
+            "still fills a gap on a contested id",
+            "does not let a legacy bare recency key move another host workspace of the same id",
+            "still adopts a host-qualified recency key, which names its own owner",
+            "writes an SSH workspace emptied by this build into the partition that owns it",
+            "restores that tombstone as an explicit empty row, not a deleted key",
+            "leaves the restored workspace un-seeded by the shared seeding predicate",
+            "does not adopt a stale populated ssh row over a tombstone in the owning partition",
+            "resurrects the legacy-transition shape exactly once and not again",
+            "publishes the tombstone rather than a row the host can read as unknown"
+          ]
+        },
+        {
+          "file": "src/renderer/src/lib/workspace-session-host-contention.test.ts",
+          "assertions": [
+            "keeps an SSH claimant out of the rotating runtime partition",
+            "does not strand the runtime co-claimant when the SSH row is written"
+          ]
+        },
+        {
+          "file": "src/shared/workspace-session-partition-owner.test.ts",
+          "assertions": [
+            "gives an SSH host its own partition, matching what the runtime already writes"
+          ]
+        }
+      ],
+      "evidenceRuns": [
+        {
+          "date": "2026-09-10",
+          "runner": "local",
+          "platform": "macos",
+          "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/ipc/ssh-host-partition-session-export.test.ts src/renderer/src/lib/workspace-session-ssh-partition-round-trip.test.ts src/renderer/src/lib/workspace-session-host-contention.test.ts src/shared/workspace-session-partition-owner.test.ts",
+          "result": "passed",
+          "durationSeconds": 1.04,
+          "summary": "56 tests passed across 4 files, including the real Store publish, the shipping write/read split round trip, the older-client empty-publish skew direction, the contested-bare-id and empty-host-row guards, and the closed-last-terminal tombstone boundary."
+        }
+      ],
+      "runtimeBudget": {
+        "p95Seconds": 30,
+        "scope": "Real Store on a temp profile plus renderer partition units; no launched app and no relay."
+      },
+      "flakeHistory": {
+        "status": "not-started",
+        "evidence": "Deterministic local validation only; no CI soak yet. The tests have no timers, network or real relay."
+      },
+      "redGreenEvidence": {
+        "status": "complete",
+        "evidence": "Against pristine main 12f53da542d, 9 of the 11 assertions that target the original defect fail: the boot read returns no tabs, the export publishes an empty tabsByWorktreePath row, the round trip ends with the tabs deleted, routing answers local instead of ssh:target-1, and an older client's empty publish deletes the tabs on pull. The remaining assertions guard the fix itself rather than main's bug: review found that gating adoption on terminal tabs stranded editor-only and browser-only SSH workspaces, destroying unsaved hot-exit drafts no other channel can recover. The gate now discovers a workspace through an exhaustive switch over the field-ownership table, and reintroducing the tabs-only gate fails 6 assertions including the write/read round trip."
+      },
+      "performanceBudget": {
+        "required": false,
+        "evidence": "Boot adds one session:get per SSH host that owns a repo, issued in parallel with the existing runtime partition reads and served from already-loaded main state. The publish fallback adds one partition read and a shallow keyed-record merge per target, which returns its input unchanged when nothing is stranded and stays the same order as the projection it feeds."
+      },
+      "knownGaps": [
+        "No live SSH host or relay is exercised; the multiplexer is faked at the request boundary.",
+        "Folder workspaces still persist to the local partition, so their half of the writer divergence in #12723 is not covered here.",
+        "Rows stranded beside a workspace the local partition already holds terminal tabs for are deliberately not recovered, and no assertion claims they are.",
+        "One-shot resurrection in the legacy-transition shape: where an older build left an empty local row while the runtime partition still holds that workspace's tabs, boot adopts them back once. Non-destructive, and non-recurrence is now asserted rather than argued - a workspace this build empties writes its empty row to the owning partition and leaves no local row behind, so there is nothing left to resurrect from.",
+        "A contested bare id is gap-filled rather than replaced, so rows stranded beside a contested workspace stay stranded. Separating them needs host-qualified keys through the tab store, which is the same open gap `workspace-session-host-contention.ts` records.",
+        "`lastVisitedAtByWorktreeId` still discovers a workspace as adoptable from a recency entry alone. Harmless now that a recency row cannot replace another host's entry for the same bare id, but it means recency is a discovery trigger and no other field of its kind is."
+      ],
+      "promotionCriteria": [
+        "Complete the CI soak requirement with no unexplained flakes.",
+        "Add coverage for a live SSH target before claiming the ssh provider is exercised end to end."
+      ],
+      "demotionRule": "Keep experimental until CI soak completes. Investigate any failure without weakening the empty-row-is-a-gap oracle, which is the assertion the data-loss fix rests on."
+    },
    {
      "id": "agent-session.history-forward-read-budget",
      "title": "Journal catch-up reads only the next page and one lookahead row",
@@ -0,0 +1,86 @@
+import type { Store } from '../persistence'
+import type { Repo } from '../../shared/repo-types'
+import { exportRemoteWorkspaceSession } from '../../shared/remote-workspace-session-projection'
+import type { RemoteWorkspaceSession } from '../../shared/remote-workspace-types'
+import type { WorkspaceSessionState } from '../../shared/workspace-session-state-types'
+import { toSshExecutionHostId } from '../../shared/execution-host'
+import { adoptStrandedHostPartitionSession } from '../../shared/workspace-session-stranded-partition-adoption'
+import { getRepoIdFromWorktreeId } from '../../shared/worktree/id'
+import {
+  resolveWorktreeExecutionHost,
+  type createRepoRowExecutionHostLookup
+} from '../../shared/worktree-execution-host-resolution'
+
+type RepoRowLookup = ReturnType<typeof createRepoRowExecutionHostLookup<Repo>>
+
+/** Which target a workspace session is exported to. */
+export type WorktreeTargetResolver = (worktreeId: string, executionHostId?: string) => string | null
+
+function targetForWorktree(
+  repoLookup: RepoRowLookup,
+  worktreeId: string,
+  executionHostId?: string
+): string | null {
+  // Why: this decides which SSH target a workspace session is exported to. The old fallback read
+  // `getRepo(id)?.connectionId`, which is host-blind — the same repo id can name rows on several
+  // hosts, so a session could be published to a machine that never owned the worktree (#11163).
+  // Unresolvable ownership exports to nobody rather than guessing.
+  const resolution = resolveWorktreeExecutionHost(repoLookup, {
+    repoId: getRepoIdFromWorktreeId(worktreeId),
+    hostId: executionHostId ?? null
+  })
+  return resolution.kind === 'resolved' ? resolution.connectionId : null
+}
+
+/**
+ * Resolve each worktree's owning connection at most once for a whole publish.
+ *
+ * Why this is shared and not per target: `targetForWorktree` computes a connection id from the
+ * repo catalog alone — only the final `=== targetId` differs — so exporting to N targets used to
+ * repeat the identical resolution N times over every worktree key. `store.getRepos()` also
+ * re-hydrates every repo row on each call, and the projection asks this question once per key of
+ * `tabsByWorktree`, `activeTabIdByWorktree`, `lastVisitedAtByWorktreeId` and
+ * `defaultTerminalTabsAppliedByWorktreeId`.
+ */
+export function createWorktreeTargetResolver(repoLookup: RepoRowLookup): WorktreeTargetResolver {
+  const resolved = new Map<string, string | null>()
+  return (worktreeId, executionHostId) => {
+    // Host id participates in resolution, so it has to participate in the key. NUL cannot appear
+    // in either id, so it is a collision-free separator.
+    const key = `${worktreeId}\u0000${executionHostId ?? ''}`
+    const cached = resolved.get(key)
+    if (cached !== undefined) {
+      return cached
+    }
+    const connectionId = targetForWorktree(repoLookup, worktreeId, executionHostId)
+    resolved.set(key, connectionId)
+    return connectionId
+  }
+}
+
+export function exportSessionForTarget(
+  resolveWorktreeTarget: WorktreeTargetResolver,
+  targetId: string,
+  session: WorkspaceSessionState
+): RemoteWorkspaceSession {
+  return exportRemoteWorkspaceSession(session, {
+    isTargetWorktree: (worktreeId, executionHostId) =>
+      resolveWorktreeTarget(worktreeId, executionHostId) === targetId
+  })
+}
+
+/**
+ * The persisted session a publish speaks for when the renderer sent none.
+ *
+ * Why not `store.getWorkspaceSession()` alone: that reads the 'local' blob, and a target's
+ * worktrees live in `ssh:<targetId>` (#12723). Publishing the local half as though it were the
+ * whole session uploaded explicit empty tab lists, and `replace-session` turned that absence into
+ * deletion on the host (#12721). Resolved per target so one target's rows can never be published
+ * under another's key when both partitions hold the same worktree id.
+ */
+export function persistedSessionForTarget(store: Store, targetId: string): WorkspaceSessionState {
+  return adoptStrandedHostPartitionSession(
+    store.getWorkspaceSession(),
+    store.getWorkspaceSession(toSshExecutionHostId(targetId))
+  )
+}
@@ -1,22 +1,20 @@
 import { ipcMain, type BrowserWindow } from 'electron'
 import type { Store } from '../persistence'
-import type { Repo } from '../../shared/repo-types'
 import { getActiveMultiplexer, getSshConnectionStore } from './ssh'
-import { exportRemoteWorkspaceSession } from '../../shared/remote-workspace-session-projection'
 import {
  REMOTE_WORKSPACE_CHANGED_NOTIFICATION,
  REMOTE_WORKSPACE_STALE_NOTIFICATION,
  type RemoteWorkspaceChangedEvent,
  type RemoteWorkspaceObservedPatchResult,
-  type RemoteWorkspaceObservedSnapshot,
-  type RemoteWorkspaceSession
+  type RemoteWorkspaceObservedSnapshot
 } from '../../shared/remote-workspace-types'
 import type { WorkspaceSessionState } from '../../shared/workspace-session-state-types'
-import { getRepoIdFromWorktreeId } from '../../shared/worktree/id'
+import { createRepoRowExecutionHostLookup } from '../../shared/worktree-execution-host-resolution'
 import {
-  createRepoRowExecutionHostLookup,
-  resolveWorktreeExecutionHost
-} from '../../shared/worktree-execution-host-resolution'
+  createWorktreeTargetResolver,
+  exportSessionForTarget,
+  persistedSessionForTarget
+} from './remote-workspace-target-session-export'
 import { getRemoteWorkspaceNamespace } from './remote-workspace-namespace'
 import { registerRemoteWorkspaceNotificationHandler } from './remote-workspace-events'
 import { CLIENT_ID } from './remote-workspace-client-identity'
@@ -107,61 +105,6 @@ function getExpectedHostObservationTokens(
  return tokens
 }

-function targetForWorktree(
-  repoLookup: ReturnType<typeof createRepoRowExecutionHostLookup<Repo>>,
-  worktreeId: string,
-  executionHostId?: string
-): string | null {
-  // Why: this decides which SSH target a workspace session is exported to. The old fallback read
-  // `getRepo(id)?.connectionId`, which is host-blind — the same repo id can name rows on several
-  // hosts, so a session could be published to a machine that never owned the worktree (#11163).
-  // Unresolvable ownership exports to nobody rather than guessing.
-  const resolution = resolveWorktreeExecutionHost(repoLookup, {
-    repoId: getRepoIdFromWorktreeId(worktreeId),
-    hostId: executionHostId ?? null
-  })
-  return resolution.kind === 'resolved' ? resolution.connectionId : null
-}
-
-/**
- * Resolve each worktree's owning connection at most once for a whole publish.
- *
- * Why this is shared and not per target: `targetForWorktree` computes a connection id from the
- * repo catalog alone — only the final `=== targetId` differs — so exporting to N targets used to
- * repeat the identical resolution N times over every worktree key. `store.getRepos()` also
- * re-hydrates every repo row on each call, and the projection asks this question once per key of
- * `tabsByWorktree`, `activeTabIdByWorktree`, `lastVisitedAtByWorktreeId` and
- * `defaultTerminalTabsAppliedByWorktreeId`.
- */
-function createWorktreeTargetResolver(
-  repoLookup: ReturnType<typeof createRepoRowExecutionHostLookup<Repo>>
-): (worktreeId: string, executionHostId?: string) => string | null {
-  const resolved = new Map<string, string | null>()
-  return (worktreeId, executionHostId) => {
-    // Host id participates in resolution, so it has to participate in the key. NUL cannot appear
-    // in either id, so it is a collision-free separator.
-    const key = `${worktreeId}\u0000${executionHostId ?? ''}`
-    const cached = resolved.get(key)
-    if (cached !== undefined) {
-      return cached
-    }
-    const connectionId = targetForWorktree(repoLookup, worktreeId, executionHostId)
-    resolved.set(key, connectionId)
-    return connectionId
-  }
-}
-
-function exportSessionForTarget(
-  resolveWorktreeTarget: (worktreeId: string, executionHostId?: string) => string | null,
-  targetId: string,
-  session: WorkspaceSessionState
-): RemoteWorkspaceSession {
-  return exportRemoteWorkspaceSession(session, {
-    isTargetWorktree: (worktreeId, executionHostId) =>
-      resolveWorktreeTarget(worktreeId, executionHostId) === targetId
-  })
-}
-
 function sendRemoteWorkspaceChanged(
  targetId: string,
  snapshot: RemoteWorkspaceObservedSnapshot,
@@ -279,7 +222,6 @@ export function registerRemoteWorkspaceHandlers(
        return []
      }

-      const workspaceSession = args.session ?? store.getWorkspaceSession()
      // One repo read, and ownership resolutions shared across targets: neither depends on the target.
      const resolveWorktreeTarget = createWorktreeTargetResolver(
        createRepoRowExecutionHostLookup(store.getRepos())
@@ -288,7 +230,11 @@ export function registerRemoteWorkspaceHandlers(
        targets.map(async (target) => {
          // Why: each target has its own revision stream. Keep same-target
          // writes queued, but do not let one slow relay block others.
-          const session = exportSessionForTarget(resolveWorktreeTarget, target.id, workspaceSession)
+          const session = exportSessionForTarget(
+            resolveWorktreeTarget,
+            target.id,
+            args.session ?? persistedSessionForTarget(store, target.id)
+          )
          const result = await queueRemoteWorkspacePatch(target.id, async () => {
            const current =
              getCachedRemoteWorkspaceSnapshot(target.id) ?? (await getRemoteSnapshot(target))
@@ -0,0 +1,233 @@
+/**
+ * What the remote-workspace export publishes when the renderer omits `session`, against the real
+ * `Store`.
+ *
+ * The shipping debounced writer takes that fallback on every session write, and it used to read the
+ * 'local' blob alone — so an SSH worktree whose tabs the main-process runtime had written to
+ * `ssh:<targetId>` was projected as an explicit empty tab list. The upload is a
+ * `replace-session` patch, which turns that absence into deletion on the host (#12721, #18173).
+ *
+ * Drives the real `Store` rather than a `getWorkspaceSession` fake: the whole defect is which
+ * partition the read reaches, and a fake answers whatever the test tells it to.
+ */
+import { mkdtempSync, realpathSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type { Repo } from '../../shared/repo-types'
+import type {
+  RemoteWorkspaceSession,
+  RemoteWorkspaceSnapshot
+} from '../../shared/remote-workspace-types'
+import type { SshTarget } from '../../shared/ssh-types'
+import type { TerminalTab } from '../../shared/terminal-tab-types'
+
+const { getActiveMultiplexerMock, getSshConnectionStoreMock } = vi.hoisted(() => ({
+  getActiveMultiplexerMock: vi.fn(),
+  getSshConnectionStoreMock: vi.fn()
+}))
+
+const ipcHandlers = new Map<string, (event: unknown, ...args: unknown[]) => unknown>()
+
+vi.mock('electron', () => ({
+  app: {
+    getPath: () => tmpdir(),
+    getName: () => 'orca-test',
+    getVersion: () => '0.0.0-test',
+    isPackaged: false,
+    on: () => {},
+    whenReady: () => Promise.resolve()
+  },
+  safeStorage: {
+    isEncryptionAvailable: () => false,
+    encryptString: (value: string) => Buffer.from(value),
+    decryptString: (value: Buffer) => value.toString()
+  },
+  ipcMain: {
+    on: () => {},
+    handle: (channel: string, handler: (event: unknown, ...args: unknown[]) => unknown) => {
+      ipcHandlers.set(channel, handler)
+    },
+    removeHandler: (channel: string) => {
+      ipcHandlers.delete(channel)
+    }
+  },
+  BrowserWindow: { getAllWindows: () => [] }
+}))
+
+vi.mock('./ssh', () => ({
+  getActiveMultiplexer: getActiveMultiplexerMock,
+  getSshConnectionStore: getSshConnectionStoreMock
+}))
+
+vi.mock('./remote-workspace-events', () => ({
+  registerRemoteWorkspaceNotificationHandler: () => () => {}
+}))
+
+const { Store } = await import('../persistence/loading-store/store')
+const { getDefaultWorkspaceSession } = await import('../../shared/constants')
+const { _resetRemoteWorkspaceCachesForTests, registerRemoteWorkspaceHandlers } =
+  await import('./remote-workspace')
+
+const TARGET_ID = 'target-1'
+const SSH_HOST_ID = `ssh:${TARGET_ID}` as const
+const REPO_ID = 'repo-remote'
+const WORKTREE_PATH = '/remote/checkout/feature'
+const WORKTREE_ID = `${REPO_ID}::${WORKTREE_PATH}`
+
+const OTHER_TARGET_ID = 'target-2'
+const OTHER_SSH_HOST_ID = `ssh:${OTHER_TARGET_ID}` as const
+const OTHER_REPO_ID = 'repo-other'
+const OTHER_WORKTREE_ID = `${OTHER_REPO_ID}::/elsewhere/checkout/main`
+
+function sshTarget(id: string, host: string): SshTarget {
+  return { id, label: id, host, port: 22, username: 'alice' }
+}
+
+const target = sshTarget(TARGET_ID, 'one.example.com')
+const otherTarget = sshTarget(OTHER_TARGET_ID, 'two.example.com')
+
+function remoteRepo(id: string, path: string, connectionId: string): Repo {
+  return { id, path, displayName: id, badgeColor: 'blue', addedAt: 1, connectionId } as Repo
+}
+
+function runtimeAuthoredTab(): TerminalTab {
+  return {
+    id: 'tab-runtime',
+    ptyId: 'pty-runtime',
+    worktreeId: WORKTREE_ID,
+    title: 'claude',
+    customTitle: null,
+    color: null,
+    sortOrder: 0,
+    createdAt: 1
+  }
+}
+
+const stores: InstanceType<typeof Store>[] = []
+let hostSnapshot: RemoteWorkspaceSnapshot
+
+afterEach(() => {
+  for (const store of stores.splice(0)) {
+    store.flush()
+  }
+  vi.restoreAllMocks()
+})
+
+beforeEach(() => {
+  _resetRemoteWorkspaceCachesForTests()
+  ipcHandlers.clear()
+  hostSnapshot = {
+    namespace: 'ns-target-1',
+    revision: 4,
+    updatedAt: 100,
+    schemaVersion: 1,
+    session: {
+      activeWorktreePath: null,
+      activeTabId: null,
+      tabsByWorktreePath: {},
+      terminalLayoutsByTabId: {}
+    }
+  }
+  getSshConnectionStoreMock.mockReset()
+  getSshConnectionStoreMock.mockReturnValue({
+    listTargets: () => [target, otherTarget],
+    getTarget: (targetId: string) =>
+      [target, otherTarget].find((candidate) => candidate.id === targetId)
+  })
+  getActiveMultiplexerMock.mockReset()
+  getActiveMultiplexerMock.mockImplementation((targetId: string) =>
+    targetId === TARGET_ID
+      ? {
+          request: (method: string, params: Record<string, unknown>) => {
+            if (method === 'workspace.get') {
+              return Promise.resolve(hostSnapshot)
+            }
+            const patch = params.patch as { session: RemoteWorkspaceSession }
+            hostSnapshot = {
+              ...hostSnapshot,
+              revision: hostSnapshot.revision + 1,
+              session: patch.session
+            }
+            return Promise.resolve({ ok: true, snapshot: hostSnapshot })
+          }
+        }
+      : undefined
+  )
+})
+
+/** The observed shape from #12721: the runtime owns the tab list in `ssh:<targetId>` while the
+ *  local blob still carries the worktree key with an empty list. */
+function createStrandedStore(): InstanceType<typeof Store> {
+  const dir = realpathSync(mkdtempSync(join(tmpdir(), 'orca-ssh-partition-export-')))
+  const store = new Store({ dataFile: join(dir, 'orca-data.json') })
+  stores.push(store)
+  store.addRepo(remoteRepo(REPO_ID, '/remote/checkout', TARGET_ID))
+  // A second populated SSH partition: the fallback has to reach the publishing target's own
+  // partition, not merely "some" partition that happens to hold tabs.
+  store.addRepo(remoteRepo(OTHER_REPO_ID, '/elsewhere/checkout', OTHER_TARGET_ID))
+  store.setWorkspaceSession({
+    ...getDefaultWorkspaceSession(),
+    tabsByWorktree: { [WORKTREE_ID]: [] }
+  })
+  store.setWorkspaceSession(
+    {
+      ...getDefaultWorkspaceSession(),
+      tabsByWorktree: { [WORKTREE_ID]: [runtimeAuthoredTab()] }
+    },
+    SSH_HOST_ID
+  )
+  store.setWorkspaceSession(
+    {
+      ...getDefaultWorkspaceSession(),
+      tabsByWorktree: {
+        [OTHER_WORKTREE_ID]: [
+          { ...runtimeAuthoredTab(), id: 'tab-other', worktreeId: OTHER_WORKTREE_ID }
+        ]
+      }
+    },
+    OTHER_SSH_HOST_ID
+  )
+  return store
+}
+
+async function publishToConnectedTarget(store: InstanceType<typeof Store>): Promise<void> {
+  registerRemoteWorkspaceHandlers(store, () => null)
+  const get = ipcHandlers.get('remoteWorkspace:get')
+  const set = ipcHandlers.get('remoteWorkspace:setForConnectedTargets')
+  if (!get || !set) {
+    throw new Error('remote workspace handlers were never registered')
+  }
+  const observed = (await get(null, { targetId: TARGET_ID })) as {
+    revision: number
+    hostObservationToken: string
+  }
+  await set(null, {
+    // The shipping debounced writer omits `session` and relies on the main-side fallback.
+    hydratedTargetIds: [TARGET_ID],
+    expectedRevisionsByTargetId: { [TARGET_ID]: observed.revision },
+    expectedHostObservationTokensByTargetId: { [TARGET_ID]: observed.hostObservationToken }
+  })
+}
+
+describe('remoteWorkspace:setForConnectedTargets session fallback', () => {
+  it('publishes tabs the runtime persisted into the target ssh partition', async () => {
+    const store = createStrandedStore()
+
+    await publishToConnectedTarget(store)
+
+    expect(hostSnapshot.session.tabsByWorktreePath[WORKTREE_PATH]?.map((tab) => tab.id)).toEqual([
+      'tab-runtime'
+    ])
+  })
+
+  it('never replaces the host snapshot with an empty list for a worktree that has tabs', async () => {
+    // The deletion step itself: `replace-session` makes an exported empty list authoritative, so
+    // publishing one for a populated worktree is what destroyed the host's copy on every launch.
+    const store = createStrandedStore()
+
+    await publishToConnectedTarget(store)
+
+    expect(hostSnapshot.session.tabsByWorktreePath[WORKTREE_PATH]).not.toEqual([])
+  })
+})
@@ -4,6 +4,7 @@ import {
  parseExecutionHostId,
  type ExecutionHostId
 } from '../../../shared/execution-host'
+import { workspaceSessionPartitionHostId } from '../../../shared/workspace-session-partition-owner'
 import { cloneWorkspaceSessionState, deleteOwnerKeyedSessionFields } from './session-owner-fields'

 // Scans the pane-key-keyed maps and the shutdown list once, removing every entry
@@ -54,11 +55,7 @@ export function workspaceSessionPartitionIdsForHost(
 }

 /** The partition the host actually owns; the others are only spill surfaces for it. */
-export function workspaceSessionOwnerPartitionForHost(
-  hostId: string | null | undefined
-): ExecutionHostId {
-  return parseExecutionHostId(hostId)?.id ?? LOCAL_EXECUTION_HOST_ID
-}
+export const workspaceSessionOwnerPartitionForHost = workspaceSessionPartitionHostId

 export function removeWorkspaceSessionOwner(
  session: WorkspaceSessionState | undefined,
@@ -48,10 +48,8 @@ export class RuntimeWorkspaceSessionController {
    }
    const resolvedWorktreeId = scope?.type === 'worktree' ? scope.worktreeId : worktreeId
    const repo = store?.getRepo?.(getRepoIdFromWorktreeId(resolvedWorktreeId))
-    // Why: SSH worktrees keep their own `ssh:<targetId>` partition here while the renderer writes
-    // them to 'local'; the shared owner map records that divergence (#12723).
    return repo
-      ? workspaceSessionPartitionHostId(getRepoExecutionHostId(repo), 'host-partition')
+      ? workspaceSessionPartitionHostId(getRepoExecutionHostId(repo))
      : LOCAL_EXECUTION_HOST_ID
  }

@@ -330,10 +330,10 @@ describe('read-time primary is the one the write path honours', () => {
    }
  }

-  it('keeps an SSH claimant in the local partition it actually persists in', () => {
-    // Why this shape: the claims catalog sorts `runtime:` before `ssh:`, so picking a primary from
-    // claimants sent the SSH workspace's rows into the runtime partition.
-    expect(buildHostIdByWorktreeId(sshVersusRuntimeState())(SHARED_ID)).toBe('local')
+  it('keeps an SSH claimant out of the rotating runtime partition', () => {
+    // Why this shape: the claims catalog sorts `runtime:` before `ssh:`, so a plain sort sent the
+    // SSH workspace's rows into the runtime partition. It now persists in its own.
+    expect(buildHostIdByWorktreeId(sshVersusRuntimeState())(SHARED_ID)).toBe(SSH_HOST)
  })

  it('does not strand the runtime co-claimant when the SSH row is written', async () => {
@@ -355,8 +355,8 @@ describe('read-time primary is the one the write path honours', () => {
    expect(runtimeWrite?.tabsByWorktree[SHARED_ID]?.map((entry) => entry.id)).toEqual([
      'runtime-tab'
    ])
-    const localWrite = set.mock.calls.find(([, hostId]) => hostId === undefined)?.[0]
-    expect(localWrite?.tabsByWorktree[SHARED_ID]?.map((entry) => entry.id)).toEqual(['ssh-tab'])
+    const sshWrite = set.mock.calls.find(([, hostId]) => hostId === SSH_HOST)?.[0]
+    expect(sshWrite?.tabsByWorktree[SHARED_ID]?.map((entry) => entry.id)).toEqual(['ssh-tab'])
  })

  it('writes a row back to the only partition that had it instead of copying it', async () => {
@@ -5,16 +5,13 @@ import {
  toRuntimeExecutionHostId,
  type ExecutionHostId
 } from '../../../shared/execution-host'
-import { parseWorkspaceKey } from '../../../shared/workspace-scope'
-import {
-  getWorktreeIdFromHostIdentity,
-  isWorktreeHostIdentity
-} from '../../../shared/worktree/host-qualified-identity'
+import { normalizeWorkspaceSessionKeyToWorkspaceId } from '../../../shared/workspace-scope'
 import { WORKSPACE_SESSION_FIELD_OWNERSHIP } from '../../../shared/workspace-session-host-field-ownership'
+import { workspaceSessionPartitionHostId } from '../../../shared/workspace-session-partition-owner'
 import {
  isWorkspaceSessionRecord,
  type WorkspaceSessionRecord
-} from './workspace-session-host-records'
+} from '../../../shared/workspace-session-host-records'
 import type { WorkspaceRuntimeOwnerProjection } from './workspace-runtime-host-ownership'
 import {
  mergeWorkspaceSessionsFromHosts,
@@ -40,10 +37,8 @@ import {
 * would let the two disagree — the catalog names `ssh:*` hosts that own no partition — and the
 * write would then copy one host's workspace into another host's partition.
 *
- * Known gaps: hosts that share a partition cannot be separated at all ('local' and every `ssh:*`
- * host persist into the 'local' blob), and the unified renderer session still holds one bucket per
- * bare id, so both workspaces display the primary's tabs. Closing either needs host-qualified keys
- * through the whole tab store.
+ * Known gap: the unified renderer session still holds one bucket per bare id, so both workspaces
+ * display the primary's tabs. Closing it needs host-qualified keys through the whole tab store.
 */

 export type WorktreeHostClaims = ReadonlyMap<string, ReadonlySet<ExecutionHostId>>
@@ -52,14 +47,9 @@ const WORKTREE_KEYED_FIELDS = (
  Object.keys(WORKSPACE_SESSION_FIELD_OWNERSHIP) as (keyof WorkspaceSessionState)[]
 ).filter((field) => WORKSPACE_SESSION_FIELD_OWNERSHIP[field] === 'worktreeKeyed')

-/** Bare worktree id behind a session key, which may be a WorkspaceKey or a host-qualified identity. */
-export function normalizeWorkspaceSessionKeyToWorktreeId(value: string): string {
-  if (isWorktreeHostIdentity(value)) {
-    return getWorktreeIdFromHostIdentity(value)
-  }
-  const scope = parseWorkspaceKey(value)
-  return scope?.type === 'worktree' ? scope.worktreeId : value
-}
+/** Bare worktree id behind a session key. Lives in shared because the partition adoption read needs
+ *  the same normalization, and two implementations of it would drift. */
+export const normalizeWorkspaceSessionKeyToWorktreeId = normalizeWorkspaceSessionKeyToWorkspaceId

 function resolveClaimedHostId(
  worktree: WorkspaceRuntimeOwnerProjection,
@@ -100,10 +90,9 @@ export function indexWorktreeHostClaims(
  return claims
 }

-/** The partition a host's session rows live in: a runtime host owns one, while 'local' and every
- *  `ssh:*` host share the 'local' blob. */
+/** The partition a host's session rows live in: every non-'local' host owns its own. */
 export function sessionPartitionHostFor(hostId: ExecutionHostId): ExecutionHostId {
-  return parseExecutionHostId(hostId)?.kind === 'runtime' ? hostId : LOCAL_EXECUTION_HOST_ID
+  return workspaceSessionPartitionHostId(hostId)
 }

 /** Distinct partitions a set of claimants spans. Fewer than two means persistence cannot tell the
@@ -112,14 +101,20 @@ export function contestedPartitionHosts(claimed: Iterable<ExecutionHostId>): Exe
  return [...new Set([...claimed].map(sessionPartitionHostFor))]
 }

-/** Stable owner of a contested id: 'local' when it is a claimant, else the lowest host id.
+/** Stable owner of a contested id: 'local' when it is a claimant, then any non-runtime host, then
+ *  the lowest host id.
 *  Deliberately not the active host — a primary that followed navigation would migrate the same
- *  rows between partitions on every workspace switch. */
+ *  rows between partitions on every workspace switch. And deliberately not plain sort order once
+ *  'local' is out: a `runtime:` environment id rotates across relay restarts, so ranking it last
+ *  keeps a re-created environment from taking a stable host's rows into its partition. */
 export function pickPrimaryHostForClaims(hostIds: Iterable<ExecutionHostId>): ExecutionHostId {
  const sorted = [...hostIds].sort()
-  return sorted.includes(LOCAL_EXECUTION_HOST_ID)
-    ? LOCAL_EXECUTION_HOST_ID
-    : (sorted[0] ?? LOCAL_EXECUTION_HOST_ID)
+  return (
+    sorted.find((hostId) => hostId === LOCAL_EXECUTION_HOST_ID) ??
+    sorted.find((hostId) => parseExecutionHostId(hostId)?.kind !== 'runtime') ??
+    sorted[0] ??
+    LOCAL_EXECUTION_HOST_ID
+  )
 }

 function definedHostIds(slices: HostSessionSlices): ExecutionHostId[] {
@@ -188,21 +183,27 @@ function shadowHostEntries(
 *
 *  `primaryHostBySessionKey` records where each key's live row came from — including the
 *  uncontested single-partition case, so the write path can put every row back in its own
- *  partition instead of re-deriving an owner that may not match. */
+ *  partition instead of re-deriving an owner that may not match. It is therefore NOT the contested
+ *  set; `contestedSessionKeys` is, and only it says a bare id names more than one workspace. */
 export function extractContestedHostSessionEntries(slices: HostSessionSlices): {
  slices: HostSessionSlices
  shadow: HostSessionSlices
  primaryHostBySessionKey: Record<string, ExecutionHostId>
+  contestedSessionKeys: Set<string>
 } {
  const shadow: HostSessionSlices = {}
  const hostIds = definedHostIds(slices)
  const hostIdsByKey = indexHostIdsBySessionKey(slices, hostIds)
  const primaryHostBySessionKey: Record<string, ExecutionHostId> = {}
+  const contestedSessionKeys = new Set<string>()
  for (const [key, owners] of hostIdsByKey) {
    primaryHostBySessionKey[key] = pickPrimaryHostForClaims(owners)
+    if (owners.length > 1) {
+      contestedSessionKeys.add(key)
+    }
  }
  if (hostIds.length < 2) {
-    return { slices, shadow, primaryHostBySessionKey }
+    return { slices, shadow, primaryHostBySessionKey, contestedSessionKeys }
  }
  const primaryByKey = new Map<string, ExecutionHostId>()
  for (const [key, owners] of hostIdsByKey) {
@@ -211,7 +212,7 @@ export function extractContestedHostSessionEntries(slices: HostSessionSlices): {
    }
  }
  if (primaryByKey.size === 0) {
-    return { slices, shadow, primaryHostBySessionKey }
+    return { slices, shadow, primaryHostBySessionKey, contestedSessionKeys }
  }
  const next: HostSessionSlices = { ...slices }
  for (const hostId of hostIds) {
@@ -225,7 +226,7 @@ export function extractContestedHostSessionEntries(slices: HostSessionSlices): {
      shadow[hostId] = result.shadow
    }
  }
-  return { slices: next, shadow, primaryHostBySessionKey }
+  return { slices: next, shadow, primaryHostBySessionKey, contestedSessionKeys }
 }

 export function mergeWorkspaceSessionsWithHostShadow(slices: HostSessionSlices): {
@@ -233,13 +234,15 @@ export function mergeWorkspaceSessionsWithHostShadow(slices: HostSessionSlices):
  slices: HostSessionSlices
  shadow: HostSessionSlices
  primaryHostBySessionKey: Record<string, ExecutionHostId>
+  contestedSessionKeys: Set<string>
 } {
  const extracted = extractContestedHostSessionEntries(slices)
  return {
    session: mergeWorkspaceSessionsFromHosts(extracted.slices),
    slices: extracted.slices,
    shadow: extracted.shadow,
-    primaryHostBySessionKey: extracted.primaryHostBySessionKey
+    primaryHostBySessionKey: extracted.primaryHostBySessionKey,
+    contestedSessionKeys: extracted.contestedSessionKeys
  }
 }

@@ -6,6 +6,7 @@ import {
  parseExecutionHostId,
  type ExecutionHostId
 } from '../../../shared/execution-host'
+import { adoptStrandedHostPartitionSession } from '../../../shared/workspace-session-stranded-partition-adoption'
 import {
  mergeWorkspaceSessionsWithHostShadow,
  normalizeWorkspaceSessionKeyToWorktreeId
@@ -99,20 +100,36 @@ function buildRuntimeHostIdByWorkspaceSessionKey(
 /** Collect the distinct runtime hosts owning any persisted repo. */
 export function listKnownRuntimeHostIds(
  repos: readonly Pick<Repo, 'connectionId' | 'executionHostId'>[]
+): ExecutionHostId[] {
+  return listKnownPartitionHostIds(repos, 'runtime')
+}
+
+/** Collect the distinct SSH hosts owning any persisted repo. Their partitions are read separately
+ *  from the runtime ones: an `ssh:*` partition is not a rival claimant of the same workspace id,
+ *  it is the other half of ONE host's session that shipping builds split in two (#12723). */
+export function listKnownSshHostIds(
+  repos: readonly Pick<Repo, 'connectionId' | 'executionHostId'>[]
+): ExecutionHostId[] {
+  return listKnownPartitionHostIds(repos, 'ssh')
+}
+
+function listKnownPartitionHostIds(
+  repos: readonly Pick<Repo, 'connectionId' | 'executionHostId'>[],
+  kind: 'ssh' | 'runtime'
 ): ExecutionHostId[] {
  const hostIds = new Set<ExecutionHostId>()
  for (const repo of repos) {
    const parsed = parseExecutionHostId(getRepoExecutionHostId(repo))
-    if (parsed?.kind === 'runtime') {
+    if (parsed?.kind === kind) {
      hostIds.add(parsed.id)
    }
  }
  return [...hostIds]
 }

-/** Boot-time hydration: fetch the local partition plus one partition per known
- *  runtime host (from loaded repos and saved runtime ids), then merge them into
- *  the unified session the hydrators expect.
+/** Boot-time hydration: fetch the local partition, one partition per known runtime host (from
+ *  loaded repos and saved runtime ids) and one per known SSH host, then merge them into the
+ *  unified session the hydrators expect.
 *
 *  Fail-soft: a partition whose fetch rejects is skipped — boot proceeds with
 *  the rest. Corrupt partitions never reach here; persistence zod-validates
@@ -131,6 +148,14 @@ export async function fetchWorkspaceSessionWithRuntimeHostOwners(
  repos: readonly Pick<Repo, 'connectionId' | 'executionHostId'>[],
  additionalRuntimeHostIds: readonly ExecutionHostId[] = []
 ): Promise<WorkspaceSessionHostRead> {
+  const readPartition = async (hostId: ExecutionHostId): Promise<WorkspaceSessionState | null> => {
+    try {
+      return await api.get(hostId)
+    } catch (err) {
+      console.warn(`[session] skipping unreadable host partition ${hostId}:`, err)
+      return null
+    }
+  }
  const slices: HostSessionSlices = {
    [LOCAL_EXECUTION_HOST_ID]: await api.get()
  }
@@ -140,18 +165,36 @@ export async function fetchWorkspaceSessionWithRuntimeHostOwners(
    ...listKnownRuntimeHostIds(repos),
    ...additionalRuntimeHostIds
  ])
-  await Promise.all(
-    [...runtimeHostIds].map(async (hostId) => {
-      try {
-        slices[hostId] = await api.get(hostId)
-      } catch (err) {
-        console.warn(`[session] skipping unreadable host partition ${hostId}:`, err)
-      }
-    })
-  )
+  const sshHostIds = listKnownSshHostIds(repos)
+  const [, sshSlices] = await Promise.all([
+    Promise.all(
+      [...runtimeHostIds].map(async (hostId) => {
+        const slice = await readPartition(hostId)
+        if (slice) {
+          slices[hostId] = slice
+        }
+      })
+    ),
+    Promise.all(sshHostIds.map((hostId) => readPartition(hostId)))
+  ])
  const merged = mergeWorkspaceSessionsWithHostShadow(slices)
+  // Why the ssh partitions stay out of `slices`: the contention split reads two slices holding one
+  // workspace id as two DIFFERENT workspaces on rival hosts and parks one of them. 'local' and
+  // `ssh:<targetId>` are the same workspace written twice, so they are reunited afterwards instead
+  // — and a workspace the merged session has no tabs for is adopted rather than read as a
+  // deletion (#12721). Routing sends the reunited rows back to the owning partition.
+  let session = merged.session
+  // Why the contested keys travel with the slice: the split parks a co-claimant's rows so the
+  // primary's write cannot erase them, but ssh slices are kept out of that claimant set on purpose.
+  // Adoption is the one place an ssh row meets a bare id another host also claims, and it cannot
+  // ask — so the verdict the merge already reached is handed to it.
+  for (const slice of sshSlices) {
+    session = adoptStrandedHostPartitionSession(session, slice, {
+      contestedSessionKeys: merged.contestedSessionKeys
+    })
+  }
  return {
-    session: merged.session,
+    session,
    // Why the merged slices and not the raw ones: a row parked out of the renderer session must not
    // still name its host as the owner, or startup builds runtime placeholders for a local row.
    runtimeHostIdByWorkspaceSessionKey: buildRuntimeHostIdByWorkspaceSessionKey(merged.slices),
@@ -74,7 +74,7 @@ function getRestoredRuntimeHostId(
  return hostId && parseExecutionHostId(hostId)?.kind === 'runtime' ? hostId : null
 }

-function getFolderWorkspaceRuntimeHostId(
+function getFolderWorkspacePartitionHostId(
  state: HostPersistenceState,
  key: string
 ): ExecutionHostId {
@@ -88,6 +88,11 @@ function getFolderWorkspaceRuntimeHostId(
    : null
  const parsed = parseExecutionHostId(workspace?.executionHostId ?? group?.executionHostId)
  if (parsed) {
+    // Why ssh still answers 'local' here while a repo-backed worktree does not: boot hydration
+    // discovers SSH partitions from the repo catalog, and a folder workspace can be the only thing
+    // an SSH target owns. Routing it to `ssh:<targetId>` would strand it behind a partition no
+    // reader enumerates. A stranded folder workspace is still adopted back out of that partition
+    // when a repo does name the host; converging its writes needs a partition census first.
    return parsed.kind === 'runtime' ? parsed.id : LOCAL_EXECUTION_HOST_ID
  }
  if (workspace && group) {
@@ -123,12 +128,12 @@ function buildRepoHostById(

 /** Map a worktree to the host partition it persists under, plus the host claims behind it.
 *
- *  Why: only `runtime:*` worktrees are partitioned out. SSH-owned worktrees stay
- *  in the 'local' partition because the SSH flow already persists them there (in
- *  the unified blob) and separately mirrors them to each target's remote
- *  snapshot — partitioning them too would double-own that data. The one exception is an id two
- *  hosts both publish: it gets a deterministic primary so the co-claimant's rows can be parked in
- *  the shadow instead of sharing one bucket with it. */
+ *  Why every non-local host and not just `runtime:*`: an SSH worktree's session is already
+ *  read-modify-written into `ssh:<targetId>` by the main-process runtime, so answering 'local'
+ *  here double-owned the data and left whichever half the readers skipped round-tripping as
+ *  absence (#12721, #12723). The one exception is an id two hosts both publish: it gets a
+ *  deterministic primary so the co-claimant's rows can be parked in the shadow instead of sharing
+ *  one bucket with it. */
 /** True only when the catalog positively says `hostId` no longer holds the workspace. An id the
 *  catalog cannot speak for yet keeps its restored partition — the same rule the shadow uses. */
 function catalogReattributedAwayFrom(
@@ -154,7 +159,7 @@ export function buildHostSessionRouting(state: HostPersistenceState): HostSessio
  const hostIdByWorktreeId = (worktreeId: string): ExecutionHostId => {
    const workspaceScope = parseWorkspaceKey(worktreeId)
    if (workspaceScope?.type === 'folder') {
-      return getFolderWorkspaceRuntimeHostId(state, worktreeId)
+      return getFolderWorkspacePartitionHostId(state, worktreeId)
    }
    const rawWorktreeId =
      workspaceScope?.type === 'worktree' ? workspaceScope.worktreeId : worktreeId
@@ -188,9 +193,7 @@ export function buildHostSessionRouting(state: HostPersistenceState): HostSessio
    if (!repoHostId) {
      return LOCAL_EXECUTION_HOST_ID
    }
-    // Why: SSH-owned worktrees stay in the 'local' partition here while the runtime writes them to
-    // `ssh:<targetId>`; the shared owner map records that divergence (#12723).
-    return workspaceSessionPartitionHostId(repoHostId, 'local-partition')
+    return workspaceSessionPartitionHostId(repoHostId)
  }
  return { hostIdByWorktreeId, claims }
 }
@@ -154,7 +154,7 @@ describe('splitWorkspaceSessionByHost', () => {
    expect(Object.keys(slices[RUNTIME_B]?.tabsByWorktree ?? {})).toEqual(['b-wt'])
  })

-  it('keeps ssh-qualified visit recency in the local slice and routes runtime-qualified keys to their partition', () => {
+  it('routes host-qualified visit recency to the partition the key names', () => {
    const state: WorkspaceSessionState = {
      ...getDefaultWorkspaceSession(),
      lastVisitedAtByWorktreeId: {
@@ -167,17 +167,17 @@ describe('splitWorkspaceSessionByHost', () => {

    const slices = splitWorkspaceSessionByHost(state, ownerByPrefix())

-    // Why local for ssh: boot hydration reads only local + runtime:* partitions,
-    // so an ssh partition would strand the recency across restarts.
-    expect(slices[LOCAL_EXECUTION_HOST_ID]?.lastVisitedAtByWorktreeId).toEqual({
-      'local-wt': 1,
-      'ssh:builder|ssh-wt': 3
-    })
+    // Why the key's own host and not 'local': the recency row has to land in the same partition as
+    // the workspace it describes, or a read that adopts one without the other reports a visit for
+    // a workspace it has no tabs for (#12721).
+    expect(slices[LOCAL_EXECUTION_HOST_ID]?.lastVisitedAtByWorktreeId).toEqual({ 'local-wt': 1 })
    expect(slices[RUNTIME_A]?.lastVisitedAtByWorktreeId).toEqual({
      'a-wt': 2,
      'runtime:env-a|a-wt': 4
    })
-    expect(slices['ssh:builder' as ExecutionHostId]).toBeUndefined()
+    expect(slices['ssh:builder' as ExecutionHostId]?.lastVisitedAtByWorktreeId).toEqual({
+      'ssh:builder|ssh-wt': 3
+    })
  })

  it('routes tab-keyed maps via the owning tab worktree (legacy + unified)', () => {
@@ -16,8 +16,9 @@ import {
  isWorkspaceSessionRecord,
  mergeWorkspaceSessionArrayField,
  mergeWorkspaceSessionRecordField,
+  worktreeIdForPaneKey,
  type WorkspaceSessionRecord
-} from './workspace-session-host-records'
+} from '../../../shared/workspace-session-host-records'

 /**
 * Split / merge the unified WorkspaceSessionState across per-host partitions.
@@ -110,16 +111,13 @@ function assignVisitRecencyByHost(
    return
  }
  for (const [key, entry] of Object.entries(value)) {
-    // Why: boot hydration reads only local + runtime:* partitions, and SSH worktree
-    // session state deliberately stays in the local partition (see buildHostIdByWorktreeId);
-    // routing ssh-qualified keys to an ssh partition would strand them across restarts.
+    // Why the qualified host wins: the key already names the host that owns the visit, so routing
+    // it anywhere else separates the recency row from the workspace it describes.
    const qualifiedHost = isWorktreeHostIdentity(key)
      ? parseExecutionHostId(key.slice(0, key.indexOf('|')))
      : null
    const host = isWorktreeHostIdentity(key)
-      ? qualifiedHost?.kind === 'runtime'
-        ? qualifiedHost.id
-        : LOCAL_EXECUTION_HOST_ID
+      ? (qualifiedHost?.id ?? LOCAL_EXECUTION_HOST_ID)
      : ctx.hostIdByWorktreeId(key)
    const slice = ensureSlice(slices, host, templates) as WorkspaceSessionRecord
    const target = (slice.lastVisitedAtByWorktreeId ??= {}) as WorkspaceSessionRecord
@@ -277,12 +275,7 @@ export function splitWorkspaceSessionByHost(
          templates,
          field,
          value,
-          (paneKey) => {
-            const separator = paneKey.lastIndexOf(':')
-            return separator > 0
-              ? ctx.worktreeIdByTabId.get(paneKey.slice(0, separator))
-              : undefined
-          },
+          (paneKey) => worktreeIdForPaneKey(ctx.worktreeIdByTabId, paneKey),
          ctx
        )
        break
@@ -0,0 +1,731 @@
+/**
+ * Boot hydration and the remote-workspace round trip for a worktree whose session the
+ * main-process runtime wrote into `ssh:<targetId>`.
+ *
+ * The renderer used to read only the local + `runtime:*` partitions, so those tabs were invisible;
+ * the export then published an explicit empty list, `replace-session` made it authoritative, and
+ * the next pull applied it as a deletion that re-poisoned the snapshot on every launch
+ * (#12721, #18173).
+ *
+ * Runs the real projection and the real pull-side merge — the failure only exists where the read,
+ * the publish and the merge meet, and each of them is individually self-consistent.
+ */
+import { describe, expect, it } from 'vitest'
+import type { SleepingAgentSessionRecord } from '../../../shared/agent-session-resume'
+import { getDefaultWorkspaceSession } from '../../../shared/constants'
+import type { ExecutionHostId } from '../../../shared/execution-host'
+import {
+  exportRemoteWorkspaceSession,
+  importRemoteWorkspaceSession
+} from '../../../shared/remote-workspace-session-projection'
+import type { TerminalTab } from '../../../shared/terminal-tab-types'
+import type { WorkspaceSessionState } from '../../../shared/workspace-session-state-types'
+import { shouldAutoCreateInitialTerminal } from '@/components/terminal/initial-terminal'
+import { mergeDirectSshRemoteWorkspaceSession } from '../hooks/remote-workspace-session-merge'
+import { fetchWorkspaceSessionWithRuntimeHostOwners } from './workspace-session-host-hydration'
+
+const TARGET_ID = 'target-1'
+const SSH_HOST_ID: ExecutionHostId = `ssh:${TARGET_ID}`
+const REPO_ID = 'repo-remote'
+const WORKTREE_PATH = '/remote/checkout/feature'
+const WORKTREE_ID = `${REPO_ID}::${WORKTREE_PATH}`
+
+const repos = [{ id: REPO_ID, connectionId: TARGET_ID, executionHostId: null }]
+
+function tab(id: string, overrides: Partial<TerminalTab> = {}): TerminalTab {
+  return {
+    id,
+    ptyId: `pty-${id}`,
+    worktreeId: WORKTREE_ID,
+    title: id,
+    customTitle: null,
+    color: null,
+    sortOrder: 0,
+    createdAt: 1,
+    ...overrides
+  }
+}
+
+function session(overrides: Partial<WorkspaceSessionState>): WorkspaceSessionState {
+  return { ...getDefaultWorkspaceSession(), ...overrides }
+}
+
+/** A session read whose partitions are exactly what persistence holds. */
+function partitionedApi(
+  partitions: Partial<Record<ExecutionHostId | 'local', WorkspaceSessionState>>
+) {
+  return {
+    get: async (hostId?: ExecutionHostId) =>
+      partitions[hostId ?? 'local'] ?? getDefaultWorkspaceSession()
+  }
+}
+
+/** The observed shape from #12721: the local blob carries the worktree key with an empty list
+ *  while the runtime owns the real list in the SSH partition. */
+function strandedPartitions(hostTabs: TerminalTab[], localTabs: TerminalTab[] = []) {
+  return {
+    local: session({ tabsByWorktree: { [WORKTREE_ID]: localTabs } }),
+    [SSH_HOST_ID]: session({
+      tabsByWorktree: { [WORKTREE_ID]: hostTabs },
+      activeTabIdByWorktree: { [WORKTREE_ID]: hostTabs[0]?.id ?? null }
+    })
+  }
+}
+
+describe('ssh host partition hydration', () => {
+  it('hydrates tabs the runtime persisted into the ssh partition', async () => {
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(strandedPartitions([tab('tab-runtime')])),
+      repos
+    )
+
+    expect(read.session.tabsByWorktree[WORKTREE_ID]?.map((entry) => entry.id)).toEqual([
+      'tab-runtime'
+    ])
+  })
+
+  it('adopts the stranded workspace rows alongside its tabs', async () => {
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(strandedPartitions([tab('tab-runtime')])),
+      repos
+    )
+
+    expect(read.session.activeTabIdByWorktree?.[WORKTREE_ID]).toBe('tab-runtime')
+  })
+
+  it('adopts a hibernated agent record the host partition alone holds', async () => {
+    // The renderer's next full write replaces the SSH partition, so a runtime-authored record the
+    // reunited session never carried would be dropped by the repair itself.
+    const partitions = strandedPartitions([tab('tab-runtime')])
+    partitions[SSH_HOST_ID] = session({
+      ...partitions[SSH_HOST_ID],
+      sleepingAgentSessionsByPaneKey: {
+        'tab-runtime:leaf-1': {
+          paneKey: 'tab-runtime:leaf-1',
+          worktreeId: WORKTREE_ID,
+          tabId: 'tab-runtime',
+          agent: 'claude',
+          providerSession: { key: 'session_id', id: 'session-1' },
+          prompt: 'resume me',
+          state: 'done',
+          capturedAt: 5,
+          updatedAt: 5
+        } satisfies SleepingAgentSessionRecord
+      }
+    })
+
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(partitionedApi(partitions), repos)
+
+    expect(
+      read.session.sleepingAgentSessionsByPaneKey?.['tab-runtime:leaf-1']?.providerSession.id
+    ).toBe('session-1')
+  })
+
+  it('leaves a workspace the local partition already holds tabs for untouched', async () => {
+    // The other direction of the same rule, and the reason adoption is only gap-filling: merging
+    // into a populated row would re-add tabs the user had closed on every launch.
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(strandedPartitions([tab('tab-runtime')], [tab('tab-local')])),
+      repos
+    )
+
+    expect(read.session.tabsByWorktree[WORKTREE_ID]?.map((entry) => entry.id)).toEqual([
+      'tab-local'
+    ])
+  })
+
+  it("leaves that workspace's other rows alone as well", async () => {
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(strandedPartitions([tab('tab-runtime')], [tab('tab-local')])),
+      repos
+    )
+
+    expect(read.session.activeTabIdByWorktree?.[WORKTREE_ID]).toBeUndefined()
+  })
+
+  it('routes the reunited workspace back to the partition that owns it', async () => {
+    const { buildHostIdByWorktreeId } = await import('./workspace-session-host-persistence')
+
+    const hostIdByWorktreeId = buildHostIdByWorktreeId({
+      repos: [{ id: REPO_ID, connectionId: TARGET_ID, executionHostId: null }],
+      worktreesByRepo: {}
+    })
+
+    expect(hostIdByWorktreeId(WORKTREE_ID)).toBe(SSH_HOST_ID)
+  })
+})
+
+describe('ssh host partition workspaces with no terminal tabs', () => {
+  /** An SSH workspace the user left with an editor open and every terminal closed. Orca does not
+   *  auto-create a terminal while other tabs exist, so this is an ordinary state — and the whole
+   *  workspace now persists to `ssh:<targetId>`, tabs or no tabs. */
+  function editorOnlyPartitions() {
+    return {
+      local: session({ tabsByWorktree: {} }),
+      [SSH_HOST_ID]: session({
+        tabsByWorktree: { [WORKTREE_ID]: [] },
+        openFilesByWorktree: {
+          [WORKTREE_ID]: [
+            {
+              filePath: '/remote/checkout/feature/src/main.ts',
+              relativePath: 'src/main.ts',
+              worktreeId: WORKTREE_ID,
+              language: 'typescript',
+              dirtyDraftContent: 'unsaved work'
+            }
+          ]
+        },
+        activeFileIdByWorktree: { [WORKTREE_ID]: '/remote/checkout/feature/src/main.ts' },
+        activeTabTypeByWorktree: { [WORKTREE_ID]: 'editor' },
+        browserTabsByWorktree: {
+          [WORKTREE_ID]: [{ id: 'browser-1', name: 'Docs', tabs: [], activeTabId: null }]
+        },
+        lastVisitedAtByWorktreeId: { [`${SSH_HOST_ID}|${WORKTREE_ID}`]: 4242 }
+      } as unknown as WorkspaceSessionState)
+    }
+  }
+
+  it('restores the open editor files', async () => {
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(editorOnlyPartitions()),
+      repos
+    )
+
+    expect(
+      read.session.openFilesByWorktree?.[WORKTREE_ID]?.map((file) => file.relativePath)
+    ).toEqual(['src/main.ts'])
+  })
+
+  it('restores an unsaved hot-exit draft, which no other channel can recover', async () => {
+    // RemoteWorkspaceSession carries terminal fields only, so the SSH host snapshot cannot
+    // round-trip editor state. Losing it here loses user-authored content outright.
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(editorOnlyPartitions()),
+      repos
+    )
+
+    expect(read.session.openFilesByWorktree?.[WORKTREE_ID]?.[0]?.dirtyDraftContent).toBe(
+      'unsaved work'
+    )
+  })
+
+  it('restores browser workspaces and the active tab type', async () => {
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(editorOnlyPartitions()),
+      repos
+    )
+
+    expect(read.session.browserTabsByWorktree?.[WORKTREE_ID]?.map((entry) => entry.id)).toEqual([
+      'browser-1'
+    ])
+    expect(read.session.activeTabTypeByWorktree?.[WORKTREE_ID]).toBe('editor')
+  })
+
+  it('restores a workspace the host partition names with no tabs row at all', async () => {
+    // Stricter than the fixtures above, which carry an empty `tabsByWorktree` key. A workspace that
+    // never had a terminal has no such key, so tab presence cannot be what discovers it.
+    const partitions = {
+      local: session({ tabsByWorktree: {} }),
+      [SSH_HOST_ID]: session({
+        tabsByWorktree: {},
+        openFilesByWorktree: {
+          [WORKTREE_ID]: [
+            {
+              filePath: '/remote/checkout/feature/README.md',
+              relativePath: 'README.md',
+              worktreeId: WORKTREE_ID,
+              language: 'markdown',
+              dirtyDraftContent: 'never saved'
+            }
+          ]
+        }
+      } as unknown as WorkspaceSessionState)
+    }
+
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(partitionedApi(partitions), repos)
+
+    expect(read.session.openFilesByWorktree?.[WORKTREE_ID]?.[0]?.dirtyDraftContent).toBe(
+      'never saved'
+    )
+  })
+
+  it('restores host-qualified visit recency, which is keyed by host and not by bare id', async () => {
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(editorOnlyPartitions()),
+      repos
+    )
+
+    expect(read.session.lastVisitedAtByWorktreeId?.[`${SSH_HOST_ID}|${WORKTREE_ID}`]).toBe(4242)
+  })
+})
+
+describe('ssh host partition rows the host has nothing for', () => {
+  /** The base half of a legacy split: `local` still holds this workspace's editor state, including
+   *  an unsaved draft, while the SSH partition holds only empty rows for it. The workspace is
+   *  adoptable (the base has no terminal tabs for it), and every worktree-keyed row it adopts is a
+   *  replacing write — so an empty host row landing on a populated base row is a real deletion. */
+  function emptyHostRowsOverBaseDraft(hostOpenFiles: boolean) {
+    const draftFile = {
+      filePath: `${WORKTREE_PATH}/src/main.ts`,
+      relativePath: 'src/main.ts',
+      worktreeId: WORKTREE_ID,
+      language: 'typescript',
+      dirtyDraftContent: 'unsaved work'
+    }
+    return {
+      local: session({
+        tabsByWorktree: {},
+        openFilesByWorktree: { [WORKTREE_ID]: [draftFile] },
+        activeTabTypeByWorktree: { [WORKTREE_ID]: 'editor' }
+      } as unknown as WorkspaceSessionState),
+      [SSH_HOST_ID]: session({
+        tabsByWorktree: { [WORKTREE_ID]: [] },
+        ...(hostOpenFiles ? { openFilesByWorktree: { [WORKTREE_ID]: [] } } : {})
+      } as unknown as WorkspaceSessionState)
+    }
+  }
+
+  it('does not let an empty host row destroy an unsaved draft the base alone holds', async () => {
+    // The host having no open files is not evidence the base's are gone. Losing this is worse than
+    // the bug the adoption exists to fix: RemoteWorkspaceSession carries terminal fields only, so
+    // nothing can recover a `dirtyDraftContent` once the read has dropped it.
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(emptyHostRowsOverBaseDraft(true) as never),
+      repos
+    )
+
+    expect(read.session.openFilesByWorktree?.[WORKTREE_ID]?.[0]?.dirtyDraftContent).toBe(
+      'unsaved work'
+    )
+  })
+
+  it('still adopts a populated host row over the base leftovers', async () => {
+    // The other side of the same rule: the guard must be about the host having nothing, not about
+    // the base having something, or adoption stops repairing the split it exists for.
+    const partitions = emptyHostRowsOverBaseDraft(false)
+    partitions[SSH_HOST_ID] = session({
+      ...partitions[SSH_HOST_ID],
+      openFilesByWorktree: {
+        [WORKTREE_ID]: [
+          {
+            filePath: `${WORKTREE_PATH}/src/host.ts`,
+            relativePath: 'src/host.ts',
+            worktreeId: WORKTREE_ID,
+            language: 'typescript'
+          }
+        ]
+      }
+    } as unknown as WorkspaceSessionState)
+
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(partitions as never),
+      repos
+    )
+
+    expect(
+      read.session.openFilesByWorktree?.[WORKTREE_ID]?.map((file) => file.relativePath)
+    ).toEqual(['src/host.ts'])
+  })
+
+  it('adopts the layout of a tab the host slice names only in unifiedTabs', async () => {
+    // `buildWorktreeIdByTabId` — the index the split routes by — resolves unified-only tabs as well
+    // as `tabsByWorktree` ones, so their tab-keyed rows are written to this partition. A read that
+    // discovered tabs from `tabsByWorktree` alone routed them in and never brought them back.
+    const partitions = {
+      local: session({ tabsByWorktree: {} }),
+      [SSH_HOST_ID]: session({
+        tabsByWorktree: {},
+        unifiedTabs: {
+          [WORKTREE_ID]: [{ id: 'tab-unified', type: 'terminal', worktreeId: WORKTREE_ID }]
+        },
+        terminalLayoutsByTabId: { 'tab-unified': { direction: 'row', panes: [] } }
+      } as unknown as WorkspaceSessionState)
+    }
+
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(partitions as never),
+      repos
+    )
+
+    expect(read.session.terminalLayoutsByTabId?.['tab-unified']).toBeDefined()
+  })
+})
+
+describe('ssh host partition adoption on a contested bare id', () => {
+  /** A worktree id is `repoId::path` with no host component, so one repo registered on two hosts
+   *  publishes the SAME id for two DIFFERENT workspaces (STA-4343). The contention split parks the
+   *  co-claimant's rows so the primary's write cannot erase them — but ssh slices are deliberately
+   *  kept out of that claimant set, on the premise that `local` and `ssh:<target>` are one workspace
+   *  written twice. A contested id is exactly where that premise fails, and adoption cannot see it
+   *  from `(base, host)` alone. */
+  const RUNTIME_HOST_ID: ExecutionHostId = 'runtime:r1'
+  const contestedRepos = [
+    { id: REPO_ID, connectionId: TARGET_ID, executionHostId: null },
+    { id: 'repo-rt', connectionId: null, executionHostId: RUNTIME_HOST_ID }
+  ]
+
+  function contestedPartitions() {
+    return {
+      local: session({
+        tabsByWorktree: {},
+        openFilesByWorktree: {
+          [WORKTREE_ID]: [
+            {
+              filePath: '/local/checkout/feature/src/main.ts',
+              relativePath: 'src/main.ts',
+              worktreeId: WORKTREE_ID,
+              language: 'typescript',
+              dirtyDraftContent: 'local unsaved work'
+            }
+          ]
+        }
+      } as unknown as WorkspaceSessionState),
+      [RUNTIME_HOST_ID]: session({
+        tabsByWorktree: { [WORKTREE_ID]: [] },
+        activeTabTypeByWorktree: { [WORKTREE_ID]: 'terminal' }
+      } as unknown as WorkspaceSessionState),
+      [SSH_HOST_ID]: session({
+        tabsByWorktree: {},
+        openFilesByWorktree: {
+          [WORKTREE_ID]: [
+            {
+              filePath: '/remote/checkout/feature/src/other.ts',
+              relativePath: 'src/other.ts',
+              worktreeId: WORKTREE_ID,
+              language: 'typescript'
+            }
+          ]
+        }
+      } as unknown as WorkspaceSessionState)
+    }
+  }
+
+  it("does not overwrite a contested workspace's own rows with the ssh workspace's", async () => {
+    // The read names `local` primary for this id and parks the runtime claimant, so routing writes
+    // whatever survives here back into local's own partition. Replacing local's rows with the SSH
+    // workspace's would persist one workspace's editor state as another's — and destroy the draft.
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(contestedPartitions() as never),
+      contestedRepos
+    )
+
+    expect(read.session.openFilesByWorktree?.[WORKTREE_ID]?.[0]?.dirtyDraftContent).toBe(
+      'local unsaved work'
+    )
+  })
+
+  it('still fills a gap on a contested id', async () => {
+    // Declining to replace is not declining to repair: a row no claimant answered is still adopted.
+    const partitions = contestedPartitions()
+    partitions[SSH_HOST_ID] = session({
+      ...partitions[SSH_HOST_ID],
+      activeFileIdByWorktree: { [WORKTREE_ID]: '/remote/checkout/feature/src/other.ts' }
+    } as unknown as WorkspaceSessionState)
+
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(partitions as never),
+      contestedRepos
+    )
+
+    expect(read.session.activeFileIdByWorktree?.[WORKTREE_ID]).toBe(
+      '/remote/checkout/feature/src/other.ts'
+    )
+  })
+
+  it('does not let a legacy bare recency key move another host workspace of the same id', async () => {
+    // `lastVisitedAtByWorktreeId` is the one worktree-keyed field whose key may name its own host,
+    // and the split has a dedicated branch for that. A bare key carries no host, so it cannot be
+    // told apart from the base's own entry for the id — replacing moves Cmd+J recency permanently.
+    const partitions = {
+      local: session({
+        tabsByWorktree: {},
+        lastVisitedAtByWorktreeId: { [WORKTREE_ID]: 1000 }
+      } as unknown as WorkspaceSessionState),
+      [SSH_HOST_ID]: session({
+        tabsByWorktree: { [WORKTREE_ID]: [] },
+        lastVisitedAtByWorktreeId: { [WORKTREE_ID]: 9999 }
+      } as unknown as WorkspaceSessionState)
+    }
+
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(partitions as never),
+      repos
+    )
+
+    expect(read.session.lastVisitedAtByWorktreeId?.[WORKTREE_ID]).toBe(1000)
+  })
+
+  it('still adopts a host-qualified recency key, which names its own owner', async () => {
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi({
+        local: session({
+          tabsByWorktree: {},
+          lastVisitedAtByWorktreeId: { [WORKTREE_ID]: 1000 }
+        } as unknown as WorkspaceSessionState),
+        [SSH_HOST_ID]: session({
+          tabsByWorktree: { [WORKTREE_ID]: [] },
+          lastVisitedAtByWorktreeId: { [`${SSH_HOST_ID}|${WORKTREE_ID}`]: 9999 }
+        } as unknown as WorkspaceSessionState)
+      } as never),
+      repos
+    )
+
+    expect(read.session.lastVisitedAtByWorktreeId?.[`${SSH_HOST_ID}|${WORKTREE_ID}`]).toBe(9999)
+    expect(read.session.lastVisitedAtByWorktreeId?.[WORKTREE_ID]).toBe(1000)
+  })
+})
+
+describe('ssh host partition write/read round trip', () => {
+  /** The two halves pinned together through the shipping write path. Testing the read against a
+   *  hand-built partition is what let an editor-only workspace fall out: the fixture asserted the
+   *  shape the read expected instead of the shape the write actually produces. */
+  async function roundTrip(payload: WorkspaceSessionState): Promise<WorkspaceSessionState> {
+    const { buildWorkspaceSessionHostSnapshots } =
+      await import('./workspace-session-host-persistence')
+    const snapshots = buildWorkspaceSessionHostSnapshots(payload, {
+      repos: [{ id: REPO_ID, connectionId: TARGET_ID, executionHostId: null }],
+      worktreesByRepo: {}
+    })
+    const partitions: Record<string, WorkspaceSessionState> = {}
+    for (const snapshot of snapshots) {
+      partitions[snapshot.hostId ?? 'local'] = snapshot.state
+    }
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(partitions as never),
+      repos
+    )
+    return read.session
+  }
+
+  it('sends an editor-only SSH workspace to its partition and reads it back', async () => {
+    const restored = await roundTrip(
+      session({
+        tabsByWorktree: {},
+        openFilesByWorktree: {
+          [WORKTREE_ID]: [
+            {
+              filePath: '/remote/checkout/feature/src/app.ts',
+              relativePath: 'src/app.ts',
+              worktreeId: WORKTREE_ID,
+              language: 'typescript',
+              dirtyDraftContent: 'work in progress'
+            }
+          ]
+        },
+        activeTabTypeByWorktree: { [WORKTREE_ID]: 'editor' }
+      } as unknown as WorkspaceSessionState)
+    )
+
+    expect(restored.openFilesByWorktree?.[WORKTREE_ID]?.[0]?.dirtyDraftContent).toBe(
+      'work in progress'
+    )
+    expect(restored.activeTabTypeByWorktree?.[WORKTREE_ID]).toBe('editor')
+  })
+
+  it('sends a terminal SSH workspace to its partition and reads it back', async () => {
+    const restored = await roundTrip(
+      session({ tabsByWorktree: { [WORKTREE_ID]: [tab('tab-live')] } })
+    )
+
+    expect(restored.tabsByWorktree[WORKTREE_ID]?.map((entry) => entry.id)).toEqual(['tab-live'])
+  })
+})
+
+describe('ssh host partition and the closed-last-terminal tombstone', () => {
+  /** Two readings of one value meet here. The terminal layer writes an explicit empty
+   *  `tabsByWorktree` row to mean "the user closed the last terminal" and reserves an ABSENT row for
+   *  "never initialized" — a real tombstone, honoured by `shouldAutoCreateInitialTerminal`. This
+   *  adoption reads an empty row on the BASE side as a gap to fill. Opposite readings, same value,
+   *  so the boundary between them is asserted rather than reasoned about: a mature product ships
+   *  exactly this defect, with a correct write side and one reader that decides seeding on a count
+   *  and never consults the record. */
+  async function roundTripSession(payload: WorkspaceSessionState): Promise<{
+    restored: WorkspaceSessionState
+    partitions: Record<string, WorkspaceSessionState>
+  }> {
+    const { buildWorkspaceSessionHostSnapshots } =
+      await import('./workspace-session-host-persistence')
+    const snapshots = buildWorkspaceSessionHostSnapshots(payload, {
+      repos: [{ id: REPO_ID, connectionId: TARGET_ID, executionHostId: null }],
+      worktreesByRepo: {}
+    })
+    const partitions: Record<string, WorkspaceSessionState> = {}
+    for (const snapshot of snapshots) {
+      partitions[snapshot.hostId ?? 'local'] = snapshot.state
+    }
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(partitions as never),
+      repos
+    )
+    return { restored: read.session, partitions }
+  }
+
+  it('writes an SSH workspace emptied by this build into the partition that owns it', async () => {
+    // The precondition the whole non-recurrence claim rests on: the tombstone lands in
+    // `ssh:<targetId>` and `local` keeps no row, so the legacy shape cannot be regenerated.
+    const { partitions } = await roundTripSession(
+      session({ tabsByWorktree: { [WORKTREE_ID]: [] } })
+    )
+
+    expect(partitions[SSH_HOST_ID]?.tabsByWorktree?.[WORKTREE_ID]).toEqual([])
+    expect(Object.hasOwn(partitions.local?.tabsByWorktree ?? {}, WORKTREE_ID)).toBe(false)
+  })
+
+  it('restores that tombstone as an explicit empty row, not a deleted key', async () => {
+    // A deleted key reads back as "never initialized" and the workspace re-seeds on every launch,
+    // which is the defect the tombstone exists to prevent. Presence is the whole signal.
+    const { restored } = await roundTripSession(session({ tabsByWorktree: { [WORKTREE_ID]: [] } }))
+
+    expect(Object.hasOwn(restored.tabsByWorktree, WORKTREE_ID)).toBe(true)
+    expect(restored.tabsByWorktree[WORKTREE_ID]).toEqual([])
+  })
+
+  it('leaves the restored workspace un-seeded by the shared seeding predicate', async () => {
+    // Asserted through the real predicate rather than by inspecting the row, because the row being
+    // right is worth nothing if the reader that acts on it disagrees.
+    const { restored } = await roundTripSession(session({ tabsByWorktree: { [WORKTREE_ID]: [] } }))
+
+    expect(
+      shouldAutoCreateInitialTerminal(
+        restored.tabsByWorktree[WORKTREE_ID]?.length ?? 0,
+        Object.hasOwn(restored.tabsByWorktree, WORKTREE_ID)
+      )
+    ).toBe(false)
+  })
+
+  it('does not adopt a stale populated ssh row over a tombstone in the owning partition', async () => {
+    // The collision stated directly. The tombstone is in `ssh:<targetId>` — where this build writes
+    // it — and adoption must neither hand stale tabs back nor read the row as a gap.
+    const partitions = {
+      local: session({ tabsByWorktree: {} }),
+      [SSH_HOST_ID]: session({ tabsByWorktree: { [WORKTREE_ID]: [] } })
+    }
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(partitions as never),
+      repos
+    )
+
+    expect(read.session.tabsByWorktree[WORKTREE_ID]).toEqual([])
+    expect(Object.hasOwn(read.session.tabsByWorktree, WORKTREE_ID)).toBe(true)
+  })
+
+  it('resurrects the legacy-transition shape exactly once and not again', async () => {
+    // The documented knownGap, and the claim that makes it acceptable. Boot 1 adopts the stranded
+    // tabs back over `local`'s empty row — non-destructive, and the repair working. The user then
+    // empties the workspace on THIS build, and boot 2 must hold the tombstone: the row now lives in
+    // the owning partition and `local` no longer names the workspace, so there is nothing left to
+    // resurrect from. A gap that recurred would be a permanent re-seed, not a one-shot.
+    const firstBoot = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi({
+        local: session({ tabsByWorktree: { [WORKTREE_ID]: [] } }),
+        [SSH_HOST_ID]: session({ tabsByWorktree: { [WORKTREE_ID]: [tab('tab-stale')] } })
+      } as never),
+      repos
+    )
+    expect(firstBoot.session.tabsByWorktree[WORKTREE_ID]?.map((entry) => entry.id)).toEqual([
+      'tab-stale'
+    ])
+
+    const { restored: secondBoot } = await roundTripSession(
+      session({ ...firstBoot.session, tabsByWorktree: { [WORKTREE_ID]: [] } })
+    )
+
+    expect(secondBoot.tabsByWorktree[WORKTREE_ID]).toEqual([])
+    expect(
+      shouldAutoCreateInitialTerminal(
+        secondBoot.tabsByWorktree[WORKTREE_ID]?.length ?? 0,
+        Object.hasOwn(secondBoot.tabsByWorktree, WORKTREE_ID)
+      )
+    ).toBe(false)
+  })
+
+  it('publishes the tombstone rather than a row the host can read as unknown', async () => {
+    // Rule 3, client-publishes -> other-client-reads. An emptied workspace must publish its empty
+    // list so a paired client sees the same state; the merge's `hostUnknown` defence covers tabs
+    // this client holds, and an empty row is exactly what it holds here.
+    const { restored } = await roundTripSession(session({ tabsByWorktree: { [WORKTREE_ID]: [] } }))
+    const published = exportRemoteWorkspaceSession(restored, {
+      isTargetWorktree: (worktreeId) => worktreeId === WORKTREE_ID
+    })
+
+    expect(published.tabsByWorktreePath[WORKTREE_PATH]).toEqual([])
+  })
+})
+
+describe('ssh host partition remote-workspace round trip', () => {
+  it('does not delete the worktree tabs across a publish and the next pull', async () => {
+    const partitions = strandedPartitions([tab('tab-runtime')])
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(partitionedApi(partitions), repos)
+    // Publish exactly what the renderer now holds, then apply it back as `replace-session` does.
+    const published = exportRemoteWorkspaceSession(read.session, {
+      isTargetWorktree: (worktreeId) => worktreeId === WORKTREE_ID
+    })
+    const pulled = importRemoteWorkspaceSession(published, {
+      resolveWorktreeId: (worktreePath) => (worktreePath === WORKTREE_PATH ? WORKTREE_ID : null),
+      executionHostId: SSH_HOST_ID
+    })
+
+    const merged = mergeDirectSshRemoteWorkspaceSession(
+      read.session,
+      pulled,
+      new Set([WORKTREE_ID]),
+      read.session.tabsByWorktree,
+      new Set(),
+      SSH_HOST_ID,
+      1
+    )
+
+    expect(merged.tabsByWorktree[WORKTREE_ID]?.map((entry) => entry.id)).toEqual(['tab-runtime'])
+  })
+
+  it('keeps the tabs when an older client publishes an empty list for them', async () => {
+    // Rule 3 skew, the dangerous direction: a client that predates this fix still reads only the
+    // local partition, so its own `replace-session` names this workspace's path with NO tabs. The
+    // merge already refuses to delete what the host has never been told about — but only for tabs
+    // this client actually holds, which before the fix it did not. Hydrating them is what arms
+    // that defence, and this client then republishes the real list and repairs the snapshot.
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(strandedPartitions([tab('tab-runtime')])),
+      repos
+    )
+    const publishedByOldClient = importRemoteWorkspaceSession(
+      {
+        activeWorktreePath: null,
+        activeTabId: null,
+        tabsByWorktreePath: { [WORKTREE_PATH]: [] },
+        terminalLayoutsByTabId: {}
+      },
+      {
+        resolveWorktreeId: (worktreePath) => (worktreePath === WORKTREE_PATH ? WORKTREE_ID : null),
+        executionHostId: SSH_HOST_ID
+      }
+    )
+
+    const merged = mergeDirectSshRemoteWorkspaceSession(
+      read.session,
+      publishedByOldClient,
+      new Set([WORKTREE_ID]),
+      read.session.tabsByWorktree,
+      new Set(),
+      SSH_HOST_ID,
+      2
+    )
+
+    expect(merged.tabsByWorktree[WORKTREE_ID]?.map((entry) => entry.id)).toEqual(['tab-runtime'])
+  })
+
+  it('publishes the stranded tabs rather than an empty list', async () => {
+    const read = await fetchWorkspaceSessionWithRuntimeHostOwners(
+      partitionedApi(strandedPartitions([tab('tab-runtime')])),
+      repos
+    )
+
+    const published = exportRemoteWorkspaceSession(read.session, {
+      isTargetWorktree: (worktreeId) => worktreeId === WORKTREE_ID
+    })
+
+    expect(published.tabsByWorktreePath[WORKTREE_PATH]?.map((entry) => entry.id)).toEqual([
+      'tab-runtime'
+    ])
+  })
+})
@@ -1,4 +1,8 @@
 import type { WorkspaceKey, WorkspaceScope } from './folder-workspace-types'
+import {
+  getWorktreeIdFromHostIdentity,
+  isWorktreeHostIdentity
+} from './worktree/host-qualified-identity'

 export function worktreeWorkspaceKey(worktreeId: string): WorkspaceKey {
  return `worktree:${worktreeId}`
@@ -20,6 +24,16 @@ export function parseWorkspaceKey(value: string): WorkspaceScope | null {
  return null
 }

+/** Bare workspace id behind a session key, which may be a WorkspaceKey, a host-qualified identity
+ *  (`ssh:target|repo::path`, used by visit recency), or already a bare id. */
+export function normalizeWorkspaceSessionKeyToWorkspaceId(value: string): string {
+  if (isWorktreeHostIdentity(value)) {
+    return getWorktreeIdFromHostIdentity(value)
+  }
+  const scope = parseWorkspaceKey(value)
+  return scope?.type === 'worktree' ? scope.worktreeId : value
+}
+
 export function isWorkspaceKey(value: string): value is WorkspaceKey {
  return parseWorkspaceKey(value) !== null
 }
@@ -19,7 +19,8 @@ export const WORKSPACE_SESSION_FIELD_OWNERSHIP = {
  activeTabId: 'global',
  browserUrlHistory: 'global',
  workspaceDocHistory: 'global',
-  // Why: SSH remains local-owned, so its connection identifiers stay in the local slice.
+  // Why global rather than per-partition: this is the client's own record of which connections it
+  // owed work to at shutdown, not state belonging to any one workspace.
  activeConnectionIdsAtShutdown: 'global',
  // Why global: keyed by runtime environment rather than by worktree, and it is this client's
  // record of what it owes those environments — the same reason SSH connection state stays local.
@@ -1,4 +1,4 @@
-import type { WorkspaceSessionState } from '../../../shared/workspace-session-state-types'
+import type { WorkspaceSessionState } from './workspace-session-state-types'

 export type WorkspaceSessionRecord = Record<string, unknown>

@@ -24,6 +24,16 @@ export function buildWorktreeIdByTabId(state: WorkspaceSessionState): Map<string
  return byTab
 }

+/** The workspace a pane key belongs to. A pane key is `<tabId>:<leafId>`; both the split and the
+ *  stranded-partition adoption resolve it here so neither can parse it its own way. */
+export function worktreeIdForPaneKey(
+  worktreeIdByTabId: Map<string, string>,
+  paneKey: string
+): string | undefined {
+  const separator = paneKey.lastIndexOf(':')
+  return separator > 0 ? worktreeIdByTabId.get(paneKey.slice(0, separator)) : undefined
+}
+
 export function buildWorktreeIdByFileId(state: WorkspaceSessionState): Map<string, string> {
  const byFile = new Map<string, string>()
  for (const files of Object.values(state.openFilesByWorktree ?? {})) {
@@ -1,29 +1,24 @@
 import { describe, expect, it } from 'vitest'
 import { workspaceSessionPartitionHostId } from './workspace-session-partition-owner'

-// Why (#12723): the renderer and the runtime used two independent owner maps for the same
-// worktree's session state. They now share one function, so the divergence is a single argument
-// and cannot drift further. Behaviour on both sides is unchanged.
+// Why (#12723): the renderer and the main-process runtime used to get different answers here for
+// the same SSH workspace, which split one session across two partitions and left whichever half a
+// reader skipped round-tripping as absence (#12721). There is one answer now.
 describe('workspaceSessionPartitionHostId', () => {
-  it('keeps runtime worktrees in their own partition on both sides', () => {
-    expect(workspaceSessionPartitionHostId('runtime:env-a', 'local-partition')).toBe(
-      'runtime:env-a'
-    )
-    expect(workspaceSessionPartitionHostId('runtime:env-a', 'host-partition')).toBe('runtime:env-a')
+  it('gives a runtime host its own partition', () => {
+    expect(workspaceSessionPartitionHostId('runtime:env-a')).toBe('runtime:env-a')
  })

-  it('keeps local worktrees local on both sides', () => {
-    expect(workspaceSessionPartitionHostId('local', 'local-partition')).toBe('local')
-    expect(workspaceSessionPartitionHostId('local', 'host-partition')).toBe('local')
+  it('gives an SSH host its own partition, matching what the runtime already writes', () => {
+    expect(workspaceSessionPartitionHostId('ssh:devbox')).toBe('ssh:devbox')
  })

-  it('records the SSH divergence as the only difference between the two models', () => {
-    expect(workspaceSessionPartitionHostId('ssh:devbox', 'local-partition')).toBe('local')
-    expect(workspaceSessionPartitionHostId('ssh:devbox', 'host-partition')).toBe('ssh:devbox')
+  it('keeps local state in the legacy local blob', () => {
+    expect(workspaceSessionPartitionHostId('local')).toBe('local')
  })

  it('falls back to the local partition for unparseable host ids', () => {
-    expect(workspaceSessionPartitionHostId(null, 'host-partition')).toBe('local')
-    expect(workspaceSessionPartitionHostId('nonsense', 'host-partition')).toBe('local')
+    expect(workspaceSessionPartitionHostId(null)).toBe('local')
+    expect(workspaceSessionPartitionHostId('nonsense')).toBe('local')
  })
 })
@@ -5,35 +5,21 @@ import {
 } from './execution-host'

 /**
- * Where an SSH-owned worktree's durable session state lives.
+ * The one partition a worktree's durable session state lives in: its own execution host.
 *
- * This is the single axis on which the renderer and the main-process runtime disagree today
- * (stablyai/orca#12723). Both sides now compute their partition through this function so the
- * divergence is one argument in one place instead of two independently drifting owner maps:
+ * This used to answer differently depending on who asked (stablyai/orca#12723). The renderer
+ * mapped SSH worktrees to the `local` blob while the main-process runtime read-modify-wrote
+ * `ssh:<targetId>`, so one workspace's session was split across two stores and neither reader
+ * reunited them. Whatever landed on the unread side did not read as unknown — it round-tripped as
+ * absence, and the replace-session upload converted that into deletion (#12721, #18173).
 *
- * - `local-partition` — the renderer's shipping model. SSH worktrees keep their session state in
- *   the `local` partition; partitioning them would double-own the data.
- * - `host-partition` — the runtime's shipping model (#12671). Pane retirement, windowless PTY
- *   handoff and orchestration fences read-modify-write `ssh:<targetId>`.
- *
- * Both partitions hold real data written by shipping builds, so neither side can simply adopt the
- * other's answer: flipping a resolver orphans whichever store it stops reading. Converging needs a
- * read-both transition (generalize `workspaceSessionPartitionIdsForHost`) and should converge on
- * `host-partition`, since Orca Remote — SSH's successor — is already partitioned as `runtime:*`.
- * Until then this function preserves today's behaviour exactly on both sides.
+ * There is no second model now: `runtime:*` and `ssh:*` each own their partition, `local` owns the
+ * legacy `workspaceSession` blob. Rows a shipping build left in `local` for an SSH worktree are
+ * still real, so the read side folds them back in — see `adoptStrandedHostPartitionSession` — and
+ * the next write returns the unified result to the owning partition.
 */
-export type WorkspaceSessionSshOwnership = 'local-partition' | 'host-partition'
-
 export function workspaceSessionPartitionHostId(
-  executionHostId: string | null | undefined,
-  sshOwnership: WorkspaceSessionSshOwnership
+  executionHostId: string | null | undefined
 ): ExecutionHostId {
-  const parsed = parseExecutionHostId(executionHostId)
-  if (parsed?.kind === 'runtime') {
-    return parsed.id
-  }
-  if (parsed?.kind === 'ssh') {
-    return sshOwnership === 'host-partition' ? parsed.id : LOCAL_EXECUTION_HOST_ID
-  }
-  return LOCAL_EXECUTION_HOST_ID
+  return parseExecutionHostId(executionHostId)?.id ?? LOCAL_EXECUTION_HOST_ID
 }
@@ -0,0 +1,293 @@
+import type { WorkspaceSessionState } from './workspace-session-state-types'
+import {
+  WORKSPACE_SESSION_FIELD_OWNERSHIP,
+  type WorkspaceSessionFieldOwnership
+} from './workspace-session-host-field-ownership'
+import { normalizeWorkspaceSessionKeyToWorkspaceId } from './workspace-scope'
+import { isWorktreeHostIdentity as isHostQualifiedSessionKey } from './worktree/host-qualified-identity'
+import {
+  buildWorktreeIdByFileId,
+  buildWorktreeIdByTabId,
+  worktreeIdForPaneKey
+} from './workspace-session-host-records'
+
+/**
+ * Fold rows a host partition holds alone back into the session the readers assemble.
+ *
+ * Shipping builds split one SSH workspace's session across two partitions: the renderer wrote
+ * `local`, the main-process runtime wrote `ssh:<targetId>` (#12723). `workspaceSessionPartitionHostId`
+ * now names a single owner, but both stores still hold real data, so every reader has to reunite
+ * them once before the write path returns the result to that owner.
+ *
+ * **A workspace is adopted whenever the host partition names it at all — not only when it has
+ * terminal tabs.** The write path routes EVERY worktree-scoped field to the owning partition, so a
+ * workspace with open editor files, browser tabs or tab groups and no terminals lives there just as
+ * completely as one with terminals. Gating on tabs would strand exactly those, and unlike terminal
+ * state they cannot be recovered from the SSH host snapshot, which carries terminal fields only —
+ * an unsaved `dirtyDraftContent` would be destroyed outright.
+ *
+ * That is why the walk below switches exhaustively over `WORKSPACE_SESSION_FIELD_OWNERSHIP` instead
+ * of listing the fields it knows about: a hand-maintained list is what let editor and browser state
+ * fall out, and a new ownership kind must not be able to fall out the same way.
+ *
+ * Adoption is told which session keys the read found **contested**. Every rule below rests on the
+ * premise that `local` and `ssh:<targetId>` are one workspace written twice — and a bare
+ * `repoId::path` id claimed by more than one host is exactly where that premise is false. The
+ * contention split cannot see it, because ssh slices are deliberately kept out of the claimant set,
+ * so the verdict is passed in instead: a contested key may still be gap-filled, never replaced.
+ * Without that, an SSH workspace's rows overwrote a different workspace's rows under the same id,
+ * and routing then wrote them into that workspace's own partition.
+ *
+ * The one thing the base keeps unconditionally is a workspace it holds **terminal tabs** for. That
+ * is the live copy the user is looking at, and merging a stale partition into it would re-add tabs
+ * they had closed on every launch. Leaving it alone keeps this a one-shot repair, at the cost of
+ * not recovering rows stranded beside a populated workspace — which are stranded on main today too,
+ * so it is never a new loss. An EMPTY tab row is not such a copy: an empty list is not evidence
+ * that anything was closed (`mergeDirectSshRemoteWorkspaceSession` argues this at length, and
+ * docs/reference/ssh-execution-boundary.md makes it general — "we could not see it" is
+ * `unverifiable`, never proof of absence). Treating it as the truth is what published an empty tab
+ * list and let `replace-session` delete the host's copy (#12721).
+ */
+
+type KeyedRecord = Record<string, unknown>
+
+const SESSION_FIELDS = Object.keys(
+  WORKSPACE_SESSION_FIELD_OWNERSHIP
+) as (keyof WorkspaceSessionState)[]
+
+function asRecord(value: unknown): KeyedRecord | null {
+  return value && typeof value === 'object' && !Array.isArray(value) ? (value as KeyedRecord) : null
+}
+
+function recordWorkspaceId(entry: unknown): string | null {
+  const worktreeId = asRecord(entry)?.worktreeId
+  return typeof worktreeId === 'string' ? worktreeId : null
+}
+
+/** A browser-workspace row is keyed by browser workspace id; its pages name the workspace. */
+function browserPagesWorkspaceId(entry: unknown): string | null {
+  const first = Array.isArray(entry) ? (entry[0] as unknown) : null
+  return recordWorkspaceId(first)
+}
+
+/** Workspaces the base holds terminal tabs for: its live copies, which adoption never touches. */
+function workspacesTheBaseOwns(base: WorkspaceSessionState): Set<string> {
+  const owned = new Set<string>()
+  for (const [key, tabs] of Object.entries(base.tabsByWorktree ?? {})) {
+    if (Array.isArray(tabs) && tabs.length > 0) {
+      owned.add(normalizeWorkspaceSessionKeyToWorkspaceId(key))
+    }
+  }
+  return owned
+}
+
+/** Every workspace the host partition names in any scoped field, minus the base's live copies. */
+function adoptableWorkspaceIds(
+  base: WorkspaceSessionState,
+  host: WorkspaceSessionState
+): Set<string> {
+  const owned = workspacesTheBaseOwns(base)
+  const adoptable = new Set<string>()
+  const consider = (value: string | null | undefined): void => {
+    if (!value) {
+      return
+    }
+    const workspaceId = normalizeWorkspaceSessionKeyToWorkspaceId(value)
+    if (!owned.has(workspaceId)) {
+      adoptable.add(workspaceId)
+    }
+  }
+  for (const field of SESSION_FIELDS) {
+    const ownership: WorkspaceSessionFieldOwnership = WORKSPACE_SESSION_FIELD_OWNERSHIP[field]
+    const value = host[field]
+    switch (ownership) {
+      case 'global':
+      case 'hostPrivate':
+      case 'tabKeyed':
+      case 'paneKeyed':
+      case 'fileKeyed':
+        // Keyed by something the workspaces below already account for.
+        break
+      case 'worktreeKeyed':
+        for (const key of Object.keys(asRecord(value) ?? {})) {
+          consider(key)
+        }
+        break
+      case 'worktreeArray':
+        for (const id of Array.isArray(value) ? (value as string[]) : []) {
+          consider(id)
+        }
+        break
+      case 'sleepingAgentKeyed':
+      case 'surfaceTombstoneKeyed':
+        for (const entry of Object.values(asRecord(value) ?? {})) {
+          consider(recordWorkspaceId(entry))
+        }
+        break
+      case 'browserWorkspaceKeyed':
+        for (const entry of Object.values(asRecord(value) ?? {})) {
+          consider(browserPagesWorkspaceId(entry))
+        }
+        break
+    }
+  }
+  return adoptable
+}
+
+/**
+ * Whether the host has nothing to say about a key. `[]`, `{}` and null/undefined all mean the host
+ * holds no rows, which is never evidence that the base's rows are wrong — the same reading the base
+ * side already gives an empty tab row, and `docs/reference/ssh-execution-boundary.md` generalises.
+ * Without this an empty host `openFilesByWorktree` row replaced a populated base one and destroyed
+ * an unsaved `dirtyDraftContent`, which no other channel can recover. The symmetric cost is that a
+ * row the host really did empty stays visible for one more launch, and a resurrected editor tab is
+ * non-destructive where a destroyed draft is not.
+ */
+function hostHasNothingFor(entry: unknown): boolean {
+  if (entry === null || entry === undefined) {
+    return true
+  }
+  if (Array.isArray(entry)) {
+    return entry.length === 0
+  }
+  return typeof entry === 'object' && Object.keys(entry as KeyedRecord).length === 0
+}
+
+function adoptRecord(
+  next: WorkspaceSessionState,
+  host: WorkspaceSessionState,
+  field: keyof WorkspaceSessionState,
+  shouldAdopt: (key: string, entry: unknown) => boolean,
+  /** Whether this key's host row may replace the base's, rather than only fill a gap. An adoptable
+   *  workspace is host-owned, so its populated rows supersede the base's leftovers — but only where
+   *  the id names one workspace and the host actually holds something. */
+  mayReplace: boolean | ((key: string) => boolean) = false
+): void {
+  const hostRecord = asRecord(host[field])
+  if (!hostRecord) {
+    return
+  }
+  const merged = { ...asRecord(next[field]) }
+  for (const [key, entry] of Object.entries(hostRecord)) {
+    const replaces =
+      (typeof mayReplace === 'function' ? mayReplace(key) : mayReplace) && !hostHasNothingFor(entry)
+    if (shouldAdopt(key, entry) && (replaces || !Object.hasOwn(merged, key))) {
+      merged[key] = entry
+    }
+  }
+  ;(next as KeyedRecord)[field] = merged
+}
+
+export type StrandedPartitionAdoptionOptions = {
+  /** Session keys the contention split found claimed by more than one partition. */
+  contestedSessionKeys?: ReadonlySet<string>
+}
+
+export function adoptStrandedHostPartitionSession(
+  base: WorkspaceSessionState,
+  host: WorkspaceSessionState | null | undefined,
+  options: StrandedPartitionAdoptionOptions = {}
+): WorkspaceSessionState {
+  if (!host) {
+    return base
+  }
+  const adoptable = adoptableWorkspaceIds(base, host)
+  if (adoptable.size === 0) {
+    return base
+  }
+  const contested = new Set<string>()
+  for (const key of options.contestedSessionKeys ?? []) {
+    contested.add(normalizeWorkspaceSessionKeyToWorkspaceId(key))
+  }
+  const adopts = (key: string): boolean =>
+    adoptable.has(normalizeWorkspaceSessionKeyToWorkspaceId(key))
+  const isContested = (key: string): boolean =>
+    contested.has(normalizeWorkspaceSessionKeyToWorkspaceId(key))
+
+  const next: WorkspaceSessionState = { ...base, tabsByWorktree: { ...base.tabsByWorktree } }
+  for (const [key, tabs] of Object.entries(host.tabsByWorktree ?? {})) {
+    if (!adopts(key) || !Array.isArray(tabs)) {
+      continue
+    }
+    // A contested id is not this workspace written twice, so the base's own row stays.
+    if (!isContested(key) || !Object.hasOwn(next.tabsByWorktree, key)) {
+      next.tabsByWorktree[key] = tabs
+    }
+  }
+  // Why the split's own indexes: they are what decided which partition each tab-, pane- and
+  // file-keyed row was written to, so reading them back through anything else lets the two walks
+  // disagree. `buildWorktreeIdByTabId` also covers unified-only tabs, whose layout and PTY records
+  // the split routes here and a `tabsByWorktree`-only walk never adopted back. Computed up front so
+  // the keyed fields do not depend on the ownership table's declaration order.
+  const worktreeIdByTabId = buildWorktreeIdByTabId(host)
+  const worktreeIdByFileId = buildWorktreeIdByFileId(host)
+  const adoptsResolved = (worktreeId: string | undefined): boolean =>
+    worktreeId !== undefined && adopts(worktreeId)
+
+  for (const field of SESSION_FIELDS) {
+    const ownership: WorkspaceSessionFieldOwnership = WORKSPACE_SESSION_FIELD_OWNERSHIP[field]
+    switch (ownership) {
+      case 'global':
+      case 'hostPrivate':
+        // 'local' owns the globals; hostPrivate is main's own per-partition fence.
+        break
+      case 'worktreeKeyed':
+        if (field !== 'tabsByWorktree') {
+          // Why a bare recency key only fills a gap: `lastVisitedAtByWorktreeId` is the one field in
+          // this kind whose key may carry a host (`<hostId>|<worktreeId>`), and the split has its
+          // own branch for that. A qualified key names its owner and cannot collide; a bare one is
+          // indistinguishable from another host's entry for the same id, and replacing moved a
+          // local workspace's Cmd+J position permanently.
+          adoptRecord(
+            next,
+            host,
+            field,
+            (key) => adopts(key),
+            (key) =>
+              !isContested(key) &&
+              (field !== 'lastVisitedAtByWorktreeId' || isHostQualifiedSessionKey(key))
+          )
+        }
+        break
+      case 'worktreeArray': {
+        const hostIds = host[field]
+        const adopted = (Array.isArray(hostIds) ? (hostIds as string[]) : []).filter(adopts)
+        if (adopted.length > 0) {
+          const baseIds = next[field]
+          ;(next as KeyedRecord)[field] = [
+            ...new Set([...(Array.isArray(baseIds) ? (baseIds as string[]) : []), ...adopted])
+          ]
+        }
+        break
+      }
+      case 'tabKeyed':
+        adoptRecord(next, host, field, (key) => adoptsResolved(worktreeIdByTabId.get(key)))
+        break
+      case 'paneKeyed':
+        adoptRecord(next, host, field, (key) =>
+          adoptsResolved(worktreeIdForPaneKey(worktreeIdByTabId, key))
+        )
+        break
+      case 'sleepingAgentKeyed':
+      case 'surfaceTombstoneKeyed':
+        // Keyed opaquely, but each record names its own workspace — the only routing left once the
+        // tab or pane it describes is gone, and the same one `splitWorkspaceSessionByHost` uses.
+        adoptRecord(next, host, field, (_key, entry) => {
+          const workspaceId = recordWorkspaceId(entry)
+          return workspaceId !== null && adopts(workspaceId)
+        })
+        break
+      case 'browserWorkspaceKeyed':
+        adoptRecord(next, host, field, (_key, entry) => {
+          const workspaceId = browserPagesWorkspaceId(entry)
+          return workspaceId !== null && adopts(workspaceId)
+        })
+        break
+      case 'fileKeyed':
+        // Routed by the open file's workspace, through the same index the split routed it by.
+        adoptRecord(next, host, field, (key) => adoptsResolved(worktreeIdByFileId.get(key)))
+        break
+    }
+  }
+  return next
+}
@@ -85,12 +85,22 @@ test.describe('SSH cold activation restore', () => {
          () =>
            orcaPage.evaluate(
              async ({ targetId, worktreeId, expectedTabIds }) => {
-                const session = await window.api.session.get()
+                // Why both partitions: an SSH worktree's session lives in `ssh:<targetId>`, and
+                // only globals like `activeConnectionIdsAtShutdown` stay in `local`. Reading
+                // `session.get()` alone asserts the partition layout rather than the invariant,
+                // which is that the state is persisted where the boot read will find it.
+                const [local, host] = await Promise.all([
+                  window.api.session.get(),
+                  window.api.session.get(`ssh:${targetId}`)
+                ])
                const persistedTabIds = new Set(
-                  (session.tabsByWorktree[worktreeId] ?? []).map((tab) => tab.id)
+                  [
+                    ...(local.tabsByWorktree[worktreeId] ?? []),
+                    ...(host.tabsByWorktree[worktreeId] ?? [])
+                  ].map((tab) => tab.id)
                )
                return (
-                  session.activeConnectionIdsAtShutdown?.includes(targetId) === true &&
+                  local.activeConnectionIdsAtShutdown?.includes(targetId) === true &&
                  expectedTabIds.every((tabId) => persistedTabIds.has(tabId))
                )
              },
@@ -251,10 +261,18 @@ test.describe('SSH cold activation restore', () => {
          () =>
            firstLaunch.page.evaluate(
              async ({ targetId, worktreeId, tabId }) => {
-                const persisted = await window.api.session.get()
+                // See the note above: the worktree's rows are in `ssh:<targetId>`, the globals in
+                // `local`.
+                const [local, host] = await Promise.all([
+                  window.api.session.get(),
+                  window.api.session.get(`ssh:${targetId}`)
+                ])
                return (
-                  persisted.activeConnectionIdsAtShutdown?.includes(targetId) === true &&
-                  persisted.tabsByWorktree[worktreeId]?.some((tab) => tab.id === tabId) === true
+                  local.activeConnectionIdsAtShutdown?.includes(targetId) === true &&
+                  [
+                    ...(local.tabsByWorktree[worktreeId] ?? []),
+                    ...(host.tabsByWorktree[worktreeId] ?? [])
+                  ].some((tab) => tab.id === tabId)
                )
              },
              { targetId: remote.targetId, worktreeId: remote.worktreeId, tabId: restoredTabId }
@@ -131,12 +131,22 @@ async function flushSessionBeforeQuit(
      () =>
        page.evaluate(
          async ({ targetId, worktreeId, tabIds }) => {
-            const persisted = await window.api.session.get()
+            // Why both partitions: an SSH worktree's rows live in `ssh:<targetId>` and only globals
+            // like `activeConnectionIdsAtShutdown` stay in `local`. Reading `session.get()` alone
+            // asserts the partition layout rather than the invariant, which is that the state is
+            // persisted where the boot read will find it.
+            const [local, host] = await Promise.all([
+              window.api.session.get(),
+              window.api.session.get(`ssh:${targetId}`)
+            ])
            const persistedIds = new Set(
-              (persisted.tabsByWorktree[worktreeId] ?? []).map((tab) => tab.id)
+              [
+                ...(local.tabsByWorktree[worktreeId] ?? []),
+                ...(host.tabsByWorktree[worktreeId] ?? [])
+              ].map((tab) => tab.id)
            )
            return (
-              persisted.activeConnectionIdsAtShutdown?.includes(targetId) === true &&
+              local.activeConnectionIdsAtShutdown?.includes(targetId) === true &&
              tabIds.every((tabId) => persistedIds.has(tabId))
            )
          },