diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-close-verdict.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-close-verdict.test.ts index 33ef8080c030..6c900f586bf1 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-close-verdict.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-close-verdict.test.ts @@ -116,14 +116,34 @@ async function runningTurn(): Promise { } /** What the settle wrote, read back from the journal the next reader opens. */ -async function settledTurn() { - await host.restoreReadableSessions([SESSION]) - const { items } = await host.journalSnapshot(SESSION) +async function settledTurn(reader = host) { + await reader.restoreReadableSessions([SESSION]) + const { items } = await reader.journalSnapshot(SESSION) const turn = items.map((item) => readAgentJournalTurn(item.body)).find(Boolean) const [settled] = [...selectStructuredAgentSettledTurns(items).values()] return { turn, settled } } +/** What the next launch reads: a host that has begun quitting opens no chat. */ +async function settledTurnAfterQuit() { + const state = hostTestState() + const reader = new StructuredAgentSessionHost({ + logger: createStructuredAgentSessionLogger(), + store: state.store, + adapter: adapter(), + journalDatabase: openTestJournalHostDatabase(state.root), + recoveryCapsule: new AgentSessionRecoveryCapsule(state.root), + claimKeyId: 'key-1', + mintSpawnToken: () => 'spawn-a', + now: () => HOST_TEST_NOW + }) + try { + return await settledTurn(reader) + } finally { + await reader.flushAllStreamedEvents() + } +} + function lastSummary(statuses: AgentSessionStatusEvent[]) { const last = statuses.at(-1) return last?.type === 'status' ? last.session : null @@ -247,7 +267,7 @@ describe('a turn cut short by closing its provider', () => { expect(statuses.findLast((event) => event.type === 'status')).toMatchObject({ session: { status: 'idle', turnOutcome: 'interruption' } }) - const { settled } = await settledTurn() + const { settled } = await settledTurnAfterQuit() expect( settled && describeNativeChatTurnStatus({ elapsedSeconds: 0, ...settled }) ).toMatchObject({ key: 'failedAfter' }) @@ -309,7 +329,7 @@ describe('a turn cut short by closing its provider', () => { await host.flushAllStreamedEvents({ trigger: 'quit' }) - const { turn } = await settledTurn() + const { turn } = await settledTurnAfterQuit() expect(turn).toMatchObject({ state: 'interrupted' }) expect(turn).not.toHaveProperty('outcome') }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts index 19af6d048d45..768dda441226 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts @@ -24,6 +24,7 @@ import { import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' import { StructuredAgentSessionIdleSweep } from './structured-agent-session-idle-sweep' import { AGENT_SESSION_NOT_ATTACHED } from './structured-agent-session-mutation-admission' +import { sessionTabListed } from './structured-agent-session-host-tabs' import { adapterSupportsRecord } from './structured-agent-session-provider-support' import { deferredStructuredAgentSessionLogger } from './structured-agent-session-logger' @@ -64,11 +65,8 @@ export function createStructuredAgentSessionConversationLifetime(host: { closeStructuredAgentSessionConversationUnderSerialize( { sessions, - closeStatus: (id) => { - const tabs = deps().store.getVisibleSessionTabIndex() - // A legacy store cannot say, so the row stays; restart is the boundary that forgets. - host.closeStatus(id, { listed: !tabs.present || tabs.sessionIds.includes(id) }) - } + // A legacy store cannot say, so the row stays; restart is the boundary that forgets. + closeStatus: (id) => host.closeStatus(id, { listed: sessionTabListed(deps().store, id) }) }, sessionId ) @@ -102,6 +100,7 @@ export function createStructuredAgentSessionConversationLifetime(host: { return { idleSweep, stopAgent, + isDisposed: (): boolean => disposed, /** Quit has begun: nothing opens a conversation or sweeps one after this. */ dispose: (): void => { disposed = true diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-tabs.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-tabs.ts index f74532ac9aef..2278d726a4ca 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-tabs.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-tabs.ts @@ -1,6 +1,7 @@ import type { AgentSessionRecord } from '../../../shared/agent-session-record' import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' +import { adapterSupportsRecord } from './structured-agent-session-provider-support' /** * Chat-tab visibility is the deletion funnel: every path that removes a chat as a user-facing @@ -37,37 +38,52 @@ export function setStructuredAgentSessionTabVisibility( return host.deps.store.setSessionTabVisibility(sessionId, visible, tabId) } +/** Whether the chat still has its tab. A legacy store with no tab index cannot say, so yes. */ +export function sessionTabListed( + store: { getVisibleSessionTabIndex: () => { present: boolean; sessionIds: string[] } }, + sessionId: string +): boolean { + const tabs = store.getVisibleSessionTabIndex() + return !tabs.present || tabs.sessionIds.includes(sessionId) +} + export type StructuredAgentSessionTab = { sessionId: string workspaceId: string agent: AgentSessionRecord['provider'] } -export function listStructuredAgentSessionTabs( - sessions: ReadonlyMap< - string, - { params: { location: { workspaceId: string }; provider: AgentSessionRecord['provider'] } } - > +/** Tabs from durable state alone: each requested id, once, in the order given, that has a record + * this host serves. Opens nothing: a chat whose history is unreadable keeps its tab and its read + * says why; one whose history is missing keeps its tab and reads empty. */ +export function listPersistedSessionTabs( + deps: { + store: { getRecord: (sessionId: string) => AgentSessionRecord | null } + adapter: Parameters[0] + }, + sessionIds: readonly string[] ): StructuredAgentSessionTab[] { - return [...sessions.entries()].map(([sessionId, session]) => ({ - sessionId, - workspaceId: session.params.location.workspaceId, - agent: session.params.provider - })) + const tabs = new Map() + for (const sessionId of sessionIds) { + const record = deps.store.getRecord(sessionId) + if (!record || tabs.has(sessionId) || !adapterSupportsRecord(deps.adapter, record)) { + continue + } + tabs.set(sessionId, { + sessionId, + workspaceId: record.location.workspaceId, + agent: record.provider + }) + } + return [...tabs.values()] } -type TabSessions = ReadonlyMap< - string, - { - child?: unknown - params: { location: { workspaceId: string }; provider: AgentSessionRecord['provider'] } - } -> +type TabSessions = ReadonlyMap /** The host's chat-tab surface; reads `host.deps` per call, so it sees the host's wrapped deps. */ export function createStructuredAgentSessionTabSurface( host: Parameters[0] & { - deps: { + deps: Parameters[0] & { store: Pick< AgentSessionRecordStore, 'getVisibleSessionTabIndex' | 'getSessionTabId' | 'showSessionTabs' @@ -78,7 +94,8 @@ export function createStructuredAgentSessionTabSurface( forgetStatus: (sessionId: string) => void ) { return { - listSessionTabs: () => listStructuredAgentSessionTabs(sessions), + /** From the record store and the given tab ids; opens no conversation. */ + listSessionTabs: (ids: readonly string[]) => listPersistedSessionTabs(host.deps, ids), getPersistedVisibleSessionTabIndex: () => host.deps.store.getVisibleSessionTabIndex(), getSessionTabId: (sessionId: string): string | null => host.deps.store.getSessionTabId(sessionId), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts index f66d2506804d..5287cc6a078c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts @@ -592,7 +592,7 @@ describe('restart', () => { const restoreReads = listRecords.mock.calls.length await host.restoreReadableSessions() - expect(host.listSessionTabs()).toEqual([ + expect(host.listSessionTabs([SESSION])).toEqual([ { sessionId: SESSION, workspaceId: 'workspace-1', agent: 'codex' } ]) const history = await host.history({ sessionId: SESSION, direction: 'tail' }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 5b5a4d19c3c1..471f01e48794 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -140,6 +140,7 @@ export class StructuredAgentSessionHost { resolveRecovery: (sessionId) => this.runtimeState.resolveRecovery(sessionId), serialize: (sessionId, task) => this.serialize(sessionId, task), hasSession: this.hasSession, + isDisposed: () => this.lifetime.isDisposed(), // Site 10: cannot overwrite a live entry — the restorer returns early on // `hasSession` inside the same serialized step as this `set`. onReadable: this.conversationDelivery.adoptOpened @@ -233,8 +234,7 @@ export class StructuredAgentSessionHost { reconcileRestartLeases = (): Promise => this.restore.reconcileRestartLeases() - restoreReadableSessions = (sessionIds?: readonly string[]): Promise => - this.restore.restoreReadableSessions(sessionIds) + restoreReadableSessions = (ids?: readonly string[]) => this.restore.restoreReadableSessions(ids) /** Make one persisted session addressable again; see `structured-agent-session-reveal`. */ revealSession = (sessionId: string): Promise => diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-persisted-tab-listing.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-persisted-tab-listing.test.ts new file mode 100644 index 000000000000..f30afc09770a --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-persisted-tab-listing.test.ts @@ -0,0 +1,125 @@ +// A restarted host lists a chat's tab from its record and the tab table, never from whether its +// history opened: a chat with no history on disk, and one whose history cannot be read, both keep +// their tab. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { isAgentSessionRefusalError } from '../../../shared/agent-session-wire-refusals' +import { closeTestJournalHostDatabases } from '../agent-session-journal/journal-host-database-test-support' +import { + createRestTestRig, + restTestChat, + sendRestTestMessage, + type RestTestRig +} from './structured-agent-session-rest-test-rig' +import { restTestOpens } from './structured-agent-session-rest-test-observations' + +let rig: RestTestRig + +beforeEach(async () => { + rig = await createRestTestRig() +}) + +afterEach(async () => { + await rig.dispose() + closeTestJournalHostDatabases() + vi.restoreAllMocks() +}) + +const tab = (sessionId: string, workspaceId = 'workspace-1') => ({ + sessionId, + workspaceId, + agent: 'codex' +}) + +describe('the tab list after a restart', () => { + it('lists every chat it is given without opening one', async () => { + await restTestChat(rig, 'session-a', { message: 'one' }) + await restTestChat(rig, 'session-b', { message: 'two' }) + await rig.crash() + const host = await rig.boot() + + expect(host.listSessionTabs(['session-a', 'session-b'])).toEqual([ + tab('session-a'), + tab('session-b') + ]) + expect(rig.adapter.historyFilePath).not.toHaveBeenCalled() + }) + + it('keeps a chat with no history on disk, which then reads empty and takes a send (T3)', async () => { + // Its first start failed before the conversation opened, so no history was ever written. + rig.adapter.acquire.mockRejectedValueOnce(new Error('spawn failed')) + vi.spyOn(console, 'warn').mockImplementation(() => undefined) + await expect(restTestChat(rig, 'session-new')).rejects.toThrow(/attach refused/) + await rig.store.setSessionTabVisibility('session-new', true) + await restTestChat(rig, 'session-old', { message: 'hello' }) + await rig.crash() + const host = await rig.boot() + const ids = ['session-new', 'session-old'] + + await host.restoreReadableSessions(ids) + + expect(host.listSessionTabs(ids)).toEqual([tab('session-new'), tab('session-old')]) + // The restore founds no history for it; the tab stands anyway. + expect(restTestOpens(rig, 'session-new')).toBe(0) + const page = await host.history({ sessionId: 'session-new', direction: 'tail' }) + expect(page.ok && page.page.items).toEqual([]) + const sent = await sendRestTestMessage(rig, 'session-new', 'first words') + expect(sent).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledOnce(), { timeout: 10_000 }) + const after = await host.history({ sessionId: 'session-new', direction: 'tail' }) + expect(JSON.stringify(after.ok && after.page)).toContain('first words') + }) + + it('keeps a chat whose history cannot be opened, and its read says why (T4)', async () => { + await restTestChat(rig, 'session-bad', { message: 'lost' }) + await restTestChat(rig, 'session-good', { message: 'kept' }) + await rig.crash() + const host = await rig.boot() + rig.adapter.historyFilePath.mockImplementation(async (sessionId) => { + if (sessionId === 'session-bad') { + throw new Error('EACCES: permission denied') + } + return null + }) + vi.spyOn(console, 'warn').mockImplementation(() => undefined) + const ids = ['session-bad', 'session-good'] + + await host.restoreReadableSessions(ids).catch(() => undefined) + + expect(host.listSessionTabs(ids)).toEqual([tab('session-bad'), tab('session-good')]) + const read = await host.history({ sessionId: 'session-bad', direction: 'tail' }).then( + () => null, + (error: unknown) => error + ) + expect(isAgentSessionRefusalError(read) && read.refusal.code).toBe( + 'agent_session_journal_unreadable' + ) + }) + + it('lists in the order it is given, once each, whatever order the records are in (T7)', async () => { + await restTestChat(rig, 'session-c') + await restTestChat(rig, 'session-a') + await restTestChat(rig, 'session-b') + await rig.crash() + const host = await rig.boot() + + expect(host.listSessionTabs(['session-b', 'session-c', 'session-a', 'session-b'])).toEqual([ + tab('session-b'), + tab('session-c'), + tab('session-a') + ]) + }) + + it('lists no chat without a record, and none this host cannot serve (T10, regression guard)', async () => { + await restTestChat(rig, 'session-served', { message: 'here' }) + await restTestChat(rig, 'session-gated', { workspaceId: 'workspace-gated', message: 'there' }) + await rig.crash() + const host = await rig.boot() + rig.unsupportedWorkspaceIds.add('workspace-gated') + const ids = ['session-served', 'session-gated', 'session-missing'] + + await host.restoreReadableSessions(ids) + + expect(host.listSessionTabs(ids)).toEqual([tab('session-served')]) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-restore.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-restore.test.ts index de2951a6292a..8805c7dcdb0e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-restore.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-restore.test.ts @@ -101,7 +101,7 @@ describe('structured session provider restore', () => { await restarted.restoreReadableSessions() - expect(restarted.listSessionTabs()).toEqual([ + expect(restarted.listSessionTabs([HOST_TEST_SESSION])).toEqual([ { sessionId: HOST_TEST_SESSION, workspaceId: 'workspace-1', agent: 'claude' } ]) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts index 8f110c3e69a2..441333324a9a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts @@ -45,6 +45,8 @@ describe('StructuredAgentSessionReadableRestorer', () => { resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), hasSession: () => false, + isDisposed: () => false, + isListed: () => true, onReadable: () => undefined }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-observations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-observations.ts new file mode 100644 index 000000000000..52b0def0cb06 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-observations.ts @@ -0,0 +1,24 @@ +// What a rest rig's run observed: how often each chat's journal opened, and the status rows its +// stream carried. + +import type { RestTestRig } from './structured-agent-session-rest-test-rig' + +export function restTestOpens(rig: RestTestRig, sessionId: string): number { + return rig.adapter.historyFilePath.mock.calls.filter(([id]) => id === sessionId).length +} + +/** The newest row the status stream carried for a session. */ +export function latestRestTestStatus(rig: RestTestRig, sessionId: string) { + for (const event of rig.statusEvents.toReversed()) { + if (event.type === 'status' && event.session.sessionId === sessionId) { + return event.session + } + if (event.type === 'snapshot') { + const found = event.sessions.find((session) => session.sessionId === sessionId) + if (found) { + return found + } + } + } + return undefined +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-rig.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-rig.ts index 0045684ccb37..2887fd7fb95e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-rig.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-rig.ts @@ -1,6 +1,8 @@ // A real host over a real store and journal, with a scripted provider and a clock the test moves, -// for the tests of a conversation that outlives its agent. The idle sweep runs on its own short -// interval; a test moves `clock.now` past the idle window and waits for the outcome. +// for the tests of a conversation that outlives its agent and of what a restarted host owes. The +// idle sweep runs on its own short interval; a test moves `clock.now` past the idle window and +// waits for the outcome. Every journal open calls `historyFilePath` once, so that mock is the open +// counter, and holding it holds the open. import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -8,6 +10,8 @@ import { join } from 'node:path' import { expect, vi, type Mock } from 'vitest' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' import type { AgentSessionMutationEnvelope, AgentSessionStatusEvent, @@ -23,6 +27,7 @@ import type { import { StructuredAgentSessionHost } from './structured-agent-session-host' import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types' import { + HOST_TEST_LOCATION, HOST_TEST_NOW, HOST_TEST_SESSION as SESSION, HOST_TEST_THREAD as THREAD, @@ -48,6 +53,8 @@ export type RestTestAdapter = { > holdsDispatch: Mock> readOptions: Mock> + /** Called once per journal open, with the session id; replace its implementation to hold one. */ + historyFilePath: Mock<(sessionId: string) => Promise> } export type RestTestRig = { @@ -63,36 +70,55 @@ export type RestTestRig = { forget: Mock readChildWork: Mock<(subject: unknown) => AgentChildWorkView[]> } + probeOwner: Mock<(record: AgentSessionRecord) => Promise> + /** Workspaces this host's adapter does not serve, as a platform gate would. */ + unsupportedWorkspaceIds: Set /** Opens a fresh host over the same store and journals: what a restart leaves behind. */ restart: (deps?: Partial) => Promise + /** A crash: nothing settles or flushes; the process's timers and handles simply stop. */ + crash: () => Promise + /** A new app run over the same files, with no flush first; the mocks and status stream start + * empty, so they hold what the new run saw and nothing the previous one did. */ + boot: (deps?: Partial) => Promise dispose: () => Promise } let ordinal = 0 let generations = 0 -export function acceptedDispatch(): AgentSessionDispatchOutcome { +/** The rig's own chat keeps the shared test thread; any other chat gets a thread of its own. */ +function restTestThread(sessionId: string): string { + return sessionId === SESSION ? THREAD : `thread-${sessionId}` +} + +export function acceptedDispatch(sessionId = SESSION): AgentSessionDispatchOutcome { ordinal += 1 return { state: 'accepted', - providerIdentity: { provider: 'codex', threadId: THREAD, turnId: `turn-${ordinal}`, ordinal } + providerIdentity: { + provider: 'codex', + threadId: restTestThread(sessionId), + turnId: `turn-${ordinal}`, + ordinal + } } } export function restTestSend( text: string, - fence = 1 + fence: number | null = 1, + sessionId = SESSION ): { envelope: AgentSessionMutationEnvelope; body: AgentJournalMessageItem } { const body = hostTestMessage(text) return { body, envelope: { - sessionId: SESSION, + sessionId, clientOperationId: hostTestOperationId(), expectedRuntimeFence: fence, payloadFingerprint: computeAgentSessionPayloadFingerprint({ method: 'agentSession.send', - sessionId: SESSION, + sessionId, fields: { body } }) } @@ -125,12 +151,14 @@ export function collectSubscriber(): { return { events, emit: (event) => events.push(event) } } +/** `root` is handed to the rig, which removes it on dispose; by default it makes its own. */ export async function createRestTestRig( - deps: Partial = {} + deps: Partial = {}, + options: { root?: string } = {} ): Promise { resetHostTestOperationIds() ordinal = 0 - const root = await mkdtemp(join(tmpdir(), 'orca-rest-')) + const root = options.root ?? (await mkdtemp(join(tmpdir(), 'orca-rest-'))) const clock = { now: HOST_TEST_NOW } const statusEvents: AgentSessionStatusEvent[] = [] const sink: RestTestRig['sink'] = { @@ -138,15 +166,19 @@ export async function createRestTestRig( forget: vi.fn(), readChildWork: vi.fn((): AgentChildWorkView[] => []) } - let store = await openTestAgentSessionRecordStore(root) + const probeOwner: RestTestRig['probeOwner'] = vi.fn(async () => ({ outcome: 'pid-absent' })) + const unsupportedWorkspaceIds = new Set() + const openStore = () => openTestAgentSessionRecordStore(root) + let store = await openStore() const adapter: RestTestAdapter = { - acquire: vi.fn(async ({ fence, spawnToken }) => ({ + acquire: vi.fn(async ({ fence, spawnToken, identity }) => ({ acquisitionGeneration: `generation-${++generations}`, process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000, spawnToken }, link: { - linkId: `link-${fence}`, - handle: { provider: 'codex' as const, threadId: THREAD }, - origin: store.getRecord(SESSION)?.providerHandleChain.length + linkId: + identity.sessionId === SESSION ? `link-${fence}` : `link-${identity.sessionId}-${fence}`, + handle: { provider: 'codex' as const, threadId: restTestThread(identity.sessionId) }, + origin: store.getRecord(identity.sessionId)?.providerHandleChain.length ? ('resumed' as const) : ('created' as const), mintedAtFence: fence, @@ -154,17 +186,21 @@ export async function createRestTestRig( } })), closeSession: vi.fn(async () => true), - dispatch: vi.fn(async () => acceptedDispatch()), + dispatch: vi.fn(async (input) => acceptedDispatch(input.sessionId)), acknowledgeSessionRelease: vi.fn(), holdsDispatch: vi.fn(() => false), - readOptions: vi.fn(async () => ({ models: [], current: { model: 'gpt-live' } })) + readOptions: vi.fn(async () => ({ models: [], current: { model: 'gpt-live' } })), + historyFilePath: vi.fn(async (_sessionId: string): Promise => null) } - const hostFor = (overrides: Partial) => - new StructuredAgentSessionHost({ + const hostFor = (overrides: Partial) => { + const host = new StructuredAgentSessionHost({ logger: createStructuredAgentSessionLogger(), store, adapter: { ...adapter, + historyFilePath: ({ identity }) => adapter.historyFilePath(identity.sessionId), + supportsCreate: (location, agent) => + agent === 'codex' && !unsupportedWorkspaceIds.has(location.workspaceId), releaseAcquisition: vi.fn(async () => true), cancelTurn: async () => ({ cancelled: true }), answerPrompt: async ({ commit }) => commit(), @@ -173,13 +209,22 @@ export async function createRestTestRig( journalDatabase: openTestJournalHostDatabase(root), claimKeyId: 'key-1', mintSpawnToken: () => 'spawn-a', - probeOwner: async () => ({ outcome: 'pid-absent' }), + probeOwner, now: () => clock.now, statusSink: sink, idleSweep: { intervalMs: SWEEP_INTERVAL_MS }, ...deps, ...overrides }) + host.subscribeStatus({ id: 'status', emit: (event) => statusEvents.push(event) }) + return host + } + const reopen = async (overrides: Partial) => { + store = await openStore() + rig.store = store + rig.host = hostFor(overrides) + return rig.host + } const rig: RestTestRig = { root, store, @@ -188,35 +233,88 @@ export async function createRestTestRig( clock, statusEvents, sink, + probeOwner, + unsupportedWorkspaceIds, restart: async (overrides = {}) => { await rig.host.flushAllStreamedEvents().catch(() => undefined) - store = await openTestAgentSessionRecordStore(root) - rig.store = store - rig.host = hostFor(overrides) - rig.host.subscribeStatus({ id: 'status', emit: (event) => statusEvents.push(event) }) - return rig.host + return reopen(overrides) + }, + crash: async () => { + const { runtimeState, lifetime, sessions } = rig.host.collaboratorsForTests() + await runtimeState.stopLeaseRenewal() + lifetime.dispose() + for (const session of sessions.values()) { + await session.journal.close() + } + sessions.clear() + }, + boot: (overrides = {}) => { + for (const mock of [adapter.historyFilePath, adapter.acquire, adapter.dispatch, probeOwner]) { + mock.mockClear() + } + sink.publish.mockClear() + sink.forget.mockClear() + statusEvents.length = 0 + return reopen(overrides) }, dispose: async () => { await rig.host.flushAllStreamedEvents().catch(() => undefined) await rm(root, { recursive: true, force: true }) } } - rig.host.subscribeStatus({ id: 'status', emit: (event) => statusEvents.push(event) }) return rig } -/** Creates the chat, lists its tab, and sends one message so its journal is on disk. */ -export async function foundRestTestChat(rig: RestTestRig): Promise { - const attached = await rig.host.attach(REST_TEST_CALLER, hostTestAttachParams(null)) +/** Creates the chat, lists its tab unless told not to, and sends `message` when one is given, + * waiting for its dispatch. */ +export async function restTestChat( + rig: RestTestRig, + sessionId: string, + options: { workspaceId?: string; listed?: boolean; message?: string } = {} +): Promise { + const fence = rig.store.getRecord(sessionId)?.lease.runtimeFence ?? null + const attached = await rig.host.attach( + REST_TEST_CALLER, + hostTestAttachParams(fence, { + envelope: { + sessionId, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: fence, + payloadFingerprint: '' + }, + location: { ...HOST_TEST_LOCATION, workspaceId: options.workspaceId ?? 'workspace-1' }, + providerHandle: { kind: 'codex', threadId: restTestThread(sessionId) } + }) + ) if (!attached.ok) { throw new Error(`attach refused: ${attached.refusal.code}`) } - await rig.store.setSessionTabVisibility(SESSION, true) - const sent = await rig.host.send(REST_TEST_CALLER, restTestSend('hello', attached.fence)) + if (options.listed !== false) { + await rig.store.setSessionTabVisibility(sessionId, true) + } + if (options.message === undefined) { + return + } + const { dispatch } = rig.adapter + const dispatched = dispatch.mock.calls.length + const params = restTestSend(options.message, attached.fence, sessionId) + const sent = await rig.host.send(REST_TEST_CALLER, params) if (!sent.ok) { throw new Error(`send refused: ${sent.refusal.code}`) } - await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalled()) + await vi.waitFor(() => expect(dispatch.mock.calls.length).toBeGreaterThan(dispatched), { + timeout: 10_000 + }) +} + +/** Creates the rig's own chat, lists its tab, and sends one message so its journal is on disk. */ +export function foundRestTestChat(rig: RestTestRig): Promise { + return restTestChat(rig, SESSION, { message: 'hello' }) +} + +/** A send from a client that has not attached this run, so it names no fence. */ +export function sendRestTestMessage(rig: RestTestRig, sessionId: string, text: string) { + return rig.host.send(REST_TEST_CALLER, restTestSend(text, null, sessionId)) } /** Runs one sweep pass now, for a test that set `idleSweep.intervalMs` out of reach. */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts index 441d4e80626d..dc3e966de82c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts @@ -37,7 +37,7 @@ const NO_OPEN_DEPS = { describe('restart journal restoration', () => { beforeEach(() => restoreRead.mockReset()) - it('bounds historical journal parsing to four sessions at a time', async () => { + it('opens one chat at a time', async () => { const gate = Promise.withResolvers() let active = 0 let peak = 0 @@ -68,16 +68,21 @@ describe('restart journal restoration', () => { resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), hasSession: () => false, + isDisposed: () => false, + isListed: () => true, onReadable: () => undefined }) - await vi.waitFor(() => expect(active).toBe(4)) - expect(restoreRead).toHaveBeenCalledTimes(4) + await vi.waitFor(() => expect(active).toBeGreaterThan(0)) + // Held long enough for a second lane, had there been one, to have started. + await new Promise((resolve) => setTimeout(resolve, 20)) + const startedWhileHeld = restoreRead.mock.calls.length gate.resolve() await restoration + expect(startedWhileHeld).toBe(1) expect(restoreRead).toHaveBeenCalledTimes(records.length) - expect(peak).toBe(4) + expect(peak).toBe(1) }) it('lets the event loop run between chats', async () => { @@ -109,6 +114,8 @@ describe('restart journal restoration', () => { resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), hasSession: () => false, + isDisposed: () => false, + isListed: () => true, onReadable: () => undefined }) ticking = false @@ -159,6 +166,8 @@ describe('restart journal restoration', () => { }, serialize: async (_sessionId, task) => task(), hasSession: () => false, + isDisposed: () => false, + isListed: () => true, onReadable: (_sessionId, readable) => { calls.push(readable === restored ? 'onReadable:restored' : 'onReadable') } @@ -186,6 +195,8 @@ describe('restart journal restoration', () => { ...bookkeeping, serialize: async (_sessionId, task) => task(), hasSession: () => false, + isDisposed: () => false, + isListed: () => true, onReadable: () => undefined }) @@ -233,6 +244,8 @@ describe('restart journal restoration', () => { resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), hasSession: () => true, + isDisposed: () => false, + isListed: () => true, onReadable: () => undefined }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts index 29ea17845d8a..3c5d0ddd67ba 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts @@ -13,7 +13,7 @@ import { setImmediate as yieldToEventLoop } from 'node:timers/promises' import type { AgentSessionRecord } from '../../../shared/agent-session-record' -import { mapWithConcurrency } from '../../../shared/map-with-concurrency' +import { mapSettledWithConcurrency } from '../../../shared/map-with-concurrency' import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' import type { OpenedStructuredAgentSessionConversation, @@ -21,7 +21,9 @@ import type { } from './structured-agent-session-conversation-open' import { restoreStructuredAgentSessionRead } from './structured-agent-session-read-restore' -const JOURNAL_RESTORE_CONCURRENCY = 4 +// One chat at a time: the restore is CPU-bound on the main thread, so more lanes only lengthen each +// event-loop turn that a user's read or send waits behind. +const JOURNAL_RESTORE_CONCURRENCY = 1 export type StructuredAgentSessionReadRestoreDeps = { openDeps: StructuredAgentSessionConversationOpenDeps & { @@ -34,6 +36,10 @@ export type StructuredAgentSessionReadRestoreDeps = { resolveRecovery: (sessionId: string) => Promise serialize: (sessionId: string, task: () => Promise) => Promise hasSession: (sessionId: string) => boolean + /** Quit has begun: the restore opens nothing more. */ + isDisposed: () => boolean + /** Whether the chat still has its tab: one closed while the restore ran stays closed. */ + isListed: (sessionId: string) => boolean onReadable: ( sessionId: string, opened: OpenedStructuredAgentSessionConversation @@ -46,6 +52,9 @@ async function restoreOneStructuredAgentSessionRead( sessionId: string, settleLeases: (sessionId: string) => Promise ): Promise { + if (input.isDisposed()) { + return + } await settleLeases(sessionId) await input.serialize(sessionId, () => restoreOneStructuredAgentSessionReadUnderSerialize(input, sessionId) @@ -54,9 +63,16 @@ async function restoreOneStructuredAgentSessionRead( /** The serialized half of the restore. */ async function restoreOneStructuredAgentSessionReadUnderSerialize( - input: Pick, + input: Pick< + StructuredAgentSessionReadRestoreDeps, + 'openDeps' | 'hasSession' | 'isDisposed' | 'isListed' | 'onReadable' + >, sessionId: string ): Promise { + // Checked under the chat's serialize, where its close and quit's teardown also run. + if (input.isDisposed() || !input.isListed(sessionId)) { + return + } if (input.hasSession(sessionId)) { // A read or a send mid-restore already opened this one. return @@ -89,9 +105,23 @@ export async function restoreStructuredAgentSessionsOnRestart( settled = false } } - await mapWithConcurrency(input.records, JOURNAL_RESTORE_CONCURRENCY, async ({ sessionId }) => { - // A journal open is synchronous SQLite: without a macrotask per chat the restore is one long task. - await yieldToEventLoop() - await restoreOneStructuredAgentSessionRead(input, sessionId, settleLeases) + const results = await mapSettledWithConcurrency( + input.records, + JOURNAL_RESTORE_CONCURRENCY, + async ({ sessionId }) => { + // A journal open is synchronous SQLite: without a macrotask per chat the restore is one long task. + await yieldToEventLoop() + await restoreOneStructuredAgentSessionRead(input, sessionId, settleLeases) + } + ) + // One chat's failure costs only that chat, which opens again when it is read. + results.forEach((result, index) => { + if (result.status === 'rejected') { + input.openDeps.logger.warn('restoring a chat for reading failed', { + scope: 'history-restore-chat', + sessionId: input.records[index]?.sessionId, + error: result.reason + }) + } }) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restore-after-listing.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restore-after-listing.test.ts new file mode 100644 index 000000000000..b77b3ff6be29 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restore-after-listing.test.ts @@ -0,0 +1,110 @@ +// The restore pass runs after the tab list is out, so the app is already usable while it works +// through the chats. A chat closed, or a host quit, before the pass reaches it stays that way. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { closeTestJournalHostDatabases } from '../agent-session-journal/journal-host-database-test-support' +import { + createRestTestRig, + restTestChat, + type RestTestRig +} from './structured-agent-session-rest-test-rig' +import { restTestOpens } from './structured-agent-session-rest-test-observations' + +const IDS = ['session-1', 'session-2', 'session-3', 'session-4', 'session-5'] +const LAST = 'session-5' + +let rig: RestTestRig +let releaseHeld = (): void => undefined + +beforeEach(async () => { + rig = await createRestTestRig() +}) + +afterEach(async () => { + releaseHeld() + await rig.dispose() + closeTestJournalHostDatabases() + vi.restoreAllMocks() +}) + +/** Boots over a crashed run and starts the pass with its first open held, so it has not reached + * the last chat. */ +async function bootWithHeldPass() { + for (const sessionId of IDS) { + await restTestChat(rig, sessionId, { message: `asked ${sessionId}` }) + } + await rig.crash() + const host = await rig.boot() + const held = Promise.withResolvers() + releaseHeld = held.resolve + rig.adapter.historyFilePath.mockImplementation(async (sessionId) => { + if (sessionId === IDS[0]) { + await held.promise + } + return null + }) + const pass = host.restoreReadableSessions(IDS) + await vi.waitFor(() => expect(rig.adapter.historyFilePath).toHaveBeenCalledOnce()) + return { host, pass, release: held.resolve } +} + +const publishedFor = (sessionId: string): number => + rig.sink.publish.mock.calls.filter(([summary]) => summary.sessionId === sessionId).length + +describe('the restore pass after the tab list', () => { + it('does not reopen a chat closed before the pass reached it', async () => { + const { host, pass, release } = await bootWithHeldPass() + + // What closing the tab does. + await host.setSessionTabVisibility(LAST, false) + await host.close(LAST, 'user-close') + const publishedAtClose = publishedFor(LAST) + release() + await pass + + expect(host.hasSession(LAST)).toBe(false) + expect(restTestOpens(rig, LAST)).toBe(0) + expect(publishedFor(LAST)).toBe(publishedAtClose) + expect(host.hasSession('session-1')).toBe(true) + }) + + it('does not open a chat it reached before quit began but opens after', async () => { + await restTestChat(rig, LAST, { message: 'asked' }) + await rig.crash() + const host = await rig.boot() + const { runtimeState, lifetime } = host.collaboratorsForTests() + // Holds the chat between the pass reaching it and its open. + const settling = Promise.withResolvers() + releaseHeld = settling.resolve + const resolveRecovery = runtimeState.resolveRecovery.bind(runtimeState) + const recovery = vi.spyOn(runtimeState, 'resolveRecovery').mockImplementation(async (id) => { + await settling.promise + return resolveRecovery(id) + }) + const pass = host.restoreReadableSessions([LAST]) + await vi.waitFor(() => expect(recovery).toHaveBeenCalled()) + + const teardown = host.flushAllStreamedEvents() + await vi.waitFor(() => expect(lifetime.isDisposed()).toBe(true)) + settling.resolve() + await teardown + await pass + + expect(restTestOpens(rig, LAST)).toBe(0) + expect(host.hasSession(LAST)).toBe(false) + }) + + it('opens nothing more, and settles no more leases, once the host is quitting', async () => { + const { host, pass, release } = await bootWithHeldPass() + const recovery = vi.spyOn(host.collaboratorsForTests().runtimeState, 'resolveRecovery') + + const teardown = host.flushAllStreamedEvents() + release() + await teardown + await pass + + expect(restTestOpens(rig, LAST)).toBe(0) + expect(host.hasSession(LAST)).toBe(false) + expect(recovery).not.toHaveBeenCalledWith(LAST) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restore-without-import.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restore-without-import.test.ts index 4fa2811472a4..239ad9ff6ff9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restore-without-import.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restore-without-import.test.ts @@ -244,6 +244,8 @@ async function restore(sessionIds: readonly string[]) { resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), hasSession: (sessionId) => sessions.has(sessionId), + isDisposed: () => false, + isListed: () => true, onReadable: (sessionId, opened) => { sessions.set(sessionId, opened.session) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restored-status-parity.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restored-status-parity.test.ts new file mode 100644 index 000000000000..d138ee459c28 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restored-status-parity.test.ts @@ -0,0 +1,92 @@ +// Status rows now arrive after the tab list, from the post-listing restore pass. Each must end +// exactly where a fresh host lands when the same chat is opened directly by a read (T8, regression +// guard: the pass is main's own open path; this pins that moving it after the list changed nothing). + +import { cp } from 'node:fs/promises' +import { afterEach, expect, it, vi } from 'vitest' +import type { AgentSessionStatusSummary } from '../../../shared/agent-session-wire' +import { closeTestJournalHostDatabases } from '../agent-session-journal/journal-host-database-test-support' +import { + createRestTestRig, + restTestChat, + type RestTestRig +} from './structured-agent-session-rest-test-rig' +import { latestRestTestStatus } from './structured-agent-session-rest-test-observations' + +const rigs: RestTestRig[] = [] + +afterEach(async () => { + for (const rig of rigs.splice(0)) { + await rig.dispose() + } + closeTestJournalHostDatabases() + vi.restoreAllMocks() +}) + +const CORPUS = ['session-settled', 'session-quiet', 'session-crashed', 'session-pending'] + +/** The chats a restart finds: two a clean quit settled, one cut off mid-turn by a crash, and one + * whose send the provider only admitted, left pending below a fence that has since moved. */ +async function seedCorpus(rig: RestTestRig): Promise { + await restTestChat(rig, 'session-settled', { message: 'finished work' }) + await restTestChat(rig, 'session-quiet') + await rig.host.flushAllStreamedEvents() + await rig.boot() + await restTestChat(rig, 'session-crashed', { message: 'cut off' }) + rig.adapter.dispatch.mockResolvedValueOnce({ state: 'admitted' }) + await restTestChat(rig, 'session-pending', { message: 'only admitted' }) + await rig.crash() + // A live lease names the link proven at its fence, so the fence moves with that link. + await rig.store.transitionHandoff('session-pending', (record) => ({ + ...record, + lease: { ...record.lease, runtimeFence: record.lease.runtimeFence + 1 }, + providerHandleChain: record.providerHandleChain.map((link, index, chain) => + index === chain.length - 1 ? { ...link, mintedAtFence: link.mintedAtFence + 1 } : link + ) + })) +} + +it('ends every chat where a direct read of it lands, and shows no pre-crash work (T8)', async () => { + const passRig = await createRestTestRig() + rigs.push(passRig) + await seedCorpus(passRig) + // One copy of the files for each boot, taken while nothing is writing. + closeTestJournalHostDatabases() + const readRoot = `${passRig.root}-read` + await cp(passRig.root, readRoot, { recursive: true }) + const readRig = await createRestTestRig({}, { root: readRoot }) + rigs.push(readRig) + + // Startup's shape: the lease check, the list, then the pass. + const passHost = await passRig.boot() + await passHost.reconcileRestartLeases() + await passHost.restoreReadableSessions(CORPUS) + + // The same run, where the window reads each chat instead. + const readHost = readRig.host + await readHost.reconcileRestartLeases() + for (const sessionId of CORPUS) { + await readHost.history({ sessionId, direction: 'tail' }) + } + + // `updatedAt` is when the row was projected, a wall-clock read that differs between two boots. + const rows = (rig: RestTestRig): Record> => + Object.fromEntries( + CORPUS.flatMap((sessionId) => { + const summary = latestRestTestStatus(rig, sessionId) + if (!summary) { + return [] + } + const { updatedAt: _updatedAt, ...row } = summary + return [[sessionId, row]] + }) + ) + await vi.waitFor(() => expect(rows(passRig)).toEqual(rows(readRig))) + for (const sessionId of CORPUS) { + expect(latestRestTestStatus(passRig, sessionId)).toBeDefined() + } + const crashedRows = passRig.statusEvents.flatMap((event) => + event.type === 'status' && event.session.sessionId === 'session-crashed' ? [event.session] : [] + ) + expect(crashedRows.map((row) => row.status)).not.toContain('working') +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts index 3046aa971f67..7a9c0ab5bfce 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts @@ -12,6 +12,7 @@ import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire' import { agentSessionRefusalError } from '../../../shared/agent-session-wire-refusals' +import { sessionTabListed } from './structured-agent-session-host-tabs' import { adapterSupportsRecord } from './structured-agent-session-provider-support' import { StructuredAgentSessionReadableRestorer } from './structured-agent-session-readable-restorer' import { StructuredAgentSessionRestartRestoreGate } from './structured-agent-session-restart-restore-gate' @@ -62,7 +63,7 @@ export function createStructuredAgentSessionHostRestore( deps: StructuredAgentSessionHostDeps, wiring: Omit< ConstructorParameters[0], - 'openDeps' | 'supportsRecord' | 'reconcile' | 'resolveRecovery' + 'openDeps' | 'supportsRecord' | 'reconcile' | 'resolveRecovery' | 'isListed' > & { reconcileLeases: (sessionId: string) => Promise resolveRecovery: (sessionId: string) => Promise @@ -77,6 +78,7 @@ export function createStructuredAgentSessionHostRestore( const restorer = new StructuredAgentSessionReadableRestorer({ openDeps: deps, supportsRecord: (record) => adapterSupportsRecord(deps.adapter, record), + isListed: (sessionId) => sessionTabListed(deps.store, sessionId), reconcile, // The next attach or send resolves recovery again, strictly, before it acts. resolveRecovery: (sessionId) => diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-send-during-startup-lease-check.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-send-during-startup-lease-check.test.ts new file mode 100644 index 000000000000..e293bec65e7c --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-send-during-startup-lease-check.test.ts @@ -0,0 +1,82 @@ +// A client that is not the desktop window (a phone, the CLI, a remote host's peer) can send before +// startup's lease check has settled the previous run's owner. The send is accepted into the chat +// at once; the agent it needs starts only after the check, through the agent start's own check. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { closeTestJournalHostDatabases } from '../agent-session-journal/journal-host-database-test-support' +import { + createRestTestRig, + restTestChat, + sendRestTestMessage, + type RestTestRig +} from './structured-agent-session-rest-test-rig' + +const CHAT = 'session-a' + +let rig: RestTestRig +// A failed assertion must not leave teardown waiting on a held probe. +let releaseProbe = (): void => undefined + +beforeEach(async () => { + rig = await createRestTestRig() +}) + +afterEach(async () => { + releaseProbe() + await rig.dispose() + closeTestJournalHostDatabases() + vi.restoreAllMocks() +}) + +describe('a send during a slow startup lease check', () => { + it('is accepted at once, and its agent starts once, after the check', async () => { + await restTestChat(rig, CHAT, { message: 'before the restart' }) + await rig.crash() + const probe = Promise.withResolvers() + const probing = Promise.withResolvers() + releaseProbe = probe.resolve + let checked = false + rig.probeOwner.mockImplementation(async () => { + probing.resolve() + await probe.promise + checked = true + return { outcome: 'pid-absent' } + }) + const host = await rig.boot() + const startupCheck = host.reconcileRestartLeases() + await probing.promise + // Each start, with whether the check had adjudicated the chat's lease by then. + const starts: { checked: boolean; unreconciled: boolean | undefined }[] = [] + const acquire = rig.adapter.acquire.getMockImplementation() + rig.adapter.acquire.mockImplementation(async (input) => { + starts.push({ checked, unreconciled: rig.store.getRecord(CHAT)?.lease.unreconciled }) + if (!acquire) { + throw new Error('the rig acquires through its own implementation') + } + return acquire(input) + }) + + const first = await sendRestTestMessage(rig, CHAT, 'first after the restart') + expect(first, JSON.stringify(first)).toMatchObject({ + ok: true, + value: { submission: { dispatchState: 'pending' } } + }) + // Queued behind the first one's delivery, which is waiting on the check. + const second = sendRestTestMessage(rig, CHAT, 'second after the restart') + // Long enough for a delivery that did not wait to have started its agent. + await new Promise((resolve) => setTimeout(resolve, 200)) + expect(starts).toEqual([]) + expect(rig.adapter.dispatch).not.toHaveBeenCalled() + + probe.resolve() + await startupCheck + expect(await second, 'second send').toMatchObject({ ok: true }) + await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledTimes(2), { + timeout: 10_000 + }) + + expect(starts).toEqual([{ checked: true, unreconciled: false }]) + const submissions = (await host.journalSnapshot(CHAT)).submissions.slice(-2) + expect(submissions.map((entry) => entry.dispatchState)).toEqual(['accepted', 'accepted']) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts index 1f8fc80c863d..10075f14568e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts @@ -431,7 +431,7 @@ describe('startup', () => { // The record is readable — the tab comes back, history answers — and nothing is running. expect(acquire).not.toHaveBeenCalled() - expect(host.listSessionTabs()).toEqual([ + expect(host.listSessionTabs([SESSION])).toEqual([ { sessionId: SESSION, workspaceId: 'workspace-1', agent: 'codex' } ]) expect((await host.history({ sessionId: SESSION, direction: 'tail' })).ok).toBe(true) diff --git a/src/main/runtime/orca-runtime-collect-mobile-visible-graph-changed-worktrees.ts b/src/main/runtime/orca-runtime-collect-mobile-visible-graph-changed-worktrees.ts index 391a33f0e247..3ffa30b8a3a7 100644 --- a/src/main/runtime/orca-runtime-collect-mobile-visible-graph-changed-worktrees.ts +++ b/src/main/runtime/orca-runtime-collect-mobile-visible-graph-changed-worktrees.ts @@ -4,6 +4,7 @@ import type { RuntimeMobileSessionTabsResult, RuntimeSyncedTab } from '../../sha import type { RuntimeLeafRecord } from './runtime-terminal-state-records' import type { PtyControllerInventory } from './runtime-pty-controller-contract' import { parseExecutionHostId } from '../../shared/execution-host' +import { getStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry' export class OrcaRuntimeWithCollectMobileVisibleGraphChangedWorktrees extends OrcaRuntimeWithSyncWindowGraph { // Why: toMobileSessionTabsResult resolves handles/titles from this.tabs and @@ -100,9 +101,11 @@ export class OrcaRuntimeWithCollectMobileVisibleGraphChangedWorktrees extends Or this.hydrateHeadlessMobileSessionTabsFromWorkspaceSession() const ptyInventory = await this.refreshMobileSessionPtyInventory() this.restoreLivePairedRendererSessionOwnedMobileTerminals(null) + // Once for every worktree's answer, after the census's await, so a change made during it counts. + const replacements = getStructuredAgentSessionHost()?.conversationReplacements?.() ?? [] const snapshots = [...this.mobileSessionTabsByWorktree.values()].map((snapshot) => this.projectMobileSessionTabsForClient( - this.toMobileSessionTabsResult(snapshot), + this.toMobileSessionTabsResult(snapshot, replacements), clientNavigationId ) ) diff --git a/src/main/runtime/orca-runtime-get-structured-agent-session-create-support.ts b/src/main/runtime/orca-runtime-get-structured-agent-session-create-support.ts index 2bf67f0b4f88..eca86c85f3f8 100644 --- a/src/main/runtime/orca-runtime-get-structured-agent-session-create-support.ts +++ b/src/main/runtime/orca-runtime-get-structured-agent-session-create-support.ts @@ -26,6 +26,9 @@ import { parseWorkspaceKey } from '../../shared/workspace-scope' import { applyStructuredCodexWorkspaceTrust } from '../agent-workspace-trust-spawn' export class OrcaRuntimeWithGetStructuredAgentSessionCreateSupport extends OrcaRuntimeWithGetWorktreePs { + // The history restore a tab restore owes, until a caller that answered with its list starts it. + protected owedStructuredAgentSessionHistoryRestore: (() => void) | null = null + async getStructuredAgentSessionCreateSupport( worktreeSelector: string, agent: 'claude' | 'codex' @@ -260,6 +263,16 @@ export class OrcaRuntimeWithGetStructuredAgentSessionCreateSupport extends OrcaR return this.structuredAgentSessionTabRestorePromise } + /** Starts the history restore the tab restore owes, once. On the next macrotask, so a caller that + * starts it as it answers has sent that answer first. */ + startStructuredAgentSessionHistoryRestore(): void { + const owed = this.owedStructuredAgentSessionHistoryRestore + this.owedStructuredAgentSessionHistoryRestore = null + if (owed) { + setImmediate(owed) + } + } + prepareStructuredAgentSessionStartupRestoration(): Promise { this.structuredAgentSessionStartupRestorePromise ??= this.prepareStructuredAgentSessionStartupRestorationOnce().catch((error) => { diff --git a/src/main/runtime/orca-runtime-prune-mobile-session-tab-group-layout.ts b/src/main/runtime/orca-runtime-prune-mobile-session-tab-group-layout.ts index 42a4dadd25ff..5d7e75d5c3e5 100644 --- a/src/main/runtime/orca-runtime-prune-mobile-session-tab-group-layout.ts +++ b/src/main/runtime/orca-runtime-prune-mobile-session-tab-group-layout.ts @@ -23,8 +23,9 @@ import type { import { buildRuntimeMobileAgentStatus } from './runtime-mobile-agent-status-builder' import { FIRST_PANE_ID } from '../../shared/pane-key' import { isTerminalLeafId, makePaneKey, parsePaneKey } from '../../shared/stable-pane-id' -import { getStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry' import { replaceConversationInSnapshot } from './structured-conversation-tab-replacement' +import { getStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry' +import type { ConversationReplacement } from '../native-chat/agent-session-wire/structured-conversation-command' import { resolveStructuredWorkerAuthority } from './structured-worker-authority' import { structuredWorkerAgentStatus } from './orchestration/structured-worker-group-addressing' @@ -76,9 +77,12 @@ export class OrcaRuntimeWithPruneMobileSessionTabGroupLayout extends OrcaRuntime /** Transforms an internal mobile session tab snapshot into a sanitized client payload, resolving launch-agent ownership and normalizing titles. */ protected toMobileSessionTabsResult( - snapshot: RuntimeMobileSessionTabsSnapshot + snapshot: RuntimeMobileSessionTabsSnapshot, + // A caller answering many worktrees in one synchronous loop derives these once for it. + replacements: readonly ConversationReplacement[] = getStructuredAgentSessionHost()?.conversationReplacements?.() ?? + [] ): RuntimeMobileSessionTabsResult { - for (const replacement of getStructuredAgentSessionHost()?.conversationReplacements?.() ?? []) { + for (const replacement of replacements) { snapshot = replaceConversationInSnapshot(snapshot, replacement) } return projectRuntimeMobileSessionTabs(snapshot, this.getMobileSessionProjectionHost()) diff --git a/src/main/runtime/orca-runtime-restore-structured-agent-session-tabs-once.ts b/src/main/runtime/orca-runtime-restore-structured-agent-session-tabs-once.ts index 61c2e0b3605f..be21e67f0aa6 100644 --- a/src/main/runtime/orca-runtime-restore-structured-agent-session-tabs-once.ts +++ b/src/main/runtime/orca-runtime-restore-structured-agent-session-tabs-once.ts @@ -62,10 +62,9 @@ export class OrcaRuntimeWithRestoreStructuredAgentSessionTabsOnce extends OrcaRu ) // Unrecorded, the profile's chats join the tabs chats opened while the import was owed left. // First: after a /clear the profile's chat would take their tab id, so seeds hit tabIdTaken. - const targets = persistedVisibleIndex.present + const listedIds = persistedVisibleIndex.present ? persistedVisibleIndex.sessionIds : [...new Set([...persistedVisibleIndex.sessionIds, ...profileIds])] - await host?.restoreReadableSessions(targets) for (const worktreeId of this.getKnownWorkspaceSessionWorktreeIds()) { this.hydrateHeadlessMobileSessionTabsFromWorkspaceSession(worktreeId, { allowAttachedWindow: true, @@ -73,7 +72,7 @@ export class OrcaRuntimeWithRestoreStructuredAgentSessionTabsOnce extends OrcaRu }) } this.hydrateHeadlessMobileSessionTabsFromWorkspaceSession() - const restored = (host?.listSessionTabs() ?? []).flatMap((session) => { + const restored = (host?.listSessionTabs(listedIds) ?? []).flatMap((session) => { if (session.agent !== 'codex' && session.agent !== 'claude') { return [] } @@ -85,16 +84,30 @@ export class OrcaRuntimeWithRestoreStructuredAgentSessionTabsOnce extends OrcaRu }) await seedStructuredAgentSessionTabIndex( host, - targets, + listedIds, restored.map((session) => session.sessionId) ) // Past the seed, projecting records nothing. - for (const replacement of host?.conversationReplacements?.() ?? []) { + // Derived once for the loop below, which stores a snapshot per tab; nothing awaits in between. + const replacements = host?.conversationReplacements?.() ?? [] + for (const replacement of replacements) { this.replaceStructuredAgentSessionTab(replacement) } + const quiet = { activate: false, notify: false, replacements } for (const session of restored) { - this.projectStructuredAgentSessionTab({ ...session, activate: false, notify: false }) + this.projectStructuredAgentSessionTab({ ...session, ...quiet }) } + // Each chat's status row arrives as its history opens; one chat's failure leaves the others and + // the list alone. Whoever answers with this list starts it, once that answer is out. + this.owedStructuredAgentSessionHistoryRestore = host + ? () => + void host.restoreReadableSessions(listedIds).catch((error: unknown) => { + host.deps.logger.warn('restoring chat history after listing failed', { + scope: 'history-restore-after-listing', + error + }) + }) + : null const wasUnverifiable = this.structuredAgentSessionInventoryUnverifiable // No host, or one still owed the records file's chats, means no one can say which chats exist; // with none on disk, empty is the answer. @@ -143,6 +156,8 @@ export class OrcaRuntimeWithRestoreStructuredAgentSessionTabsOnce extends OrcaRu activate: boolean notify?: boolean replacesSessionId?: string + /** The /clear replacements, derived once by a caller projecting many tabs in one loop. */ + replacements?: readonly ConversationReplacement[] }): void { const existing = this.mobileSessionTabsByWorktree.get(input.workspaceId) const id = `agent-session:${input.sessionId}` @@ -219,7 +234,7 @@ export class OrcaRuntimeWithRestoreStructuredAgentSessionTabsOnce extends OrcaRu ...(existing?.tabGroupLayout ? { tabGroupLayout: existing.tabGroupLayout } : {}), tabs } - const stored = this.storeMobileSessionSnapshot(input.workspaceId, snapshot) + const stored = this.storeMobileSessionSnapshot(input.workspaceId, snapshot, input.replacements) if (input.notify !== false) { this.emitMobileSessionTabsSnapshot(stored) } diff --git a/src/main/runtime/orca-runtime-runtime-id.ts b/src/main/runtime/orca-runtime-runtime-id.ts index 5a471276d85d..c96ae4ff362c 100644 --- a/src/main/runtime/orca-runtime-runtime-id.ts +++ b/src/main/runtime/orca-runtime-runtime-id.ts @@ -100,12 +100,13 @@ export class OrcaRuntimeWithRuntimeId { protected mobileSessionTabsByWorktree = new Map() - /** Single host writer for mobile session snapshots; versions are total-order stamps. */ + /** Single host writer for mobile snapshots; total-order versions; a loop passes `replacements`. */ protected storeMobileSessionSnapshot( worktreeId: string, - snapshot: RuntimeMobileSessionTabsSnapshot + snapshot: RuntimeMobileSessionTabsSnapshot, + replacements = getStructuredAgentSessionHost()?.conversationReplacements?.() ?? [] ): RuntimeMobileSessionTabsSnapshot { - for (const replacement of getStructuredAgentSessionHost()?.conversationReplacements?.() ?? []) { + for (const replacement of replacements) { snapshot = replaceConversationInSnapshot(snapshot, replacement) } const existing = this.mobileSessionTabsByWorktree.get(worktreeId) diff --git a/src/main/runtime/orca-runtime-structured-session-restore.test.ts b/src/main/runtime/orca-runtime-structured-session-restore.test.ts index 92c39dda2835..0e1fdbc9b76b 100644 --- a/src/main/runtime/orca-runtime-structured-session-restore.test.ts +++ b/src/main/runtime/orca-runtime-structured-session-restore.test.ts @@ -51,7 +51,7 @@ describe('structured session cold restoration', () => { expect(restoreReadableSessions).not.toHaveBeenCalled() }) - it('loads records, inventories PTYs, restores ownership, then projects tabs exactly once', async () => { + it('loads records, inventories PTYs, restores ownership, projects tabs, then restores history, once', async () => { const runtime = new OrcaRuntimeService() const hydrate = vi.fn() const refresh = vi.fn(async () => new Set()) @@ -83,6 +83,12 @@ describe('structured session cold restoration', () => { const second = runtime.restoreStructuredAgentSessionTabs() expect(second).toBe(first) await Promise.all([first, second]) + // The history pass is owed, not started: the caller that answers with the list starts it. + expect(restoreReadableSessions).not.toHaveBeenCalled() + runtime.startStructuredAgentSessionHistoryRestore() + runtime.startStructuredAgentSessionHistoryRestore() + expect(restoreReadableSessions).not.toHaveBeenCalled() + await new Promise((resolve) => setImmediate(resolve)) expect(hydrate).toHaveBeenCalledWith('workspace-1', { allowAttachedWindow: true, @@ -102,8 +108,9 @@ describe('structured session cold restoration', () => { expect(reconcileRestartLeases.mock.invocationCallOrder[0]).toBeLessThan( restoreReadableSessions.mock.invocationCallOrder[0] ?? Infinity ) - expect(restoreReadableSessions.mock.invocationCallOrder[0]).toBeLessThan( - hydrate.mock.invocationCallOrder[0] ?? Infinity + // Tabs are projected first; history opens after, so no chat's history holds the list. + expect(hydrate.mock.invocationCallOrder[0]).toBeLessThan( + restoreReadableSessions.mock.invocationCallOrder[0] ?? Infinity ) }) @@ -144,8 +151,11 @@ describe('structured session cold restoration', () => { } as never) await runtime.restoreStructuredAgentSessionTabs() + runtime.startStructuredAgentSessionHistoryRestore() - expect(restoreReadableSessions).toHaveBeenCalledWith(['session-survives-rollback']) + await vi.waitFor(() => + expect(restoreReadableSessions).toHaveBeenCalledWith(['session-survives-rollback']) + ) }) it('treats an empty durable visible-session index as authoritative', async () => { @@ -198,8 +208,9 @@ describe('structured session cold restoration', () => { } as never) await runtime.restoreStructuredAgentSessionTabs() + runtime.startStructuredAgentSessionHistoryRestore() - expect(restoreReadableSessions).toHaveBeenCalledWith([]) + await vi.waitFor(() => expect(restoreReadableSessions).toHaveBeenCalledWith([])) }) it('normalizes a restored tab id and removes it when closed', async () => { @@ -357,7 +368,9 @@ describe('structured session cold restoration', () => { sessionId: 'restored-claude', agent: 'claude', activate: false, - notify: false + notify: false, + // Derived once by the restore for every tab it projects. + replacements: [] }) const restored = await runtime.listMobileSessionTabs('id:workspace-1') diff --git a/src/main/runtime/rpc/methods/session-tabs-inventory.ts b/src/main/runtime/rpc/methods/session-tabs-inventory.ts index be3a1d2de42c..354e23770912 100644 --- a/src/main/runtime/rpc/methods/session-tabs-inventory.ts +++ b/src/main/runtime/rpc/methods/session-tabs-inventory.ts @@ -5,7 +5,7 @@ import type { RpcContext } from '../core' import { projectSessionTabAgentStatus } from './session-tab-agent-status-projection' import { projectSessionTabBrowserPlacements } from './session-tab-browser-placement-projection' import { createSessionTabsRetirementProofDelta } from './session-tabs-retirement-proof-delta' -import { restoreStructuredTabsIfSupported } from './structured-session-tab-restore' +import { answerAfterStructuredTabRestore } from './structured-session-tab-restore' type SessionTabsInventory = { snapshots: RuntimeMobileSessionTabsResult[] @@ -230,37 +230,30 @@ export async function subscribeSessionTabsInventory( } let collected: Awaited> | undefined try { - // Why: restore after registering, so an unsubscribe or socket close while it runs still finds the stream. - const restoring = restoreStructuredTabsIfSupported(context) - if (restoring) { - await restoring - if (closed) { - return - } - } - for (let attempt = 1; !collected; attempt += 1) { - censusInvalidated = false - const candidate = await collectSessionTabsInventory( - { ...context, signal: inventoryController.signal }, - true - ) - if (closed) { - return - } - if (censusInvalidated) { + // Why: restore after registering, so an unsubscribe or socket close while it runs still finds the + // stream. The snapshots emit right after this settles, before the history restore it starts. + collected = await answerAfterStructuredTabRestore(context, async () => { + for (let attempt = 1; !closed; attempt += 1) { + censusInvalidated = false + const candidate = await collectSessionTabsInventory( + { ...context, signal: inventoryController.signal }, + true + ) + if (!censusInvalidated || closed) { + return candidate + } clearBufferedChanges() if (attempt === MAX_CENSUS_COLLECTION_ATTEMPTS) { throw new Error('session_tabs_inventory_unstable') } - } else { - collected = candidate } - } + return undefined + }) } catch (error) { runtime.cleanupSubscription(subscriptionId) throw error } - if (closed) { + if (closed || !collected) { return } const { inventory, changeSequence } = collected diff --git a/src/main/runtime/rpc/methods/session-tabs.ts b/src/main/runtime/rpc/methods/session-tabs.ts index d549daffbaf8..d51252eaa439 100644 --- a/src/main/runtime/rpc/methods/session-tabs.ts +++ b/src/main/runtime/rpc/methods/session-tabs.ts @@ -15,7 +15,7 @@ import { import { SESSION_TAB_MARKDOWN_METHODS } from './session-tab-markdown-methods' import { SESSION_TAB_MUTATION_METHODS } from './session-tab-mutation-methods' import { createSessionTabsRetirementProofDelta } from './session-tabs-retirement-proof-delta' -import { restoreStructuredTabsIfSupported } from './structured-session-tab-restore' +import { answerAfterStructuredTabRestore } from './structured-session-tab-restore' import { assertLegacyAiVaultResumeCommandAllowed } from '../../../ai-vault/structured-session-ownership' import { SessionTabsUnsubscribeAllParams } from '../../../../shared/rpc-contract/session-tabs-params' import { SESSION_TABS_SPLIT_GROUP_PLACEMENT_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' @@ -24,22 +24,20 @@ export const SESSION_TAB_METHODS = [ defineMethod({ name: 'session.tabs.list', params: WorktreeTabSelector, - handler: async (params, { runtime, pairedDeviceId, clientKind, clientCapabilities }) => { - await restoreStructuredTabsIfSupported({ runtime, clientKind, clientCapabilities }) - return projectSessionTabsForClient( - await runtime.listMobileSessionTabs(params.worktree, pairedDeviceId), - clientKind, - clientCapabilities + handler: (params, { runtime, pairedDeviceId, clientKind, clientCapabilities }) => + answerAfterStructuredTabRestore({ runtime, clientKind, clientCapabilities }, async () => + projectSessionTabsForClient( + await runtime.listMobileSessionTabs(params.worktree, pairedDeviceId), + clientKind, + clientCapabilities + ) ) - } }), defineMethod({ name: 'session.tabs.listAll', params: null, - handler: async (_params, context) => { - await restoreStructuredTabsIfSupported(context) - return listSessionTabsInventory(context) - } + handler: (_params, context) => + answerAfterStructuredTabRestore(context, () => listSessionTabsInventory(context)) }), ...SESSION_TAB_MUTATION_METHODS, ...SESSION_TAB_CLOSE_METHODS, @@ -142,39 +140,46 @@ export const SESSION_TAB_METHODS = [ register(explicitWorktreeId) } try { - await restoreStructuredTabsIfSupported({ runtime, clientKind, clientCapabilities }) - if (released) { - return - } - const initial = await runtime.listMobileSessionTabs(params.worktree, pairedDeviceId) - if (released) { - return - } - if (!subscriptionId) { - register(initial.worktree) - if (released) { - return - } - } - const subscribedWorktree = initial.worktree - const withProofDelta = createSessionTabsRetirementProofDelta(clientCapabilities) - emit({ - type: 'snapshot', - ...withProofDelta(projectSessionTabsForClient(initial, clientKind, clientCapabilities)) - }) - if (released) { - return - } - stopListening = runtime.onMobileSessionTabsChanged((snapshot) => { - if (snapshot.worktree === subscribedWorktree) { + // The first snapshot emits inside this, before the history restore it starts. + await answerAfterStructuredTabRestore( + { runtime, clientKind, clientCapabilities }, + async () => { + if (released) { + return + } + const initial = await runtime.listMobileSessionTabs(params.worktree, pairedDeviceId) + if (released) { + return + } + if (!subscriptionId) { + register(initial.worktree) + if (released) { + return + } + } + const subscribedWorktree = initial.worktree + const withProofDelta = createSessionTabsRetirementProofDelta(clientCapabilities) emit({ - type: 'updated', + type: 'snapshot', ...withProofDelta( - projectSessionTabsForClient(snapshot, clientKind, clientCapabilities) + projectSessionTabsForClient(initial, clientKind, clientCapabilities) ) }) + if (released) { + return + } + stopListening = runtime.onMobileSessionTabsChanged((snapshot) => { + if (snapshot.worktree === subscribedWorktree) { + emit({ + type: 'updated', + ...withProofDelta( + projectSessionTabsForClient(snapshot, clientKind, clientCapabilities) + ) + }) + } + }, pairedDeviceId) } - }, pairedDeviceId) + ) } catch (error) { // A stream already ended by its release must not also report an error. if (released) { diff --git a/src/main/runtime/rpc/methods/structured-session-tab-restore.ts b/src/main/runtime/rpc/methods/structured-session-tab-restore.ts index 81ecde983603..d5c20fe85c8c 100644 --- a/src/main/runtime/rpc/methods/structured-session-tab-restore.ts +++ b/src/main/runtime/rpc/methods/structured-session-tab-restore.ts @@ -7,7 +7,7 @@ import { supportsStructuredAgentSessions } from './structured-agent-session-poli * of each chat, and gating on capability left it with nothing to project after a desktop restart — * no chat and no prompt. Restoring spawns no provider child for a cleanly closed session, and a * host with no saved sessions returns before building anything. */ -export function restoreStructuredTabsIfSupported( +function restoreStructuredTabsIfSupported( context: Pick ): Promise | undefined { const shouldRestore = context.clientKind === 'mobile' || supportsStructuredAgentSessions(context) @@ -17,3 +17,22 @@ export function restoreStructuredTabsIfSupported( // Nothing to restore: callers skip the await, so a stream's setup keeps its timing. return undefined } + +/** The tab restore, then the caller's answer. The chat history restore the tab restore owes starts + * only once that answer has settled (sent, failed or abandoned), so the answer's own work, such as + * the PTY census, never waits behind it. The first caller to settle starts it. */ +export async function answerAfterStructuredTabRestore( + context: Pick, + answer: () => Promise +): Promise { + const restoring = restoreStructuredTabsIfSupported(context) + if (!restoring) { + return answer() + } + try { + await restoring + return await answer() + } finally { + context.runtime.startStructuredAgentSessionHistoryRestore?.() + } +} diff --git a/src/main/runtime/structured-agent-session-startup-tab-restore.test.ts b/src/main/runtime/structured-agent-session-startup-tab-restore.test.ts index ac6df4c4b2b3..8e2636d805e5 100644 --- a/src/main/runtime/structured-agent-session-startup-tab-restore.test.ts +++ b/src/main/runtime/structured-agent-session-startup-tab-restore.test.ts @@ -1,7 +1,7 @@ -// With native chat on, the renderer's startup also awaits the chat tab restore (`session.tabs.listAll`), -// which reads every chat whose tab was open at quit. That read must not wait on record-store -// bookkeeping: with a saved tab index it writes nothing, and a store that cannot be written costs -// a bounded number of failed writes, not one per chat. +// With native chat on, the renderer's startup also awaits the chat tab list (`session.tabs.listAll`). +// The list opens no chat; a restore it does not wait for then reads every chat whose tab was open at +// quit. Neither may wait on record-store bookkeeping: with a saved tab index they write nothing, and +// a store that cannot be written costs a bounded number of failed writes, not one per chat. import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -248,6 +248,14 @@ function startupRuntime(options: { afterInstall?: () => void; profileChats?: str } } +/** Starts the history restore the listing owes, as the caller that answers with it does, and joins + * it. */ +async function historyRestored(runtime: OrcaRuntimeService): Promise { + runtime.startStructuredAgentSessionHistoryRestore() + await new Promise((resolve) => setImmediate(resolve)) + await getStructuredAgentSessionHost()!.restoreReadableSessions() +} + async function expectHistory(sessionId: string): Promise { const host = getStructuredAgentSessionHost() expect(JSON.stringify((await host!.journalSnapshot(sessionId)).items)).toContain(PROMPT) @@ -292,6 +300,7 @@ describe('restoring the chat tabs open at quit', () => { }) await expect(runtime.restoreStructuredAgentSessionTabs()).resolves.toBeUndefined() + await historyRestored(runtime) expect(published().map((tab) => tab.id)).toEqual([ `agent-session:${CHAT_A}`, @@ -337,6 +346,7 @@ describe('restoring the chat tabs open at quit', () => { await runtime.prepareStructuredAgentSessionStartupRestoration() const prepared = writes.refused await runtime.restoreStructuredAgentSessionTabs() + await historyRestored(runtime) expect(published()).toHaveLength(count) expect(prepared).toBe(1) @@ -383,6 +393,7 @@ describe('restoring the chat tabs open at quit', () => { await runtime.prepareStructuredAgentSessionStartupRestoration() const prepared = writes.refused await runtime.restoreStructuredAgentSessionTabs() + await historyRestored(runtime) expect(published()).toHaveLength(2) expect(prepared).toBe(1) @@ -418,6 +429,7 @@ describe('restoring the chat tabs open at quit', () => { writes.failing = true await expect(runtime.restoreStructuredAgentSessionTabs()).resolves.toBeUndefined() + await historyRestored(runtime) expect(published()).toHaveLength(2) await expectHistory(CHAT_A) diff --git a/src/main/runtime/structured-agent-session-tab-listing-startup.test.ts b/src/main/runtime/structured-agent-session-tab-listing-startup.test.ts new file mode 100644 index 000000000000..337640862a23 --- /dev/null +++ b/src/main/runtime/structured-agent-session-tab-listing-startup.test.ts @@ -0,0 +1,385 @@ +// The chat tab list a client asks for at startup answers from records and the tab table. It waits +// for no chat's history; each chat's history opens after the answer. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../shared/protocol-version' +import type { RuntimeMobileSessionTabsResult } from '../../shared/runtime-types' +import { closeTestJournalHostDatabases } from '../native-chat/agent-session-journal/journal-host-database-test-support' +import { setStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry' +import { + createRestTestRig, + restTestChat, + type RestTestRig +} from '../native-chat/agent-session-wire/structured-agent-session-rest-test-rig' +import { latestRestTestStatus } from '../native-chat/agent-session-wire/structured-agent-session-rest-test-observations' +import { AgentSessionStoreTransactions } from './agent-session-store-transactions' +import { OrcaRuntimeService } from './orca-runtime' +import { RpcDispatcher } from './rpc/dispatcher' +import { SESSION_TAB_METHODS } from './rpc/methods/session-tabs' + +type ListingInternals = { + store: { getWorkspaceSession: () => unknown } + getClientSettings(): { experimentalStructuredNativeChat: boolean } + supportsAuthoritativeSessionTabsInventory(): boolean + hasPersistedStructuredAgentSessionStore(): boolean + getKnownWorkspaceSessionWorktreeIds(): Set + hydrateHeadlessMobileSessionTabsFromWorkspaceSession(): Set + refreshMobileSessionPtyRecords(): Promise | null> + refreshMobileSessionPtyInventory(): Promise + ensureStructuredAgentSessionHost(): Promise +} + +const CONTEXT = { + clientKind: 'runtime' as const, + clientCapabilities: [STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY] +} + +let rig: RestTestRig +// Anything a test holds is let go before teardown, so a failed assertion cannot hang it. +let releaseHeld = (): void => undefined + +beforeEach(async () => { + rig = await createRestTestRig() +}) + +afterEach(async () => { + releaseHeld() + // The listing starts the history restore after its answer, without awaiting it; it finishes + // before the rig removes its files. + await new Promise((resolve) => setImmediate(resolve)) + await rig.host.restoreReadableSessions() + setStructuredAgentSessionHost(null) + await rig.dispose() + closeTestJournalHostDatabases() + vi.restoreAllMocks() +}) + +/** A restarted runtime whose structured host is the rig's current one. */ +function restartedRuntime(workspaceSession: unknown = null) { + const runtime = new OrcaRuntimeService() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: these members exist on the runtime; they are protected, not absent. + const internal = runtime as unknown as ListingInternals + internal.store = { getWorkspaceSession: () => workspaceSession } + internal.getClientSettings = () => ({ experimentalStructuredNativeChat: true }) + // The PTY census behind an authoritative inventory is not what this pins. + internal.supportsAuthoritativeSessionTabsInventory = () => false + internal.hasPersistedStructuredAgentSessionStore = () => true + internal.getKnownWorkspaceSessionWorktreeIds = () => new Set() + internal.hydrateHeadlessMobileSessionTabsFromWorkspaceSession = () => new Set() + internal.refreshMobileSessionPtyRecords = async () => new Set() + // The census an answer takes after the tab restore: daemon, git and SSH round trips in the app. + internal.refreshMobileSessionPtyInventory = async () => { + await new Promise((resolve) => setTimeout(resolve, 20)) + return null + } + internal.ensureStructuredAgentSessionHost = async () => { + setStructuredAgentSessionHost(rig.host) + } + const dispatcher = new RpcDispatcher({ runtime, methods: SESSION_TAB_METHODS }) + const listAll = async (): Promise => { + const response = await dispatcher.dispatch( + { id: 'list-all', authToken: 'tok', method: 'session.tabs.listAll' }, + CONTEXT + ) + if (!response.ok) { + throw new Error(`listAll refused: ${JSON.stringify(response.error)}`) + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: listAll answers with the inventory shape its handler returns. + const { snapshots } = response.result as { snapshots: RuntimeMobileSessionTabsResult[] } + return chatIds(snapshots) + } + /** Mobile's stream: resolves with its first snapshots, and what had run when they were sent. */ + const subscribeAll = (seen: () => unknown) => + new Promise<{ ids: string[]; atAnswer: unknown }>((resolve, reject) => { + void dispatcher.dispatchStreaming( + { id: 'subscribe-all', authToken: 'tok', method: 'session.tabs.subscribeAll' }, + (message) => { + const frame = JSON.parse(message) + if (!frame.ok) { + reject(new Error(JSON.stringify(frame.error))) + } else if (frame.result?.type === 'snapshots') { + resolve({ ids: chatIds(frame.result.snapshots), atAnswer: seen() }) + } + }, + { ...CONTEXT, connectionId: 'connection-1' } + ) + }) + return { runtime, listAll, subscribeAll } +} + +function chatIds(snapshots: RuntimeMobileSessionTabsResult[]): string[] { + return snapshots.flatMap((snapshot) => + snapshot.tabs.flatMap((tab) => (tab.type === 'agent-session' ? [tab.sessionId] : [])) + ) +} + +/** Resolves with the answer, or with 'still waiting' once `ms` pass without one. */ +function within(promise: Promise, ms = 2_000): Promise { + return Promise.race([ + promise, + new Promise<'still waiting'>((resolve) => setTimeout(() => resolve('still waiting'), ms)) + ]) +} + +describe('listing chat tabs at startup', () => { + it('answers before any chat opens, then opens each one (T1)', async () => { + const ids = ['session-1', 'session-2', 'session-3'] + for (const sessionId of ids) { + await restTestChat(rig, sessionId, { message: sessionId }) + } + await rig.crash() + await rig.boot() + const opens = Promise.withResolvers() + releaseHeld = opens.resolve + let opened = 0 + rig.adapter.historyFilePath.mockImplementation(async () => { + await opens.promise + opened += 1 + return null + }) + const { listAll } = restartedRuntime() + + expect(await within(listAll())).toEqual(ids) + expect(opened).toBe(0) + + opens.resolve() + await vi.waitFor(() => ids.forEach((id) => expect(rig.host.hasSession(id)).toBe(true))) + expect(opened).toBe(3) + }) + + it.each(['listAll', 'subscribeAll'] as const)( + 'starts the history restore only after %s has answered, census and all', + async (method) => { + const ids = ['session-1', 'session-2', 'session-3', 'session-4', 'session-5'] + for (const sessionId of ids) { + await restTestChat(rig, sessionId, { message: sessionId }) + } + await rig.crash() + await rig.boot() + const restore = vi.spyOn(rig.host, 'restoreReadableSessions') + const { runtime, listAll, subscribeAll } = restartedRuntime() + const seen = () => ({ + started: restore.mock.calls.length, + opens: rig.adapter.historyFilePath.mock.calls.length + }) + + const answered = + method === 'listAll' ? { ids: await listAll(), atAnswer: seen() } : await subscribeAll(seen) + + expect(answered).toEqual({ ids, atAnswer: { started: 0, opens: 0 } }) + await vi.waitFor(() => ids.forEach((id) => expect(rig.host.hasSession(id)).toBe(true))) + expect(restore).toHaveBeenCalledOnce() + runtime.cleanupSubscriptionsForConnection('connection-1') + } + ) + + it.each(['listAll', 'subscribeAll'] as const)( + 'derives the /clear replacements once per loop of %s, whatever the tab and worktree count', + async (method) => { + const ids = Array.from({ length: 9 }, (_, index) => `session-${index + 1}`) + for (const [index, sessionId] of ids.entries()) { + await restTestChat(rig, sessionId, { workspaceId: `workspace-${(index % 3) + 1}` }) + } + await rig.crash() + await rig.boot() + const derived = vi.spyOn(rig.host, 'conversationReplacements') + const { runtime, listAll, subscribeAll } = restartedRuntime() + const answer = async () => + method === 'listAll' ? listAll() : (await subscribeAll(() => undefined)).ids + + // The first answer also restores the tabs (one loop over every tab, one over every worktree); + // the second only lists them. + for (const loops of [2, 1]) { + derived.mockClear() + expect((await answer()).toSorted()).toEqual(ids.toSorted()) + expect(derived).toHaveBeenCalledTimes(loops) + runtime.cleanupSubscriptionsForConnection('connection-1') + } + } + ) + + it('answers with a /clear committed after an earlier listing', async () => { + await restTestChat(rig, 'session-cleared') + await restTestChat(rig, 'session-fresh', { listed: false }) + await rig.crash() + await rig.boot() + const { listAll } = restartedRuntime() + expect(await listAll()).toEqual(['session-cleared']) + + const record = rig.store.getRecord('session-cleared')! + await rig.store.setConversationCommand('session-cleared', record.lease.runtimeFence, { + command: 'clear', + state: 'completed', + phase: 'committed', + operationId: 'clear-1', + callerKey: 'client-1', + replacementSessionId: 'session-fresh' + }) + + expect(await listAll()).toEqual(['session-fresh']) + }) + + it('answers with a cleared chat revealed while the listing waits on its census', async () => { + await restTestChat(rig, 'session-cleared') + await restTestChat(rig, 'session-fresh', { listed: false }) + await rig.crash() + await rig.boot() + const { runtime, listAll } = restartedRuntime() + expect(await listAll()).toEqual(['session-cleared']) + const record = rig.store.getRecord('session-cleared')! + await rig.store.setConversationCommand('session-cleared', record.lease.runtimeFence, { + command: 'clear', + state: 'completed', + phase: 'committed', + operationId: 'clear-1', + callerKey: 'client-1', + replacementSessionId: 'session-fresh' + }) + expect(await listAll()).toEqual(['session-fresh']) + // An authoritative census that could not reach every host collects a second time; hold that one. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: these runtime members exist; they are protected. + const internal = runtime as unknown as { + supportsAuthoritativeSessionTabsInventory(): boolean + getAuthoritativeSessionTabsInventoryEpoch(): string + refreshMobileSessionPtyInventory(): Promise + } + internal.supportsAuthoritativeSessionTabsInventory = () => true + internal.getAuthoritativeSessionTabsInventoryEpoch = () => 'epoch-1' + let collections = 0 + const inSecond = Promise.withResolvers() + const held = Promise.withResolvers() + releaseHeld = held.resolve + internal.refreshMobileSessionPtyInventory = async () => { + collections += 1 + if (collections === 2) { + inSecond.resolve() + await held.promise + } + return null + } + const listing = listAll() + await inSecond.promise + + // agentSession.reveal from another client: the cleared chat opened from its history. + const revealed = await rig.host.revealSession('session-cleared') + await runtime.publishStructuredAgentSessionTab({ ...revealed, activate: true }) + held.resolve() + + expect(await listing).toEqual(['session-cleared']) + expect(await listAll()).toEqual(['session-cleared']) + }) + + it('writes nothing to the record store while it lists', async () => { + const ids = ['session-1', 'session-2', 'session-3', 'session-4', 'session-5'] + for (const sessionId of ids) { + await restTestChat(rig, sessionId, { message: sessionId }) + } + await rig.crash() + await rig.boot() + // Settled first, so a write here could only be the listing's own. + await rig.host.reconcileRestartLeases() + const writes = vi.spyOn(AgentSessionStoreTransactions.prototype, 'transact') + const { listAll } = restartedRuntime() + + expect(await listAll()).toEqual(ids) + // Every write, a tab's visibility included, is one store transaction. + expect(writes).not.toHaveBeenCalled() + }) + + it('lists every chat when four fail to open, and the others still get status rows (T5)', async () => { + // The first four failing: each failure must cost only its own chat, never the chats after it. + const ids = ['session-1', 'session-2', 'session-3', 'session-4', 'session-5', 'session-6'] + const failing = ids.slice(0, 4) + for (const sessionId of ids) { + await restTestChat(rig, sessionId, { message: `asked ${sessionId}` }) + } + await rig.crash() + await rig.boot() + rig.adapter.historyFilePath.mockImplementation(async (sessionId) => { + if (failing.includes(sessionId)) { + throw new Error('EACCES: permission denied') + } + return null + }) + const warn = vi.spyOn(rig.host.deps.logger, 'warn').mockImplementation(() => undefined) + const restore = vi.spyOn(rig.host, 'restoreReadableSessions') + const { listAll } = restartedRuntime() + + expect(await within(listAll())).toEqual(ids) + await vi.waitFor(() => expect(restore).toHaveBeenCalledOnce()) + await restore.mock.results[0]?.value + for (const sessionId of ['session-5', 'session-6']) { + expect(rig.host.hasSession(sessionId)).toBe(true) + expect(latestRestTestStatus(rig, sessionId)).toMatchObject({ + latestPrompt: `asked ${sessionId}` + }) + } + const failures = warn.mock.calls.filter( + ([message]) => message === 'restoring a chat for reading failed' + ) + expect(failures.map(([, detail]) => detail)).toEqual( + failing.map((sessionId) => + expect.objectContaining({ + sessionId, + error: expect.objectContaining({ message: 'EACCES: permission denied' }) + }) + ) + ) + expect(await listAll()).toEqual(ids) + expect(restore).toHaveBeenCalledOnce() + }) + + it('lists in tab-table order, not the order the chats were created (T7)', async () => { + for (const sessionId of ['session-c', 'session-a', 'session-b']) { + await restTestChat(rig, sessionId, { listed: false }) + } + for (const sessionId of ['session-b', 'session-c', 'session-a']) { + await rig.store.setSessionTabVisibility(sessionId, true) + } + await rig.crash() + await rig.boot() + const { listAll } = restartedRuntime() + + expect(await within(listAll())).toEqual(['session-b', 'session-c', 'session-a']) + }) + + it('lists in the saved window order when the profile has no tab table (T7)', async () => { + for (const sessionId of ['session-c', 'session-a', 'session-b']) { + await restTestChat(rig, sessionId, { listed: false }) + } + await rig.crash() + const host = await rig.boot() + vi.spyOn(host, 'getPersistedVisibleSessionTabIndex').mockReturnValue({ + present: false, + sessionIds: [] + }) + const savedTab = (sessionId: string, sortOrder: number) => ({ + id: `agent-session:${sessionId}`, + entityId: sessionId, + groupId: 'group-1', + worktreeId: 'workspace-1', + contentType: 'agent-session', + label: 'Codex Chat', + customLabel: null, + color: null, + sortOrder, + createdAt: 1 + }) + const { listAll } = restartedRuntime({ + activeRepoId: null, + activeWorktreeId: 'workspace-1', + activeTabId: null, + tabsByWorktree: {}, + terminalLayoutsByTabId: {}, + unifiedTabs: { + 'workspace-1': [ + savedTab('session-a', 0), + savedTab('session-b', 1), + savedTab('session-c', 2) + ] + } + }) + + expect(await within(listAll())).toEqual(['session-a', 'session-b', 'session-c']) + }) +})