diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-teardown.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-teardown.ts index a4e53d2f4c7..2a1c25ae0da 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-teardown.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-teardown.ts @@ -18,6 +18,11 @@ export type StructuredAgentSessionTeardownPhase = { /** Quit must not wait indefinitely on an in-flight handoff; see `drain-handoffs` below. */ const HANDOFF_DRAIN_TIMEOUT_MS = 5_000 +/** The marker write goes through the store's transaction queue, which an operation wedged at quit + * can still be occupying. Giving up costs the user one click on the next launch; waiting costs + * them a quit that never finishes, so this bookkeeping never gets to gate the shutdown. */ +const RESUME_MARKER_RECORD_TIMEOUT_MS = 2_000 + /** Eight steps at ten seconds each would outlast the global quit deadline, and a quit that dies * mid-eviction leaves the lease unreleased — the exact state restart has to clean up. Bounded * well below that deadline so the phases after this one still get to run. */ @@ -55,8 +60,23 @@ export function structuredAgentSessionHostTeardownPhases(collaborators: { handoffs: { stopTuiHistoryCatchup: () => void; drain: () => Promise } tasks: { drainAttaches: () => Promise } evictOwnedSessions: () => Promise + recordResumeMarkers: () => Promise }): StructuredAgentSessionTeardownPhase[] { return [ + // FIRST, because `evict-owned-sessions` settles every running turn to `interrupted` — after it + // the live signal this reads is gone. Its failure is swallowed rather than collected: losing a + // resume offer is a nuisance, and bookkeeping must never be what stops a quit. + { + name: 'record-resume-markers', + run: () => + withTimeout( + collaborators.recordResumeMarkers().catch((error: unknown) => { + console.warn('[structured-agent-session] recording resume markers failed', error) + }), + RESUME_MARKER_RECORD_TIMEOUT_MS, + undefined + ) + }, { name: 'dispose-holds', run: () => collaborators.holds.dispose() }, { name: 'stop-lease-renewal', run: () => collaborators.runtimeState.stopLeaseRenewal() }, { name: 'stop-tui-catchup', run: () => collaborators.handoffs.stopTuiHistoryCatchup() }, 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 8fcdef95480..af51b30808f 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 @@ -52,6 +52,11 @@ import type { StructuredAgentSessionStatusSubscriber } from './structured-agent- import { StructuredAgentSessionEventRecovery } from './structured-agent-session-event-recovery' import { StructuredAgentSessionBackgroundTaskChannel } from './structured-agent-session-background-task-channel' import { StructuredAgentSessionClientDelivery } from './structured-agent-session-client-delivery' +import type { AgentSessionResumeTrigger } from '../../../shared/agent-session-resume-marker' +import { + createStructuredAgentSessionRestartResume, + type StructuredAgentSessionRestartResume +} from './structured-agent-session-restart-resume-host' export type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types' export class StructuredAgentSessionHost { @@ -76,6 +81,8 @@ export class StructuredAgentSessionHost { private readonly holds: StructuredAgentSessionHolds private readonly eventRecovery: StructuredAgentSessionEventRecovery private readonly backgroundTasks: StructuredAgentSessionBackgroundTaskChannel + /** Public because the RPC surface addresses it directly; see the restart-resume collaborator. */ + readonly restartResume: StructuredAgentSessionRestartResume constructor(readonly deps: StructuredAgentSessionHostDeps) { this.backgroundTasks = new StructuredAgentSessionBackgroundTaskChannel( @@ -145,6 +152,11 @@ export class StructuredAgentSessionHost { attachContext: () => this.attachContext(), onBarrierError: (sessionId, error) => deps.onEventSinkError?.({ sessionId, error }) }) + this.restartResume = createStructuredAgentSessionRestartResume(deps, this.sessions, { + revealSession: this.revealSession, + hold: this.hold, + now: this.now + }) this.runtimeState.startLeaseRenewal() } @@ -244,7 +256,7 @@ export class StructuredAgentSessionHost { flushStreamedEvents = (sessionId: string): Promise => this.runtimeState.flushEventSink(sessionId) - async flushAllStreamedEvents(): Promise { + async flushAllStreamedEvents(options?: { trigger?: AgentSessionResumeTrigger }): Promise { const retainSessionIds = new Set() await tearDownStructuredAgentSessionHost({ phases: structuredAgentSessionHostTeardownPhases({ @@ -253,7 +265,8 @@ export class StructuredAgentSessionHost { handoffs: this.handoffs, tasks: this.tasks, evictOwnedSessions: () => - evictOwnedStructuredAgentSessions(this.lifetimeContext(), retainSessionIds) + evictOwnedStructuredAgentSessions(this.lifetimeContext(), retainSessionIds), + recordResumeMarkers: () => this.restartResume.recordMarkers(options?.trigger ?? 'quit') }), sessions: this.sessions, retainSessionIds, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts new file mode 100644 index 00000000000..429a468f6c3 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts @@ -0,0 +1,122 @@ +// The host's restart-resume surface: what teardown records, what may resume, and the one call that +// resumes it. +// +// Assembled here rather than on the host for the same reason holds and handoffs were — the host is +// a coordinator, and a marker set that has to open journals before it can adjudicate them reads +// better next to the predicate it feeds. + +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import type { AgentSessionResumeTrigger } from '../../../shared/agent-session-resume-marker' +import { + latestStructuredAgentSessionPrompt, + newestStructuredAgentSessionTurnId +} from '../../../shared/structured-agent-session-projection' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import { adapterSupportsRecord } from './structured-agent-session-provider-support' +import { + structuredAgentSessionResumableSet, + type StructuredAgentSessionResumeCandidate +} from './structured-agent-session-restart-resume-set' +import { + resumeStructuredAgentSessionsFromRestart, + StructuredAgentSessionResumeAdmission, + type StructuredAgentSessionResumeOutcome +} from './structured-agent-session-restart-resume-runner' +import { structuredAgentSessionsWorkingAtTeardown } from './structured-agent-session-working-at-teardown' + +type LiveSession = { journal: AgentSessionJournal; hasProviderChild: boolean } + +/** The host capabilities this needs, named so the collaborator cannot quietly grow more. */ +type RestartResumeSurfaces = { + revealSession: (sessionId: string) => Promise<{ readable: boolean }> + /** The resume-capable hold; see the runner for why a hold and not a send. */ + hold: (sessionId: string, holderId: string) => Promise + now: () => number +} + +export type StructuredAgentSessionRestartResume = { + recordMarkers: (trigger: AgentSessionResumeTrigger) => Promise + list: () => Promise + resume: ( + sessionIds: readonly string[] | undefined, + owner: string + ) => Promise + dismiss: () => Promise +} + +export function createStructuredAgentSessionRestartResume( + deps: { store: AgentSessionRecordStore; adapter: StructuredAgentSessionAdapter }, + /** The host's LIVE session map — the only honest answer to "was this actually working". */ + sessions: ReadonlyMap, + surfaces: RestartResumeSurfaces +): StructuredAgentSessionRestartResume { + const admission = new StructuredAgentSessionResumeAdmission() + const itemsFor = (sessionId: string) => sessions.get(sessionId)?.journal.snapshot().items ?? [] + + const list = async (): Promise => { + const now = surfaces.now() + const markers = deps.store.resumeMarkers.list(now) + // A marked session this launch has not opened yet cannot answer for its own turn, and an + // unreadable journal leaves the predicate with one record instead of two — which refuses. + for (const marker of markers) { + if (!sessions.has(marker.sessionId)) { + await surfaces.revealSession(marker.sessionId).catch(() => null) + } + } + return structuredAgentSessionResumableSet({ + markers, + getRecord: deps.store.getRecord, + supportsRecord: (record) => adapterSupportsRecord(deps.adapter, record), + journalTurnId: (sessionId) => newestStructuredAgentSessionTurnId(itemsFor(sessionId)), + latestPrompt: (sessionId) => latestStructuredAgentSessionPrompt(itemsFor(sessionId)), + now + }) + } + + return { + recordMarkers: (trigger) => + deps.store.resumeMarkers.record( + structuredAgentSessionsWorkingAtTeardown({ + sessions, + getRecord: deps.store.getRecord, + trigger, + now: surfaces.now() + }), + surfaces.now() + ), + list, + /** + * Turning the offer down, which SPENDS the markers. + * + * A prompt that returns at every launch is worse than the problem it solves. Nothing is lost by + * spending them: the first resume-capable hold on a childless session re-acquires the provider + * at the same proved cursor, so opening the chat still resumes it. Every live marker goes, not + * just the eligible ones, so an ineligible marker cannot make the prompt reappear either. + */ + dismiss: async () => { + const markers = deps.store.resumeMarkers.list(surfaces.now()) + for (const marker of markers) { + await deps.store.resumeMarkers.consume(marker.sessionId) + } + return markers.length + }, + resume: async (sessionIds, owner) => { + const requested = sessionIds ? new Set(sessionIds) : null + // Re-derived, never taken from the caller: a client may name any session id, and only the + // predicate decides which of them is allowed a provider child. + const candidates = (await list()).filter( + (candidate) => !requested || requested.has(candidate.sessionId) + ) + return resumeStructuredAgentSessionsFromRestart( + { + admission, + consumeMarker: (sessionId) => deps.store.resumeMarkers.consume(sessionId), + resume: (sessionId) => surfaces.hold(sessionId, `restart-resume:${sessionId}`) + }, + candidates, + owner + ) + } + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-runner.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-runner.ts new file mode 100644 index 00000000000..0d8d58a8804 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-runner.ts @@ -0,0 +1,107 @@ +// Spending the markers: the one path that turns a resumable candidate back into a live agent. +// +// The manual "Resume" button and the automatic setting both land here, so the two can never drift +// into different eligibility or different double-fire protection. +// +// Resume itself is a HOLD, not a send. The first resume-capable hold on a childless session +// re-acquires the provider at the cursor the record already proved — Claude's `resume` + +// `resumeSessionAt`, Codex's thread id — which is native continuation. Nothing re-sends the user's +// prompt: that is what makes an agent redo work it already finished. + +import { forEachWithConcurrency } from '../../../shared/map-with-concurrency' +import type { StructuredAgentSessionResumeCandidate } from './structured-agent-session-restart-resume-set' + +/** Providers are expensive to start and 20-30 marked chats is an ordinary morning. Resumes go out + * a few at a time so a launch cannot spawn every app-server at once. */ +export const STRUCTURED_AGENT_SESSION_RESUME_CONCURRENCY = 3 + +export const STRUCTURED_AGENT_SESSION_RESUME_IN_PROGRESS = + 'agent_session_resume_already_in_progress' + +export type StructuredAgentSessionResumeOutcome = { + sessionId: string + outcome: 'resumed' | 'refused' + /** Refusal code; `agent_session_resume_already_in_progress` names the live owner in `owner`. */ + reason?: string + owner?: string +} + +/** + * One resume per session at a time, whoever is asking. + * + * Two surfaces can reach for the same chat at once — the banner's "Resume all" and a user clicking + * one row — and both would otherwise take a hold, race the acquisition, and leave the loser's + * refusal looking like a real failure. The second caller is told who holds it instead. + */ +export class StructuredAgentSessionResumeAdmission { + private readonly owners = new Map() + + liveOwner(sessionId: string): string | null { + return this.owners.get(sessionId) ?? null + } + + async run(sessionId: string, owner: string, task: () => Promise): Promise { + const live = this.owners.get(sessionId) + if (live !== undefined) { + throw Object.assign(new Error(STRUCTURED_AGENT_SESSION_RESUME_IN_PROGRESS), { owner: live }) + } + this.owners.set(sessionId, owner) + try { + return await task() + } finally { + this.owners.delete(sessionId) + } + } +} + +export type StructuredAgentSessionResumeRunnerDeps = { + admission: StructuredAgentSessionResumeAdmission + /** Spends the marker durably. False means another launch already took it. */ + consumeMarker: (sessionId: string) => Promise + /** Takes the resume-capable hold that re-acquires the provider child. */ + resume: (sessionId: string) => Promise + concurrency?: number +} + +export async function resumeStructuredAgentSessionsFromRestart( + deps: StructuredAgentSessionResumeRunnerDeps, + candidates: readonly StructuredAgentSessionResumeCandidate[], + owner: string +): Promise { + const outcomes: StructuredAgentSessionResumeOutcome[] = [] + await forEachWithConcurrency( + candidates, + deps.concurrency ?? STRUCTURED_AGENT_SESSION_RESUME_CONCURRENCY, + async (candidate) => { + outcomes.push(await resumeOne(deps, candidate.sessionId, owner)) + } + ) + return outcomes +} + +async function resumeOne( + deps: StructuredAgentSessionResumeRunnerDeps, + sessionId: string, + owner: string +): Promise { + try { + return await deps.admission.run(sessionId, owner, async () => { + // Consumed BEFORE the hold, not after it succeeds. A crash between the two costs one resume + // the user can start by hand; the other order costs them the same agent running twice. + if (!(await deps.consumeMarker(sessionId))) { + return { sessionId, outcome: 'refused' as const, reason: 'agent_session_resume_consumed' } + } + await deps.resume(sessionId) + return { sessionId, outcome: 'resumed' as const } + }) + } catch (error) { + const reason = error instanceof Error ? error.message : String(error) + const live = (error as { owner?: string }).owner + return { + sessionId, + outcome: 'refused', + reason, + ...(live === undefined ? {} : { owner: live }) + } + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-set.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-set.ts new file mode 100644 index 00000000000..660b8b534fe --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-set.ts @@ -0,0 +1,81 @@ +// Which of the previous generation's markers a launch may actually act on. +// +// Every clause here exists to refuse, and the bias is deliberate: a session resumed that should not +// have been spends the user's tokens and can make an agent redo destructive work it already +// finished. A session missed is an annoyance. When any input is ambiguous this answers "no". +// +// Two INDEPENDENT records must concur. The marker is teardown's word; the journal's own turn record +// is the session's word. One without the other proves nothing — a marker whose journal never opened +// that turn is a marker for work that did not exist, and a journal turn with no marker is the +// stale-`running`-row case this whole mechanism exists to refuse. + +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { agentSessionProviderHandleChainHead } from '../../../shared/agent-session-provider-handle' +import { agentSessionProviderHandleKey } from '../../../shared/agent-session-provider-handle' +import { + isExpiredAgentSessionResumeMarker, + type AgentSessionResumeMarker, + type AgentSessionResumeTrigger +} from '../../../shared/agent-session-resume-marker' +import { isResumableStructuredAgentSessionRecord } from './structured-agent-session-resume-eligibility' + +export type StructuredAgentSessionResumeCandidate = { + sessionId: string + workspaceId: string + agent: AgentSessionRecord['provider'] + turnId: string + trigger: AgentSessionResumeTrigger + recordedAt: number + /** The prompt the row quotes, so the user recognises the chat before resuming it. */ + latestPrompt: string +} + +export type StructuredAgentSessionResumeSetInput = { + markers: readonly AgentSessionResumeMarker[] + getRecord: (sessionId: string) => AgentSessionRecord | null + supportsRecord: (record: AgentSessionRecord) => boolean + /** The newest turn id in that session's journal, whatever state it settled in; null when the + * journal could not be read. Deliberately not the live-turn reader: eviction has already + * rewritten that turn to `interrupted` by the time this runs. */ + journalTurnId: (sessionId: string) => string | null + latestPrompt: (sessionId: string) => string + now: number +} + +export function structuredAgentSessionResumableSet( + input: StructuredAgentSessionResumeSetInput +): StructuredAgentSessionResumeCandidate[] { + const candidates: StructuredAgentSessionResumeCandidate[] = [] + for (const marker of input.markers) { + if (isExpiredAgentSessionResumeMarker(marker, input.now)) { + continue + } + const record = input.getRecord(marker.sessionId) + if (!record || !input.supportsRecord(record)) { + continue + } + // The lease must be free and adjudicated. A contested or still-reconciling record is somebody + // else's to resolve, and resuming into it is how a session gets two writers. + if (!isResumableStructuredAgentSessionRecord(record)) { + continue + } + // A cursor that changed since teardown is a different conversation than the one we marked. + const head = agentSessionProviderHandleChainHead(record.providerHandleChain) + if (!head || agentSessionProviderHandleKey(head.handle) !== marker.providerHandleKey) { + continue + } + if (input.journalTurnId(marker.sessionId) !== marker.turnId) { + continue + } + candidates.push({ + sessionId: marker.sessionId, + workspaceId: record.location.workspaceId, + agent: record.provider, + turnId: marker.turnId, + trigger: marker.trigger, + recordedAt: marker.recordedAt, + latestPrompt: input.latestPrompt(marker.sessionId) + }) + } + return candidates +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume.test.ts new file mode 100644 index 00000000000..d2c0b8cd303 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume.test.ts @@ -0,0 +1,494 @@ +// The restart-resume safety rules, stated as refusals. +// +// Every negative case here is a session that MUST NOT get a provider child back. Resuming one that +// was not working spends the user's tokens and can make an agent redo destructive work it already +// finished; missing one is an annoyance. Each test removes exactly one input from an otherwise +// resumable session, so deleting the matching guard turns that test red. + +import { describe, expect, it, vi } from 'vitest' +import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { + AGENT_SESSION_RESUME_MARKER_TTL_MS, + type AgentSessionResumeMarker +} from '../../../shared/agent-session-resume-marker' +import { newestStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-live-turn' +import { structuredAgentSessionResumableSet } from './structured-agent-session-restart-resume-set' +import { + resumeStructuredAgentSessionsFromRestart, + StructuredAgentSessionResumeAdmission, + STRUCTURED_AGENT_SESSION_RESUME_CONCURRENCY, + STRUCTURED_AGENT_SESSION_RESUME_IN_PROGRESS +} from './structured-agent-session-restart-resume-runner' +import { structuredAgentSessionsWorkingAtTeardown } from './structured-agent-session-working-at-teardown' +import { createStructuredAgentSessionRestartResume } from './structured-agent-session-restart-resume-host' + +const SESSION = 'session-working-1' +const THREAD = 'thread-1' +const HANDLE_KEY = `codex:${JSON.stringify(THREAD)}` +const NOW = 1_700_000_000_000 + +function turnItem( + turnId: string, + state: 'running' | 'completed' | 'interrupted' +): AgentJournalRenderItem { + return { + itemId: `turn:${turnId}`, + revision: 1, + body: { kind: 'turn', turnId, state }, + sequence: 1, + observedAt: NOW + } +} + +function record(overrides: { chain?: AgentSessionRecord['providerHandleChain'] } = {}) { + return { + schemaVersion: 2, + sessionId: SESSION, + location: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'git-worktree' + }, + provider: 'codex', + providerHandleChain: overrides.chain ?? [ + { + linkId: 'link-1', + handle: { provider: 'codex', threadId: THREAD }, + origin: 'created', + mintedAtFence: 1, + observedAt: NOW + } + ], + accountHome: { variable: 'CODEX_HOME', path: '/home/codex' }, + lease: { + sessionId: SESSION, + runtimeKind: 'native', + runtimeFence: 1, + handoffStage: null, + provenHandleLinkId: null, + ownerProcess: null, + reservedSpawnToken: null, + leaseDeadlineAt: NOW, + lastRenewedAt: NOW, + handoffOperationId: null, + journalCheckpoint: null, + claimKeyId: 'key-1', + claimStatus: 'released', + unreconciled: false, + deathEvidence: null + }, + createdAt: NOW, + updatedAt: NOW + } as unknown as AgentSessionRecord +} + +function journal(items: AgentJournalRenderItem[], isReadOnly = false) { + return { isReadOnly, snapshot: () => ({ items, submissions: [] }) } as never +} + +function marker(overrides: Partial = {}): AgentSessionResumeMarker { + return { + sessionId: SESSION, + turnId: 'turn-1', + recordedAt: NOW, + trigger: 'quit', + providerHandleKey: HANDLE_KEY, + ...overrides + } +} + +function resumableSet(input: { + markers: AgentSessionResumeMarker[] + items?: AgentJournalRenderItem[] + chain?: AgentSessionRecord['providerHandleChain'] + now?: number +}) { + const items = input.items ?? [turnItem('turn-1', 'interrupted')] + return structuredAgentSessionResumableSet({ + markers: input.markers, + getRecord: () => record(input.chain === undefined ? {} : { chain: input.chain }), + supportsRecord: () => true, + journalTurnId: () => newestStructuredAgentSessionTurnId(items), + latestPrompt: () => 'fix the auth bug', + now: input.now ?? NOW + }) +} + +describe('deriving what was working at teardown', () => { + it('marks a session this host was running a turn for', () => { + const markers = structuredAgentSessionsWorkingAtTeardown({ + sessions: new Map([ + [SESSION, { journal: journal([turnItem('turn-1', 'running')]), hasProviderChild: true }] + ]), + getRecord: () => record(), + trigger: 'quit', + now: NOW + }) + + expect(markers).toEqual([ + { + sessionId: SESSION, + turnId: 'turn-1', + recordedAt: NOW, + trigger: 'quit', + providerHandleKey: HANDLE_KEY + } + ]) + }) + + it('carries the update trigger so the surface can say the restart was not the user choice', () => { + const [recorded] = structuredAgentSessionsWorkingAtTeardown({ + sessions: new Map([ + [SESSION, { journal: journal([turnItem('turn-1', 'running')]), hasProviderChild: true }] + ]), + getRecord: () => record(), + trigger: 'update', + now: NOW + }) + + expect(recorded?.trigger).toBe('update') + }) + + it('marks nothing for an idle session', () => { + expect( + structuredAgentSessionsWorkingAtTeardown({ + sessions: new Map([[SESSION, { journal: journal([]), hasProviderChild: true }]]), + getRecord: () => record(), + trigger: 'quit', + now: NOW + }) + ).toEqual([]) + }) + + it('marks nothing for a turn that completed before the quit', () => { + expect( + structuredAgentSessionsWorkingAtTeardown({ + sessions: new Map([ + [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), hasProviderChild: true }] + ]), + getRecord: () => record(), + trigger: 'quit', + now: NOW + }) + ).toEqual([]) + }) + + // The user's stated fear. A journal restored for READING carries whatever `running` row an older + // crash left behind, and it is the live `hasProviderChild` — not that row — that decides. + it('marks nothing for a stale running row this host was not executing', () => { + expect( + structuredAgentSessionsWorkingAtTeardown({ + sessions: new Map([ + [SESSION, { journal: journal([turnItem('turn-1', 'running')]), hasProviderChild: false }] + ]), + getRecord: () => record(), + trigger: 'quit', + now: NOW + }) + ).toEqual([]) + }) + + it('marks nothing for a session that never proved a provider cursor', () => { + expect( + structuredAgentSessionsWorkingAtTeardown({ + sessions: new Map([ + [SESSION, { journal: journal([turnItem('turn-1', 'running')]), hasProviderChild: true }] + ]), + getRecord: () => record({ chain: [] }), + trigger: 'quit', + now: NOW + }) + ).toEqual([]) + }) +}) + +describe('the resumable set', () => { + it('offers a genuinely working session exactly once', () => { + const candidates = resumableSet({ markers: [marker()] }) + + expect(candidates).toHaveLength(1) + expect(candidates[0]).toMatchObject({ + sessionId: SESSION, + turnId: 'turn-1', + trigger: 'quit', + latestPrompt: 'fix the auth bug' + }) + }) + + // No marker means no teardown ever observed this session working, whatever its journal says. + it('offers nothing for a stale running row with no marker', () => { + expect(resumableSet({ markers: [], items: [turnItem('turn-1', 'running')] })).toEqual([]) + }) + + it('refuses when the marker and the journal name different turns', () => { + expect( + resumableSet({ + markers: [marker({ turnId: 'turn-9' })], + items: [turnItem('turn-1', 'interrupted')] + }) + ).toEqual([]) + }) + + it('refuses when the journal cannot answer for any turn', () => { + expect(resumableSet({ markers: [marker()], items: [] })).toEqual([]) + }) + + it('refuses when the session has no resume cursor', () => { + expect(resumableSet({ markers: [marker()], chain: [] })).toEqual([]) + }) + + // A cursor that moved since teardown is a different conversation than the one we marked. + it('refuses when the resume cursor drifted after the marker was written', () => { + expect( + resumableSet({ markers: [marker({ providerHandleKey: 'codex:"other-thread"' })] }) + ).toEqual([]) + }) + + it('refuses a marker that has outlived its expiry', () => { + expect( + resumableSet({ markers: [marker()], now: NOW + AGENT_SESSION_RESUME_MARKER_TTL_MS + 1 }) + ).toEqual([]) + }) +}) + +describe('the restart-resume surface', () => { + function surface(input: { + markers?: AgentSessionResumeMarker[] + sessions?: Map + }) { + const live = new Map((input.markers ?? [marker()]).map((entry) => [entry.sessionId, entry])) + const recorded: AgentSessionResumeMarker[][] = [] + const held: string[] = [] + const store = { + getRecord: () => record(), + resumeMarkers: { + list: () => [...live.values()], + record: async (markers: readonly AgentSessionResumeMarker[]) => { + recorded.push([...markers]) + live.clear() + markers.forEach((entry) => live.set(entry.sessionId, entry)) + }, + consume: async (sessionId: string) => live.delete(sessionId) + } + } + const sessions = + input.sessions ?? + new Map([ + [ + SESSION, + { journal: journal([turnItem('turn-1', 'interrupted')]), hasProviderChild: false } + ] + ]) + return { + restartResume: createStructuredAgentSessionRestartResume( + { store, adapter: { supportsCreate: () => true } } as never, + sessions as never, + { + revealSession: async () => ({ readable: true }), + hold: async (sessionId: string) => { + held.push(sessionId) + }, + now: () => NOW + } + ), + live, + recorded, + held + } + } + + it('offers and resumes an eligible session', async () => { + const { restartResume, held } = surface({}) + + expect(await restartResume.list()).toHaveLength(1) + await restartResume.resume(undefined, 'modal') + + expect(held).toEqual([SESSION]) + }) + + // Turning the prompt down must not leave anything that can bring it back next launch — including + // a marker that was never eligible in the first place. + it('spends every live marker on dismiss, eligible or not', async () => { + const ineligible = marker({ sessionId: 'session-working-2', turnId: 'turn-elsewhere' }) + const { restartResume, live } = surface({ markers: [marker(), ineligible] }) + + expect(await restartResume.dismiss()).toBe(2) + + expect(live.size).toBe(0) + expect(await restartResume.list()).toEqual([]) + }) + + // The client names ids; only the host decides which of them may have a provider child. + it('resumes nothing for a session id the caller invented', async () => { + const { restartResume, held } = surface({}) + + const outcomes = await restartResume.resume(['session-not-offered-1'], 'modal') + + expect(outcomes).toEqual([]) + expect(held).toEqual([]) + }) + + // Quitting while the prompt is open: the offered session has no provider child in THIS + // generation, so teardown mints no marker for it and the replace-the-whole-set write clears the + // old one. The offer is discarded rather than resurrected, and nothing can double-fire. + it('leaves no marker behind when the user quits with the offer still open', async () => { + const { restartResume, live, recorded } = surface({}) + + await restartResume.recordMarkers('quit') + + expect(recorded).toEqual([[]]) + expect(live.size).toBe(0) + }) + + it('re-marks a session whose resume is already running when the next quit lands', async () => { + const { restartResume, recorded } = surface({ + sessions: new Map([ + [SESSION, { journal: journal([turnItem('turn-2', 'running')]), hasProviderChild: true }] + ]) + }) + + await restartResume.recordMarkers('update') + + expect(recorded[0]).toEqual([ + { + sessionId: SESSION, + turnId: 'turn-2', + recordedAt: NOW, + trigger: 'update', + providerHandleKey: HANDLE_KEY + } + ]) + }) +}) + +describe('spending a marker', () => { + function runner(overrides: { resume?: () => Promise; concurrency?: number } = {}) { + const consumed = new Set() + const resume = overrides.resume ?? vi.fn(async () => {}) + return { + resume, + consumed, + deps: { + admission: new StructuredAgentSessionResumeAdmission(), + // Stands in for the durable store: the first caller spends it, later ones find it gone. + consumeMarker: async (sessionId: string) => { + if (consumed.has(sessionId)) { + return false + } + consumed.add(sessionId) + return true + }, + resume, + ...(overrides.concurrency === undefined ? {} : { concurrency: overrides.concurrency }) + } + } + } + + const candidate = (sessionId: string) => ({ + sessionId, + workspaceId: 'workspace-1', + agent: 'codex' as const, + turnId: 'turn-1', + trigger: 'quit' as const, + recordedAt: NOW, + latestPrompt: '' + }) + + it('resumes a candidate once and reports it', async () => { + const { deps, resume } = runner() + + const outcomes = await resumeStructuredAgentSessionsFromRestart( + deps, + [candidate(SESSION)], + 'banner' + ) + + expect(outcomes).toEqual([{ sessionId: SESSION, outcome: 'resumed' }]) + expect(resume).toHaveBeenCalledOnce() + }) + + // A second relaunch finds the marker already spent; nothing may run again. + it('refuses a marker a previous launch already consumed', async () => { + const { deps, resume } = runner() + await resumeStructuredAgentSessionsFromRestart(deps, [candidate(SESSION)], 'first-launch') + + const outcomes = await resumeStructuredAgentSessionsFromRestart( + deps, + [candidate(SESSION)], + 'second-launch' + ) + + expect(outcomes).toEqual([ + { sessionId: SESSION, outcome: 'refused', reason: 'agent_session_resume_consumed' } + ]) + expect(resume).toHaveBeenCalledOnce() + }) + + it('spends the marker before it submits, so a crash mid-resume cannot double-fire', async () => { + const order: string[] = [] + const { deps, consumed } = runner({ + resume: async () => { + order.push(`consumed:${consumed.has(SESSION)}`) + throw new Error('provider died mid-resume') + } + }) + + const outcomes = await resumeStructuredAgentSessionsFromRestart( + deps, + [candidate(SESSION)], + 'banner' + ) + + expect(order).toEqual(['consumed:true']) + expect(outcomes[0]).toMatchObject({ outcome: 'refused' }) + // Still spent after the failure: the next launch must not retry it on its own. + expect(consumed.has(SESSION)).toBe(true) + }) + + it('refuses a second concurrent resume and names the live owner', async () => { + let release = (): void => {} + const blocked = new Promise((resolve) => { + release = resolve + }) + const { deps } = runner({ resume: () => blocked }) + + const first = resumeStructuredAgentSessionsFromRestart(deps, [candidate(SESSION)], 'banner') + await vi.waitFor(() => expect(deps.admission.liveOwner(SESSION)).toBe('banner')) + const second = await resumeStructuredAgentSessionsFromRestart(deps, [candidate(SESSION)], 'row') + release() + await first + + expect(second).toEqual([ + { + sessionId: SESSION, + outcome: 'refused', + reason: STRUCTURED_AGENT_SESSION_RESUME_IN_PROGRESS, + owner: 'banner' + } + ]) + }) + + it('staggers instead of starting every provider at once', async () => { + let live = 0 + let peak = 0 + const { deps } = runner({ + resume: async () => { + live += 1 + peak = Math.max(peak, live) + await new Promise((resolve) => setTimeout(resolve, 5)) + live -= 1 + } + }) + const candidates = Array.from({ length: 12 }, (_, index) => + candidate(`session-staggered-${index}`) + ) + + const outcomes = await resumeStructuredAgentSessionsFromRestart(deps, candidates, 'banner') + + // Unbounded fan-out would peak at all 12 — which is the spawn storm this exists to prevent. + expect(peak).toBe(STRUCTURED_AGENT_SESSION_RESUME_CONCURRENCY) + expect(outcomes).toHaveLength(candidates.length) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-teardown-handoff-drain.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-teardown-handoff-drain.test.ts index 12d3e6dfe43..479e8f0f186 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-teardown-handoff-drain.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-teardown-handoff-drain.test.ts @@ -155,9 +155,12 @@ describe('structured agent-session host teardown', () => { runtimeState: { stopLeaseRenewal: () => undefined, flushAllEventSinks: noop }, handoffs: { stopTuiHistoryCatchup: () => undefined, drain: noop }, tasks: { drainAttaches: noop }, - evictOwnedSessions: noop + evictOwnedSessions: noop, + recordResumeMarkers: noop }) expect(phases.map((phase) => phase.name)).toEqual([ + // First: eviction settles every running turn, so the live signal is gone after it. + 'record-resume-markers', 'dispose-holds', 'stop-lease-renewal', 'stop-tui-catchup', @@ -177,7 +180,8 @@ describe('structured agent-session host teardown', () => { vi.useFakeTimers() try { const teardown = host.flushAllStreamedEvents() - await vi.advanceTimersByTimeAsync(5_000) + // Covers both bounded phases: the resume-marker write gives up first, then the handoff drain. + await vi.advanceTimersByTimeAsync(10_000) await expect(teardown).resolves.toBeUndefined() } finally { vi.useRealTimers() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-working-at-teardown.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-working-at-teardown.ts new file mode 100644 index 00000000000..0402113ac68 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-working-at-teardown.ts @@ -0,0 +1,63 @@ +// Which sessions were genuinely working when this process went away. +// +// Read off the LIVE host state, never off a persisted status field. That distinction is the whole +// safety argument: a `running` turn row left behind by an older crash is still sitting in that +// session's journal, and a rule that trusted it would hand a provider child back to work nobody is +// doing. A crashed generation leaves no entry in this map, so it can never produce a marker. +// +// Three facts have to line up for one marker, and each rules out a different false positive: +// this host is running the child (not a journal we merely opened for reading), the journal's newest +// turn is actually running (not one that completed before quit), and the session has a provider +// cursor to resume onto (not a conversation that never proved a thread). + +import { agentSessionProviderHandleChainHead } from '../../../shared/agent-session-provider-handle' +import { agentSessionProviderHandleKey } from '../../../shared/agent-session-provider-handle' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { + AgentSessionResumeMarker, + AgentSessionResumeTrigger +} from '../../../shared/agent-session-resume-marker' +import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-live-turn' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' + +type WorkingCandidateSession = { + journal: AgentSessionJournal + /** Only this host generation's own child counts. A restored-for-reading journal has none. */ + hasProviderChild: boolean +} + +export function structuredAgentSessionsWorkingAtTeardown(input: { + sessions: ReadonlyMap + getRecord: (sessionId: string) => AgentSessionRecord | null + trigger: AgentSessionResumeTrigger + now: number +}): AgentSessionResumeMarker[] { + const markers: AgentSessionResumeMarker[] = [] + for (const [sessionId, session] of input.sessions) { + if (!session.hasProviderChild) { + continue + } + // A journal this host cannot read tells us nothing about what the turn was doing. + if (session.journal.isReadOnly) { + continue + } + const turnId = activeStructuredAgentSessionTurnId(session.journal.snapshot().items) + if (!turnId) { + continue + } + const head = agentSessionProviderHandleChainHead( + input.getRecord(sessionId)?.providerHandleChain ?? [] + ) + if (!head) { + continue + } + markers.push({ + sessionId, + turnId, + recordedAt: input.now, + trigger: input.trigger, + providerHandleKey: agentSessionProviderHandleKey(head.handle) + }) + } + return markers +} diff --git a/src/main/runtime/agent-session-claim-key-retention.ts b/src/main/runtime/agent-session-claim-key-retention.ts new file mode 100644 index 00000000000..c6b1729db84 --- /dev/null +++ b/src/main/runtime/agent-session-claim-key-retention.ts @@ -0,0 +1,30 @@ +// How long a retired claim key stays verifiable, and the two reads that depend on it. +// +// Split off the record store so that class stays within its size budget; the rule itself is +// unchanged: a rotation must never strand an agent that is still running under the old key. + +import type { AgentSessionStoreState } from './agent-session-record-store-file' + +export const AGENT_SESSION_CLAIM_KEY_RETENTION_MS = 30 * 24 * 60 * 60 * 1000 + +export function isAgentSessionClaimKeyVerifiable( + state: AgentSessionStoreState, + keyId: string, + now: number +): boolean { + const retired = state.retiredClaimKeys.find((entry) => entry.keyId === keyId) + return !retired || now - retired.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS +} + +export function retireAgentSessionClaimKey( + state: AgentSessionStoreState, + keyId: string, + now: number +): void { + if (!state.retiredClaimKeys.some((entry) => entry.keyId === keyId)) { + state.retiredClaimKeys.push({ keyId, retiredAt: now }) + } + state.retiredClaimKeys = state.retiredClaimKeys.filter( + (entry) => now - entry.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS + ) +} diff --git a/src/main/runtime/agent-session-record-store-file.ts b/src/main/runtime/agent-session-record-store-file.ts index 7eb3cacdc7c..9d39871e109 100644 --- a/src/main/runtime/agent-session-record-store-file.ts +++ b/src/main/runtime/agent-session-record-store-file.ts @@ -8,8 +8,8 @@ */ import { createHash } from 'node:crypto' -import { chmod, mkdir, readFile, rm } from 'node:fs/promises' -import { dirname, join } from 'node:path' +import { readFile } from 'node:fs/promises' +import { join } from 'node:path' import { agentSessionOperationKey, isAgentSessionOperationRow, @@ -20,12 +20,10 @@ import { isAgentSessionRecord, type AgentSessionRecord } from '../../shared/agent-session-record' -import { - copyFileDurable, - durableWriteTempPath, - renameDurable, - writeTempFileDurable -} from '../durable-file-write' +import type { AgentSessionResumeMarker } from '../../shared/agent-session-resume-marker' +import { parseAgentSessionResumeMarkers } from './agent-session-resume-marker-file' +import { agentSessionStoreBackupPath as backupPath } from './agent-session-record-store-write' +export { saveAgentSessionStore } from './agent-session-record-store-write' import { parseVisibleSessionIds } from './agent-session-visible-tab-index' import { serializeAgentSessionStoreState } from './agent-session-store-serialization' @@ -47,6 +45,11 @@ export type AgentSessionStoreState = { visibleSessionIds: Set /** True once this store has committed the visibility index field. */ visibleSessionIdsIndexPresent: boolean + /** Teardown's record of the sessions that were working when the app went away. Kept beside the + * records rather than on them: a marker is a consumed-once obligation with its own expiry, not + * part of a session's durable identity, and quit writes every session's marker in ONE + * transaction — a per-journal write would race the bounded quit deadline 20-30 times over. */ + resumeMarkers: Map } export type LoadedAgentSessionStore = { @@ -64,8 +67,6 @@ export function agentSessionStorePath(directory: string): string { return join(directory, AGENT_SESSION_STORE_FILE_NAME) } -const backupPath = (filePath: string): string => `${filePath}.bak` - function emptyState(hostId: string): AgentSessionStoreState { return { schemaVersion: AGENT_SESSION_STORE_SCHEMA_VERSION, @@ -75,7 +76,8 @@ function emptyState(hostId: string): AgentSessionStoreState { retiredClaimKeys: [], unreadableRecords: new Map(), visibleSessionIds: new Set(), - visibleSessionIdsIndexPresent: false + visibleSessionIdsIndexPresent: false, + resumeMarkers: new Map() } } @@ -225,6 +227,9 @@ function parseState( } state.visibleSessionIdsIndexPresent = visibleSessionIds.present visibleSessionIds.ids.forEach((sessionId) => state.visibleSessionIds.add(sessionId)) + state.resumeMarkers = parseAgentSessionResumeMarkers( + (parsed as { resumeMarkers?: unknown }).resumeMarkers + ) return { state, needsRewrite } } @@ -311,35 +316,3 @@ export async function loadAgentSessionStore( needsRewrite: false } } - -/** - * Commit the whole state. The live path is never absent: the new content is made durable in a temp - * file first, a validated primary is COPIED to the backup, and only then does the rename publish it. - * Backup recovery keeps the known-good backup in place while publishing the repaired primary. - * - * The old ordering renamed the live file aside before writing the new one, so a death in that - * window left the profile with a backup and no primary — which is exactly the state that wedged a - * real profile. Copy, don't move. - */ -export async function saveAgentSessionStore( - filePath: string, - state: AgentSessionStoreState, - options: { primaryStatus: 'validated' | 'unusable-or-absent' } -): Promise { - const directory = dirname(filePath) - await mkdir(directory, { recursive: true, mode: 0o700 }) - await chmod(directory, 0o700) - const tmpPath = durableWriteTempPath(filePath) - try { - await writeTempFileDurable(tmpPath, serializeAgentSessionStoreState(state), 0o600) - // Only a primary parsed under the transaction lock may replace the backup. During recovery the - // primary is corrupt or absent, so the known-good backup must survive until publication. - if (options.primaryStatus === 'validated') { - await copyFileDurable(filePath, backupPath(filePath)) - } - await renameDurable(tmpPath, filePath) - } catch (error) { - await rm(tmpPath, { force: true }).catch(() => {}) - throw error - } -} diff --git a/src/main/runtime/agent-session-record-store-write.ts b/src/main/runtime/agent-session-record-store-write.ts new file mode 100644 index 00000000000..76f68f03de3 --- /dev/null +++ b/src/main/runtime/agent-session-record-store-write.ts @@ -0,0 +1,48 @@ +/** + * Committing the durable agent-session store. + * + * Split from the read/parse half so neither outgrows its budget. The ordering below is the whole + * reason this is its own file: the live path is never absent. New content is made durable in a temp + * file first, a validated primary is COPIED to the backup, and only then does the rename publish it. + * Backup recovery keeps the known-good backup in place while publishing the repaired primary. + * + * The old ordering renamed the live file aside before writing the new one, so a death in that + * window left the profile with a backup and no primary — which is exactly the state that wedged a + * real profile. Copy, don't move. + */ + +import { chmod, mkdir, rm } from 'node:fs/promises' +import { dirname } from 'node:path' +import { + copyFileDurable, + durableWriteTempPath, + renameDurable, + writeTempFileDurable +} from '../durable-file-write' +import type { AgentSessionStoreState } from './agent-session-record-store-file' +import { serializeAgentSessionStoreState } from './agent-session-store-serialization' + +export const agentSessionStoreBackupPath = (filePath: string): string => `${filePath}.bak` + +export async function saveAgentSessionStore( + filePath: string, + state: AgentSessionStoreState, + options: { primaryStatus: 'validated' | 'unusable-or-absent' } +): Promise { + const directory = dirname(filePath) + await mkdir(directory, { recursive: true, mode: 0o700 }) + await chmod(directory, 0o700) + const tmpPath = durableWriteTempPath(filePath) + try { + await writeTempFileDurable(tmpPath, serializeAgentSessionStoreState(state), 0o600) + // Only a primary parsed under the transaction lock may replace the backup. During recovery the + // primary is corrupt or absent, so the known-good backup must survive until publication. + if (options.primaryStatus === 'validated') { + await copyFileDurable(filePath, agentSessionStoreBackupPath(filePath)) + } + await renameDurable(tmpPath, filePath) + } catch (error) { + await rm(tmpPath, { force: true }).catch(() => {}) + throw error + } +} diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index e3655b273fc..efc108e5ed0 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -16,6 +16,11 @@ import { type AgentSessionMutationOperationAdmission, type AgentSessionOperationAdmission } from './agent-session-operation-admission' +import { createAgentSessionResumeMarkerStore } from './agent-session-resume-marker-store' +import { + isAgentSessionClaimKeyVerifiable, + retireAgentSessionClaimKey +} from './agent-session-claim-key-retention' import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication' import { classifyObservedAgentSessionSpawnToken } from '../../shared/agent-session-lease-adjudication' import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' @@ -71,11 +76,15 @@ import { export const AGENT_SESSION_LEASE_TTL_MS = 30_000, AGENT_SESSION_LEASE_RENEW_INTERVAL_MS = 10_000 -/** Retired claim keys stay verifiable this long so a rotation cannot strand a running agent. */ -export const AGENT_SESSION_CLAIM_KEY_RETENTION_MS = 30 * 24 * 60 * 60 * 1000 +export { AGENT_SESSION_CLAIM_KEY_RETENTION_MS } from './agent-session-claim-key-retention' export class AgentSessionRecordStore { - private constructor(private readonly transactions: AgentSessionStoreTransactionQueue) {} + /** Teardown's record of what was working; spent once by the resume that uses one. */ + readonly resumeMarkers: ReturnType + + private constructor(private readonly transactions: AgentSessionStoreTransactionQueue) { + this.resumeMarkers = createAgentSessionResumeMarkerStore(transactions) + } static async open(args: { directory: string; hostId: string }): Promise { const filePath = agentSessionStorePath(args.directory) @@ -159,10 +168,8 @@ export class AgentSessionRecordStore { listOperationRows = (): AgentSessionOperationRow[] => [...this.state.operations.values()] - isClaimKeyVerifiable(keyId: string, now: number): boolean { - const retired = this.state.retiredClaimKeys.find((entry) => entry.keyId === keyId) - return !retired || now - retired.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS - } + isClaimKeyVerifiable = (keyId: string, now: number): boolean => + isAgentSessionClaimKeyVerifiable(this.state, keyId, now) /** Spawn tokens observed on the host with no matching lease. Stop them; never adopt them. */ listOrphanSpawnTokens(observedTokens: readonly string[]): string[] { @@ -320,16 +327,8 @@ export class AgentSessionRecordStore { replaceSessionOptions = (args: AgentSessionOptionsReplacement): Promise => this.mutate(args.sessionId, (record) => replaceAgentSessionRecordOptions(record, args)) - async retireClaimKey(keyId: string, now: number): Promise { - await this.transact(() => { - if (!this.state.retiredClaimKeys.some((entry) => entry.keyId === keyId)) { - this.state.retiredClaimKeys.push({ keyId, retiredAt: now }) - } - this.state.retiredClaimKeys = this.state.retiredClaimKeys.filter( - (entry) => now - entry.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS - ) - }) - } + retireClaimKey = (keyId: string, now: number): Promise => + this.transact(() => retireAgentSessionClaimKey(this.state, keyId, now)) private async mutate( sessionId: string, diff --git a/src/main/runtime/agent-session-recovery-publish-fault.test.ts b/src/main/runtime/agent-session-recovery-publish-fault.test.ts index f3f2ba91d2a..64ec740ac6f 100644 --- a/src/main/runtime/agent-session-recovery-publish-fault.test.ts +++ b/src/main/runtime/agent-session-recovery-publish-fault.test.ts @@ -49,7 +49,8 @@ function state(generation: number): AgentSessionStoreState { retiredClaimKeys: [{ keyId: `generation-${generation}`, retiredAt: generation }], unreadableRecords: new Map(), visibleSessionIds: new Set(), - visibleSessionIdsIndexPresent: false + visibleSessionIdsIndexPresent: false, + resumeMarkers: new Map() } } diff --git a/src/main/runtime/agent-session-reservation-admission.test.ts b/src/main/runtime/agent-session-reservation-admission.test.ts index 80bcb77cfa5..12924da19d5 100644 --- a/src/main/runtime/agent-session-reservation-admission.test.ts +++ b/src/main/runtime/agent-session-reservation-admission.test.ts @@ -71,7 +71,8 @@ function storeState(records: readonly AgentSessionRecord[] = []): AgentSessionSt retiredClaimKeys: [], unreadableRecords: new Map(), visibleSessionIds: new Set(), - visibleSessionIdsIndexPresent: true + visibleSessionIdsIndexPresent: true, + resumeMarkers: new Map() } } diff --git a/src/main/runtime/agent-session-resume-marker-file.ts b/src/main/runtime/agent-session-resume-marker-file.ts new file mode 100644 index 00000000000..50711462503 --- /dev/null +++ b/src/main/runtime/agent-session-resume-marker-file.ts @@ -0,0 +1,25 @@ +// Reading resume markers back off disk. +// +// Advisory state, so a malformed entry is DROPPED rather than failing the load. Every other section +// of the store file refuses the whole file on a bad row, which is right for a lease and wrong for +// this: a resume marker must never be able to cost a user their entire session store. + +import { + isAgentSessionResumeMarker, + type AgentSessionResumeMarker +} from '../../shared/agent-session-resume-marker' + +export function parseAgentSessionResumeMarkers( + value: unknown +): Map { + const markers = new Map() + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + return markers + } + for (const [sessionId, entry] of Object.entries(value)) { + if (isAgentSessionResumeMarker(entry) && entry.sessionId === sessionId) { + markers.set(sessionId, entry) + } + } + return markers +} diff --git a/src/main/runtime/agent-session-resume-marker-store.test.ts b/src/main/runtime/agent-session-resume-marker-store.test.ts new file mode 100644 index 00000000000..91e99aa60f0 --- /dev/null +++ b/src/main/runtime/agent-session-resume-marker-store.test.ts @@ -0,0 +1,122 @@ +// Markers on disk: they survive a restart, they are spent exactly once, they expire, and a +// malformed one can never cost the user their session store. + +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { + AGENT_SESSION_RESUME_MARKER_TTL_MS, + type AgentSessionResumeMarker +} from '../../shared/agent-session-resume-marker' +import { AgentSessionRecordStore } from './agent-session-record-store' +import { agentSessionStorePath } from './agent-session-record-store-file' + +const NOW = 1_700_000_000_000 +const SESSION = 'session-working-1' + +function marker(overrides: Partial = {}): AgentSessionResumeMarker { + return { + sessionId: SESSION, + turnId: 'turn-1', + recordedAt: NOW, + trigger: 'quit', + providerHandleKey: 'codex:"thread-1"', + ...overrides + } +} + +let root: string +let directory: string + +async function openStore(): Promise { + return AgentSessionRecordStore.open({ directory, hostId: 'local' }) +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-resume-marker-')) + directory = join(root, 'agent-sessions') +}) + +afterEach(async () => { + await rm(root, { recursive: true, force: true }) +}) + +describe('durable resume markers', () => { + // The transaction queue compares the state it is about to write against a snapshot; a markers-only + // write is the one change that has no other field moving with it. + it('persists a markers-only transaction across a restart', async () => { + const store = await openStore() + await store.resumeMarkers.record([marker()], NOW) + + const reopened = await openStore() + + expect(reopened.resumeMarkers.list(NOW)).toEqual([marker()]) + }) + + it('replaces the whole set, so an earlier generation leaves nothing behind', async () => { + const store = await openStore() + await store.resumeMarkers.record([marker(), marker({ sessionId: 'session-working-2' })], NOW) + + await store.resumeMarkers.record([marker({ sessionId: 'session-working-2' })], NOW) + + expect((await openStore()).resumeMarkers.list(NOW).map((entry) => entry.sessionId)).toEqual([ + 'session-working-2' + ]) + }) + + it('spends a marker exactly once', async () => { + const store = await openStore() + await store.resumeMarkers.record([marker()], NOW) + + expect(await store.resumeMarkers.consume(SESSION)).toBe(true) + expect(await store.resumeMarkers.consume(SESSION)).toBe(false) + expect((await openStore()).resumeMarkers.list(NOW)).toEqual([]) + }) + + it('stops reporting a marker once it has expired', async () => { + const store = await openStore() + await store.resumeMarkers.record([marker()], NOW) + + expect(store.resumeMarkers.list(NOW + AGENT_SESSION_RESUME_MARKER_TTL_MS + 1)).toEqual([]) + }) + + it('treats a marker from the future as expired rather than immortal', async () => { + const store = await openStore() + await store.resumeMarkers.record([marker({ recordedAt: NOW + 60_000 })], NOW + 60_000) + + expect(store.resumeMarkers.list(NOW)).toEqual([]) + }) + + // Every other section of this file refuses the whole store on a malformed entry. A resume marker + // is advisory, and must never be able to make a user's sessions unreadable. + it('drops a malformed marker instead of failing the load', async () => { + const store = await openStore() + await store.resumeMarkers.record([marker()], NOW) + const filePath = agentSessionStorePath(directory) + const parsed = JSON.parse(await readFile(filePath, 'utf-8')) + parsed.resumeMarkers = { + [SESSION]: { sessionId: SESSION, turnId: 42, trigger: 'nonsense' }, + 'session-working-2': marker({ sessionId: 'session-working-2' }) + } + await writeFile(filePath, JSON.stringify(parsed)) + + const reopened = await openStore() + + expect(reopened.resumeMarkers.list(NOW).map((entry) => entry.sessionId)).toEqual([ + 'session-working-2' + ]) + }) + + it('reads a profile written before markers existed', async () => { + const store = await openStore() + // Recorded first only to create the file; the field is then stripped to model an older profile. + await store.resumeMarkers.record([marker()], NOW) + const filePath = agentSessionStorePath(directory) + const parsed = JSON.parse(await readFile(filePath, 'utf-8')) + delete parsed.resumeMarkers + await writeFile(filePath, JSON.stringify(parsed)) + + expect((await openStore()).resumeMarkers.list(NOW)).toEqual([]) + }) +}) diff --git a/src/main/runtime/agent-session-resume-marker-store.ts b/src/main/runtime/agent-session-resume-marker-store.ts new file mode 100644 index 00000000000..d8f11275850 --- /dev/null +++ b/src/main/runtime/agent-session-resume-marker-store.ts @@ -0,0 +1,39 @@ +// Resume-marker reads and writes, kept off the record store class so that surface stays within its +// size budget. Built straight on the transaction queue, so a marker write is the same whole-file +// atomic commit every other durable session fact gets. + +import { + isExpiredAgentSessionResumeMarker, + type AgentSessionResumeMarker +} from '../../shared/agent-session-resume-marker' +import type { AgentSessionStoreTransactionQueue } from './agent-session-store-transaction-queue' + +export type AgentSessionResumeMarkerStore = { + list: (now: number) => AgentSessionResumeMarker[] + record: (markers: readonly AgentSessionResumeMarker[], now: number) => Promise + consume: (sessionId: string) => Promise +} + +export function createAgentSessionResumeMarkerStore( + queue: AgentSessionStoreTransactionQueue +): AgentSessionResumeMarkerStore { + return { + // Expired markers are filtered on READ as well as on write: reporting one must never depend on + // a prune having already run. + list: (now) => + [...queue.state.resumeMarkers.values()].filter( + (marker) => !isExpiredAgentSessionResumeMarker(marker, now) + ), + // Replaces the whole set: quit records every working session at once, so anything still here + // from an earlier generation is stale by definition. + record: (markers, now) => + queue.transact(() => { + queue.state.resumeMarkers = new Map( + markers + .filter((marker) => !isExpiredAgentSessionResumeMarker(marker, now)) + .map((marker) => [marker.sessionId, marker]) + ) + }), + consume: (sessionId) => queue.transact(() => queue.state.resumeMarkers.delete(sessionId)) + } +} diff --git a/src/main/runtime/agent-session-store-serialization.ts b/src/main/runtime/agent-session-store-serialization.ts index cda3e44d7ff..abe89fa42c2 100644 --- a/src/main/runtime/agent-session-store-serialization.ts +++ b/src/main/runtime/agent-session-store-serialization.ts @@ -16,5 +16,9 @@ export function serializeAgentSessionStoreState(state: AgentSessionStoreState): if (state.visibleSessionIdsIndexPresent) { serialized.visibleSessionIds = [...state.visibleSessionIds] } + // Emitted only when present, so a profile that never recorded one keeps its existing bytes. + if (state.resumeMarkers.size > 0) { + serialized.resumeMarkers = Object.fromEntries(state.resumeMarkers) + } return JSON.stringify(serialized) } diff --git a/src/main/runtime/agent-session-store-transaction-queue.ts b/src/main/runtime/agent-session-store-transaction-queue.ts index 060ec3bd633..0c1926d2084 100644 --- a/src/main/runtime/agent-session-store-transaction-queue.ts +++ b/src/main/runtime/agent-session-store-transaction-queue.ts @@ -39,12 +39,15 @@ function agentSessionStoreStateChanged( retiredClaimKeys: AgentSessionStoreState['retiredClaimKeys'], unreadableRecords: AgentSessionStoreState['unreadableRecords'], visibleSessionIds: AgentSessionStoreState['visibleSessionIds'], - visibleSessionIdsIndexPresent: AgentSessionStoreState['visibleSessionIdsIndexPresent'] + visibleSessionIdsIndexPresent: AgentSessionStoreState['visibleSessionIdsIndexPresent'], + resumeMarkers: AgentSessionStoreState['resumeMarkers'] ): boolean { return ( !mapEntriesMatch(state.records, records) || !mapEntriesMatch(state.operations, operations) || !mapEntriesMatch(state.unreadableRecords, unreadableRecords) || + // Without this a markers-only transaction compares equal and is never written to disk. + !mapEntriesMatch(state.resumeMarkers, resumeMarkers) || state.visibleSessionIdsIndexPresent !== visibleSessionIdsIndexPresent || state.visibleSessionIds.size !== visibleSessionIds.size || [...state.visibleSessionIds].some((id) => !visibleSessionIds.has(id)) || @@ -101,6 +104,7 @@ export class AgentSessionStoreTransactionQueue { const unreadableRecords = new Map(this.state.unreadableRecords) const visibleSessionIds = new Set(this.state.visibleSessionIds) const visibleSessionIdsIndexPresent = this.state.visibleSessionIdsIndexPresent + const resumeMarkers = new Map(this.state.resumeMarkers) try { // The lost commit may have granted a higher fence than the backup records show. Rather // than refuse forever, raise every recovered fence clear of anything that commit could @@ -120,7 +124,8 @@ export class AgentSessionStoreTransactionQueue { retiredClaimKeys, unreadableRecords, visibleSessionIds, - visibleSessionIdsIndexPresent + visibleSessionIdsIndexPresent, + resumeMarkers ) ) { return result @@ -141,6 +146,7 @@ export class AgentSessionStoreTransactionQueue { this.state.unreadableRecords = unreadableRecords this.state.visibleSessionIds = visibleSessionIds this.state.visibleSessionIdsIndexPresent = visibleSessionIdsIndexPresent + this.state.resumeMarkers = resumeMarkers throw error } }) diff --git a/src/main/runtime/rpc/methods/structured-agent-session-restart-resume.ts b/src/main/runtime/rpc/methods/structured-agent-session-restart-resume.ts new file mode 100644 index 00000000000..cd40bc85aad --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-restart-resume.ts @@ -0,0 +1,48 @@ +// `agentSession.restartResumable` / `agentSession.restartResume` — the restart-resume offer. +// +// Both reach for records on disk this process may not have opened yet, so they build the host the +// way hold and reveal do. Listing is read-only and takes nothing live; resuming goes through the +// host's single resume path, which re-derives eligibility rather than trusting the ids it is given. + +import { defineMethod } from '../core' +import { + ensureStructuredHostInstalled, + requireStructuredHost, + structuredCallerFor +} from './structured-agent-session-gate' +import { RestartResumableParams, RestartResumeParams } from './structured-agent-session-schemas' + +export const STRUCTURED_AGENT_SESSION_RESTART_RESUME_METHODS = [ + defineMethod({ + name: 'agentSession.restartResumable', + params: RestartResumableParams, + handler: async (_params, ctx) => { + await ensureStructuredHostInstalled(ctx) + return { sessions: await requireStructuredHost(ctx).restartResume.list() } + } + }), + defineMethod({ + // Spends the markers without resuming; see the collaborator for why turning the offer down + // consumes it rather than leaving it to return at every launch. + name: 'agentSession.restartResumableDismiss', + params: RestartResumableParams, + handler: async (_params, ctx) => { + await ensureStructuredHostInstalled(ctx) + return { dismissed: await requireStructuredHost(ctx).restartResume.dismiss() } + } + }), + defineMethod({ + name: 'agentSession.restartResume', + params: RestartResumeParams, + handler: async (params, ctx) => { + await ensureStructuredHostInstalled(ctx) + const host = requireStructuredHost(ctx) + return { + results: await host.restartResume.resume( + params.sessionIds, + structuredCallerFor(ctx).callerKey + ) + } + } + }) +] diff --git a/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts b/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts index 7725e70bded..6ae8d8d04a4 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts @@ -17,6 +17,8 @@ export { MutationEnvelope, OptionsParams, RespondParams, + RestartResumableParams, + RestartResumeParams, RewindParams, SendParams, SessionId, diff --git a/src/main/runtime/rpc/methods/structured-agent-session.test.ts b/src/main/runtime/rpc/methods/structured-agent-session.test.ts index 73e3113aea8..f45b961083a 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.test.ts @@ -162,7 +162,7 @@ describe('capability gating', () => { } // Bump deliberately: the whole agentSession.* surface is behind the structured capability, // so an additive method is invisible to old clients and needs no protocol bump. - expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(22) + expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(25) }) it('hides the surface from a declared client that did not advertise it', async () => { diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index f1d0fc59ec5..a6f09a18de4 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -34,6 +34,7 @@ import { } from './structured-agent-session-create' import { STRUCTURED_AGENT_SESSION_HOLD_METHODS } from './structured-agent-session-hold' import { STRUCTURED_AGENT_SESSION_REVEAL_METHODS } from './structured-agent-session-reveal' +import { STRUCTURED_AGENT_SESSION_RESTART_RESUME_METHODS } from './structured-agent-session-restart-resume' import { resolveUncommittedStructuredCreate } from './structured-agent-session-precommit-refusal' import { bindStructuredAgentSessionStream, @@ -326,5 +327,6 @@ export const STRUCTURED_AGENT_SESSION_METHODS = [ }), ...STRUCTURED_AGENT_SESSION_HOLD_METHODS, ...STRUCTURED_AGENT_SESSION_REVEAL_METHODS, + ...STRUCTURED_AGENT_SESSION_RESTART_RESUME_METHODS, ...STRUCTURED_AGENT_SESSION_STATUS_METHODS ] diff --git a/src/main/runtime/structured-agent-session-runtime.ts b/src/main/runtime/structured-agent-session-runtime.ts index 92643fcae57..467a88a1e65 100644 --- a/src/main/runtime/structured-agent-session-runtime.ts +++ b/src/main/runtime/structured-agent-session-runtime.ts @@ -10,6 +10,7 @@ import { existsSync } from 'node:fs' import { join } from 'node:path' import type { AgentSessionRecord } from '../../shared/agent-session-record' +import type { AgentSessionResumeTrigger } from '../../shared/agent-session-resume-marker' import { createCodexStructuredLaunchResolver } from '../codex/codex-structured-launch-resolution' import { CodexStructuredSessionAdapter, @@ -111,6 +112,15 @@ export const CLAUDE_STRUCTURED_AUTH_POLICY_REQUIRED = */ const pendingTeardown = new Set() +/** Why the app is going away, for the resume markers teardown stamps. A module-level latch rather + * than an argument because the quit path's call to `stopStructuredAgentSessionRuntime()` is + * asserted verbatim by the startup-ordering ratchet. */ +let teardownTrigger: AgentSessionResumeTrigger = 'quit' + +export function setStructuredAgentSessionTeardownTrigger(trigger: AgentSessionResumeTrigger): void { + teardownTrigger = trigger +} + export function ensureStructuredAgentSessionHost( deps: StructuredAgentSessionRuntimeDeps ): Promise { @@ -139,7 +149,10 @@ export async function waitForStructuredAgentSessionRecovery(): Promise { * A teardown that fails is RETRIED by the next stop rather than forgotten: the * host keeps every journal whose close rejected, and this is the only handle * onto that host once the module slot is cleared. */ -export async function stopStructuredAgentSessionRuntime(): Promise { +export async function stopStructuredAgentSessionRuntime(options?: { + trigger?: AgentSessionResumeTrigger +}): Promise { + const trigger = options?.trigger ?? teardownTrigger const pending = installing installing = null setStructuredAgentSessionHost(null) @@ -153,7 +166,7 @@ export async function stopStructuredAgentSessionRuntime(): Promise { const failures: unknown[] = [] for (const runtime of outstanding) { try { - await tearDownRuntime(runtime) + await tearDownRuntime(runtime, trigger) } catch (error) { pendingTeardown.add(runtime) failures.push(error) @@ -167,7 +180,10 @@ export async function stopStructuredAgentSessionRuntime(): Promise { } } -async function tearDownRuntime(installed: InstalledRuntime): Promise { +async function tearDownRuntime( + installed: InstalledRuntime, + trigger: AgentSessionResumeTrigger +): Promise { // Drain an in-flight recovery before stopping children; recovery may still // be writing lifecycle rows or acquiring a replacement child. await installed.waitForRecovery() @@ -185,7 +201,7 @@ async function tearDownRuntime(installed: InstalledRuntime): Promise { // A row a child delivers during that backstop close is not captured, and was not captured // under the old order either. The drain below keeps a late callback from outliving the runtime. try { - await installed.host.flushAllStreamedEvents() + await installed.host.flushAllStreamedEvents({ trigger }) } catch (error) { failures.push(error) } diff --git a/src/main/startup/main-process-quit.ts b/src/main/startup/main-process-quit.ts index a136981c873..abc10099e8d 100644 --- a/src/main/startup/main-process-quit.ts +++ b/src/main/startup/main-process-quit.ts @@ -8,7 +8,10 @@ import { beginSshShutdown } from '../ipc/ssh-shutdown-drain' import { agentHookServer } from '../agent-hooks/server' import { wslHookRelayManager } from '../agent-hooks/wsl-hook-relay-manager' import { removeManagedAgentHooksAsync } from '../agent-hooks/managed-agent-hook-controls' -import { stopStructuredAgentSessionRuntime } from '../runtime/structured-agent-session-runtime' +import { + setStructuredAgentSessionTeardownTrigger, + stopStructuredAgentSessionRuntime +} from '../runtime/structured-agent-session-runtime' import { awaitRuntimeFileWatcherUnsubscribes } from '../runtime/orca-runtime-files' import { clearRuntimeMetadataIfOwned } from '../runtime/runtime-metadata' import { shutdownPairedRuntimeBrowserClientHosts } from '../browser/paired-runtime-browser-client-host-runtime' @@ -133,6 +136,9 @@ function installWillQuitHandler(): void { state.pluginMarketplaceInstaller = null const pluginHostShutdown = state.pluginService?.dispose() ?? Promise.resolve() const codexBackfillRecoveryShutdown = stopCodexStateDbBackfillRecoveries() + // Why before the stop: teardown stamps each working session's resume marker with why the app + // went away, and an update install is a restart the user never chose. + setStructuredAgentSessionTeardownTrigger(updateQuitInProgress ? 'update' : 'quit') const structuredAgentSessionShutdown = stopStructuredAgentSessionRuntime() state.pluginService = null setUnreadDockBadgeCount(0) diff --git a/src/renderer/src/app-shell/AppRootSurfaces.tsx b/src/renderer/src/app-shell/AppRootSurfaces.tsx index 31de9b1ce8d..59e6c67ee5c 100644 --- a/src/renderer/src/app-shell/AppRootSurfaces.tsx +++ b/src/renderer/src/app-shell/AppRootSurfaces.tsx @@ -9,6 +9,7 @@ import { MarkdownTemplatePicker } from '../components/editor/MarkdownTemplatePic import RecentTabSwitcher from '../components/tab-bar/RecentTabSwitcher' import { SkillFreshnessUpdateDialog } from '../components/skills/SkillFreshnessUpdateDialog' import { StarNagCard } from '../components/StarNagCard' +import { NativeChatResumeOnRestartModal } from '../components/NativeChatResumeOnRestartModal' import { StarNagAgentValueMomentObserver } from '../components/star-nag/StarNagAgentValueMomentObserver' import { StarNagToastHost } from '../components/star-nag/StarNagToastHost' import { TelemetryFirstLaunchSurface } from '../components/TelemetryFirstLaunchSurface' @@ -296,6 +297,9 @@ export function AppRootSurfaces(props: { + + + diff --git a/src/renderer/src/components/NativeChatResumeOnRestartModal.tsx b/src/renderer/src/components/NativeChatResumeOnRestartModal.tsx new file mode 100644 index 00000000000..e75af04cce8 --- /dev/null +++ b/src/renderer/src/components/NativeChatResumeOnRestartModal.tsx @@ -0,0 +1,277 @@ +import { useCallback, useEffect, useState } from 'react' +import { Play, RotateCcw } from 'lucide-react' +import { toast } from 'sonner' +import { Button } from './ui/button' +import { Checkbox } from './ui/checkbox' +import { + Dialog, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle +} from './ui/dialog' +import { useAppStore } from '../store' +import { callStructuredAgentSession } from '@/runtime/structured-agent-session-client' +import { translate } from '@/i18n/i18n' + +/** + * What would resume, shown before anything runs. + * + * The list is the point. Resuming a chat that was not working spends tokens and can make an agent + * redo work it already finished, so the user sees exactly which chats the last teardown recorded as + * mid-turn and decides. The checkbox is the opt-in to skipping this prompt in future — it removes + * the PROMPT, never a safety check: automatic mode calls the same RPC, which re-derives the same + * predicate and staggers the same way. + * + * Turning the offer down spends the markers. A prompt that returns at every launch is worse than + * the problem it solves, and nothing is lost: opening a chat takes a resume-capable hold, which + * re-acquires the provider at the same cursor. + */ + +type ResumeCandidate = { + sessionId: string + workspaceId: string + agent: 'claude' | 'codex' + trigger: 'quit' | 'update' + latestPrompt: string +} + +type ResumeOutcome = { sessionId: string; outcome: 'resumed' | 'refused' } + +// Structured sessions run on the machine hosting the runtime; both launch resolvers refuse anything +// else, so there is no remote target to aim this at. +const LOCAL = { kind: 'local' } as const + +function announceResumed(count: number): void { + if (count <= 0) { + return + } + toast( + count === 1 + ? translate('auto.components.NativeChatResumeOnRestartModal.resumedOne', 'Resumed 1 chat') + : translate( + 'auto.components.NativeChatResumeOnRestartModal.resumedMany', + 'Resumed {{value0}} chats', + { + value0: count + } + ) + ) +} + +export function NativeChatResumeOnRestartModal(): React.JSX.Element | null { + const structuredEnabled = useAppStore( + (store) => store.settings?.experimentalStructuredNativeChat === true + ) + const autoResume = useAppStore((store) => store.settings?.nativeChatResumeWorkOnRestart === true) + const updateSettings = useAppStore((store) => store.updateSettings) + const [candidates, setCandidates] = useState([]) + const [dontAskAgain, setDontAskAgain] = useState(false) + const [busy, setBusy] = useState(false) + const [resolved, setResolved] = useState(false) + + useEffect(() => { + if (!structuredEnabled || resolved) { + return + } + let cancelled = false + // Fetched after mount, never awaited by startup: the workspace is usable first. + void (async () => { + try { + const offered = await callStructuredAgentSession<{ sessions: ResumeCandidate[] }>( + LOCAL, + 'agentSession.restartResumable' + ) + if (cancelled || offered.sessions.length === 0) { + return + } + if (autoResume) { + // Identical call to the buttons below; the host re-derives eligibility either way. + const result = await callStructuredAgentSession<{ results: ResumeOutcome[] }>( + LOCAL, + 'agentSession.restartResume', + {} + ) + // Automatic must never be silent: someone who ticked the box months ago still sees this. + announceResumed(result.results.filter((entry) => entry.outcome === 'resumed').length) + return + } + setCandidates(offered.sessions) + } catch { + // A host that cannot answer offers nothing. There is no failure worth a modal of its own. + } + })() + return () => { + cancelled = true + } + }, [autoResume, resolved, structuredEnabled]) + + /** Applied on whichever action the user takes, so the box means the same thing either way. */ + const persistPreference = useCallback(async (): Promise => { + if (dontAskAgain) { + await updateSettings({ nativeChatResumeWorkOnRestart: true }).catch(() => undefined) + } + }, [dontAskAgain, updateSettings]) + + const resume = useCallback( + async (sessionIds?: string[]): Promise => { + setBusy(true) + try { + await persistPreference() + const result = await callStructuredAgentSession<{ results: ResumeOutcome[] }>( + LOCAL, + 'agentSession.restartResume', + sessionIds ? { sessionIds } : {} + ) + const settled = new Set(result.results.map((entry) => entry.sessionId)) + const remaining = candidates.filter((candidate) => !settled.has(candidate.sessionId)) + announceResumed(result.results.filter((entry) => entry.outcome === 'resumed').length) + setCandidates(remaining) + if (remaining.length === 0) { + setResolved(true) + } + } finally { + setBusy(false) + } + }, + [candidates, persistPreference] + ) + + /** Any close is a decline, and a decline spends the markers so this cannot return every launch. */ + const decline = useCallback(async (): Promise => { + setResolved(true) + await persistPreference() + await callStructuredAgentSession(LOCAL, 'agentSession.restartResumableDismiss', {}).catch( + () => undefined + ) + }, [persistPreference]) + + if (!structuredEnabled || resolved || candidates.length === 0) { + return null + } + + const interruptedByUpdate = candidates.some((candidate) => candidate.trigger === 'update') + + return ( + { + if (!next && !busy) { + void decline() + } + }} + > + {/* Height is capped, never the data: seeing WHICH chats would resume is the whole point, so + the list scrolls inside the dialog while the header and the primary action stay put. */} + + + + + {translate( + 'auto.components.NativeChatResumeOnRestartModal.title', + 'Resume interrupted chats?' + )} + + + {interruptedByUpdate + ? translate( + 'auto.components.NativeChatResumeOnRestartModal.updateBody', + 'These chats were mid-turn when Orca installed an update. Resuming continues each agent where it left off, without re-sending your prompt.' + ) + : translate( + 'auto.components.NativeChatResumeOnRestartModal.body', + 'These chats were mid-turn when Orca closed. Resuming continues each agent where it left off, without re-sending your prompt.' + )} + + + +
    + {candidates.map((candidate) => ( +
  • +
    +

    + {candidate.latestPrompt.trim() || + translate( + 'auto.components.NativeChatResumeOnRestartModal.untitled', + 'Untitled chat' + )} +

    +

    + {candidate.agent} · {candidate.workspaceId} +

    +
    + +
    • + ))} +
+ + {/* Says the quiet part: declining is not destructive, because opening the chat still + re-acquires it at the same cursor. */} +

+ {translate( + 'auto.components.NativeChatResumeOnRestartModal.notNowHint', + 'Not now keeps everything — opening a chat later still picks it up where it left off.' + )} +

+ + + + + + + + + + ) +} diff --git a/src/renderer/src/components/settings/NativeChatExperimentalSetting.tsx b/src/renderer/src/components/settings/NativeChatExperimentalSetting.tsx index 85d27dae2e3..05e2405d5cb 100644 --- a/src/renderer/src/components/settings/NativeChatExperimentalSetting.tsx +++ b/src/renderer/src/components/settings/NativeChatExperimentalSetting.tsx @@ -20,6 +20,7 @@ export function NativeChatExperimentalSetting({ }: NativeChatExperimentalSettingProps): React.JSX.Element { const nativeChatEnabled = settings.experimentalNativeChat === true const structuredNativeChatEnabled = settings.experimentalStructuredNativeChat === true + const resumeOnRestartEnabled = settings.nativeChatResumeWorkOnRestart === true const defaultView: NativeChatDefaultView = settings.openAgentTabsInChatByDefault === true ? 'native-chat' : 'terminal-chat' @@ -150,6 +151,36 @@ export function NativeChatExperimentalSetting({ /> ) : null} + + {/* Only structured sessions have a resume cursor to continue from. */} + {defaultView === 'native-chat' && structuredNativeChatEnabled ? ( +
+
+ +

+ {translate( + 'auto.components.settings.ExperimentalPane.nativeChat.resumeCopy', + 'When Orca quits or installs an update, chats that were mid-turn are offered again on the next launch. Turn this on to resume them automatically instead of choosing from the list.' + )} +

+
+ + updateSettings({ nativeChatResumeWorkOnRestart: !resumeOnRestartEnabled }) + } + /> +
+ ) : null} ) : null} diff --git a/src/renderer/src/i18n/en-runtime-required.json b/src/renderer/src/i18n/en-runtime-required.json index 8fb5d19446c..c9344bd8da0 100644 --- a/src/renderer/src/i18n/en-runtime-required.json +++ b/src/renderer/src/i18n/en-runtime-required.json @@ -1176,6 +1176,10 @@ "a05bcdaf57": "Agents View", "f63ea281e3": "Threaded left-sidebar feed for agent completions and blocking states.", "fb82ea1d7a": "Automatically materialize configured files or folders into newly created worktrees.", + "nativeChat": { + "resumeCopy": "When Orca quits or installs an update, chats that were mid-turn are offered again on the next launch. On, they resume without asking and Orca tells you afterwards — the same thing as ticking \"Don't ask again\" in that prompt. Off, you choose from the list each time.", + "resumeTitle": "Resume working chats automatically after a restart" + }, "newWorktreeCardStyle": { "copy": "Preview updated worktree-card layout, metadata placement, card-display menu options, and status presentation." } @@ -2618,9 +2622,6 @@ }, "components": { "native-chat": { - "approval": { - "cancel": "Cancel" - }, "composer": { "effort": "Effort" }, diff --git a/src/renderer/src/i18n/locales/en.json b/src/renderer/src/i18n/locales/en.json index 44f74db10cb..74eb5fbe44b 100644 --- a/src/renderer/src/i18n/locales/en.json +++ b/src/renderer/src/i18n/locales/en.json @@ -1919,6 +1919,22 @@ "SelectedTextCopyMenu": { "9b40d7b018": "Copy" }, + "NativeChatResumeOnRestartModal": { + "title": "Resume interrupted chats?", + "body": "These chats were mid-turn when Orca closed. Resuming continues each agent where it left off, without re-sending your prompt.", + "updateBody": "These chats were mid-turn when Orca installed an update. Resuming continues each agent where it left off, without re-sending your prompt.", + "resume": "Resume", + "resumeAll": "Resume all", + "resuming": "Resuming…", + "notNow": "Not now", + "untitled": "Untitled chat", + "listLabel": "Chats that would resume", + "notNowHint": "Not now keeps everything — opening a chat later still picks it up where it left off.", + "dontAskAgain": "Don't ask again — resume automatically next time", + "dontAskAgainHint": "Qualifying chats will resume on their own after a restart, and Orca will tell you when it happens. You can turn this off in Settings → Experimental → Chat UI.", + "resumedOne": "Resumed 1 chat", + "resumedMany": "Resumed {{value0}} chats" + }, "StarNagCard": { "92b0f9d921": "is authenticated and try again.", "cd8c34aac1": "gh", @@ -7041,7 +7057,10 @@ "structuredTitle": "Use updated structured native chat", "structuredCopy": "Opt in to the host-owned structured chat runtime for Codex and Claude. Off keeps the existing terminal-backed chat path.", "structuredScope": "Local sessions only for now. WSL and remote execution hosts (including SSH) continue to use terminal chat, and Windows falls back to it unless Orca can read process start times.", - "structuredToggleLabel": "Toggle updated structured native chat" + "structuredToggleLabel": "Toggle updated structured native chat", + "resumeTitle": "Resume working chats automatically after a restart", + "resumeCopy": "When Orca quits or installs an update, chats that were mid-turn are offered again on the next launch. On, they resume without asking and Orca tells you afterwards — the same thing as ticking \"Don't ask again\" in that prompt. Off, you choose from the list each time.", + "resumeToggleLabel": "Toggle automatic resume after a restart" }, "agentDashboard": { "title": "Agent Dashboard", diff --git a/src/shared/agent-session-resume-marker.ts b/src/shared/agent-session-resume-marker.ts new file mode 100644 index 00000000000..214c6ad394a --- /dev/null +++ b/src/shared/agent-session-resume-marker.ts @@ -0,0 +1,61 @@ +// What a teardown recorded about a session that was genuinely working when the app went away. +// +// A marker is written ONLY by the teardown path, from the live runtime — never derived from a +// persisted `running` row, which survives a crash and would resurrect work nobody is doing. It is +// the first of two records a resume needs: the journal's own turn record has to name the same turn +// before anything is handed a provider child again. +// +// Markers are transient obligations, so each one carries the two ways it can die: it is consumed on +// the resume that uses it, and it expires on its own if no launch ever does. + +/** Why the app went away. Recorded because an update install is a restart the user did not choose, + * and the surface that offers the resume says so. */ +export const AGENT_SESSION_RESUME_TRIGGERS = ['quit', 'update'] as const +export type AgentSessionResumeTrigger = (typeof AGENT_SESSION_RESUME_TRIGGERS)[number] + +/** A marker older than this is ignored and pruned: relaunching a week later must not restart a turn + * the user has long since forgotten, and an obligation with no expiry strands forever. */ +export const AGENT_SESSION_RESUME_MARKER_TTL_MS = 24 * 60 * 60 * 1000 + +export type AgentSessionResumeMarker = { + sessionId: string + /** The turn that was running when teardown observed it, from the live journal. */ + turnId: string + /** Execution host's clock at teardown. */ + recordedAt: number + trigger: AgentSessionResumeTrigger + /** Key of the provider handle this session had proved at teardown. The launch resolver reads its + * own copy off the record, so this is the CONCURRING record — it proves the cursor has not + * drifted since, never the cursor a spawn is aimed at. */ + providerHandleKey: string +} + +const MAX_FIELD_LENGTH = 512 + +function isMarkerField(value: unknown): value is string { + return typeof value === 'string' && value.length > 0 && value.length <= MAX_FIELD_LENGTH +} + +export function isAgentSessionResumeMarker(value: unknown): value is AgentSessionResumeMarker { + if (typeof value !== 'object' || value === null) { + return false + } + const marker = value as Partial + return ( + isMarkerField(marker.sessionId) && + isMarkerField(marker.turnId) && + isMarkerField(marker.providerHandleKey) && + Number.isSafeInteger(marker.recordedAt) && + (marker.recordedAt as number) >= 0 && + (marker.trigger === 'quit' || marker.trigger === 'update') + ) +} + +export function isExpiredAgentSessionResumeMarker( + marker: AgentSessionResumeMarker, + now: number +): boolean { + // A marker from the future is a clock that moved backwards, not a fresh one; treat it as expired + // rather than let it outlive every TTL. + return now < marker.recordedAt || now - marker.recordedAt > AGENT_SESSION_RESUME_MARKER_TTL_MS +} diff --git a/src/shared/default-global-settings.ts b/src/shared/default-global-settings.ts index c11d05dc753..e7d372f93da 100644 --- a/src/shared/default-global-settings.ts +++ b/src/shared/default-global-settings.ts @@ -129,6 +129,7 @@ export function buildDefaultSettings(args: { openAgentTabsInChatByDefault: false, experimentalNativeChat: false, experimentalStructuredNativeChat: false, + nativeChatResumeWorkOnRestart: false, nativeChatSessionOptions: {}, openInApplications: [...DEFAULT_OPEN_IN_APPLICATIONS], rightSidebarOpenByDefault: true, diff --git a/src/shared/global-settings-types.ts b/src/shared/global-settings-types.ts index 6369033c892..9fcb2bdea51 100644 --- a/src/shared/global-settings-types.ts +++ b/src/shared/global-settings-types.ts @@ -217,6 +217,9 @@ export type GlobalSettings = { experimentalNativeChat?: boolean /** Opt-in updated structured runtime; off keeps the existing PTY-backed native chat path. */ experimentalStructuredNativeChat?: boolean + /** Opt-in: resume working structured chats automatically on the next launch. Off still offers + * the list, so the user sees exactly what would run before anything spends tokens. */ + nativeChatResumeWorkOnRestart?: boolean /** Last explicit native-chat model + option selections; live panes need an applied/dispatched record before showing a value. */ nativeChatSessionOptions?: PersistedNativeChatSessionOptions /** Extra launcher rows for the worktree "Open in" submenu. VS Code is always shown first. */ diff --git a/src/shared/rpc-contract/rpc-params-catalog.generated.ts b/src/shared/rpc-contract/rpc-params-catalog.generated.ts index deae9bc0eaa..3b62d74027c 100644 --- a/src/shared/rpc-contract/rpc-params-catalog.generated.ts +++ b/src/shared/rpc-contract/rpc-params-catalog.generated.ts @@ -470,6 +470,8 @@ import { HoldParams, OptionsParams, RespondParams, + RestartResumableParams, + RestartResumeParams, RewindParams, SendParams, SetOptionParams, @@ -570,6 +572,9 @@ export const RPC_PARAMS_BY_METHOD = { 'agentSession.requestHandoff': HandoffParams, 'agentSession.respondToApproval': RespondParams, 'agentSession.respondToQuestion': RespondParams, + 'agentSession.restartResumable': RestartResumableParams, + 'agentSession.restartResumableDismiss': RestartResumableParams, + 'agentSession.restartResume': RestartResumeParams, 'agentSession.reveal': OptionsParams, 'agentSession.rewind': RewindParams, 'agentSession.send': SendParams, diff --git a/src/shared/rpc-contract/structured-agent-session-params.ts b/src/shared/rpc-contract/structured-agent-session-params.ts index edbda12d6d9..8afca124482 100644 --- a/src/shared/rpc-contract/structured-agent-session-params.ts +++ b/src/shared/rpc-contract/structured-agent-session-params.ts @@ -18,6 +18,9 @@ export const MAX_BLOCKS = 64 export const MAX_OPTION_LABEL = 512 +/** One relaunch cannot offer more chats than a profile plausibly holds. */ +export const MAX_RESTART_RESUME_SESSIONS = 512 + export const SessionId = z .string() .max(MAX_ID_LENGTH) @@ -227,6 +230,16 @@ export const HoldParams = z .object({ sessionId: SessionId, holderId: Identifier('Invalid holder id') }) .strict() +/** A launch's offer to resume what the last teardown recorded as working. No arguments: the set is + * the host's to derive, never a client's to assert. */ +export const RestartResumableParams = z.object({}).strict() + +/** Omitting `sessionIds` takes the whole offered set; naming them takes that subset. Either way the + * host re-derives eligibility, so an id a client invents is simply not in the set. */ +export const RestartResumeParams = z + .object({ sessionIds: z.array(SessionId).max(MAX_RESTART_RESUME_SESSIONS).optional() }) + .strict() + export const HistoryParams = z .object({ sessionId: SessionId, diff --git a/src/shared/structured-agent-session-live-turn.ts b/src/shared/structured-agent-session-live-turn.ts index 319a6d7d579..041b216c7e1 100644 --- a/src/shared/structured-agent-session-live-turn.ts +++ b/src/shared/structured-agent-session-live-turn.ts @@ -21,6 +21,21 @@ export function activeStructuredAgentSessionTurnId( return null } +/** The newest turn's id whatever state it ended in. Restart resume compares this against the + * teardown marker, and by then eviction has already settled that turn to `interrupted` — so the + * running-only reader above would answer null for exactly the sessions this has to identify. */ +export function newestStructuredAgentSessionTurnId( + items: readonly AgentJournalRenderItem[] +): string | null { + for (let index = items.length - 1; index >= 0; index -= 1) { + const turn = readAgentJournalTurn(items[index]?.body) + if (turn) { + return turn.turnId + } + } + return null +} + /** * Whether the newest thing the active turn produced is the model's own reasoning. * diff --git a/src/shared/structured-agent-session-projection.ts b/src/shared/structured-agent-session-projection.ts index 68e8caf22eb..f82541aa684 100644 --- a/src/shared/structured-agent-session-projection.ts +++ b/src/shared/structured-agent-session-projection.ts @@ -20,7 +20,8 @@ import { sha256 } from './sha256' // Re-exported so the live-turn readers' existing consumers keep one import site. export { activeStructuredAgentSessionToolCall, - activeStructuredAgentSessionTurnId + activeStructuredAgentSessionTurnId, + newestStructuredAgentSessionTurnId } from './structured-agent-session-live-turn' function boundedText(payload: { head: string; truncated: boolean; byteLength: number }): string {