perf(layout): re-project only the written partition in the layout stream (#28845)

The layout stream (layout.subscribe, still unused) re-loaded every session
partition and ran the load's save-and-diff report after each write. A save
now names the partition it wrote, the stream re-loads only that partition
without the report, and the renderer layout client no longer re-subscribes
on a timer. Adds a latency spec variant with a subscriber attached.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

src/main/persistence-host-partitioned-sessions.test.ts

@@ -314,6 +314,20 @@ describe('Store host-partitioned workspace sessions', () => {
     expect(store.getWorkspaceSession('local').activeRepoId).toBe('repo-local')
   })
 
+  it('names the one partition each session write changed', async () => {
+    const store = await createStore()
+    store.setWorkspaceSession(makeHostSession('repo-a'), 'runtime:env-a')
+    const written: (string | null)[] = []
+    store.onWorkspaceSessionWritten((hostId) => written.push(hostId))
+
+    store.setWorkspaceSession(makeHostSession('repo-local'), 'local')
+    store.setWorkspaceSession(makeHostSession('repo-a2'), 'runtime:env-a')
+    store.patchWorkspaceSession({ activeTabId: 'tab-a' }, 'runtime:env-a')
+    store.removeWorkspaceSessionHost('runtime:env-a')
+
+    expect(written).toEqual(['local', 'runtime:env-a', 'runtime:env-a', 'runtime:env-a'])
+  })
+
   it('patches a single host partition without touching the others', async () => {
     const store = await createStore()
     store.setWorkspaceSession(makeHostSession('repo-local'), 'local')

src/main/persistence/loading-store/session-host-partitions.ts

@@ -93,9 +93,11 @@ export class SessionHostPartitionOperations {
     const { [hostId]: _removed, ...remaining } = partitions
     runtime.state.workspaceSessionsByHostId = remaining
     invalidateLocalWorktreeMetadataPruneInputs()
-    scheduleSave(this[sessionHostPartitionOperationsContext].scheduling, [
-      'workspaceSessionsByHostId'
-    ])
+    scheduleSave(
+      this[sessionHostPartitionOperationsContext].scheduling,
+      ['workspaceSessionsByHostId'],
+      hostId
+    )
   }
 
   readTerminalScrollbackSnapshot(ref: string): string | null {
@@ -170,7 +172,8 @@ export function removeWorkspaceSessionOwnerInPartition(
   }
   scheduleSave(
     owner[sessionHostPartitionOperationsContext].scheduling,
-    resolved === LOCAL_EXECUTION_HOST_ID ? ['workspaceSession'] : ['workspaceSessionsByHostId']
+    resolved === LOCAL_EXECUTION_HOST_ID ? ['workspaceSession'] : ['workspaceSessionsByHostId'],
+    resolved
   )
 }
 
@@ -234,9 +237,11 @@ export function setHostWorkspaceSession(
     ...owner[sessionHostPartitionOperationsContext].runtime.state.workspaceSessionsByHostId,
     [hostId]: withRequiredWorkspaceSessionMaps(orderTerminalRowsByTabGroups(pruned))
   }
-  scheduleSave(owner[sessionHostPartitionOperationsContext].scheduling, [
-    'workspaceSessionsByHostId'
-  ])
+  scheduleSave(
+    owner[sessionHostPartitionOperationsContext].scheduling,
+    ['workspaceSessionsByHostId'],
+    hostId
+  )
 }
 
 /**

src/main/persistence/loading-store/session-snapshot-operations.ts

@@ -128,7 +128,8 @@ export class SessionSnapshotOperations {
     }
     scheduleSave(
       scheduling,
-      hostId === LOCAL_EXECUTION_HOST_ID ? ['workspaceSession'] : ['workspaceSessionsByHostId']
+      hostId === LOCAL_EXECUTION_HOST_ID ? ['workspaceSession'] : ['workspaceSessionsByHostId'],
+      hostId
     )
   }
 

src/main/persistence/loading-store/store-runtime-state.ts

@@ -1,6 +1,7 @@
 import { removeStaleDurableWriteTempFiles } from '../../durable-file-write'
 import type { GlobalSettings } from '../../../shared/global-settings-types'
 import type { PersistedState } from '../../../shared/persisted-state-types'
+import type { WorkspaceSessionWriteListener } from './workspace-session-write-listeners'
 import type { ActiveViewPreference } from '../../active-view-preference'
 import {
   getProfileTerminalScrollbackSnapshotRoot,
@@ -84,7 +85,7 @@ export class StoreRuntimeState {
   >()
   uiChangeListeners = new Set<(ui: PersistedState['ui']) => void>()
   /** Observers of persisted workspace-session writes; see notifyWorkspaceSessionWritten. */
-  readonly workspaceSessionWriteListeners = new Set<() => void>()
+  readonly workspaceSessionWriteListeners = new Set<WorkspaceSessionWriteListener>()
   projectHostOperations: ProjectHostPersistenceOperations | null = null
   projectGroupOperations: ProjectGroupPersistenceOperations | null = null
   folderWorkspaceOperations: FolderWorkspacePersistenceOperations | null = null

src/main/persistence/loading-store/workspace-session-snapshot-publication.ts

@@ -124,11 +124,12 @@ export function setLocalWorkspaceSession(
     remappedManualUnread.changed
   ) {
     // UI remaps include protected fields, so keep the complete serializer boundary.
-    scheduleSave(context.scheduling)
+    scheduleSave(context.scheduling, undefined, LOCAL_EXECUTION_HOST_ID)
   } else {
     scheduleSave(
       context.scheduling,
-      remappedLeases.changed ? ['workspaceSession', 'sshRemotePtyLeases'] : ['workspaceSession']
+      remappedLeases.changed ? ['workspaceSession', 'sshRemotePtyLeases'] : ['workspaceSession'],
+      LOCAL_EXECUTION_HOST_ID
     )
   }
 }

src/main/persistence/loading-store/workspace-session-write-listeners.ts

@@ -1,19 +1,27 @@
+import type { ExecutionHostId } from '../../../shared/execution-host'
 import type { StoreRuntimeState } from './store-runtime-state'
 
+/** Gets the one partition a write changed, or null when any partition may have changed. */
+export type WorkspaceSessionWriteListener = (hostId: ExecutionHostId | null) => void
+
 const WORKSPACE_SESSION_DOMAINS: readonly string[] = [
   'workspaceSession',
   'workspaceSessionsByHostId'
 ]
 
-/** `dirtyDomains` undefined means a full save, which may carry an in-place session edit. */
+/**
+ * `dirtyDomains` undefined means a full save, which may carry an in-place session edit.
+ * `hostId` names the only partition the write changed; without it every partition is suspect.
+ */
 export function notifyWorkspaceSessionWritten(
   runtime: Pick<StoreRuntimeState, 'workspaceSessionWriteListeners'>,
-  dirtyDomains?: readonly string[]
+  dirtyDomains?: readonly string[],
+  hostId?: ExecutionHostId
 ): void {
   if (dirtyDomains && !dirtyDomains.some((domain) => WORKSPACE_SESSION_DOMAINS.includes(domain))) {
     return
   }
   for (const listener of runtime.workspaceSessionWriteListeners) {
-    listener()
+    listener(hostId ?? null)
   }
 }

src/main/persistence/loading-store/write-scheduling.ts

@@ -1,7 +1,11 @@
 import type { StoreRuntimeState } from './store-runtime-state'
 import type { PrimaryStateWriteOperations } from './primary-state-writes'
 import { enqueueWrite } from './primary-state-writes'
-import { notifyWorkspaceSessionWritten } from './workspace-session-write-listeners'
+import type { ExecutionHostId } from '../../../shared/execution-host'
+import {
+  notifyWorkspaceSessionWritten,
+  type WorkspaceSessionWriteListener
+} from './workspace-session-write-listeners'
 
 const SAVE_DEBOUNCE_MS = 1_000
 const SAVE_MAX_WAIT_MS = 5_000
@@ -40,7 +44,7 @@ export class WriteSchedulingOperations {
   }
 
   /** Called after each in-memory workspace-session write; listeners must not throw or block. */
-  onWorkspaceSessionWritten(listener: () => void): () => void {
+  onWorkspaceSessionWritten(listener: WorkspaceSessionWriteListener): () => void {
     const { runtime } = this[writeSchedulingOperationsContext]
     runtime.workspaceSessionWriteListeners.add(listener)
     return () => {
@@ -49,12 +53,14 @@ export class WriteSchedulingOperations {
   }
 }
 
+/** `writtenHostId` names the one session partition this save changed, if it changed only one. */
 export function scheduleSave(
   owner: WriteSchedulingOperations,
-  dirtyDomains?: readonly string[]
+  dirtyDomains?: readonly string[],
+  writtenHostId?: ExecutionHostId
 ): void {
   const { runtime, writes } = owner[writeSchedulingOperationsContext]
-  notifyWorkspaceSessionWritten(runtime, dirtyDomains)
+  notifyWorkspaceSessionWritten(runtime, dirtyDomains, writtenHostId)
   runtime.automationListProjectionCache = null
   const trackedDomains = runtime.dirtyProfileStateDomains
   if (dirtyDomains === undefined) {

src/main/runtime/workspace-layout-stream.test.ts

@@ -3,39 +3,77 @@ import { LOCAL_EXECUTION_HOST_ID, type ExecutionHostId } from '../../shared/exec
 import type { WorkspaceSessionState } from '../../shared/workspace-session-state-types'
 import {
   localDesktopSession,
-  relaySshSession
+  relaySshSession,
+  serverRuntimeSession
 } from '../../shared/workspace-layout/workspace-layout-profile.test-fixture'
 import {
   emptySession,
   FOLDER_KEY,
   GIT_KEY,
+  SERVER_KEY,
   SSH_KEY
 } from '../../shared/workspace-layout/workspace-layout-session.test-fixture'
 import type { WorkspaceLayoutEvent } from '../../shared/workspace-layout/workspace-layout-stream-frames'
+import { loadWorkspaceLayout } from '../../shared/workspace-layout/workspace-layout-load'
+import { nameBasedLoadContext } from '../../shared/workspace-layout/workspace-layout-minted-ids'
+import {
+  publishWorkspaceLayout,
+  type PublishedWorkspaceLayout
+} from '../../shared/workspace-layout/workspace-layout-published'
 import { WorkspaceLayoutStream } from './workspace-layout-stream'
 
 const SSH_HOST: ExecutionHostId = 'ssh:target-1'
+const SERVER_HOST: ExecutionHostId = 'runtime:env-1'
 
 function fakeStore(partitions: Map<ExecutionHostId, WorkspaceSessionState>) {
-  const listeners = new Set<() => void>()
-  const onWorkspaceSessionWritten = vi.fn((listener: () => void) => {
+  const listeners = new Set<(hostId: ExecutionHostId | null) => void>()
+  const onWorkspaceSessionWritten = vi.fn((listener: (hostId: ExecutionHostId | null) => void) => {
     listeners.add(listener)
     return () => listeners.delete(listener)
   })
+  const getWorkspaceSession = vi.fn((hostId: ExecutionHostId) => partitions.get(hostId)!)
   return {
     store: {
-      getWorkspaceSession: (hostId: ExecutionHostId) => partitions.get(hostId)!,
+      getWorkspaceSession,
       getWorkspaceSessionHostIds: () => [...partitions.keys()],
       onWorkspaceSessionWritten
     },
     listeners,
+    /** A save that names the partition it wrote. */
     write(hostId: ExecutionHostId, session: WorkspaceSessionState) {
       partitions.set(hostId, session)
-      listeners.forEach((listener) => listener())
+      listeners.forEach((listener) => listener(hostId))
+    },
+    /** A save that names no partition, so every partition may have changed. */
+    writeUnnamed(hostId: ExecutionHostId, session: WorkspaceSessionState) {
+      partitions.set(hostId, session)
+      listeners.forEach((listener) => listener(null))
+    },
+    /** A removed host's partition, named by the save that drops it. */
+    removePartition(hostId: ExecutionHostId) {
+      partitions.delete(hostId)
+      listeners.forEach((listener) => listener(hostId))
     }
   }
 }
 
+/** Today's projection: every partition loaded with its save-and-diff report, then filtered by home. */
+function fullProjection(
+  partitions: Map<ExecutionHostId, WorkspaceSessionState>,
+  homeHostId: (key: string) => ExecutionHostId | null
+): [string, PublishedWorkspaceLayout][] {
+  const projected = new Map<string, PublishedWorkspaceLayout>()
+  for (const [hostId, session] of partitions) {
+    const { layout } = loadWorkspaceLayout(hostId, session, nameBasedLoadContext())
+    for (const [key, workspace] of Object.entries(layout.workspaces)) {
+      if (homeHostId(key) === hostId) {
+        projected.set(key, publishWorkspaceLayout(workspace, hostId))
+      }
+    }
+  }
+  return [...projected]
+}
+
 function setup() {
   const partitions = new Map<ExecutionHostId, WorkspaceSessionState>([
     [LOCAL_EXECUTION_HOST_ID, localDesktopSession()],
@@ -51,7 +89,7 @@ function setup() {
     store: () => fake.store,
     homeHostId: (key) => homes[key] ?? null
   })
-  return { ...fake, stream, homes }
+  return { ...fake, stream, homes, partitions }
 }
 
 const flush = () => new Promise<void>((resolve) => queueMicrotask(resolve))
@@ -199,4 +237,86 @@ describe('WorkspaceLayoutStream', () => {
       expect(JSON.stringify(later.tabs)).toBe(JSON.stringify(earlier.tabs.slice(1)))
     }
   })
+
+  it('re-projects only the partition a write names', async () => {
+    const { stream, store, write, writeUnnamed } = setup()
+    stream.subscribe(() => {})
+    store.getWorkspaceSession.mockClear()
+    write(SSH_HOST, relaySshSession('target-1'))
+    await flush()
+    expect(store.getWorkspaceSession.mock.calls).toEqual([[SSH_HOST]])
+    store.getWorkspaceSession.mockClear()
+    writeUnnamed(SSH_HOST, relaySshSession('target-1'))
+    await flush()
+    expect(store.getWorkspaceSession.mock.calls).toEqual([[LOCAL_EXECUTION_HOST_ID], [SSH_HOST]])
+  })
+
+  it('publishes what a full re-projection with the load report would, write after write', async () => {
+    const { stream, write, writeUnnamed, removePartition, homes, partitions } = setup()
+    const homeHostId = (key: string) => homes[key] ?? null
+    let cache = new Map<string, PublishedWorkspaceLayout>()
+    const { snapshot } = stream.subscribe((event) => {
+      if (event.type === 'removed') {
+        cache.delete(event.key)
+      } else {
+        cache.set(event.key, event.layout)
+      }
+    })
+    cache = new Map(snapshot.map((entry) => [entry.key, entry.layout]))
+    const expectFullProjection = async (): Promise<void> => {
+      await flush()
+      const expected = fullProjection(partitions, homeHostId)
+      expect(Object.fromEntries(cache)).toEqual(Object.fromEntries(expected))
+      // A late subscriber's snapshot keeps today's order too.
+      const late = stream.subscribe(() => {})
+      expect(late.snapshot.map((entry) => [entry.key, entry.layout])).toEqual(expected)
+      late.unsubscribe()
+    }
+    await expectFullProjection()
+
+    const local = localDesktopSession()
+    local.tabGroups![GIT_KEY]![0]!.tabOrder.reverse()
+    write(LOCAL_EXECUTION_HOST_ID, local)
+    await expectFullProjection()
+
+    // A named write and an unnamed one in the same task.
+    const ssh = relaySshSession('target-1')
+    delete ssh.tabsByWorktree[SSH_KEY]
+    delete ssh.unifiedTabs![SSH_KEY]
+    write(SSH_HOST, ssh)
+    writeUnnamed(LOCAL_EXECUTION_HOST_ID, localDesktopSession())
+    await expectFullProjection()
+
+    // A new partition, a home that moves between partitions, then the partition removed.
+    homes[SERVER_KEY] = SERVER_HOST
+    write(SERVER_HOST, serverRuntimeSession('env-1'))
+    await expectFullProjection()
+    homes[GIT_KEY] = SERVER_HOST
+    write(SSH_HOST, relaySshSession('target-1'))
+    await expectFullProjection()
+    homes[GIT_KEY] = LOCAL_EXECUTION_HOST_ID
+    removePartition(SERVER_HOST)
+    await expectFullProjection()
+  })
+
+  it('re-projects every partition after a failed projection', async () => {
+    const { stream, store, write } = setup()
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
+    const events: WorkspaceLayoutEvent[] = []
+    stream.subscribe((event) => events.push(event))
+    store.getWorkspaceSession.mockImplementationOnce(() => {
+      throw new Error('read failed')
+    })
+    const local = localDesktopSession()
+    local.tabGroups![GIT_KEY]![0]!.tabOrder.reverse()
+    write(LOCAL_EXECUTION_HOST_ID, local)
+    await flush()
+    expect(events).toEqual([])
+    // The local edit the failed projection missed is published by a write elsewhere.
+    write(SSH_HOST, relaySshSession('target-1'))
+    await flush()
+    expect(events.map((event) => [event.type, event.key])).toEqual([['workspace', GIT_KEY]])
+    expect(warn).toHaveBeenCalledTimes(1)
+    warn.mockRestore()
+  })
 })

src/main/runtime/workspace-layout-stream.ts

@@ -1,9 +1,10 @@
 // The read-only layout stream behind `layout.subscribe`: each workspace's published layout,
-// re-projected from the Store's sessions after every write while anyone subscribes. Nothing reads
-// it until a client opts in; with no subscriber it does no work.
+// re-projected from the Store's sessions after every write while anyone subscribes. A write that
+// names its partition re-projects only that partition. Nothing reads it until a client opts in;
+// with no subscriber it does no work.
 
 import type { ExecutionHostId } from '../../shared/execution-host'
-import { loadWorkspaceLayout } from '../../shared/workspace-layout/workspace-layout-load'
+import { loadWorkspaceLayoutWithoutReport } from '../../shared/workspace-layout/workspace-layout-load'
 import { stableJson } from '../../shared/workspace-layout/workspace-layout-load-report'
 import {
   publishWorkspaceLayout,
@@ -26,8 +27,11 @@ export class WorkspaceLayoutStream {
   private readonly listeners = new Set<LayoutListener>()
   /** Workspace key → its published layout, with its JSON for the by-value comparison. */
   private published = new Map<string, Published>()
+  /** Partition → every workspace it holds, its own copy of a workspace it does not own included. */
+  private projectedPartitions = new Map<ExecutionHostId, Map<string, Published>>()
   private stopWatching: (() => void) | null = null
-  private dirty = false
+  /** Partitions to re-project at the next flush; 'all' when a write did not name its partition. */
+  private dirty: Set<ExecutionHostId> | 'all' | null = null
   private failureLogged = false
 
   constructor(
@@ -45,9 +49,10 @@ export class WorkspaceLayoutStream {
   } {
     if (this.listeners.size === 0) {
       this.stopWatching =
-        this.deps.store()?.onWorkspaceSessionWritten?.(() => this.markDirty()) ?? null
+        this.deps.store()?.onWorkspaceSessionWritten?.((hostId) => this.markDirty(hostId)) ?? null
       this.published = new Map()
-      this.dirty = true
+      this.projectedPartitions = new Map()
+      this.dirty = 'all'
     }
     // A write not yet published reaches existing listeners first, so the snapshot is current.
     this.flush()
@@ -62,28 +67,38 @@ export class WorkspaceLayoutStream {
         this.stopWatching?.()
         this.stopWatching = null
         this.published.clear()
+        this.projectedPartitions.clear()
       }
     }
   }
 
-  private markDirty(): void {
-    if (this.dirty) {
-      return
+  private markDirty(hostId: ExecutionHostId | null): void {
+    const scheduled = this.dirty !== null
+    if (hostId === null) {
+      this.dirty = 'all'
+    } else if (this.dirty === null) {
+      this.dirty = new Set([hostId])
+    } else if (this.dirty !== 'all') {
+      this.dirty.add(hostId)
+    }
+    if (!scheduled) {
+      // Coalesces a burst of writes in one task into one projection.
+      queueMicrotask(() => this.flush())
     }
-    this.dirty = true
-    // Coalesces a burst of writes in one task into one projection.
-    queueMicrotask(() => this.flush())
   }
 
   private flush(): void {
-    if (!this.dirty) {
+    const dirty = this.dirty
+    if (dirty === null) {
       return
     }
-    this.dirty = false
+    this.dirty = null
     // A failed projection is logged once; it never reaches the write that notified it.
     try {
-      this.reconcile()
+      this.reconcile(dirty)
     } catch (error) {
+      // The cache may be part-updated; the next write re-projects every partition.
+      this.projectedPartitions.clear()
       this.logFailure('projection', error)
     }
   }
@@ -95,28 +110,43 @@ export class WorkspaceLayoutStream {
     }
   }
 
-  private project(): Map<string, Published> {
-    const store = this.deps.store()
+  private projectPartition(hostId: ExecutionHostId): Map<string, Published> {
     const projected = new Map<string, Published>()
-    for (const hostId of store?.getWorkspaceSessionHostIds?.() ?? []) {
-      const session = store?.getWorkspaceSession?.(hostId)
-      if (!session) {
-        continue
-      }
-      const { layout } = loadWorkspaceLayout(hostId, session, nameBasedLoadContext())
-      for (const [key, workspace] of Object.entries(layout.workspaces)) {
+    const session = this.deps.store()?.getWorkspaceSession?.(hostId)
+    if (!session) {
+      return projected
+    }
+    const { layout } = loadWorkspaceLayoutWithoutReport(hostId, session, nameBasedLoadContext())
+    for (const [key, workspace] of Object.entries(layout.workspaces)) {
+      const published = publishWorkspaceLayout(workspace, hostId)
+      projected.set(key, { layout: published, json: stableJson(published) })
+    }
+    return projected
+  }
+
+  private project(dirty: Set<ExecutionHostId> | 'all'): Map<string, Published> {
+    const hostIds = this.deps.store()?.getWorkspaceSessionHostIds?.() ?? []
+    const partitions = new Map<ExecutionHostId, Map<string, Published>>()
+    for (const hostId of hostIds) {
+      const cached =
+        dirty === 'all' || dirty.has(hostId) ? undefined : this.projectedPartitions.get(hostId)
+      partitions.set(hostId, cached ?? this.projectPartition(hostId))
+    }
+    this.projectedPartitions = partitions
+    const projected = new Map<string, Published>()
+    for (const [hostId, workspaces] of partitions) {
+      for (const [key, workspace] of workspaces) {
         // Only the owning partition's copy is published; a stray copy elsewhere is not shown.
         if (this.deps.homeHostId(key) === hostId) {
-          const layout = publishWorkspaceLayout(workspace, hostId)
-          projected.set(key, { layout, json: stableJson(layout) })
+          projected.set(key, workspace)
         }
       }
     }
     return projected
   }
 
-  private reconcile(): void {
-    const next = this.project()
+  private reconcile(dirty: Set<ExecutionHostId> | 'all'): void {
+    const next = this.project(dirty)
     const events: WorkspaceLayoutEvent[] = []
     for (const key of this.published.keys()) {
       if (!next.has(key)) {

src/renderer/src/runtime/runtime-layout-client.test.ts

@@ -142,15 +142,6 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta
     return host.streams[count - 1]!
   }
 
-  /** Whether a retry opens exactly `delayMs` after the previous loss. */
-  async function expectRetryAfter(delayMs: number): Promise<FakeStream> {
-    const count = host.streams.length
-    await vi.advanceTimersByTimeAsync(delayMs - 1)
-    expect(host.streams).toHaveLength(count)
-    await vi.advanceTimersByTimeAsync(1)
-    return opened(count + 1)
-  }
-
   function snapshot(workspaces: { key: string; layout: PublishedWorkspaceLayout }[] = []) {
     return { type: 'snapshot', subscriptionId: 'layout-1', workspaces }
   }
@@ -186,7 +177,7 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta
     expect(stream.unsubscribed).toBe(false)
   })
 
-  it('after a lost stream, resubscribes and the new snapshot replaces the cache', async () => {
+  it('after a lost stream, keeps the last state and never reopens it', async () => {
     subscribe()
     const first = await opened(1)
     first.send(
@@ -197,36 +188,27 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta
     )
     first.drop()
     expect(first.unsubscribed).toBe(true)
-    // The last state stays until the runtime sends a new one; nothing is cleared or guessed.
-    expect(cache).toEqual({ a: layout('a', ['t1', 't2']), b: layout('b') })
-    const second = await expectRetryAfter(250)
-    // A late frame from the dropped stream changes nothing.
+    expect(status).toEqual({ state: 'lost' })
+    await vi.advanceTimersByTimeAsync(60_000)
+    expect(host.streams).toHaveLength(1)
+    // The last state stays; a late frame from the dropped stream changes nothing.
     first.send({ type: 'removed', key: 'a' })
-    expect(cache).toHaveProperty('a')
-    second.send(snapshot([{ key: 'a', layout: layout('a', ['t2']) }]))
-    expect(cache).toEqual({ a: layout('a', ['t2']) })
+    expect(cache).toEqual({ a: layout('a', ['t1', 't2']), b: layout('b') })
   })
 
-  it('backs off on failed subscribes and error frames, and a snapshot alone does not reset it', async () => {
-    host.rejectNextSubscribe()
-    subscribe()
-    await vi.advanceTimersByTimeAsync(0)
-    expect(host.streams).toHaveLength(0)
-    const first = await expectRetryAfter(250)
-    first.fail('stream failed')
-    expect(first.unsubscribed).toBe(true)
-    // A host that snapshots then drops at once keeps backing off instead of looping fast.
-    let stream = await expectRetryAfter(500)
-    for (const delay of [1000, 2000, 4000, 5000, 5000]) {
-      stream.send(snapshot())
-      stream.drop()
-      stream = await expectRetryAfter(delay)
-    }
-    // A change after the snapshot proves the stream stayed up; the next loss starts over.
-    stream.send(snapshot())
-    stream.send({ type: 'workspace', key: 'a', layout: layout('a') })
-    stream.drop()
-    await expectRetryAfter(250)
+  it.each([
+    ['a failed subscribe', () => host.rejectNextSubscribe(), async () => {}],
+    ['an error frame', () => {}, async () => (await opened(1)).fail('stream failed')]
+  ])('stops after %s without retrying', async (_case, before, fail) => {
+    before()
+    const subscription = subscribe()
+    await fail()
+    await vi.advanceTimersByTimeAsync(60_000)
+    expect(status).toEqual({ state: 'lost' })
+    expect(host.streams.length).toBeLessThanOrEqual(1)
+    expect(host.streams.every((stream) => stream.unsubscribed)).toBe(true)
+    subscription.close()
+    expect(status).toEqual({ state: 'lost' })
   })
 
   it.each([
@@ -260,7 +242,7 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta
     expect(status).toMatchObject({ state: 'refused', code: 'method_not_found' })
   })
 
-  it('close unsubscribes, stops retrying and ignores later frames', async () => {
+  it('close unsubscribes and ignores later frames', async () => {
     const subscription = subscribe()
     const stream = await opened(1)
     stream.send(snapshot())
@@ -274,15 +256,6 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta
     expect(cache).toEqual({})
   })
 
-  it('close while waiting to retry cancels the retry', async () => {
-    const subscription = subscribe()
-    ;(await opened(1)).drop()
-    expect(status).toEqual({ state: 'retrying' })
-    subscription.close()
-    await vi.advanceTimersByTimeAsync(10_000)
-    expect(host.streams).toHaveLength(1)
-  })
-
   it('close before the subscribe settles releases the stream it then gets', async () => {
     const subscription = subscribe()
     subscription.close()

src/renderer/src/runtime/runtime-layout-client.ts

@@ -11,20 +11,19 @@ import {
 import type { RuntimeClientTarget } from '../../../shared/runtime-client-target'
 import { subscribeRuntimeRpc } from './runtime-rpc-subscribe'
 
-/** `retrying`: waiting to reopen a lost stream. `refused`: the host answered that it will never
- *  serve this stream, so the client stopped. */
+/** `lost`: the stream ended without a refusal and the client stopped. `refused`: the host answered
+ *  that it will never serve this stream, so the client stopped. */
 export type RuntimeLayoutStreamStatus =
-  | { state: 'connecting' | 'live' | 'retrying' | 'closed' }
+  | { state: 'connecting' | 'live' | 'lost' | 'closed' }
   | { state: 'refused'; code: string }
 
-const RETRY_BASE_MS = 250
-const RETRY_MAX_MS = 5000
 // An older host without the method, or a caller scope denied it: a retry gets the same answer.
 const REFUSAL_CODES = ['method_not_found', 'forbidden'] as const
 
 /**
- * Opens the stream and keeps it open: after a lost stream it resubscribes, and the new snapshot
- * replaces the cache. `onFrame` gets a snapshot first, then that stream's changes.
+ * Opens the stream once; `onFrame` gets a snapshot first, then that stream's changes. A local
+ * stream ends only on close, document retirement or a refusal, so nothing reopens it. A remote
+ * stream that drops reports `lost`; reopening it belongs to the connection's reconnect.
  */
 export function subscribeRuntimeLayout(
   target: RuntimeClientTarget,
@@ -35,8 +34,6 @@ export function subscribeRuntimeLayout(
   // Bumped whenever a stream is given up, so its late callbacks change nothing.
   let generation = 0
   let handle: { unsubscribe: () => void } | null = null
-  let retryTimer: ReturnType<typeof setTimeout> | null = null
-  let failures = 0
   let state: RuntimeLayoutStreamStatus['state'] = 'connecting'
 
   const setStatus = (status: RuntimeLayoutStreamStatus): void => {
@@ -52,10 +49,6 @@ export function subscribeRuntimeLayout(
 
   const stop = (status: RuntimeLayoutStreamStatus): void => {
     release()
-    if (retryTimer !== null) {
-      clearTimeout(retryTimer)
-      retryTimer = null
-    }
     setStatus(status)
   }
 
@@ -67,18 +60,7 @@ export function subscribeRuntimeLayout(
       console.warn('[workspace-layout] layout stream lost:', error)
     }
     const refusal = REFUSAL_CODES.find((code) => hasRuntimeRpcErrorCode(error, code))
-    if (refusal) {
-      stop({ state: 'refused', code: refusal })
-      return
-    }
-    release()
-    setStatus({ state: 'retrying' })
-    const delay = Math.min(RETRY_BASE_MS * 2 ** failures, RETRY_MAX_MS)
-    failures += 1
-    retryTimer = setTimeout(() => {
-      retryTimer = null
-      void open()
-    }, delay)
+    stop(refusal ? { state: 'refused', code: refusal } : { state: 'lost' })
   }
 
   const receive = (streamGeneration: number, frame: WorkspaceLayoutStreamFrame | null): void => {
@@ -95,9 +77,6 @@ export function subscribeRuntimeLayout(
     } else if (state !== 'live') {
       // The host sends the snapshot first; a change before it has nothing to replace.
       return
-    } else {
-      // A change after the snapshot proves the stream stays up, unlike a snapshot then a drop.
-      failures = 0
     }
     onFrame(frame)
   }
@@ -124,7 +103,7 @@ export function subscribeRuntimeLayout(
   void open()
   return {
     close: () => {
-      if (state !== 'closed' && state !== 'refused') {
+      if (state === 'connecting' || state === 'live') {
         stop({ state: 'closed' })
       }
     }

src/shared/workspace-layout/workspace-layout-load.ts

@@ -22,6 +22,7 @@ import {
 } from './workspace-layout-load-bindings'
 import { diffStoredSession } from './workspace-layout-load-report'
 import type {
+  LoadedWorkspaceLayout,
   WorkspaceLayoutLoadContext,
   WorkspaceLayoutLoadResult
 } from './workspace-layout-load-types'
@@ -120,6 +121,16 @@ export function loadWorkspaceLayout(
   stored: WorkspaceSessionState,
   context: WorkspaceLayoutLoadContext
 ): WorkspaceLayoutLoadResult {
+  const loaded = loadWorkspaceLayoutWithoutReport(hostId, stored, context)
+  return { ...loaded, changes: diffStoredSession(stored, saveWorkspaceLayout(loaded)) }
+}
+
+/** The same load without the save and diff, for a reader that never writes the result back. */
+export function loadWorkspaceLayoutWithoutReport(
+  hostId: ExecutionHostId,
+  stored: WorkspaceSessionState,
+  context: WorkspaceLayoutLoadContext
+): LoadedWorkspaceLayout {
   // The Loader edits its own copy; the Store's object stays untouched.
   const session = structuredClone(stored)
   const desktopView: DesktopLayoutView = {
@@ -174,6 +185,5 @@ export function loadWorkspaceLayout(
   loadLeaves(layout, { session, layouts }, { view: desktopView, facts, carried }, context)
   unbindDuplicateTerminals(layout)
   layout = applyLegacySurfaceTombstones(layout, session, carried)
-  const loaded = { layout, desktopView, facts, carried }
-  return { ...loaded, changes: diffStoredSession(stored, saveWorkspaceLayout(loaded)) }
+  return { layout, desktopView, facts, carried }
 }

tests/e2e/workspace-layout-oracle-latency.spec.ts

@@ -2,8 +2,9 @@
  * Measures command → drawn update for layout commands, idle and with three panes flooding output
  * (`yes`), one of them in a hidden worktree, and time to first drawn after relaunching a large
  * profile. Measuring only: it records each number in the oracle report and never fails on one,
- * so a later budget can gate on it. Runs with no layout subscriber, so main is measured as users
- * run it. The committed main baseline is workspace-layout-oracle-latency-baseline.json.
+ * so a later budget can gate on it. The plain run has no layout subscriber, so main is measured as
+ * users run it; the subscribed run holds one `layout.subscribe` stream open in the window, as the
+ * layout client will. The committed main baseline is workspace-layout-oracle-latency-baseline.json.
  *
  * Each command is timed twice. The in-app clock (helpers/layout-latency-probe.ts) stops in the
  * window at React's commit and at the next frame. The outer clock stops when the test sees the
@@ -12,7 +13,7 @@
 
 import { mkdirSync, writeFileSync } from 'node:fs'
 import path from 'node:path'
-import type { Page, TestInfo } from '@stablyai/playwright-test'
+import type { ElectronApplication, Page, TestInfo } from '@stablyai/playwright-test'
 import { expect, test } from './helpers/orca-app'
 import { createRestartSession } from './helpers/orca-restart'
 import { bootstrapFirstLaunch, seededRepoPathOrSkip } from './helpers/terminal-restart-persistence'
@@ -57,6 +58,53 @@ const LARGE_PROFILE = { hiddenWorktrees: 12, tabsPerWorktree: 3, activePanes: 4
 type Condition = 'idle' | 'flood'
 type Sample = { outerMs: number; inApp: InAppSample }
 type Samples = Record<string, Sample[]>
+/** Main's memory at a phase's start and end, in MB; no GC is forced, so read it as a trend. */
+type MainMemory = { heapUsedMb: [number, number]; rssMb: [number, number] }
+/** What the window's layout stream received during a phase. */
+type StreamTraffic = { frames: number; kb: number }
+
+declare global {
+  // oxlint-disable-next-line typescript-eslint/consistent-type-definitions -- declaration merging requires interface
+  interface Window {
+    __layoutStreamTraffic?: { frames: number; bytes: number; errors: number }
+  }
+}
+
+async function readMainMemory(
+  app: ElectronApplication
+): Promise<{ heapUsed: number; rss: number }> {
+  return app.evaluate(() => {
+    const { heapUsed, rss } = process.memoryUsage()
+    return { heapUsed, rss }
+  })
+}
+
+/** Holds one `layout.subscribe` stream open in the window for the rest of the run. */
+async function attachLayoutSubscriber(page: Page): Promise<void> {
+  await page.evaluate(async () => {
+    const traffic = { frames: 0, bytes: 0, errors: 0 }
+    window.__layoutStreamTraffic = traffic
+    await window.api.runtime.subscribe({ method: 'layout.subscribe', params: {} }, (response) => {
+      if (!response.ok) {
+        traffic.errors += 1
+        return
+      }
+      traffic.frames += 1
+      traffic.bytes += JSON.stringify(response.result).length
+    })
+  })
+  await expect
+    .poll(() => page.evaluate(() => window.__layoutStreamTraffic?.frames ?? 0))
+    .toBeGreaterThan(0)
+}
+
+async function readStreamTraffic(page: Page): Promise<{ frames: number; bytes: number }> {
+  const traffic = await page.evaluate(() => window.__layoutStreamTraffic ?? null)
+  if (traffic && traffic.errors > 0) {
+    throw new Error(`layout stream reported ${traffic.errors} error frame(s)`)
+  }
+  return { frames: traffic?.frames ?? 0, bytes: traffic?.bytes ?? 0 }
+}
 
 /** Reads can briefly fail while the window re-publishes its terminals; only reads are retried. */
 async function read<T>(client: RuntimeClient, method: string, params: unknown): Promise<T> {
@@ -243,9 +291,11 @@ type Stats = ReturnType<typeof stats>
 type StartupSample = { firstDrawnMs: number; eventLoopDelayMax: number }
 type LatencyReport =
   | {
-      scenario: 'latency'
+      scenario: 'latency' | 'latency-subscribed'
       summary: Record<string, { n: number; inAppCommit: Stats; inAppDrawn: Stats; outer: Stats }>
       eventLoopDelay: Record<string, EventLoopDelaySummary>
+      mainMemory: Record<Condition, MainMemory>
+      streamTraffic?: Record<Condition, StreamTraffic>
       closeConfirms: Record<Condition, number>
       samples: Samples
     }
@@ -271,9 +321,7 @@ function writeLatencyReport(testInfo: TestInfo, { samples, ...summary }: Latency
   )
 }
 
-// oxlint-disable-next-line no-empty-pattern -- The measurement owns its launch.
-test('layout oracle: command to drawn latency', async ({}, testInfo) => {
-  test.setTimeout(90 * 60_000)
+async function measureLatency(testInfo: TestInfo, subscribed: boolean): Promise<void> {
   const repoPath = seededRepoPathOrSkip()
   const session = createRestartSession(testInfo)
   const { app, page } = await session.launch()
@@ -284,12 +332,22 @@ test('layout oracle: command to drawn latency', async ({}, testInfo) => {
     const worktree = `id:${worktreeId}`
     const client = new RuntimeClient(session.userDataDir, 30_000)
     await waitForBoundPanes(page, 1)
-    const phases: Record<Condition, { from: number; to: number; confirms: number }> = {
-      idle: { from: Date.now(), to: 0, confirms: 0 },
-      flood: { from: 0, to: 0, confirms: 0 }
+    if (subscribed) {
+      await attachLayoutSubscriber(page)
     }
-    phases.idle.confirms = await measureCondition(page, client, worktreeId, 'idle', samples)
-    phases.idle.to = Date.now()
+    type Mark = { at: number; heapUsed: number; rss: number; frames: number; bytes: number }
+    const mark = async (): Promise<Mark> => ({
+      at: Date.now(),
+      ...(await readMainMemory(app)),
+      ...(await readStreamTraffic(page))
+    })
+    type Phase = { from: Mark; to: Mark; confirms: number }
+    const measurePhase = async (condition: Condition): Promise<Phase> => {
+      const from = await mark()
+      const confirms = await measureCondition(page, client, worktreeId, condition, samples)
+      return { from, to: await mark(), confirms }
+    }
+    const idle = await measurePhase('idle')
 
     // Two flooding tabs beside the measured one and one in a worktree that is never shown.
     const list = await read<RuntimeTerminalListResult>(client, 'terminal.list', { worktree })
@@ -305,9 +363,7 @@ test('layout oracle: command to drawn latency', async ({}, testInfo) => {
     }
     await client.call('terminal.focus', { terminal: measured.handle, navigation: 'host' })
     await waitForActiveTerminalManager(page)
-    phases.flood.from = Date.now()
-    phases.flood.confirms = await measureCondition(page, client, worktreeId, 'flood', samples)
-    phases.flood.to = Date.now()
+    const flood = await measurePhase('flood')
 
     const summary = Object.fromEntries(
       Object.entries(samples).map(([action, values]) => [
@@ -320,18 +376,30 @@ test('layout oracle: command to drawn latency', async ({}, testInfo) => {
         }
       ])
     )
-    const eventLoopDelay = Object.fromEntries(
-      Object.entries(phases).map(([condition, phase]) => [
-        condition,
-        summarizeEventLoopDelay(session.userDataDir, 'main', phase.from, phase.to)
-      ])
-    )
-    const closeConfirms = { idle: phases.idle.confirms, flood: phases.flood.confirms }
+    const perCondition = <T>(read: (phase: Phase) => T): Record<Condition, T> => ({
+      idle: read(idle),
+      flood: read(flood)
+    })
+    const mb = (bytes: number): number => Math.round(bytes / 1024 / 102.4) / 10
     writeLatencyReport(testInfo, {
-      scenario: 'latency',
+      scenario: subscribed ? 'latency-subscribed' : 'latency',
       summary,
-      eventLoopDelay,
-      closeConfirms,
+      eventLoopDelay: perCondition(({ from, to }) =>
+        summarizeEventLoopDelay(session.userDataDir, 'main', from.at, to.at)
+      ),
+      mainMemory: perCondition(({ from, to }) => ({
+        heapUsedMb: [mb(from.heapUsed), mb(to.heapUsed)],
+        rssMb: [mb(from.rss), mb(to.rss)]
+      })),
+      ...(subscribed
+        ? {
+            streamTraffic: perCondition(({ from, to }) => ({
+              frames: to.frames - from.frames,
+              kb: Math.round((to.bytes - from.bytes) / 1024)
+            }))
+          }
+        : {}),
+      closeConfirms: perCondition(({ confirms }) => confirms),
       samples
     })
   } finally {
@@ -341,6 +409,18 @@ test('layout oracle: command to drawn latency', async ({}, testInfo) => {
     await session.close(app)
     await session.dispose()
   }
+}
+
+// oxlint-disable-next-line no-empty-pattern -- The measurement owns its launch.
+test('layout oracle: command to drawn latency', async ({}, testInfo) => {
+  test.setTimeout(90 * 60_000)
+  await measureLatency(testInfo, false)
+})
+
+// oxlint-disable-next-line no-empty-pattern -- The measurement owns its launch.
+test('layout oracle: command to drawn latency with a layout subscriber', async ({}, testInfo) => {
+  test.setTimeout(90 * 60_000)
+  await measureLatency(testInfo, true)
 })
 
 // oxlint-disable-next-line no-empty-pattern -- The measurement owns its launches.