The window's layout stream reconnects after a drop; a retried quit still sends resume records (dark, layout-owner build only) (#30380)

* chore(layout): the drift check's ignore comment names every main-only kind (dark, layout-owner build only)

* fix(layout): a quit stage retried after a failed one still sends the resume records (dark, layout-owner build only)

* feat(layout): the window's local layout stream reopens after a loss, and its snapshot redraws every workspace (dark, layout-owner build only)
This commit is contained in:
Jinwoo Hong
2026-10-18 10:25:37 -07:00
committed by GitHub
parent 447c4a0fb2
commit 93ed192025
9 changed files with 298 additions and 36 deletions
@@ -195,7 +195,8 @@ export function windowSaveDivergences(
...rawPaths.keys()
])
const result: WindowSaveDivergence[] = []
// Ignored, as main's own writes: pane bindings, resume records, default-tab marks, closed-tab records, row owners.
// Ignored, as main's own writes: pane bindings, resume records, default-tab marks, closed-tab records, row owners,
// clientHostedBrowserPagesByWorkspace and the topology revision.
for (const workspace of workspaces) {
const paths = [
...diffWorkspaceLayouts(was?.workspaces[workspace], now?.workspaces[workspace]),
@@ -0,0 +1,75 @@
import { describe, expect, it, vi } from 'vitest'
import type { SleepingAgentSessionRecord } from '../../../shared/agent-session-resume'
import type { AppState } from '../store/types'
import { createQuitSleepRecordCapture } from './quit-sleep-record-capture'
import { createShutdownCheckpointPersist } from './shutdown-checkpoint-persist'
const RECORD: SleepingAgentSessionRecord = {
paneKey: 'tab-1:11111111-1111-4111-8111-111111111111',
worktreeId: 'wt-1',
agent: 'claude',
providerSession: { key: 'session_id', id: 'session-1' },
prompt: '',
state: 'working',
capturedAt: 1,
updatedAt: 1
}
/** A window holding tab-1, whose quit capture records RECORD (and then finds it unchanged). */
function windowStore() {
let records: AppState['sleepingAgentSessionsByPaneKey'] = {}
const state = (): AppState =>
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the capture reads only these fields.
({
tabsByWorktree: { 'wt-1': [{ id: 'tab-1' }] },
sleepingAgentSessionsByPaneKey: records,
captureAllSleepingAgentSessions: () => {
records = { [RECORD.paneKey]: RECORD }
}
}) as unknown as AppState
return { getState: state }
}
function persist(stage: (args: { sleepRecords?: SleepingAgentSessionRecord[] }) => void) {
const capture = createQuitSleepRecordCapture(windowStore())
return createShutdownCheckpointPersist({
shouldCaptureSession: () => true,
captureTerminalBuffers: vi.fn(),
captureSleepingAgentSessions: capture.capture,
onSleepRecordsStaged: capture.staged,
buildSessionSnapshots: () => [],
buildUiPatch: () => ({}),
hasDirtyOpenFiles: () => false,
isDegradableShutdownInProgress: () => false,
stageBeforeUnloadSync: stage
})
}
describe('the quit capture’s resume records in a layout-owner build', () => {
it('sends the first capture’s records again on a stage retried after a failure', () => {
const staged: (SleepingAgentSessionRecord[] | undefined)[] = []
let fail = true
const checkpoint = persist(({ sleepRecords }) => {
staged.push(sleepRecords)
if (fail) {
throw new Error('stage failed')
}
})
expect(() => checkpoint.run()).toThrow('stage failed')
fail = false
checkpoint.run()
expect(staged).toEqual([[RECORD], [RECORD]])
})
it('sends nothing again once a stage carried them', () => {
const staged: (SleepingAgentSessionRecord[] | undefined)[] = []
const checkpoint = persist(({ sleepRecords }) => staged.push(sleepRecords))
checkpoint.run()
checkpoint.run()
expect(staged).toEqual([[RECORD], undefined])
})
})
@@ -0,0 +1,27 @@
import type { AppState } from '../store/types'
import type { SleepingAgentSessionRecord } from '../../../shared/agent-session-resume'
import { changedSleepingRecords } from '../store/terminals/terminal-sleep-layout-owner'
/**
* A layout-owner build's quit capture: the resume records it changed of tabs the window holds,
* since the last stage that carried them, so a stage retried after a failed one still sends the
* first capture's records.
*/
export function createQuitSleepRecordCapture(store: { getState: () => AppState }): {
capture: () => SleepingAgentSessionRecord[]
staged: () => void
} {
let unstagedSince: AppState['sleepingAgentSessionsByPaneKey'] | null = null
return {
capture: () => {
const previous = unstagedSince ?? store.getState().sleepingAgentSessionsByPaneKey
unstagedSince = previous
store.getState().captureAllSleepingAgentSessions('quit')
const state = store.getState()
return changedSleepingRecords(state, previous, state.sleepingAgentSessionsByPaneKey).recorded
},
staged: () => {
unstagedSince = null
}
}
}
@@ -17,6 +17,8 @@ export type ShutdownCheckpointPersistDeps = {
captureTerminalBuffers: () => void
/** Returns the records it changed when main records them (a layout-owner build). */
captureSleepingAgentSessions: () => SleepingAgentSessionRecord[] | void
/** The stage carried the capture's records: a retry's capture may diff from here. */
onSleepRecordsStaged?: () => void
buildSessionSnapshots: () => WorkspaceSessionHostSnapshot[]
buildUiPatch: () => Partial<PersistedUIState>
hasDirtyOpenFiles: () => boolean
@@ -82,6 +84,9 @@ export function createShutdownCheckpointPersist(
ui: deps.buildUiPatch(),
...(sleepRecords && sleepRecords.length > 0 && !degraded ? { sleepRecords } : {})
})
if (!degraded) {
deps.onSleepRecordsStaged?.()
}
} catch (error) {
// A durable-only stage has nothing safer left to fall back to.
if (degraded) {
@@ -20,8 +20,7 @@ import {
createShutdownCheckpointGuard
} from '../lib/shutdown-checkpoint-guard'
import { createShutdownCheckpointPersist } from './shutdown-checkpoint-persist'
import { changedSleepingRecords } from '../store/terminals/terminal-sleep-layout-owner'
import type { SleepingAgentSessionRecord } from '../../../shared/agent-session-resume'
import { createQuitSleepRecordCapture } from './quit-sleep-record-capture'
import { shutdownBufferCaptures } from '../components/terminal-pane/shutdown-buffer-captures'
import {
dispatchWindowCloseRequest,
@@ -36,14 +35,6 @@ import {
ORCA_RENDERER_UNLOAD_PREVENTED_EVENT
} from '../../../shared/renderer-shutdown-events'
/** The quit capture, returning the resume records it changed of tabs the window holds. */
function captureQuitSleepRecordsForMain(): SleepingAgentSessionRecord[] {
const previous = useAppStore.getState().sleepingAgentSessionsByPaneKey
useAppStore.getState().captureAllSleepingAgentSessions('quit')
const state = useAppStore.getState()
return changedSleepingRecords(state, previous, state.sleepingAgentSessionsByPaneKey).recorded
}
// Why: bound the resume-record loss window on a hard kill to ~1 min; capture skips unchanged records so per-tick cost is negligible.
const SLEEPING_AGENT_RESUME_CAPTURE_INTERVAL_MS = 60_000
@@ -77,6 +68,9 @@ export function useAppSessionPersistence(): void {
// two firings, PTY exit events can arrive and unmount TerminalPanes,
// emptying shutdownBufferCaptures. The guard prevents the second call
// from overwriting the good session data with an empty snapshot.
const quitSleepRecords = ORCA_LAYOUT_OWNER_IS_MAIN
? createQuitSleepRecordCapture(useAppStore)
: null
const shutdownCheckpointPersist = createShutdownCheckpointPersist({
shouldCaptureSession: () => shouldPersistWorkspaceSession(useAppStore.getState()),
captureTerminalBuffers: () => {
@@ -93,9 +87,10 @@ export function useAppSessionPersistence(): void {
// map so a daemon/session death while the app is closed can still
// cold-restore via the agent's resume command (#5232).
// Switch point: the stage writes only the capture's changes into main's records.
captureSleepingAgentSessions: ORCA_LAYOUT_OWNER_IS_MAIN
? captureQuitSleepRecordsForMain
captureSleepingAgentSessions: quitSleepRecords
? quitSleepRecords.capture
: () => useAppStore.getState().captureAllSleepingAgentSessions('quit'),
onSleepRecordsStaged: quitSleepRecords?.staged,
// Why: re-read state after capture() calls populated scrollback buffers
// into the store via Zustand setters. The shouldCaptureSession read is
// only for the gating flags and would miss those updates.
@@ -203,29 +203,35 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta
expect(positions).toEqual([3, 4, 5, 6])
})
it('after a lost stream, keeps the last state and never reopens it', async () => {
subscribe()
const first = await opened(1)
first.send(
snapshot([
{ key: 'a', layout: layout('a', ['t1', 't2']) },
{ key: 'b', layout: layout('b') }
])
)
first.drop()
expect(first.unsubscribed).toBe(true)
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).toEqual({ a: layout('a', ['t1', 't2']), b: layout('b') })
})
it.skipIf(target.kind === 'local')(
'after a lost stream, keeps the last state and never reopens it',
async () => {
subscribe()
const first = await opened(1)
first.send(
snapshot([
{ key: 'a', layout: layout('a', ['t1', 't2']) },
{ key: 'b', layout: layout('b') }
])
)
first.drop()
expect(first.unsubscribed).toBe(true)
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).toEqual({ a: layout('a', ['t1', 't2']), b: layout('b') })
}
)
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) => {
if (target.kind === 'local') {
return
}
before()
const subscription = subscribe()
await fail()
@@ -289,3 +295,94 @@ describe.each(targets)('runtime layout client over the %s transport', (_name, ta
await vi.waitFor(() => expect(stream.unsubscribed).toBe(true))
})
})
describe('the local layout stream after a loss', () => {
let host: FakeHost
let cache: WorkspaceLayoutCache
let statuses: RuntimeLayoutStreamStatus[]
beforeEach(() => {
vi.useFakeTimers()
vi.spyOn(console, 'warn').mockImplementation(() => {})
host = installFakeHost({ kind: 'local' })
cache = EMPTY_WORKSPACE_LAYOUT_CACHE
statuses = []
})
afterEach(() => {
vi.useRealTimers()
vi.restoreAllMocks()
vi.unstubAllGlobals()
})
const subscribe = () =>
subscribeRuntimeLayout(
{ kind: 'local' },
{},
(frame) => {
cache = applyLayoutFrame(cache, frame)
},
(next) => {
statuses.push(next)
}
)
async function opened(count: number): Promise<FakeStream> {
await vi.waitFor(() => expect(host.streams).toHaveLength(count))
await vi.advanceTimersByTimeAsync(0)
return host.streams[count - 1]!
}
const snapshot = (workspaces: { key: string; layout: PublishedWorkspaceLayout }[]) => ({
type: 'snapshot',
subscriptionId: 'layout-1',
workspaces
})
it('reopens it, and the new snapshot replaces the cache', async () => {
subscribe()
const first = await opened(1)
first.send(
snapshot([
{ key: 'a', layout: layout('a') },
{ key: 'b', layout: layout('b') }
])
)
first.drop()
expect(statuses.at(-1)).toEqual({ state: 'lost' })
await vi.advanceTimersByTimeAsync(1_000)
const second = await opened(2)
expect(statuses.at(-1)).toEqual({ state: 'connecting' })
second.send(snapshot([{ key: 'a', layout: layout('a', ['t1']) }]))
expect(statuses.at(-1)).toEqual({ state: 'live' })
expect(cache).toEqual({ a: layout('a', ['t1']) })
// A late frame from the dropped stream changes nothing.
first.send({ type: 'removed', key: 'a' })
expect(cache).toEqual({ a: layout('a', ['t1']) })
})
it('waits longer after each failed reopen, and close stops a pending one', async () => {
const subscription = subscribe()
;(await opened(1)).drop()
host.rejectNextSubscribe()
await vi.advanceTimersByTimeAsync(1_000)
expect(statuses.at(-1)).toEqual({ state: 'lost' })
await vi.advanceTimersByTimeAsync(1_999)
expect(host.streams).toHaveLength(1)
await vi.advanceTimersByTimeAsync(1)
await opened(2)
host.streams[1]!.drop()
subscription.close()
expect(statuses.at(-1)).toEqual({ state: 'closed' })
await vi.advanceTimersByTimeAsync(60_000)
expect(host.streams).toHaveLength(2)
})
it('never reopens after a refusal', async () => {
subscribe()
;(await opened(1)).fail('no layout stream', 'method_not_found')
await vi.advanceTimersByTimeAsync(60_000)
expect(host.streams).toHaveLength(1)
})
})
@@ -21,9 +21,13 @@ export type RuntimeLayoutStreamStatus =
// 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
/** First wait before a lost local stream reopens; it doubles to the cap until a snapshot comes. */
const LOCAL_REOPEN_DELAY_MS = 1_000
const LOCAL_REOPEN_DELAY_CAP_MS = 30_000
/**
* 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
* Opens the stream; `onFrame` gets a snapshot first, then that stream's changes. A lost local
* stream reports `lost`, then reopens, and its snapshot replaces what the caller holds. A remote
* stream that drops reports `lost`; reopening it belongs to the connection's reconnect.
*/
export function subscribeRuntimeLayout(
@@ -38,6 +42,8 @@ export function subscribeRuntimeLayout(
let generation = 0
let handle: { unsubscribe: () => void } | null = null
let state: RuntimeLayoutStreamStatus['state'] = 'connecting'
let reopenTimer: ReturnType<typeof setTimeout> | null = null
let reopenDelay = LOCAL_REOPEN_DELAY_MS
const setStatus = (status: RuntimeLayoutStreamStatus): void => {
state = status.state
@@ -64,6 +70,15 @@ export function subscribeRuntimeLayout(
}
const refusal = REFUSAL_CODES.find((code) => hasRuntimeRpcErrorCode(error, code))
stop(refusal ? { state: 'refused', code: refusal } : { state: 'lost' })
// Why: layout.apply still reaches main without the stream, so only main's own changes would
// go unseen until a relaunch.
if (!refusal && target.kind === 'local') {
reopenTimer = setTimeout(() => {
reopenTimer = null
void open()
}, reopenDelay)
reopenDelay = Math.min(reopenDelay * 2, LOCAL_REOPEN_DELAY_CAP_MS)
}
}
const receive = (streamGeneration: number, value: unknown): void => {
@@ -90,6 +105,7 @@ export function subscribeRuntimeLayout(
return
}
if (frame.type === 'snapshot') {
reopenDelay = LOCAL_REOPEN_DELAY_MS
setStatus({ state: 'live' })
} else if (state !== 'live') {
// The host sends the snapshot first; a change before it has nothing to replace.
@@ -120,7 +136,11 @@ export function subscribeRuntimeLayout(
void open()
return {
close: () => {
if (state === 'connecting' || state === 'live') {
if (reopenTimer) {
clearTimeout(reopenTimer)
reopenTimer = null
stop({ state: 'closed' })
} else if (state === 'connecting' || state === 'live') {
stop({ state: 'closed' })
}
}
@@ -21,7 +21,11 @@ import { reportWindowSessionWriteActivity } from '../lib/window-session-write-ac
import { createStoreSessionMockApi } from '../store/slices/store-session-test-harness'
import { createTestStore } from '../store/slices/store-test-helpers'
import type { LayoutFrameSlices } from '../store/workspace-layout-frame-slices'
import { afterLayoutFrame, endLayoutFramePositions } from './layout-frame-position'
import {
afterLayoutFrame,
endLayoutFramePositions,
layoutFramesFlowing
} from './layout-frame-position'
import { holdTabPaneTree, requestTabPaneTreeRedraw } from './layout-owner-pane-intents'
import { openWorkspaceLayoutFrameSync } from './workspace-layout-frame-sync'
@@ -344,6 +348,36 @@ describe('the window drawn from layout frames', () => {
sync.close()
})
it('after a lost stream reopens, its snapshot redraws every workspace from main', async () => {
const { store, sync, layouts } = await bootWithFrames('local', localDesktopSession())
sync.start()
await flushMicrotasks()
stream.onStatus!({ state: 'lost' })
expect(layoutFramesFlowing()).toBe(false)
// Drawn while no frame came; main's snapshot does not have it.
const drawn = { type: 'leaf' as const, leafId: 'drawn-by-the-window' }
store.setState({
terminalLayoutsByTabId: {
...store.getState().terminalLayoutsByTabId,
'tab-shell': { ...store.getState().terminalLayoutsByTabId['tab-shell']!, root: drawn }
}
})
stream.onStatus!({ state: 'connecting' })
expect(layoutFramesFlowing()).toBe(true)
const renamed = structuredClone(layouts)
renamed[GIT_KEY]!.tabs.find((tab) => tab.id === 'tab-unbound')!.customTitle = 'from main'
stream.onFrame!(snapshot(renamed))
stream.onStatus!({ state: 'live' })
await flushMicrotasks()
const rows = store.getState().tabsByWorktree[GIT_KEY]!
expect(rows.find((tab) => tab.id === 'tab-unbound')!.customTitle).toBe('from main')
expect(store.getState().terminalLayoutsByTabId['tab-shell']!.root).not.toEqual(drawn)
expect(sync.terminalTab(GIT_KEY, 'tab-shell')).toBeDefined()
sync.close()
})
it('shows a live title change of the focused pane at once, even while a write is owed', async () => {
const { store, sync, layouts } = await bootWithFrames('local', localDesktopSession())
sync.start()
@@ -259,6 +259,10 @@ export function openWorkspaceLayoutFrameSync(store: FrameSyncStore): WorkspaceLa
(frame) => {
cache = applyLayoutFrame(cache, frame)
if (frame.type === 'snapshot') {
// A reopened stream's snapshot redraws every workspace: the window drew alone meanwhile.
if (snapshotReceived) {
shown = {}
}
snapshotReceived = true
live = started
}
@@ -268,7 +272,11 @@ export function openWorkspaceLayoutFrameSync(store: FrameSyncStore): WorkspaceLa
if (status.state === 'lost' || status.state === 'refused') {
console.warn('[layout-owner] layout stream stopped:', status)
}
if (status.state !== 'connecting' && status.state !== 'live') {
if (status.state === 'connecting' && !streamOpen) {
// The local stream reopens after a loss; replies wait for its frames again.
streamOpen = true
beginLayoutFramePositions()
} else if (status.state !== 'connecting' && status.state !== 'live') {
streamOpen = false
endLayoutFramePositions()
}