diff --git a/src/main/claude/claude-history-window-source-budget.test.ts b/src/main/claude/claude-history-window-source-budget.test.ts index 998589297e5..90d74b94e37 100644 --- a/src/main/claude/claude-history-window-source-budget.test.ts +++ b/src/main/claude/claude-history-window-source-budget.test.ts @@ -210,17 +210,18 @@ describe('Claude provider history source budget', () => { }) it('replays a concurrent repair at the grown size, not the pinned one', async () => { - const tail = `${JSON.stringify({ + const grown = `${JSON.stringify({ type: 'user', uuid: 'grown', parentUuid: 'latest', sessionId: 'provider', message: { role: 'user', content: 'appended' } - })}\n${JSON.stringify({ type: 'last-prompt', sessionId: 'provider', leafUuid: 'grown' })}\n` - // No marker yet: the proof's first attempt fails, and the retry is what sees - // both the repair AND the record the window has to report. - await writeFile(state.path, SOURCE.slice(0, SOURCE.lastIndexOf('{"type":"last-prompt"'))) - state.growth = tail + })}\n` + const torn = Math.floor(grown.length / 2) + // A torn last record: the proof's first attempt fails, and the retry is what + // sees both the repair AND the record the window has to report. + await writeFile(state.path, SOURCE + grown.slice(0, torn)) + state.growth = grown.slice(torn) const result = await read() diff --git a/src/main/claude/claude-structured-acquisition-launch.ts b/src/main/claude/claude-structured-acquisition-launch.ts index 5e2f4de9b70..7b7f3d3203b 100644 --- a/src/main/claude/claude-structured-acquisition-launch.ts +++ b/src/main/claude/claude-structured-acquisition-launch.ts @@ -4,7 +4,6 @@ import { type StructuredAgentSessionAcquireInput } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import { withAgentSessionCreatePhase } from '../observability/agent-session-instrumentation' -import type { ClaudeRewindAttempt } from './claude-structured-rewind' import type { ClaudeStructuredLaunch } from './claude-structured-launch-resolution' import { cancelClaudeAcquisitionAttempt, @@ -29,9 +28,8 @@ export async function resolveClaudeAcquisitionLaunch(args: { callbacks: ClaudeAcquireCallbacks previous: ClaudeAcquisitionAttempt | undefined attempt: ClaudeAcquisitionAttempt - rewind: ClaudeRewindAttempt }): Promise { - const { input, deps, sessions, acquisitions, exits, callbacks, previous, attempt, rewind } = args + const { input, deps, sessions, acquisitions, exits, callbacks, previous, attempt } = args const sessionId = input.identity.sessionId return withAgentSessionCreatePhase('auth_settle', input.recordPhase, async () => { if (previous && !(await cancelClaudeAcquisitionAttempt(previous))) { @@ -65,7 +63,7 @@ export async function resolveClaudeAcquisitionLaunch(args: { providerHandle: { kind: 'claude' as const, sessionId: resumeSession.providerSessionId, - leafUuid: resumeSession.leafUuid + leafUuid: resumeSession.turnEndLeafUuid } } : input.identity @@ -76,7 +74,6 @@ export async function resolveClaudeAcquisitionLaunch(args: { ? error : new AgentSessionPreSpawnError(error) }) - rewind.applyLaunch(launch, deps) acquisitions.assertCurrent(sessionId, attempt) return launch }) diff --git a/src/main/claude/claude-structured-dispatch-test-support.ts b/src/main/claude/claude-structured-dispatch-test-support.ts index 83971a38d13..605c9072805 100644 --- a/src/main/claude/claude-structured-dispatch-test-support.ts +++ b/src/main/claude/claude-structured-dispatch-test-support.ts @@ -9,8 +9,8 @@ export function sessionFor(send: Mock = vi.fn().mockResolvedValue(undefined)): C return { connection: { send } as unknown as ClaudeSession['connection'], providerSessionId: 'provider-session', - claudeConfigDir: '/accounts/claude', leafUuid: null, + turnEndLeafUuid: null, fence: 1, acquisitionGeneration: 'generation-1', prompts: {} as ClaudeSession['prompts'], diff --git a/src/main/claude/claude-structured-history-window.test.ts b/src/main/claude/claude-structured-history-window.test.ts index b8e487a1481..851831b86f1 100644 --- a/src/main/claude/claude-structured-history-window.test.ts +++ b/src/main/claude/claude-structured-history-window.test.ts @@ -8,6 +8,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { structuredAgentSessionSendBody } from '../../shared/structured-agent-session-outbox' import { structuredAgentSessionPayloadFingerprint } from '../../shared/structured-agent-session-mutation' import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' +import { reconcileSubmissions } from '../native-chat/agent-session-journal/journal-submission-reconciler' import { claudeProviderHistoryWindowFromJsonl, resolveClaudeProviderHistoryWindow @@ -281,3 +282,70 @@ describe('claudeProviderHistoryWindowFromJsonl', () => { expect(read(contents, 'anchor', true).turnInFlight).toBe(true) }) }) + +describe('a crash between Claude saving a prompt and Orca recording its echo', () => { + const row = (type: string, uuid: string, parentUuid: string | null, extra: Row = {}): Row => ({ + type, + uuid, + parentUuid, + isSidechain: false, + sessionId: PROVIDER_SESSION, + ...extra + }) + const marker = (leafUuid: string): Row => ({ + type: 'last-prompt', + sessionId: PROVIDER_SESSION, + leafUuid + }) + const side = (type: string): Row => ({ type, sessionId: PROVIDER_SESSION }) + // Shaped like a real 2.1.280 transcript: after a turn, Claude's marker names its stop-hook + // summary, and a crash mid-turn leaves the next prompt after the marker with no newer marker. + const CRASHED_MID_TURN = [ + side('queue-operation'), + row('attachment', 'hook-start', null, { attachment: { type: 'hook_success' } }), + prompt('alpha', 'hook-start', [{ type: 'text', text: 'ALPHA' }]), + row('attachment', 'alpha-context', 'alpha', { attachment: { type: 'date' } }), + marker('alpha-context'), + side('ai-title'), + row('assistant', 'alpha-reply', 'alpha-context', { + message: { role: 'assistant', content: [{ type: 'text', text: 'ALPHA' }] } + }), + row('attachment', 'alpha-hook', 'alpha-reply', { attachment: { type: 'hook_success' } }), + row('system', 'alpha-stop-summary', 'alpha-hook', { subtype: 'stop_hook_summary' }), + marker('alpha-stop-summary'), + side('queue-operation'), + prompt('bravo', 'alpha-stop-summary', [{ type: 'text', text: 'BRAVO' }]) + ] + .map((entry) => JSON.stringify(entry)) + .join('\n') + + it('reconciles the prompt Claude already holds as accepted, not undelivered', () => { + // The durable anchor is Orca's last completed turn: the reply it saw on the live stream. + const window = read(`${CRASHED_MID_TURN}\n`, 'alpha-reply') + expect(window).toMatchObject({ boundaryConsistent: true }) + expect(window.items.map((item) => item.providerItemId)).toEqual(['bravo']) + + const [verdict] = reconcileSubmissions({ + history: window, + submissions: [ + { + clientMessageId: 'bravo-send', + fence: 1, + payloadFingerprint: sendFingerprint('BRAVO'), + dispatchState: 'unknown', + providerItemId: null, + reason: null, + submittedAt: 0, + resolvedAt: null + } + ] + }) + expect(verdict).toMatchObject({ clientMessageId: 'bravo-send', outcome: 'accepted' }) + }) + + it('ends the conversation at the last main-chain row, never a trailing sidechain row', () => { + const subagent = row('assistant', 'subagent-reply', null, { isSidechain: true }) + const window = read(`${CRASHED_MID_TURN}\n${JSON.stringify(subagent)}\n`, 'alpha-reply') + expect(window.items.map((item) => item.providerItemId)).toEqual(['bravo']) + }) +}) diff --git a/src/main/claude/claude-structured-history-window.ts b/src/main/claude/claude-structured-history-window.ts index c82f1a5feb0..00ae8692025 100644 --- a/src/main/claude/claude-structured-history-window.ts +++ b/src/main/claude/claude-structured-history-window.ts @@ -5,7 +5,9 @@ // conversation Orca is about to resume. Absence here is not an inference about a // dead child — it is the content of the next turn's context. // -// The window is anchored on the leaf uuid Orca durably recorded for the session. +// The window is anchored on the leaf uuid Orca durably recorded for the session +// and walks back to it from the file's last transcript row, which is where a +// resume by session id continues; Claude's marker lags a crash mid-turn. // Without that anchor the read has no proven start, and the branch proof is what // decides whether the file we just read still descends from it: a fork, a // compaction, a sibling branch, or a torn tail all fail the proof, and every one diff --git a/src/main/claude/claude-structured-launch-resolution.test.ts b/src/main/claude/claude-structured-launch-resolution.test.ts index 33ef5f8705c..ac5a68445da 100644 --- a/src/main/claude/claude-structured-launch-resolution.test.ts +++ b/src/main/claude/claude-structured-launch-resolution.test.ts @@ -130,7 +130,7 @@ describe('claude structured launch resolution', () => { expect(first.env).toMatchObject({ [CLAUDE_SESSION_STATE_EVENTS_ENV]: '1' }) }) - it('resumes the session and leaf at the durable chain head', async () => { + it('resumes the durable chain head by session id and carries its leaf as bookkeeping', async () => { const launch = await resolverFor( record({ providerHandleChain: [ @@ -152,7 +152,8 @@ describe('claude structured launch resolution', () => { resumed: true }) expect(launch.options.resume).toBe('provider-current') - expect(launch.options.resumeSessionAt).toBe('leaf-current') + // Claude owns where the conversation continues; a stored leaf would cut or branch it. + expect(launch.options).not.toHaveProperty('resumeSessionAt') expect(launch.options.sessionId).toBeUndefined() }) @@ -164,24 +165,22 @@ describe('claude structured launch resolution', () => { expect(launch.env).toMatchObject({ [CLAUDE_SESSION_STATE_EVENTS_ENV]: '1' }) }) - it('refuses a durable journal leaf that diverged before resume resolution', async () => { - const resolve = resolverFor( - record({ - providerHandleChain: [ - { - handle: { - provider: 'claude', - sessionId: 'provider-current', - leafUuid: 'leaf-current' - } - } - ] as AgentSessionRecord['providerHandleChain'] - }) - ) + it('launches when only the bookkeeping leaf moved, and refuses a changed session', async () => { + const resolve = resolverFor(RESUMABLE) - await expect(resolve({ identity: identityAt('leaf-stale') })).rejects.toThrow( - 'durable resume identity changed before spawn' - ) + // A failed turn-end or exit write leaves the identity's leaf behind the record's. + await expect(resolve({ identity: identityAt('leaf-stale') })).resolves.toMatchObject({ + providerSessionId: 'provider-current', + resumeLeafUuid: 'leaf-current' + }) + await expect( + resolve({ + identity: { + ...IDENTITY, + providerHandle: { kind: 'claude', sessionId: 'provider-other', leafUuid: 'leaf-current' } + } + }) + ).rejects.toThrow('durable resume identity changed before spawn') }) it('keeps session-only resume when the durable handle has no leaf', async () => { @@ -200,7 +199,7 @@ describe('claude structured launch resolution', () => { )({ identity: identityAt(null) }) expect(launch.options.resume).toBe('provider-current') - expect(launch.options.resumeSessionAt).toBeUndefined() + expect(launch.options).not.toHaveProperty('resumeSessionAt') }) // Agent Permissions is stored as the bypass flag inside the launch arguments, so presence of diff --git a/src/main/claude/claude-structured-launch-resolution.ts b/src/main/claude/claude-structured-launch-resolution.ts index 3dab0c23a2d..56516011775 100644 --- a/src/main/claude/claude-structured-launch-resolution.ts +++ b/src/main/claude/claude-structured-launch-resolution.ts @@ -44,8 +44,6 @@ export type ClaudeStructuredSdkOptions = Pick< | 'allowDangerouslySkipPermissions' | 'sessionId' | 'resume' - | 'resumeSessionAt' - | 'resumeDropsTurn' > /** @@ -95,6 +93,7 @@ export type ClaudeStructuredLaunch = { env?: Record claudeConfigDir: string providerSessionId: string + /** The previous head leaf, carried into the publication link; never a resume argument. */ resumeLeafUuid: string | null resumed: boolean } @@ -186,8 +185,7 @@ export function createClaudeStructuredLaunchResolver( if ( head?.handle.provider === 'claude' && (identity.providerHandle.kind !== 'claude' || - identity.providerHandle.sessionId !== head.handle.sessionId || - identity.providerHandle.leafUuid !== head.handle.leafUuid) + identity.providerHandle.sessionId !== head.handle.sessionId) ) { throw new Error('claude durable resume identity changed before spawn') } @@ -244,11 +242,9 @@ export function createClaudeStructuredLaunchResolver( ...CLAUDE_STRUCTURED_BASE_OPTIONS, ...permission, extraArgs: { ...CLAUDE_STRUCTURED_BASE_OPTIONS.extraArgs, ...permission.extraArgs }, + // Claude owns where a resumed conversation continues; the stored leaf is Orca's bookkeeping. ...(head?.handle.provider === 'claude' - ? { - resume: providerSessionId, - ...(head.handle.leafUuid === null ? {} : { resumeSessionAt: head.handle.leafUuid }) - } + ? { resume: providerSessionId } : { sessionId: providerSessionId }) }, cwd: await deps.resolveWorkspacePath(record.location.workspaceId), diff --git a/src/main/claude/claude-structured-options.test.ts b/src/main/claude/claude-structured-options.test.ts index 9b962fc33d7..f364db6e80e 100644 --- a/src/main/claude/claude-structured-options.test.ts +++ b/src/main/claude/claude-structured-options.test.ts @@ -20,8 +20,8 @@ function sessionFor(setModel: ClaudeSession['connection']['setModel']): ClaudeSe supportedModels: async (): Promise => [] } as ClaudeSession['connection'], providerSessionId: 'provider-session', - claudeConfigDir: '/accounts/claude', leafUuid: null, + turnEndLeafUuid: null, fence: 1, acquisitionGeneration: 'generation-1', prompts: {} as ClaudeSession['prompts'], diff --git a/src/main/claude/claude-structured-resume-point.test.ts b/src/main/claude/claude-structured-resume-point.test.ts new file mode 100644 index 00000000000..a11ffc15450 --- /dev/null +++ b/src/main/claude/claude-structured-resume-point.test.ts @@ -0,0 +1,127 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { ClaudeStructuredSessionEvent } from './claude-structured-session-adapter' +import { + adapterFor, + fakeClaude, + identityFor, + PROVIDER_SESSION_ID, + recordingJournalSink, + tick +} from './claude-structured-session-test-support' + +/** A resumed owner that completed one turn (`a4`) and then saw the next turn's prompt. */ +async function ownerMidSecondTurn(persisted: unknown[]) { + const claude = fakeClaude() + const adapter = adapterFor( + claude, + { resumed: true, resumeLeafUuid: 'a3', options: { resume: PROVIDER_SESSION_ID } }, + [], + persisted + ) + await adapter.acquire({ + identity: identityFor(), + fence: 7, + spawnToken: 'spawn-7', + events: recordingJournalSink() + }) + const frame = (message: Record) => + claude.connections[0]!.handlers.onMessage?.({ session_id: PROVIDER_SESSION_ID, ...message }) + frame({ type: 'user', uuid: 'u4' }) + frame({ type: 'assistant', uuid: 'a4' }) + frame({ type: 'result', subtype: 'success', uuid: 'a4-result' }) + frame({ type: 'user', uuid: 'u5' }) + return { adapter, claude, frame } +} + +const lastCompletedTurn = { providerSessionId: PROVIDER_SESSION_ID, leafUuid: 'a4', fence: 7 } + +afterEach(() => { + vi.restoreAllMocks() +}) + +describe('Claude resume point is the last completed turn on every exit path', () => { + it('on close', async () => { + const persisted: unknown[] = [] + const { adapter } = await ownerMidSecondTurn(persisted) + await expect(adapter.closeSession('session-1')).resolves.toBe(true) + expect(persisted).toEqual([expect.objectContaining(lastCompletedTurn)]) + }) + + it('on an unexpected exit', async () => { + const persisted: unknown[] = [] + const { adapter, claude } = await ownerMidSecondTurn(persisted) + claude.connections[0]!.handlers.onExit?.(new Error('claude crashed')) + await adapter.drainObservedExits() + await tick() + expect(persisted).toEqual([expect.objectContaining(lastCompletedTurn)]) + }) + + it('on an acquisition release', async () => { + const persisted: unknown[] = [] + const { adapter } = await ownerMidSecondTurn(persisted) + await expect(adapter.releaseAcquisition({ sessionId: 'session-1' })).resolves.toBe(true) + expect(persisted).toEqual([expect.objectContaining(lastCompletedTurn)]) + }) + + it('carries the launch leaf forward when no turn completed', async () => { + const persisted: unknown[] = [] + const claude = fakeClaude() + const adapter = adapterFor( + claude, + { resumed: true, resumeLeafUuid: 'a3', options: { resume: PROVIDER_SESSION_ID } }, + [], + persisted + ) + const acquisition = await adapter.acquire({ + identity: identityFor(), + fence: 7, + spawnToken: 'spawn-7', + events: recordingJournalSink() + }) + expect(acquisition.link.handle).toMatchObject({ leafUuid: 'a3' }) + await expect(adapter.closeSession('session-1')).resolves.toBe(true) + expect(persisted).toEqual([expect.objectContaining({ leafUuid: 'a3' })]) + }) + + it('saves only real messages: hook, attachment, and result frames never become the leaf', async () => { + const persisted: unknown[] = [] + const { adapter, frame } = await ownerMidSecondTurn(persisted) + frame({ type: 'assistant', uuid: 'a5' }) + // Claude's transcript chains these after a reply; the live stream never adopts them. + frame({ type: 'system', subtype: 'stop_hook_summary', uuid: 'a5-hook-summary' }) + frame({ type: 'attachment', uuid: 'a5-attachment' }) + frame({ type: 'result', subtype: 'success', uuid: 'a5-result' }) + await expect(adapter.closeSession('session-1')).resolves.toBe(true) + expect(persisted).toEqual([expect.objectContaining({ leafUuid: 'a5' })]) + }) + + it('on an unexpected exit whose durable write fails, still ends the session and logs', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const events: ClaudeStructuredSessionEvent[] = [] + const claude = fakeClaude() + const adapter = adapterFor( + claude, + { resumed: true, resumeLeafUuid: 'a3', options: { resume: PROVIDER_SESSION_ID } }, + events, + [], + undefined, + async () => { + throw new Error('record write failed') + } + ) + await adapter.acquire({ + identity: identityFor(), + fence: 7, + spawnToken: 'spawn-7', + events: recordingJournalSink() + }) + claude.connections[0]!.handlers.onExit?.(new Error('claude crashed')) + await adapter.drainObservedExits() + await tick() + expect(events.at(-1)).toMatchObject({ type: 'ended', cause: 'unexpected-exit' }) + expect(warn).toHaveBeenCalledWith( + '[claude-resume-point] exit cursor was not persisted:', + expect.objectContaining({ sessionId: 'session-1', error: expect.any(Error) }) + ) + }) +}) diff --git a/src/main/claude/claude-structured-resume-point.ts b/src/main/claude/claude-structured-resume-point.ts new file mode 100644 index 00000000000..e8124f52d27 --- /dev/null +++ b/src/main/claude/claude-structured-resume-point.ts @@ -0,0 +1,56 @@ +import type { + ClaudeSession, + ClaudeStructuredSessionAdapterDeps +} from './claude-structured-session-state' + +/** + * Record a completed turn: its leaf becomes the one close and exit persist, and the durable point + * advances in place so an owner that dies before its close path runs keeps it. Writes run one at a + * time, and a failure is only logged: this is bookkeeping and must never fail the turn. + */ +export function persistClaudeTurnResumePoint( + sessionId: string, + session: ClaudeSession, + deps: Pick +): void { + if (session.closeFinalization || session.closeFinalized) { + return + } + session.turnEndLeafUuid = session.leafUuid + const leafUuid = session.turnEndLeafUuid + const persist = deps.persistResumePoint + if (!persist || leafUuid === null || session.resumePointWrite?.leafUuid === leafUuid) { + return + } + const previous = session.resumePointWrite?.settled ?? Promise.resolve() + const write: NonNullable = { + leafUuid, + settled: previous + .then(() => + persist({ + sessionId, + providerSessionId: session.providerSessionId, + leafUuid, + fence: session.fence + }) + ) + .catch((error: unknown) => { + console.warn('[claude-resume-point] turn-end resume point was not persisted:', { + sessionId, + leafUuid, + error + }) + // Forget the failed leaf so the next turn end retries it even when the leaf has not moved. + if (session.resumePointWrite === write) { + session.resumePointWrite = undefined + } + }) + } + session.resumePointWrite = write +} + +/** Close and exit persist the last completed turn, after any in-flight turn-end write settles. */ +export async function settledClaudeTurnEndLeaf(session: ClaudeSession): Promise { + await session.resumePointWrite?.settled + return session.turnEndLeafUuid +} diff --git a/src/main/claude/claude-structured-rewind.test.ts b/src/main/claude/claude-structured-rewind.test.ts deleted file mode 100644 index 5642e6cd088..00000000000 --- a/src/main/claude/claude-structured-rewind.test.ts +++ /dev/null @@ -1,207 +0,0 @@ -import { describe, expect, it, vi } from 'vitest' -import { - adapterFor, - fakeClaude, - identityFor, - PROVIDER_SESSION_ID -} from './claude-structured-session-test-support' -import { ClaudeRewindAttempt } from './claude-structured-rewind' -import { AgentSessionRewindRefusal } from '../native-chat/agent-session-wire/structured-agent-session-adapter' - -const intent = { targetUuid: 'kept', previousLeafUuid: 'tip', dropsTurn: 'drop' } -const proofLaunch = { - providerSessionId: PROVIDER_SESSION_ID, - claudeConfigDir: '/claude', - options: {}, - resumed: true, - resumeLeafUuid: 'tip', - cwd: '/workspace', - pathToClaudeCodeExecutable: 'claude' -} - -describe('Claude rewind acquisition', () => { - it('executes a cursor resume in place and proves the exact target before publication', async () => { - const fake = fakeClaude() - const proof = vi.fn(async (_input: { intentionalRewindUuid?: string }) => 'kept') - const adapter = adapterFor( - fake, - { resumed: true, resumeLeafUuid: 'tip' }, - [], - [], - undefined, - proof - ) - try { - const acquired = await adapter.acquire({ - identity: identityFor(), - fence: 7, - spawnToken: 'spawn', - rewind: intent - }) - expect(acquired.link.handle).toMatchObject({ - provider: 'claude', - sessionId: PROVIDER_SESSION_ID, - leafUuid: 'kept' - }) - expect(fake.connections[0]!.launch.options).toMatchObject({ - resume: PROVIDER_SESSION_ID, - resumeSessionAt: 'kept', - resumeDropsTurn: 'drop' - }) - expect(fake.connections[0]!.launch.options).not.toHaveProperty('forkSession') - expect(proof).toHaveBeenCalledWith( - expect.objectContaining({ previousLeafUuid: 'tip', intentionalRewindUuid: 'kept' }) - ) - await adapter.closeSession('session-1') - await adapter.acquire({ identity: identityFor(), fence: 8, spawnToken: 'spawn-next' }) - expect(fake.connections[1]!.launch.options).not.toHaveProperty('resumeDropsTurn') - expect( - proof.mock.calls.filter(([input]) => input.intentionalRewindUuid !== undefined) - ).toHaveLength(1) - } finally { - await adapter.closeAll() - } - }) - it('recognizes the documented refusal and closes the failed child without retry', async () => { - const fake = fakeClaude() - const openConnection = fake.openConnection - fake.openConnection = async (launch, handlers) => { - const connection = await openConnection(launch, handlers) - const initialize = connection.initializationResult - connection.initializationResult = async (...args) => { - const result = await initialize(...args) - handlers?.onMessage?.({ - type: 'result', - subtype: 'error_during_execution', - session_id: PROVIDER_SESSION_ID, - errors: ['Resume rejected by --resume-drops-turn: additional prompt observed'] - }) - return result - } - return connection - } - const proof = vi.fn(async (_input: { intentionalRewindUuid?: string }) => 'kept') - const adapter = adapterFor(fake, { resumed: true }, [], [], undefined, proof) - await expect( - adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn', rewind: intent }) - ).rejects.toMatchObject({ rewindReason: 'provider-refused' }) - expect(fake.connections).toHaveLength(1) - expect(fake.connections[0]?.closed).toBe(true) - expect(proof).not.toHaveBeenCalled() - await adapter.closeAll() - }) - it('consumes proof authorization even if its first read fails', async () => { - const proof = vi.fn(async () => { - throw new Error('torn transcript') - }) - const attempt = new ClaudeRewindAttempt(intent) - const launch = { - providerSessionId: PROVIDER_SESSION_ID, - claudeConfigDir: '/claude', - options: {}, - resumed: true, - resumeLeafUuid: 'tip', - cwd: '/workspace', - pathToClaudeCodeExecutable: 'claude' - } - await expect(attempt.prove(launch, { readTranscriptLeaf: proof })).rejects.toBeInstanceOf( - AgentSessionRewindRefusal - ) - expect(await attempt.prove(launch, { readTranscriptLeaf: proof })).toBeNull() - expect(proof).toHaveBeenCalledTimes(1) - }) - it('never persists success for a mismatching leaf', async () => { - const onProved = vi.fn(async () => {}) - const attempt = new ClaudeRewindAttempt(intent, onProved) - await expect( - attempt.prove(proofLaunch, { readTranscriptLeaf: async () => 'other' }) - ).rejects.toMatchObject({ rewindReason: 'proof-mismatch' }) - expect(onProved).not.toHaveBeenCalled() - }) - it('preserves commit failure as unknown and consumes the override before persisting', async () => { - const diskError = new Error('record write failed') - const onProved = vi.fn(async () => { - throw diskError - }) - const proof = vi.fn(async () => 'kept') - const attempt = new ClaudeRewindAttempt(intent, onProved) - const launch = { - providerSessionId: PROVIDER_SESSION_ID, - claudeConfigDir: '/claude', - options: {}, - resumed: true, - resumeLeafUuid: 'tip', - cwd: '/workspace', - pathToClaudeCodeExecutable: 'claude' - } - await expect(attempt.prove(launch, { readTranscriptLeaf: proof })).rejects.toBe(diskError) - expect(onProved).toHaveBeenCalledWith('kept') - expect(await attempt.prove(launch, { readTranscriptLeaf: proof })).toBeNull() - expect(proof).toHaveBeenCalledTimes(1) - }) - it('checkpoints the proved target before late acquisition failure without persisting a stale cursor', async () => { - const fake = fakeClaude() - const launch = { resumed: true, resumeLeafUuid: 'tip' } - const persisted: unknown[] = [] - const proof = vi.fn(async () => 'kept') - const adapter = adapterFor(fake, launch, [], persisted, undefined, proof) - const onProved = vi.fn(async (leafUuid: string) => { - launch.resumeLeafUuid = leafUuid - fake.connections[0]!.closed = true - }) - try { - await expect( - adapter.acquire({ - identity: identityFor(), - fence: 7, - spawnToken: 'spawn', - rewind: { ...intent, onProved } - }) - ).rejects.toThrow('exited while being acquired') - expect(onProved).toHaveBeenCalledWith('kept') - expect(persisted).toEqual([]) - const acquired = await adapter.acquire({ - identity: identityFor(), - fence: 8, - spawnToken: 'retry' - }) - expect(acquired.link.handle).toMatchObject({ leafUuid: 'kept' }) - expect(fake.connections[1]!.launch.options).not.toHaveProperty('resumeDropsTurn') - expect(proof).toHaveBeenCalledTimes(1) - } finally { - await adapter.closeAll() - } - }) - it('restores an interrupted unproved rewind only after exact ordinary branch proof', async () => { - const fake = fakeClaude() - const proof = vi.fn(async (_input: { intentionalRewindUuid?: string }) => 'kept') - const restored = vi.fn(async () => {}) - const adapter = adapterFor( - fake, - { resumed: true, resumeLeafUuid: 'tip' }, - [], - [], - undefined, - proof - ) - const input = { - identity: identityFor(), - fence: 7, - spawnToken: 'spawn', - rewindRecovery: { leafUuid: 'tip', onProved: restored } - } - try { - await expect(adapter.acquire(input)).rejects.toMatchObject({ rewindReason: 'proof-mismatch' }) - expect(restored).not.toHaveBeenCalled() - proof.mockResolvedValue('tip') - await adapter.acquire({ ...input, fence: 8, spawnToken: 'retry' }) - expect(restored).toHaveBeenCalledOnce() - expect(proof).toHaveBeenCalledWith(expect.objectContaining({ previousLeafUuid: 'tip' })) - for (const [request] of proof.mock.calls) { - expect(request).not.toHaveProperty('intentionalRewindUuid') - } - } finally { - await adapter.closeAll() - } - }) -}) diff --git a/src/main/claude/claude-structured-rewind.ts b/src/main/claude/claude-structured-rewind.ts deleted file mode 100644 index 79587327a09..00000000000 --- a/src/main/claude/claude-structured-rewind.ts +++ /dev/null @@ -1,118 +0,0 @@ -import { AgentSessionRewindRefusal } from '../native-chat/agent-session-wire/structured-agent-session-adapter' - -export function claudeRewindRefusalFromMessage( - message: Record -): AgentSessionRewindRefusal | null { - return message.type === 'result' && - message.subtype === 'error_during_execution' && - Array.isArray(message.errors) && - message.errors.some( - (error) => - typeof error === 'string' && error.startsWith('Resume rejected by --resume-drops-turn:') - ) - ? new AgentSessionRewindRefusal('provider-refused') - : null -} - -import type { StructuredAgentSessionAcquireInput } from '../native-chat/agent-session-wire/structured-agent-session-adapter' -import type { ClaudeStructuredLaunch } from './claude-structured-launch-resolution' -import type { ClaudeStructuredSessionAdapterDeps } from './claude-structured-session-state' - -type Intent = NonNullable - -/** The proof authorization exists only for this acquisition's first proof attempt. */ -export class ClaudeRewindAttempt { - private refusal: AgentSessionRewindRefusal | null = null - constructor( - private intent: Intent | undefined, - private readonly onProved?: (leafUuid: string) => Promise - ) {} - - observe(message: Record): AgentSessionRewindRefusal | null { - if (!this.intent) { - return null - } - this.refusal ??= claudeRewindRefusalFromMessage(message) - return this.refusal - } - - applyLaunch( - launch: ClaudeStructuredLaunch, - deps: Pick - ): void { - if (!this.intent) { - return - } - if (!launch.resumed || !deps.readTranscriptLeaf) { - throw new AgentSessionRewindRefusal('unsupported') - } - launch.options = { - ...launch.options, - resume: launch.providerSessionId, - resumeSessionAt: this.intent.targetUuid, - ...(this.intent.dropsTurn ? { resumeDropsTurn: this.intent.dropsTurn } : {}) - } - launch.resumeLeafUuid = this.intent.targetUuid - } - - async prove( - launch: ClaudeStructuredLaunch, - deps: Pick - ): Promise { - const intent = this.intent - this.clear() - if (this.refusal) { - throw this.refusal - } - if (!intent) { - return null - } - let leaf: string | null - try { - leaf = await deps.readTranscriptLeaf!({ - providerSessionId: launch.providerSessionId, - previousLeafUuid: intent.previousLeafUuid, - intentionalRewindUuid: intent.targetUuid, - claudeConfigDir: launch.claudeConfigDir - }) - if (leaf !== intent.targetUuid) { - throw new AgentSessionRewindRefusal('proof-mismatch') - } - } catch (error) { - throw error instanceof AgentSessionRewindRefusal - ? error - : new AgentSessionRewindRefusal('proof-mismatch') - } - // Persistence failure is an unknown outcome, never evidence that the provider refused. - await this.onProved?.(leaf) - return leaf - } - - clear(): void { - this.intent = undefined - } -} - -/** An interrupted, unproved rewind restores its original cursor without ancestor authorization. */ -export async function proveClaudeRewindRecovery( - recovery: StructuredAgentSessionAcquireInput['rewindRecovery'], - launch: ClaudeStructuredLaunch, - deps: Pick -): Promise { - if (!recovery) { - return null - } - if (!launch.resumed || launch.resumeLeafUuid !== recovery.leafUuid || !deps.readTranscriptLeaf) { - throw new AgentSessionRewindRefusal('proof-mismatch') - } - const leaf = await deps.readTranscriptLeaf({ - providerSessionId: launch.providerSessionId, - previousLeafUuid: recovery.leafUuid, - claudeConfigDir: launch.claudeConfigDir - }) - if (leaf !== recovery.leafUuid) { - throw new AgentSessionRewindRefusal('proof-mismatch') - } - await recovery.onProved() - return leaf -} diff --git a/src/main/claude/claude-structured-session-acquisition.ts b/src/main/claude/claude-structured-session-acquisition.ts index a10a2dca671..47eaddb49d9 100644 --- a/src/main/claude/claude-structured-session-acquisition.ts +++ b/src/main/claude/claude-structured-session-acquisition.ts @@ -1,4 +1,3 @@ -import { ClaudeRewindAttempt, proveClaudeRewindRecovery } from './claude-structured-rewind' import { AgentSessionPreSpawnError, type AgentSessionAcquisition, @@ -45,6 +44,7 @@ import { } from './claude-structured-session-state' import { resolveClaudeAcquisitionError } from './claude-structured-session-close' import { readClaudeTranscriptEntryUuid } from './claude-tui-exit' +import { persistClaudeTurnResumePoint } from './claude-structured-resume-point' import { withAgentSessionCreatePhase } from '../observability/agent-session-instrumentation' import { resolveClaudeAcquisitionLaunch } from './claude-structured-acquisition-launch' import { @@ -92,7 +92,6 @@ export async function acquireClaudeSession({ createClaudeJournalFailureHandler({ attempt, initDeadline, callbacks, sessionId }) ) - const rewind = new ClaudeRewindAttempt(input.rewind, input.rewind?.onProved) const onMessage = (message: Record): void => { const init = readClaudeInit(message) if (readClaudeFrameString(message, 'session_id') !== expectedProviderSessionId) { @@ -103,11 +102,6 @@ export async function acquireClaudeSession({ } return } - const refusal = rewind.observe(message) - if (refusal) { - initDeadline.reject(refusal) - return - } if (init) { initDeadline.resolve(init) // Every turn opens with an init frame naming the model the CLI is actually @@ -122,6 +116,10 @@ export async function acquireClaudeSession({ if (liveSession) { liveSession.leafUuid = observedLeafUuid observeClaudeFastModeFacts(liveSession, message) + // Recording a turn end is an owner action; a result that trails the child's exit has no owner. + if (message.type === 'result' && sessions.get(sessionId) === liveSession) { + persistClaudeTurnResumePoint(sessionId, liveSession, deps) + } } const turnOrigin = liveSession ? resolveClaudeReplayTurn(liveSession, message, (settlement) => @@ -161,8 +159,7 @@ export async function acquireClaudeSession({ exits, callbacks, previous, - attempt, - rewind + attempt }) expectedProviderSessionId = launch.providerSessionId observedLeafUuid = launch.resumeLeafUuid @@ -241,9 +238,6 @@ export async function acquireClaudeSession({ diagnostic: claudeAuthDiagnostic(init, settings) }) ) - observedLeafUuid = (await rewind.prove(launch, deps)) ?? observedLeafUuid - observedLeafUuid = - (await proveClaudeRewindRecovery(input.rewindRecovery, launch, deps)) ?? observedLeafUuid const process = await claudeProcessIdentity( { ...input, pid: connection.pid }, deps.readProcessStartTime @@ -257,8 +251,8 @@ export async function acquireClaudeSession({ connection, init, initialization, - claudeConfigDir: launch.claudeConfigDir, leafUuid: observedLeafUuid, + turnEndLeafUuid: launch.resumeLeafUuid, fence: input.fence, effort: readClaudeSettingsEffort(settings), ...claudeStructuredSessionPublicationOptions(acquisitionOptions), @@ -304,7 +298,6 @@ export async function acquireClaudeSession({ acquisitions.deleteIfCurrent(sessionId, attempt) throw acquisitionError } finally { - rewind.clear() attempt.finish() } } diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index bfb34fa972a..738d9465d3c 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -60,8 +60,11 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda supportsLocation = supportsClaudeStructuredLocation - rewindSupport: NonNullable = () => - this.deps.readTranscriptLeaf ? { supported: true } : { supported: false, reason: 'unsupported' } + // Orca's marker-based rewind proof can never pass on the real binary; rewind returns via a fork. + rewindSupport: NonNullable = () => ({ + supported: false, + reason: 'unsupported' + }) acquire = (input: StructuredAgentSessionAcquireInput): Promise => acquireClaudeSession({ @@ -133,9 +136,14 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda settleClaudeExitedSession(exit.session) return } - // Persist the transcript-derived cursor before publishing the lifecycle + // Persist the last completed turn before publishing the lifecycle // event that lets the host release and reacquire this exact child. - await persistClaudeSessionHandle(sessionId, exit.session, this.deps).catch(() => undefined) + await persistClaudeSessionHandle(sessionId, exit.session, this.deps).catch( + (error: unknown) => { + // Recovery still publishes: the record keeps its last durable point, and the loss is logged. + console.warn('[claude-resume-point] exit cursor was not persisted:', { sessionId, error }) + } + ) if (this.exits.get(sessionId) !== exit) { settleClaudeExitedSession(exit.session) return @@ -280,7 +288,6 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda sessions: this.sessions, acquisitions: this.acquisitions, ...(this.deps.persistHandle ? { persistHandle: this.deps.persistHandle } : {}), - ...(this.deps.readTranscriptLeaf ? { readTranscriptLeaf: this.deps.readTranscriptLeaf } : {}), ...(this.deps.onBackgroundTasksChanged ? { onBackgroundTasksChanged: this.deps.onBackgroundTasksChanged } : {}), diff --git a/src/main/claude/claude-structured-session-close.test.ts b/src/main/claude/claude-structured-session-close.test.ts index 46bda78cacd..f90af6dfbad 100644 --- a/src/main/claude/claude-structured-session-close.test.ts +++ b/src/main/claude/claude-structured-session-close.test.ts @@ -62,7 +62,6 @@ describe('Claude published session close lifecycle', () => { events, [], undefined, - undefined, persistHandle, (_sessionId, state) => backgroundStates.push(state) ) diff --git a/src/main/claude/claude-structured-session-close.ts b/src/main/claude/claude-structured-session-close.ts index f6a58c36357..0ce448b4e9a 100644 --- a/src/main/claude/claude-structured-session-close.ts +++ b/src/main/claude/claude-structured-session-close.ts @@ -17,7 +17,7 @@ import type { ClaudePromptRegistry } from './claude-structured-prompt-replies' import type { AgentSessionBackgroundTaskState } from '../../shared/agent-session-wire' import { closeProcessRegistry } from '../../shared/child-process/close-process-registry' import { retireClaudeDispatchWaiters } from './claude-structured-dispatch' -import { readClaudeTranscriptLeafWithReproof } from './claude-transcript-branch-proof' +import { settledClaudeTurnEndLeaf } from './claude-structured-resume-point' export function claudeAcquisitionCleanupError( connection: ClaudeStreamJsonConnection | null | undefined, @@ -80,11 +80,6 @@ type CloseClaudePublishedSessionInput = { sessionId: string, state: AgentSessionBackgroundTaskState | null ) => void - readTranscriptLeaf?: (input: { - providerSessionId: string - previousLeafUuid: string | null - claudeConfigDir: string - }) => Promise } async function finalizeClaudePublishedSession( @@ -113,29 +108,14 @@ async function finalizeClaudePublishedSession( if (session.backgroundTasks.clear()) { input.onBackgroundTasksChanged?.(input.sessionId, null) } - try { - const transcriptLeaf = input.readTranscriptLeaf - ? await readClaudeTranscriptLeafWithReproof({ - readTranscriptLeaf: input.readTranscriptLeaf, - providerSessionId: session.providerSessionId, - previousLeafUuid: session.leafUuid, - claudeConfigDir: session.claudeConfigDir - }) - : null - if (transcriptLeaf) { - session.leafUuid = transcriptLeaf - } - } catch { - // Keep the last observed main-transcript frame when the durable tail is - // unavailable or proves a stale/divergent branch. - } + const leafUuid = await settledClaudeTurnEndLeaf(session) const persistence = session.closePersistence ?? (session.closePersistence = (async () => { await input.persistHandle?.({ sessionId: input.sessionId, providerSessionId: session.providerSessionId, - leafUuid: session.leafUuid, + leafUuid, fence: session.fence }) })()) @@ -164,7 +144,7 @@ async function finalizeClaudePublishedSession( type: 'handle', sessionId: input.sessionId, providerSessionId: session.providerSessionId, - leafUuid: session.leafUuid, + leafUuid, fence: session.fence }) } catch (error) { @@ -240,11 +220,6 @@ export function closeClaudePublishedSessionForDeps( sessionId: string, state: AgentSessionBackgroundTaskState | null ) => void - readTranscriptLeaf?: (input: { - providerSessionId: string - previousLeafUuid: string | null - claudeConfigDir: string - }) => Promise } ): Promise { return closeClaudePublishedSession({ sessions, sessionId, ...deps }) @@ -265,11 +240,6 @@ export async function closeClaudeSession(input: { sessionId: string, state: AgentSessionBackgroundTaskState | null ) => void - readTranscriptLeaf?: (input: { - providerSessionId: string - previousLeafUuid: string | null - claudeConfigDir: string - }) => Promise }): Promise { const attempt = input.acquisitions.get(input.sessionId) if (!(await cancelClaudeAcquisitionAttempt(attempt))) { diff --git a/src/main/claude/claude-structured-session-exit-lifecycle.ts b/src/main/claude/claude-structured-session-exit-lifecycle.ts index 20cbc1f3125..f8a4f6bf8e6 100644 --- a/src/main/claude/claude-structured-session-exit-lifecycle.ts +++ b/src/main/claude/claude-structured-session-exit-lifecycle.ts @@ -1,4 +1,4 @@ -import { readClaudeTranscriptLeafWithReproof } from './claude-transcript-branch-proof' +import { settledClaudeTurnEndLeaf } from './claude-structured-resume-point' import type { ClaudeSession, ClaudeSessionExit, @@ -30,27 +30,13 @@ export async function drainClaudeObservedExits( export async function persistClaudeSessionHandle( sessionId: string, session: ClaudeSession, - deps: Pick + deps: Pick ): Promise { - try { - const transcriptLeaf = deps.readTranscriptLeaf - ? await readClaudeTranscriptLeafWithReproof({ - readTranscriptLeaf: deps.readTranscriptLeaf, - providerSessionId: session.providerSessionId, - previousLeafUuid: session.leafUuid, - claudeConfigDir: session.claudeConfigDir - }) - : null - if (transcriptLeaf) { - session.leafUuid = transcriptLeaf - } - } catch { - // An unavailable tail must not overwrite the last observed leaf. - } + const leafUuid = await settledClaudeTurnEndLeaf(session) await deps.persistHandle?.({ sessionId, providerSessionId: session.providerSessionId, - leafUuid: session.leafUuid, + leafUuid, fence: session.fence }) } diff --git a/src/main/claude/claude-structured-session-publication.ts b/src/main/claude/claude-structured-session-publication.ts index 2434f522661..bca70c9db76 100644 --- a/src/main/claude/claude-structured-session-publication.ts +++ b/src/main/claude/claude-structured-session-publication.ts @@ -11,8 +11,9 @@ export function createClaudeSessionPublication(input: { connection: ClaudeSession['connection'] init: ClaudeInitObservation initialization?: unknown - claudeConfigDir: string leafUuid: string | null + /** The launch's stored leaf: a frame seen before publication is not a completed turn. */ + turnEndLeafUuid: string | null fence: number acquisitionGeneration: string resumed: boolean @@ -51,8 +52,8 @@ export function createClaudeSessionPublication(input: { session: { connection: input.connection, providerSessionId: input.init.providerSessionId, - claudeConfigDir: input.claudeConfigDir, leafUuid: input.leafUuid, + turnEndLeafUuid: input.turnEndLeafUuid, fence: input.fence, acquisitionGeneration: input.acquisitionGeneration, prompts: input.prompts, diff --git a/src/main/claude/claude-structured-session-recovery.test.ts b/src/main/claude/claude-structured-session-recovery.test.ts index bd8f8cb40b2..098b796e485 100644 --- a/src/main/claude/claude-structured-session-recovery.test.ts +++ b/src/main/claude/claude-structured-session-recovery.test.ts @@ -5,7 +5,6 @@ import { type ClaudeStructuredSessionAdapterDeps, type ClaudeStructuredSessionEvent } from './claude-structured-session-adapter' -import { ClaudeTranscriptPreviousCursorMissingError } from './claude-transcript-branch-proof' import { adapterFor, fakeClaude, @@ -15,13 +14,13 @@ import { tick } from './claude-structured-session-test-support' -describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { +describe('ClaudeStructuredSessionAdapter close and exit recovery', () => { it('shares concurrent close finalization and emits lifecycle once', async () => { const claude = fakeClaude() const events: ClaudeStructuredSessionEvent[] = [] const persistence = Promise.withResolvers() const persistHandle = vi.fn(() => persistence.promise) - const adapter = adapterFor(claude, {}, events, [], undefined, undefined, persistHandle) + const adapter = adapterFor(claude, {}, events, [], undefined, persistHandle) const journalSink: StructuredAgentSessionEventSink = { appendItem: () => {}, appendTombstone: () => {}, @@ -108,7 +107,7 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { .fn>() .mockRejectedValueOnce(persistenceError) .mockResolvedValueOnce(undefined) - const adapter = adapterFor(claude, {}, [], [], undefined, undefined, persistHandle) + const adapter = adapterFor(claude, {}, [], [], undefined, persistHandle) await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) await expect(adapter.closeSession('session-1')).rejects.toBe(persistenceError) @@ -117,7 +116,7 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { expect(persistHandle).toHaveBeenCalledTimes(2) }) - it('persists only the last transcript-entry uuid before graceful close', async () => { + it('persists the last completed turn message before graceful close', async () => { const claude = fakeClaude() const events: ClaudeStructuredSessionEvent[] = [] const persistedHandles: unknown[] = [] @@ -159,117 +158,18 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { expect(claude.connections[0].closeCount).toBe(1) }) - it('prefers a validated durable transcript leaf at graceful close', async () => { - const claude = fakeClaude() - const persistedHandles: unknown[] = [] - const readTranscriptLeaf = vi.fn().mockResolvedValue('durable-tail') - const adapter = adapterFor(claude, {}, [], persistedHandles, undefined, readTranscriptLeaf) - await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) - claude.connections[0].handlers.onMessage?.({ - type: 'assistant', - session_id: PROVIDER_SESSION_ID, - uuid: 'observed-tail' - }) - - await adapter.closeSession('session-1') - - expect(readTranscriptLeaf).toHaveBeenCalledWith({ - providerSessionId: PROVIDER_SESSION_ID, - previousLeafUuid: 'observed-tail', - claudeConfigDir: '/accounts/claude' - }) - expect(persistedHandles.at(-1)).toMatchObject({ leafUuid: 'durable-tail' }) - }) - - it('passes the pinned Claude account home to transcript validation', async () => { - const claude = fakeClaude() - const persistedHandles: unknown[] = [] - const readTranscriptLeaf = vi.fn().mockResolvedValue('durable-tail') - const adapter = adapterFor( - claude, - { claudeConfigDir: '/accounts/selected' }, - [], - persistedHandles, - undefined, - readTranscriptLeaf - ) - await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) - claude.connections[0].handlers.onMessage?.({ - type: 'assistant', - session_id: PROVIDER_SESSION_ID, - uuid: 'observed-tail' - }) - - await adapter.closeSession('session-1') - - expect(readTranscriptLeaf).toHaveBeenCalledWith({ - providerSessionId: PROVIDER_SESSION_ID, - previousLeafUuid: 'observed-tail', - claudeConfigDir: '/accounts/selected' - }) - }) - - it('re-proves from the transcript root when the observed cursor is missing', async () => { - const claude = fakeClaude() - const persistedHandles: unknown[] = [] - const readTranscriptLeaf = vi - .fn() - .mockRejectedValueOnce(new ClaudeTranscriptPreviousCursorMissingError()) - .mockResolvedValueOnce('reproved-main-leaf') - const adapter = adapterFor(claude, {}, [], persistedHandles, undefined, readTranscriptLeaf) - await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) - claude.connections[0].handlers.onMessage?.({ - type: 'assistant', - session_id: PROVIDER_SESSION_ID, - uuid: 'observed-tail' - }) - - await adapter.closeSession('session-1') - - expect(readTranscriptLeaf).toHaveBeenNthCalledWith(1, { - providerSessionId: PROVIDER_SESSION_ID, - previousLeafUuid: 'observed-tail', - claudeConfigDir: '/accounts/claude' - }) - expect(readTranscriptLeaf).toHaveBeenNthCalledWith(2, { - providerSessionId: PROVIDER_SESSION_ID, - previousLeafUuid: null, - claudeConfigDir: '/accounts/claude' - }) - expect(persistedHandles.at(-1)).toMatchObject({ leafUuid: 'reproved-main-leaf' }) - }) - - it('keeps the observed leaf when transcript validation proves a sibling branch', async () => { - const claude = fakeClaude() - const persistedHandles: unknown[] = [] - const readTranscriptLeaf = vi - .fn() - .mockRejectedValue(new Error('latest marker is on a sibling branch')) - const adapter = adapterFor(claude, {}, [], persistedHandles, undefined, readTranscriptLeaf) - await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) - claude.connections[0].handlers.onMessage?.({ - type: 'assistant', - session_id: PROVIDER_SESSION_ID, - uuid: 'observed-tail' - }) - - await adapter.closeSession('session-1') - - expect(readTranscriptLeaf).toHaveBeenCalledTimes(1) - expect(persistedHandles.at(-1)).toMatchObject({ leafUuid: 'observed-tail' }) - }) - - it('persists the last transcript leaf before an unexpected first-hand exit', async () => { + it('persists the last completed turn, not a half-turn prompt, on an unexpected exit', async () => { const claude = fakeClaude() const persistedHandles: unknown[] = [] const events: ClaudeStructuredSessionEvent[] = [] const adapter = adapterFor(claude, {}, events, persistedHandles) await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) - claude.connections[0].handlers.onMessage?.({ - type: 'assistant', - session_id: PROVIDER_SESSION_ID, - uuid: 'crash-leaf' - }) + const frame = (message: Record) => + claude.connections[0].handlers.onMessage?.({ session_id: PROVIDER_SESSION_ID, ...message }) + frame({ type: 'assistant', uuid: 'crash-leaf' }) + frame({ type: 'result', uuid: 'result-frame-uuid' }) + // The next turn's prompt is live, but its turn never completed. + frame({ type: 'user', uuid: 'half-turn-prompt' }) claude.connections[0].handlers.onExit?.( new Error('claude stream-json exited (code 1): crashed unexpectedly') @@ -290,81 +190,6 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { }) }) - it('derives the crash cursor from the validated transcript tail', async () => { - const claude = fakeClaude() - const persistedHandles: unknown[] = [] - const adapter = adapterFor( - claude, - {}, - [], - persistedHandles, - undefined, - vi.fn().mockResolvedValue('durable-crash-leaf') - ) - await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) - claude.connections[0].handlers.onMessage?.({ - type: 'assistant', - session_id: PROVIDER_SESSION_ID, - uuid: 'stale-observed-tail' - }) - claude.connections[0].handlers.onExit?.( - new Error('claude stream-json exited (signal SIGKILL): crashed') - ) - await tick() - - expect(persistedHandles.at(-1)).toMatchObject({ leafUuid: 'durable-crash-leaf' }) - }) - - it('re-proves a first-hand crash cursor from the transcript root after stale validation', async () => { - const claude = fakeClaude() - const persistedHandles: unknown[] = [] - const readTranscriptLeaf = vi - .fn() - .mockRejectedValueOnce(new ClaudeTranscriptPreviousCursorMissingError()) - .mockResolvedValueOnce('reproved-crash-leaf') - const adapter = adapterFor(claude, {}, [], persistedHandles, undefined, readTranscriptLeaf) - await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) - claude.connections[0].handlers.onMessage?.({ - type: 'assistant', - session_id: PROVIDER_SESSION_ID, - uuid: 'stale-observed-tail' - }) - claude.connections[0].handlers.onExit?.(new Error('crashed')) - await tick() - - expect(readTranscriptLeaf).toHaveBeenNthCalledWith(1, { - providerSessionId: PROVIDER_SESSION_ID, - previousLeafUuid: 'stale-observed-tail', - claudeConfigDir: '/accounts/claude' - }) - expect(readTranscriptLeaf).toHaveBeenNthCalledWith(2, { - providerSessionId: PROVIDER_SESSION_ID, - previousLeafUuid: null, - claudeConfigDir: '/accounts/claude' - }) - expect(persistedHandles.at(-1)).toMatchObject({ leafUuid: 'reproved-crash-leaf' }) - }) - - it('keeps the observed crash leaf when transcript validation proves a sibling branch', async () => { - const claude = fakeClaude() - const persistedHandles: unknown[] = [] - const readTranscriptLeaf = vi - .fn() - .mockRejectedValue(new Error('latest marker is on a sibling branch')) - const adapter = adapterFor(claude, {}, [], persistedHandles, undefined, readTranscriptLeaf) - await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) - claude.connections[0].handlers.onMessage?.({ - type: 'assistant', - session_id: PROVIDER_SESSION_ID, - uuid: 'observed-crash-tail' - }) - claude.connections[0].handlers.onExit?.(new Error('crashed')) - await tick() - - expect(readTranscriptLeaf).toHaveBeenCalledTimes(1) - expect(persistedHandles.at(-1)).toMatchObject({ leafUuid: 'observed-crash-tail' }) - }) - it('publishes lifecycle recovery even when crash-cursor persistence fails', async () => { const claude = fakeClaude() const events: ClaudeStructuredSessionEvent[] = [] @@ -374,7 +199,6 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { events, [], undefined, - undefined, vi.fn().mockRejectedValue(new Error('store unavailable')) ) await adapter.acquire({ identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }) @@ -452,7 +276,7 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { expect(events.filter((event) => event.type === 'ended')).toHaveLength(1) }) - it('launches the first replacement from the settled retained transcript cursor', async () => { + it('launches the first replacement from the settled retained turn-end cursor', async () => { const claude = fakeClaude() const events: ClaudeStructuredSessionEvent[] = [] const persistedHandles: unknown[] = [] @@ -461,7 +285,6 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { appendTombstone: () => {}, publish: () => {} } - const readTranscriptLeaf = vi.fn().mockResolvedValue('durable-retained-leaf') let durableLeafUuid: string | null = null const resolveLaunch = vi.fn(async ({ identity }) => { if ( @@ -484,7 +307,7 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { } return { pathToClaudeCodeExecutable: 'claude', - options: { resume: PROVIDER_SESSION_ID, resumeSessionAt: durableLeafUuid }, + options: { resume: PROVIDER_SESSION_ID }, cwd: '/work/repo', claudeConfigDir: '/accounts/claude', providerSessionId: PROVIDER_SESSION_ID, @@ -504,7 +327,6 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { onEvent: (event) => events.push(event), readProcessStartTime: async () => 1_700_000_000_000, now: () => 1_700_000_000_500, - readTranscriptLeaf, persistHandle }) const firstAcquisition = await adapter.acquire({ @@ -538,6 +360,7 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { session_id: PROVIDER_SESSION_ID, uuid: 'observed-retained-leaf' }) + first.handlers.onMessage?.({ type: 'result', session_id: PROVIDER_SESSION_ID }) first.close = vi .fn<() => Promise>() .mockResolvedValueOnce(false) @@ -570,23 +393,17 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { { sessionId: 'session-1', providerSessionId: PROVIDER_SESSION_ID, - leafUuid: 'durable-retained-leaf', + leafUuid: 'observed-retained-leaf', fence: 7 } ]) - expect(readTranscriptLeaf).toHaveBeenCalledOnce() - expect(readTranscriptLeaf).toHaveBeenCalledWith({ - providerSessionId: PROVIDER_SESSION_ID, - previousLeafUuid: 'observed-retained-leaf', - claudeConfigDir: '/accounts/claude' - }) expect(resolveLaunch).toHaveBeenNthCalledWith(2, { identity: { ...identityFor(), providerHandle: { kind: 'claude', sessionId: PROVIDER_SESSION_ID, - leafUuid: 'durable-retained-leaf' + leafUuid: 'observed-retained-leaf' } } }) @@ -606,15 +423,12 @@ describe('ClaudeStructuredSessionAdapter transcript-derived recovery', () => { handle: { provider: 'claude', sessionId: PROVIDER_SESSION_ID, - leafUuid: 'durable-retained-leaf' + leafUuid: 'observed-retained-leaf' }, origin: 'resumed', mintedAtFence: 8 }) - expect(claude.connections[1]?.launch.options).toMatchObject({ - resume: PROVIDER_SESSION_ID, - resumeSessionAt: 'durable-retained-leaf' - }) + expect(claude.connections[1]?.launch.options).toEqual({ resume: PROVIDER_SESSION_ID }) expect(claude.connections).toHaveLength(2) }) }) diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts index de697192074..c333f629582 100644 --- a/src/main/claude/claude-structured-session-state.ts +++ b/src/main/claude/claude-structured-session-state.ts @@ -96,14 +96,13 @@ export type ClaudeStructuredSessionAdapterDeps = { leafUuid: string | null fence: number }) => Promise - /** Read the durable transcript branch after a child has flushed its final rows. */ - readTranscriptLeaf?: (input: { + /** Advance the durable resume point in place at a turn end; bookkeeping, never a turn failure. */ + persistResumePoint?: (input: { + sessionId: string providerSessionId: string - previousLeafUuid: string | null - intentionalRewindUuid?: string - /** Account-scoped Claude config root that owns this provider session. */ - claudeConfigDir: string - }) => Promise + leafUuid: string + fence: number + }) => Promise } export type ClaudeDispatchWaiter = { @@ -128,9 +127,10 @@ export type ClaudeDispatchWaiter = { export type ClaudeSession = { connection: ClaudeStreamJsonConnection providerSessionId: string - /** Durable transcript files live under this account's `projects` directory. */ - claudeConfigDir: string + /** Latest main-chain message seen on the live stream, mid-turn included. */ leafUuid: string | null + /** `leafUuid` at the last completed turn; the only leaf close and exit persist. */ + turnEndLeafUuid: string | null fence: number acquisitionGeneration: string prompts: ClaudePromptRegistry @@ -160,6 +160,8 @@ export type ClaudeSession = { dispatchSequence: number /** Fences overlapping option writes so a late completion cannot restore stale state. */ optionMutationSequence: number + /** Latest resume point written at a turn end; close and exit persist after it settles. */ + resumePointWrite?: { leafUuid: string; settled: Promise } /** Shared durable-close write; a failed write clears this for a retry. */ closePersistence?: Promise /** Shared full close/finalization operation; a failed operation clears this for a retry. */ diff --git a/src/main/claude/claude-structured-session-test-support.ts b/src/main/claude/claude-structured-session-test-support.ts index dbff80182d4..1e6e43c4e7e 100644 --- a/src/main/claude/claude-structured-session-test-support.ts +++ b/src/main/claude/claude-structured-session-test-support.ts @@ -202,7 +202,6 @@ export function adapterFor( events: ClaudeStructuredSessionEvent[] = [], persistedHandles: unknown[] = [], initTimeoutMs?: number, - readTranscriptLeaf?: ClaudeStructuredSessionAdapterDeps['readTranscriptLeaf'], persistHandle?: ClaudeStructuredSessionAdapterDeps['persistHandle'], onBackgroundTasksChanged?: ClaudeStructuredSessionAdapterDeps['onBackgroundTasksChanged'], onDispatchSettledLate?: ClaudeStructuredSessionAdapterDeps['onDispatchSettledLate'] @@ -229,8 +228,7 @@ export function adapterFor( persistedHandles.push(handle) }), ...(onBackgroundTasksChanged ? { onBackgroundTasksChanged } : {}), - ...(onDispatchSettledLate ? { onDispatchSettledLate } : {}), - ...(readTranscriptLeaf ? { readTranscriptLeaf } : {}) + ...(onDispatchSettledLate ? { onDispatchSettledLate } : {}) }) } @@ -248,7 +246,6 @@ export async function acquired( undefined, undefined, undefined, - undefined, onDispatchSettledLate ) await adapter.acquire({ diff --git a/src/main/claude/claude-structured-turn-resume-point.test.ts b/src/main/claude/claude-structured-turn-resume-point.test.ts new file mode 100644 index 00000000000..ab707461197 --- /dev/null +++ b/src/main/claude/claude-structured-turn-resume-point.test.ts @@ -0,0 +1,227 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + agentSessionLeaseFixture, + agentSessionRecordFixture +} from '../../shared/agent-session-record.test-fixture' +import type { AgentSessionRecord } from '../../shared/agent-session-record' +import { LOCAL_EXECUTION_HOST_ID } from '../../shared/execution-host' +import type { AgentSessionRecordStore } from '../runtime/agent-session-record-store' +import { reviseAgentSessionClaudeResumePoint } from '../runtime/agent-session-provider-handle-transition' +import { createClaudeStructuredLaunchResolver } from './claude-structured-launch-resolution' +import { + ClaudeStructuredSessionAdapter, + type ClaudeStructuredSessionAdapterDeps, + type ClaudeStructuredSessionEvent +} from './claude-structured-session-adapter' +import { + fakeClaude, + identityFor, + PROVIDER_SESSION_ID, + recordingJournalSink, + tick +} from './claude-structured-session-test-support' + +const FENCE = 7 + +function liveRecord(): AgentSessionRecord { + const record = agentSessionRecordFixture( + agentSessionLeaseFixture({ runtimeKind: 'native', runtimeFence: FENCE }) + ) + return { + ...record, + location: { ...record.location, executionHostId: LOCAL_EXECUTION_HOST_ID }, + providerHandleChain: [ + { + linkId: 'created-link', + origin: 'created', + mintedAtFence: 1, + observedAt: 500, + handle: { provider: 'claude', sessionId: PROVIDER_SESSION_ID, leafUuid: null } + }, + { + linkId: 'published-link', + origin: 'resumed', + mintedAtFence: FENCE, + observedAt: 1_000, + handle: { provider: 'claude', sessionId: PROVIDER_SESSION_ID, leafUuid: 'resumed-at' } + } + ] + } +} + +/** An owner whose durable record is updated the way the runtime adapter updates it. */ +async function liveOwner( + persistResumePoint?: ClaudeStructuredSessionAdapterDeps['persistResumePoint'] +) { + const store = { record: liveRecord() } + const claude = fakeClaude() + const events: ClaudeStructuredSessionEvent[] = [] + const persistedHandles: unknown[] = [] + const adapter = new ClaudeStructuredSessionAdapter({ + resolveLaunch: async () => ({ + pathToClaudeCodeExecutable: 'claude', + options: { resume: PROVIDER_SESSION_ID }, + cwd: '/work/repo', + claudeConfigDir: '/accounts/claude', + providerSessionId: PROVIDER_SESSION_ID, + resumeLeafUuid: 'resumed-at', + resumed: true + }), + onEvent: (event) => events.push(event), + openConnection: claude.openConnection, + readProcessStartTime: async () => 1_700_000_000_000, + now: () => 1_700_000_000_500, + persistHandle: async (handle) => { + persistedHandles.push(handle) + }, + persistResumePoint: + persistResumePoint ?? + (async ({ providerSessionId, leafUuid, fence }) => { + store.record = reviseAgentSessionClaudeResumePoint({ + record: store.record, + fence, + providerSessionId, + leafUuid, + now: 2_000 + }) + }) + }) + await adapter.acquire({ + identity: identityFor(), + fence: FENCE, + spawnToken: 'spawn-7', + events: recordingJournalSink() + }) + const connection = claude.connections[0]! + const frame = (message: Record) => + connection.handlers.onMessage?.({ session_id: PROVIDER_SESSION_ID, ...message }) + const turn = async (userUuid: string, assistantUuid: string) => { + frame({ type: 'user', uuid: userUuid }) + frame({ type: 'assistant', uuid: assistantUuid }) + frame({ type: 'system', subtype: 'stop_hook_summary', uuid: `${assistantUuid}-hook` }) + frame({ type: 'result', subtype: 'success', uuid: `${assistantUuid}-result` }) + await tick() + } + return { adapter, claude, store, events, persistedHandles, frame, turn } +} + +afterEach(() => { + vi.restoreAllMocks() +}) + +describe('Claude durable resume point at turn end', () => { + it('advances after each completed turn without growing the handle chain', async () => { + const { adapter, store, turn } = await liveOwner() + try { + await turn('u1', 'a1') + expect(store.record.providerHandleChain.at(-1)?.handle).toMatchObject({ leafUuid: 'a1' }) + await turn('u2', 'a2') + expect(store.record.providerHandleChain).toHaveLength(2) + expect(store.record.providerHandleChain.at(-1)).toMatchObject({ + linkId: 'published-link', + handle: { leafUuid: 'a2' } + }) + } finally { + await adapter.closeAll() + } + }) + + it('keeps the last completed turn after a crash that ran no close or exit', async () => { + const { store, frame, turn } = await liveOwner() + await turn('u1', 'a1') + await turn('u2', 'a2') + // The next turn starts, then the host dies before its reply or any close path. + frame({ type: 'user', uuid: 'u3' }) + await tick() + const head = store.record.providerHandleChain.at(-1)!.handle + expect(head).toMatchObject({ leafUuid: 'a2' }) + + const resolve = createClaudeStructuredLaunchResolver({ + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the resolver reads only getRecord from its store. + store: { getRecord: () => store.record } as unknown as AgentSessionRecordStore, + resolveWorkspacePath: async (id) => `/repos/${id}`, + resolveCommand: () => '/usr/local/bin/claude', + resolveAuthPolicy: () => ({ stripAuthEnv: false }) + }) + const launch = await resolve({ + identity: { + ...identityFor(store.record.sessionId), + providerHandle: { kind: 'claude', sessionId: PROVIDER_SESSION_ID, leafUuid: 'a2' } + } + }) + // Bookkeeping only: Claude continues from the end of its own conversation. + expect(launch).toMatchObject({ resumeLeafUuid: 'a2', options: { resume: PROVIDER_SESSION_ID } }) + expect(launch.options).not.toHaveProperty('resumeSessionAt') + }) + + it('lands the close cursor after, never under, an in-flight turn-end write', async () => { + const write = Promise.withResolvers() + const { adapter, persistedHandles, turn } = await liveOwner(() => write.promise) + await turn('u1', 'a1') + const closed = adapter.closeSession('session-1') + await tick() + expect(persistedHandles).toEqual([]) + write.resolve() + await expect(closed).resolves.toBe(true) + expect(persistedHandles).toEqual([expect.objectContaining({ leafUuid: 'a1' })]) + }) + + it('never fails or holds up a turn when the write fails', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const persist = vi.fn(async () => { + throw new Error('record write failed') + }) + const { adapter, events, persistedHandles, turn } = await liveOwner(persist) + await turn('u1', 'a1') + await turn('u2', 'a2') + expect(persist).toHaveBeenCalledTimes(2) + expect( + events.filter((event) => event.type === 'message' && event.message.type === 'result') + ).toHaveLength(2) + expect(warn).toHaveBeenCalledWith( + '[claude-resume-point] turn-end resume point was not persisted:', + expect.objectContaining({ leafUuid: 'a2' }) + ) + await expect(adapter.closeSession('session-1')).resolves.toBe(true) + expect(persistedHandles).toEqual([expect.objectContaining({ leafUuid: 'a2' })]) + }) + + it('retries a failed write at the next turn end even when the leaf has not moved', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + let failing = true + const persist = vi.fn(async () => { + if (failing) { + throw new Error('record write failed') + } + }) + const { adapter, frame, turn } = await liveOwner(persist) + try { + await turn('u1', 'a1') + expect(persist).toHaveBeenCalledTimes(1) + failing = false + // A turn that ends without a new message, such as an interrupted one, keeps the same leaf. + frame({ type: 'result', subtype: 'error_during_execution', uuid: 'a1-result-2' }) + await tick() + expect(persist).toHaveBeenCalledTimes(2) + expect(persist).toHaveBeenLastCalledWith(expect.objectContaining({ leafUuid: 'a1' })) + } finally { + await adapter.closeAll() + } + }) + + it('ignores a result that trails the child exit', async () => { + const persist = vi.fn(async () => {}) + const { adapter, claude, frame, persistedHandles, turn } = await liveOwner(persist) + await turn('u1', 'a1') + expect(persist).toHaveBeenCalledTimes(1) + claude.connections[0]!.handlers.onExit?.(new Error('claude crashed')) + frame({ type: 'user', uuid: 'u2' }) + frame({ type: 'assistant', uuid: 'a2' }) + frame({ type: 'result', subtype: 'success', uuid: 'a2-result' }) + await adapter.drainObservedExits() + await tick() + // A result with no owner behind it is not a completed turn; the exit keeps the last one. + expect(persist).toHaveBeenCalledTimes(1) + expect(persistedHandles).toEqual([expect.objectContaining({ leafUuid: 'a1' })]) + }) +}) diff --git a/src/main/claude/claude-transcript-branch-graph.test.ts b/src/main/claude/claude-transcript-branch-graph.test.ts index e77ac6996ae..90eb5b6bbd8 100644 --- a/src/main/claude/claude-transcript-branch-graph.test.ts +++ b/src/main/claude/claude-transcript-branch-graph.test.ts @@ -1,19 +1,41 @@ // The graph's own contract, independent of which bytes a reader fed it. import { describe, expect, it } from 'vitest' -import { createBranchProof } from './claude-transcript-branch-graph' +import { + ClaudeTranscriptMarkerMissingError, + createBranchProof +} from './claude-transcript-branch-graph' const row = (uuid: string, parentUuid: string | null): string => JSON.stringify({ type: 'user', uuid, parentUuid, sessionId: 'provider' }) -function build(lines: string[], previousLeafUuid: string | null) { - const builder = createBranchProof({ providerSessionId: 'provider', previousLeafUuid }) +function build(lines: string[], previousLeafUuid: string | null, tip?: 'marker' | 'file-tail') { + const builder = createBranchProof({ + providerSessionId: 'provider', + previousLeafUuid, + ...(tip === undefined ? {} : { tip }) + }) for (const [index, line] of lines.entries()) { builder.add(line, index, true) } return builder } +describe('createBranchProof file-tail tip', () => { + const MARKERLESS = [row('anchor', null), row('mid', 'anchor'), row('leaf', 'mid')] + + it('proves the tip from the last main-chain row with no last-prompt row anywhere', () => { + expect(build(MARKERLESS, 'anchor', 'file-tail').finish()).toEqual({ + leafUuid: 'leaf', + relation: 'descendant' + }) + }) + + it('still requires the marker on the same bytes in marker mode', () => { + expect(() => build(MARKERLESS, 'anchor').finish()).toThrow(ClaudeTranscriptMarkerMissingError) + }) +}) + describe('createBranchProof ancestry chain', () => { const MARKER = JSON.stringify({ type: 'last-prompt', sessionId: 'provider', leafUuid: 'leaf' }) diff --git a/src/main/claude/claude-transcript-branch-graph.ts b/src/main/claude/claude-transcript-branch-graph.ts index 56cf62de2b8..e739da07975 100644 --- a/src/main/claude/claude-transcript-branch-graph.ts +++ b/src/main/claude/claude-transcript-branch-graph.ts @@ -87,16 +87,33 @@ function proveAppendOrder(nodes: Map): void { } } +/** Rows Claude's own loader can end a conversation on; titles and markers carry no chain. */ +const TRANSCRIPT_TAIL_TYPES: ReadonlySet = new Set([ + 'user', + 'assistant', + 'system', + 'attachment' +]) + type BranchProofInput = { providerSessionId: string previousLeafUuid: string | null intentionalRewindUuid?: string + /** + * Which row is the branch tip. `file-tail` proves from the file's last + * main-chain row: Claude writes its `last-prompt` marker only sporadically, so + * a marker tip hides rows Claude already holds after a crash. Without an + * eligible tail row the marker rules apply unchanged. Default: `marker`. + */ + tip?: 'marker' | 'file-tail' } function createBranchProof(input: BranchProofInput) { const nodes = new Map() let leafUuid: string | null = null let leafMarkerLineIndex = -1 + let tailUuid: string | null = null + let tailLineIndex = -1 return { add, finish, ancestryChain } function add(line: string, index: number, terminated: boolean): void { @@ -156,9 +173,23 @@ function createBranchProof(input: BranchProofInput) { lineIndex: existing?.lineIndex ?? index, disallowedLeaf }) + if ( + input.tip === 'file-tail' && + TRANSCRIPT_TAIL_TYPES.has(row.type) && + row.isSidechain !== true && + row.parent_tool_use_id == null + ) { + tailUuid = uuid + tailLineIndex = index + } } function finish(): ClaudeTranscriptBranchProof { + if (input.tip === 'file-tail' && tailUuid) { + // The last main-chain row supersedes any marker; the marker lags crashes. + leafUuid = tailUuid + leafMarkerLineIndex = tailLineIndex + } if (!leafUuid) { throw new ClaudeTranscriptMarkerMissingError() } diff --git a/src/main/claude/claude-transcript-branch-proof.ts b/src/main/claude/claude-transcript-branch-proof.ts index c8d330e0955..38641043b70 100644 --- a/src/main/claude/claude-transcript-branch-proof.ts +++ b/src/main/claude/claude-transcript-branch-proof.ts @@ -32,7 +32,8 @@ export type ClaudeTranscriptBranchAncestry = { chain: string[] } -type AncestryInput = BranchProofInput & { +/** Replay always proves from the file tail, so a caller-supplied tip has no meaning here. */ +type AncestryInput = Omit & { /** The ancestry walk stops here; the proof is what established it is reachable. */ ancestryAnchorUuid: string /** The FIRST record carrying each chain uuid, in file order. */ @@ -150,10 +151,11 @@ export async function proveClaudeTranscriptBranch( } /** - * Prove the branch, then replay the anchor..leaf records off the SAME pinned - * bytes. Two bounded passes instead of one whole-file string: the graph pass - * retains uuid/parentUuid only, and the replay pass hands each chain record to - * the caller once and keeps nothing, so neither pass holds the transcript. + * Prove the branch from the file's last transcript row back to the anchor, then + * replay the anchor..tail records off the SAME pinned bytes. Two bounded passes + * instead of one whole-file string: the graph pass retains uuid/parentUuid only, + * and the replay pass hands each chain record to the caller once and keeps + * nothing, so neither pass holds the transcript. * * The replay runs only after `finish()` succeeds, so a growth retry can never * emit a record twice. @@ -165,7 +167,7 @@ export async function replayClaudeTranscriptBranchAncestry( input.transcriptPath, input.maxRecordBytes, async (readLines) => { - const builder = createBranchProof(input) + const builder = createBranchProof({ ...input, tip: 'file-tail' }) let index = 0 for await (const record of readLines()) { builder.add(record.line, index++, record.terminated) @@ -187,7 +189,7 @@ export async function replayClaudeTranscriptBranchAncestry( export function replayClaudeTranscriptBranchAncestryFromJsonl( input: AncestryInput & { contents: string } ): ClaudeTranscriptBranchAncestry { - const builder = createBranchProof(input) + const builder = createBranchProof({ ...input, tip: 'file-tail' }) const lines = input.contents.split('\n') for (const [index, line] of lines.entries()) { builder.add(line, index, index < lines.length - 1) @@ -200,37 +202,3 @@ export function replayClaudeTranscriptBranchAncestryFromJsonl( } return { proof, chain } } - -/** Re-run a durable branch proof from the transcript root when a sampled cursor is stale. */ -export async function readClaudeTranscriptLeafWithReproof(input: { - readTranscriptLeaf: (input: { - providerSessionId: string - previousLeafUuid: string | null - claudeConfigDir: string - }) => Promise - claudeConfigDir: string - providerSessionId: string - previousLeafUuid: string | null -}): Promise { - try { - return await input.readTranscriptLeaf({ - providerSessionId: input.providerSessionId, - previousLeafUuid: input.previousLeafUuid, - claudeConfigDir: input.claudeConfigDir - }) - } catch (error) { - // A missing cursor can be stale after compaction and is safe to re-prove from the root. A torn - // tail is still being written; dropping the cursor would make a later sibling look admissible. - if ( - input.previousLeafUuid === null || - !(error instanceof ClaudeTranscriptPreviousCursorMissingError) - ) { - throw error - } - return input.readTranscriptLeaf({ - providerSessionId: input.providerSessionId, - previousLeafUuid: null, - claudeConfigDir: input.claudeConfigDir - }) - } -} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition.ts index 87c63454b24..8d8426da437 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition.ts @@ -1,5 +1,4 @@ import { isDeepStrictEqual } from 'node:util' -import { claudeRewindAcquisitionProofs } from './structured-rewind-claude-proof' import type { AgentSessionRecord } from '../../../shared/agent-session-record' import { AgentSessionPreSpawnError, @@ -17,7 +16,6 @@ export async function acquireOwner( input: AttachFlowInput, record: AgentSessionRecord ): Promise<{ record: AgentSessionRecord; acquisitionGeneration: string | null }> { - const { store, rewind, now } = input const fence = record.lease.runtimeFence const spawnToken = record.lease.reservedSpawnToken if (!spawnToken) { @@ -39,7 +37,6 @@ export async function acquireOwner( } const acquired = await input.adapter.acquire({ identity: journalIdentityFor(record, input.params), - ...claudeRewindAcquisitionProofs({ store, record, rewind, now }), fence, // Retries must recover the original reservation, not mint a second child. spawnToken, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index 94d03ddd32b..94bad859e46 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -44,12 +44,6 @@ export class AgentSessionAcquisitionRefusal extends Error { } } -export class AgentSessionRewindRefusal extends AgentSessionAcquisitionRefusal { - constructor(readonly rewindReason: AgentSessionRewindReason) { - super(`agent_session_rewind:${rewindReason}`) - } -} - export class AgentSessionPromptUnavailableError extends Error { constructor(itemId: string) { super(`The provider is no longer waiting on ${itemId}.`) @@ -129,14 +123,6 @@ export type StructuredAgentSessionLifecycleEvent = { export type StructuredAgentSessionAcquireInput = { identity: AgentSessionJournalIdentity - rewind?: { - targetUuid: string - previousLeafUuid: string - dropsTurn?: string - onProved?: (leafUuid: string) => Promise - } - /** Recovery restores an unproved rewind's original cursor with ordinary branch proof. */ - rewindRecovery?: { leafUuid: string; onProved: () => Promise } fence: number spawnToken: string options?: Readonly> diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts index 3abfd42b8aa..9fcc99e41b0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts @@ -1,12 +1,9 @@ import { settlePostAcquisitionAttachFailure } from './structured-agent-session-attach-failure' -import { rewindRefusal } from './structured-rewind-refusal' import { - AgentSessionRewindRefusal, AgentSessionAcquisitionExitUnprovenError, AgentSessionAcquisitionRootExitObservedError, AgentSessionAcquisitionRefusal, isAgentSessionPreSpawnError, - type StructuredAgentSessionAcquireInput, type StructuredAgentSessionAdapter } from './structured-agent-session-adapter' // The host supplies owner authority; this flow reserves, proves, and publishes the session. @@ -45,7 +42,6 @@ import { import type { ProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' export type AttachFlowInput = { - rewind?: StructuredAgentSessionAcquireInput['rewind'] store: AgentSessionRecordStore adapter: StructuredAgentSessionAdapter journalRoot: string @@ -214,9 +210,6 @@ export async function performAttach( ) } } - if (error instanceof AgentSessionRewindRefusal) { - return rewindRefusal(error.rewindReason) - } if (error instanceof AgentSessionAcquisitionRefusal) { return { ok: false, refusal: { code: error.code, message: error.message } } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts index c85536ff679..2ceb8862571 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts @@ -1,4 +1,3 @@ -import type { StructuredAgentSessionAcquireInput } from './structured-agent-session-adapter' import { recoverStructuredRewind } from './structured-rewind-recovery' import { recoverInterruptedCompaction } from './structured-compaction-recovery' // The host's attach, lifted out of the host class. @@ -39,8 +38,7 @@ export function attachStructuredAgentSession( context: StructuredAgentSessionAttachContext, callerKey: string, params: AgentSessionAttachParams, - admitRecoveryTicket?: () => boolean, - rewind?: StructuredAgentSessionAcquireInput['rewind'] + admitRecoveryTicket?: () => boolean ): Promise> { const sessionId = params.envelope.sessionId const run = (recordPhase?: AgentSessionCreatePhaseRecorder) => @@ -82,7 +80,6 @@ export function attachStructuredAgentSession( context.runtimeState.probeOwner(sessionId) ) const attached = await performAttach({ - rewind, store: context.deps.store, adapter: context.deps.adapter, journalRoot: context.deps.journalRoot, @@ -155,16 +152,14 @@ export function attachStructuredAgentSession( hasProviderChild: true, acquisitionGeneration: acquisitionGeneration ?? previous?.acquisitionGeneration ?? null }) - if (!rewind) { - await recoverStructuredRewind( - context.deps.store, - sessionId, - attached.journal, - fence, - context.deps.adapter, - context.now - ) - } + await recoverStructuredRewind( + context.deps.store, + sessionId, + attached.journal, + fence, + context.deps.adapter, + context.now + ) await recoverInterruptedCompaction(context.deps.store, sessionId, attached.journal, fence) if (attached.recovery) { context.subscribers.reset(sessionId, attached.journal, attached.recovery.reset, fence) 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 index 21ab253c46c..390f3df1d0a 100644 --- 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 @@ -5,10 +5,9 @@ // into different eligibility or different double-fire protection. // // Resume itself acquires a provider child, not a new 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. +// childless session re-acquires the provider's own conversation — Claude's `resume` by session id, +// 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' diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.test.ts index e022f640403..19ae18dd5f4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.test.ts @@ -9,7 +9,6 @@ import { } from '../../../shared/agent-session-journal-item-key' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' -import { AgentSessionRewindRefusal } from './structured-agent-session-adapter' import { StructuredAgentSessionHost } from './structured-agent-session-host' import type { StructuredAgentSessionAdapter, @@ -36,7 +35,6 @@ let adapter: StructuredAgentSessionAdapter let acquires: StructuredAgentSessionAcquireInput[] const rewind = vi.fn>() const recoverRewind = vi.fn>() -let failClaude = false beforeEach(async () => { resetHostTestOperationIds() @@ -50,7 +48,6 @@ beforeEach(async () => { } ] }) - failClaude = false acquires = [] directory = await mkdtemp(join(tmpdir(), 'orca-rewind-')) store = await AgentSessionRecordStore.open({ @@ -58,19 +55,11 @@ beforeEach(async () => { hostId: 'local' }) adapter = { - supportsCreate: (_location, agent) => agent === 'codex' || agent === 'claude', + supportsCreate: (_location, agent) => agent === 'codex', supportsLocation: () => true, acquire: async (input) => { acquires.push(input) - if (input.rewind && failClaude) { - throw new AgentSessionRewindRefusal('provider-refused') - } - if (input.rewind) { - await input.rewind.onProved?.(input.rewind.targetUuid) - } - await input.rewindRecovery?.onProved() sink = input.events! - const handle = input.identity.providerHandle return { process: { hostId: 'local', @@ -84,14 +73,7 @@ beforeEach(async () => { mintedAtFence: input.fence, observedAt: HOST_TEST_NOW, origin: acquires.length === 1 ? 'created' : 'resumed', - handle: - handle.kind === 'claude' - ? { - provider: 'claude', - sessionId: handle.sessionId, - leafUuid: input.rewind?.targetUuid ?? 'tip' - } - : { provider: 'codex', threadId: HOST_TEST_THREAD } + handle: { provider: 'codex', threadId: HOST_TEST_THREAD } } } }, @@ -122,22 +104,14 @@ afterEach(async () => { await rm(directory, { recursive: true, force: true }) }) -async function seed(provider: 'codex' | 'claude' = 'codex', acceptedSubmissions = false) { - const params = - provider === 'codex' - ? hostTestAttachParams(null) - : hostTestAttachParams(null, { - provider, - agent: provider, - accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/claude' }, - providerHandle: { kind: 'claude', sessionId: 'claude-session', leafUuid: 'tip' } - }) - expect(await host.attach(caller, params)).toMatchObject({ ok: true }) - const keys = ['kept', 'drop', 'tip'].map((uuid) => - provider === 'codex' - ? { provider, threadId: HOST_TEST_THREAD, turnId: uuid, ordinal: 0 } - : { provider, sessionId: 'claude-session', uuid } - ) +async function seed(acceptedSubmissions = false) { + expect(await host.attach(caller, hostTestAttachParams(null))).toMatchObject({ ok: true }) + const keys = ['kept', 'drop', 'tip'].map((turnId) => ({ + provider: 'codex' as const, + threadId: HOST_TEST_THREAD, + turnId, + ordinal: 0 + })) let selectedItemId = agentJournalItemKey(keys[1]!) for (const [i, identity] of keys.entries()) { const body = { @@ -196,32 +170,14 @@ function params( } describe('host rewind', () => { - it.each(['codex', 'claude'] as const)( - 'resolves accepted %s user submissions to provider targets', - async (provider) => { - const target = await seed(provider, true) - expect(target.startsWith('orca:')).toBe(true) - expect(await host.rewind(caller, params(target))).toMatchObject({ ok: true }) - expect(host.journalSnapshot(HOST_TEST_SESSION).items).toHaveLength(1) - if (provider === 'codex') { - expect(rewind).toHaveBeenCalledWith(expect.objectContaining({ beforeTurnId: 'drop' })) - } else { - expect(acquires[1]?.rewind).toMatchObject({ targetUuid: 'kept', dropsTurn: 'drop' }) - } - } - ) - - it('retains the preceding accepted Claude prompt when rewinding its assistant response', async () => { - await seed('claude', true) - const target = agentJournalItemKey({ - provider: 'claude', - sessionId: 'claude-session', - uuid: 'tip' - }) + it('resolves accepted codex user submissions to provider targets', async () => { + const target = await seed(true) + expect(target.startsWith('orca:')).toBe(true) expect(await host.rewind(caller, params(target))).toMatchObject({ ok: true }) - expect(acquires[1]?.rewind).toMatchObject({ targetUuid: 'drop' }) - expect(host.journalSnapshot(HOST_TEST_SESSION).items).toHaveLength(2) + expect(host.journalSnapshot(HOST_TEST_SESSION).items).toHaveLength(1) + expect(rewind).toHaveBeenCalledWith(expect.objectContaining({ beforeTurnId: 'drop' })) }) + it('finishes a durable provider success on reattach without repeating the provider mutation', async () => { const target = await seed() const request = params(target) @@ -306,43 +262,6 @@ describe('host rewind', () => { expect(await host.rewind(caller, request)).toMatchObject({ ok: true, replayed: true }) expect(rewind).toHaveBeenCalledTimes(1) }) - it('reacquires Claude at the retained cursor with the same session and a new lease fence', async () => { - const target = await seed('claude') - const before = store.getRecord(HOST_TEST_SESSION)!.lease.runtimeFence - expect(await host.rewind(caller, params(target))).toMatchObject({ ok: true }) - const emit = vi.fn() - const unsubscribe = host.subscribe({ id: 'after-rewind', sessionId: HOST_TEST_SESSION, emit }) - emit.mockClear() - sink.appendItem( - { provider: 'claude', sessionId: 'claude-session', uuid: 'next' }, - hostTestMessage('next') - ) - sink.publish() - await host.flushStreamedEvents(HOST_TEST_SESSION) - expect(emit).toHaveBeenCalledWith(expect.objectContaining({ type: 'batch' })) - unsubscribe() - expect(acquires[1]?.rewind).toMatchObject({ - targetUuid: 'kept', - previousLeafUuid: 'tip', - dropsTurn: 'drop' - }) - expect(store.getRecord(HOST_TEST_SESSION)!.lease.runtimeFence).toBeGreaterThan(before) - expect(store.getRecord(HOST_TEST_SESSION)!.lease.ownerProcess?.pid).toBe(4002) - expect(host.journalSnapshot(HOST_TEST_SESSION).items).toHaveLength(2) - }) - it('recovers a Claude refusal with one plain resume and preserves the journal', async () => { - const target = await seed('claude') - failClaude = true - const before = host.journalSnapshot(HOST_TEST_SESSION) - expect(await host.rewind(caller, params(target))).toMatchObject({ - ok: false, - refusal: { rewindReason: 'provider-refused' } - }) - expect(acquires).toHaveLength(3) - expect(acquires[2]?.rewind).toBeUndefined() - expect(host.journalSnapshot(HOST_TEST_SESSION)).toEqual(before) - expect(store.getRecord(HOST_TEST_SESSION)!.lease.claimStatus).toBe('live') - }) it('refuses a rewind racing an active turn before provider execution', async () => { const target = await seed() sink.appendItem( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts index c301845c7c1..f6372c3d8ed 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts @@ -19,7 +19,6 @@ import { admitAndRunAgentSessionMutation } from './structured-agent-session-muta import { conversationCommandBlocked } from './structured-conversation-command-admission' import { rewindRefusal } from './structured-rewind-refusal' import { persistRewindRecord, recoverStructuredRewind } from './structured-rewind-recovery' -import { replaceClaudeRewindOwner } from './structured-rewind-claude-owner' import { mergeRetainedHostLifecycleRows } from './structured-rewind-retained-host-rows' export async function rewindStructuredAgentSession( @@ -100,7 +99,6 @@ export async function rewindStructuredAgentSession( return rewindRefusal('invalid-target') } let boundary = selected - let claude: Parameters[3] | undefined if (key.provider === 'codex' && head.provider === 'codex') { if (key.threadId !== head.threadId) { return rewindRefusal('invalid-target') @@ -114,32 +112,6 @@ export async function rewindStructuredAgentSession( readAgentJournalTurn(item.body)?.turnId === key.turnId ) }) - } else if (key.provider === 'claude' && head.provider === 'claude') { - if (key.sessionId !== head.sessionId) { - return rewindRefusal('invalid-target') - } - const previous = snapshot.items - .slice(0, boundary) - .map((item) => parseAgentJournalItemKey(providerKey(item.itemId))) - .findLast( - (identity) => - identity?.provider === 'claude' && identity.sessionId === key.sessionId - ) - if (previous?.provider !== 'claude') { - return rewindRefusal('invalid-target') - } - const prompts = snapshot.items - .slice(boundary) - .filter((item) => item.body.kind === 'message' && item.body.role === 'user') - const prompt = - prompts.length === 1 - ? parseAgentJournalItemKey(providerKey(prompts[0]!.itemId)) - : null - claude = { - targetUuid: previous.uuid, - previousLeafUuid: head.leafUuid ?? '', - ...(prompt?.provider === 'claude' ? { dropsTurn: prompt.uuid } : {}) - } } else { return rewindRefusal('invalid-target') } @@ -168,44 +140,39 @@ export async function rewindStructuredAgentSession( } await persistRewindRecord(store, sessionId, ctx.fence, prepared) ctx.publish() - const provider = claude - ? await replaceClaudeRewindOwner(attachContext, caller.callerKey, params, claude) - : await ctx.adapter.rewind!({ - sessionId, - fence: ctx.fence, - beforeTurnId: key.provider === 'codex' ? key.turnId : '', - onPrepared: async (items) => { - const retained = mergeRetainedHostLifecycleRows( - prepared.retained, - items.map(({ identity, body }) => ({ - itemId: agentJournalItemKey(identity), - body, - observedAt: ctx.now() - })) - ) - if ( - retained.length > 10_000 || - Buffer.byteLength(JSON.stringify(retained), 'utf8') > - AGENT_SESSION_HISTORY_MAX_PAGE_BYTES - ) { - throw new Error('agent_session_rewind:history-limit') - } - prepared = { ...prepared, retained } - await persistRewindRecord(store, sessionId, ctx.fence, prepared) - }, - onReverted: async () => { - await persistRewindRecord(store, sessionId, ctx.fence, { - ...prepared, - providerApplied: true - }) - } + const provider = await ctx.adapter.rewind!({ + sessionId, + fence: ctx.fence, + beforeTurnId: key.provider === 'codex' ? key.turnId : '', + onPrepared: async (items) => { + const retained = mergeRetainedHostLifecycleRows( + prepared.retained, + items.map(({ identity, body }) => ({ + itemId: agentJournalItemKey(identity), + body, + observedAt: ctx.now() + })) + ) + if ( + retained.length > 10_000 || + Buffer.byteLength(JSON.stringify(retained), 'utf8') > + AGENT_SESSION_HISTORY_MAX_PAGE_BYTES + ) { + throw new Error('agent_session_rewind:history-limit') + } + prepared = { ...prepared, retained } + await persistRewindRecord(store, sessionId, ctx.fence, prepared) + }, + onReverted: async () => { + await persistRewindRecord(store, sessionId, ctx.fence, { + ...prepared, + providerApplied: true }) + } + }) const fence = store.getRecord(sessionId)!.lease.runtimeFence if (!provider.ok) { - const reason = - 'reason' in provider - ? provider.reason - : (provider.refusal.rewindReason ?? 'outcome-unknown') + const reason = provider.reason if (reason !== 'outcome-unknown') { await persistRewindRecord(store, sessionId, fence, { ...prepared, diff --git a/src/main/native-chat/agent-session-wire/structured-rewind-claude-owner.ts b/src/main/native-chat/agent-session-wire/structured-rewind-claude-owner.ts deleted file mode 100644 index 13d2b51f284..00000000000 --- a/src/main/native-chat/agent-session-wire/structured-rewind-claude-owner.ts +++ /dev/null @@ -1,78 +0,0 @@ -import { agentSessionProviderHandleChainHead } from '../../../shared/agent-session-provider-handle' -import { createHash } from 'node:crypto' -import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' -import type { AgentSessionRewindParams } from '../../../shared/agent-session-rewind' -import type { StructuredAgentSessionAcquireInput } from './structured-agent-session-adapter' -import { attachFingerprintFields } from './structured-agent-session-attach' -import type { StructuredAgentSessionAttachContext } from './structured-agent-session-attach-context' -import { attachStructuredAgentSession } from './structured-agent-session-attach-orchestration' -import { rewindRefusal } from './structured-rewind-refusal' - -/** Runs within the rewind's session queue; acquisition still uses the normal reservation CAS. */ -export async function replaceClaudeRewindOwner( - context: StructuredAgentSessionAttachContext, - callerKey: string, - params: AgentSessionRewindParams, - rewind: NonNullable -): Promise<{ ok: true; items?: never } | ReturnType> { - const sessionId = params.envelope.sessionId - const session = context.sessions.get(sessionId)! - if (!(await context.deps.adapter.closeSession?.(sessionId))) { - return rewindRefusal('outcome-unknown') - } - session.hasProviderChild = false - context.publishStatus?.(sessionId) - const head = agentSessionProviderHandleChainHead( - context.deps.store.getRecord(sessionId)!.providerHandleChain - )?.handle - if (head?.provider !== 'claude' || !head.leafUuid) { - return rewindRefusal('invalid-target') - } - rewind = { ...rewind, previousLeafUuid: head.leafUuid } - const attach = async (intent: typeof rewind | undefined, stage: string) => { - const current = context.deps.store.getRecord(sessionId)! - const operationId = `${params.envelope.clientOperationId.split('-')[0]}-${createHash('sha256') - .update(JSON.stringify([callerKey, params.envelope.clientOperationId, stage])) - .digest('hex') - .slice(0, 32)}` - const attachParams = { - ...session.params, - envelope: { - sessionId, - clientOperationId: operationId, - expectedRuntimeFence: current.lease.runtimeFence, - payloadFingerprint: '' - } - } - attachParams.envelope.payloadFingerprint = computeAgentSessionPayloadFingerprint({ - method: 'agentSession.attach', - sessionId, - fields: attachFingerprintFields(attachParams) - }) - return attachStructuredAgentSession( - { - ...context, - serialize: (_id, run) => run() - }, - callerKey, - attachParams, - undefined, - intent - ) - } - const result = await attach(rewind, 'rewind') - if (result.ok) { - return { ok: true } as const - } - if ( - result.refusal.rewindReason === 'provider-refused' || - result.refusal.rewindReason === 'proof-mismatch' - ) { - const recovered = await attach(undefined, 'resume') - if (!recovered.ok) { - return rewindRefusal('outcome-unknown') - } - return rewindRefusal(result.refusal.rewindReason) - } - return rewindRefusal(result.refusal.rewindReason ?? 'outcome-unknown') -} diff --git a/src/main/native-chat/agent-session-wire/structured-rewind-claude-proof.test.ts b/src/main/native-chat/agent-session-wire/structured-rewind-claude-proof.test.ts deleted file mode 100644 index 7803dc7245d..00000000000 --- a/src/main/native-chat/agent-session-wire/structured-rewind-claude-proof.test.ts +++ /dev/null @@ -1,90 +0,0 @@ -import { describe, expect, it } from 'vitest' -import { agentSessionRecordFixture } from '../../../shared/agent-session-record.test-fixture' -import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' -import { claudeRewindAcquisitionProofs } from './structured-rewind-claude-proof' - -function setup() { - let current = agentSessionRecordFixture() - current.providerHandleChain = current.providerHandleChain.map((link) => ({ - ...link, - handle: { provider: 'claude', sessionId: 'provider-session-alpha-1', leafUuid: 'tip' } - })) - current.rewind = { - operationId: 'rewind-operation', - callerKey: 'desktop', - itemId: 'selected', - expectedEpoch: 'old-epoch', - phase: 'prepared', - retained: [] - } - const store: Pick = { - transitionHandoff: async (_sessionId, transition) => { - current = transition(current) - return current - } - } - return { - store, - record: () => current, - setFence: () => { - current = { ...current, lease: { ...current.lease, runtimeFence: 8 } } - } - } -} - -describe('Claude rewind durable proof checkpoints', () => { - it('atomically checkpoints the exact target and resumable head before owner publication', async () => { - const state = setup() - const proofs = claudeRewindAcquisitionProofs({ - store: state.store, - record: state.record(), - now: () => 3_000, - rewind: { previousLeafUuid: 'tip', targetUuid: 'kept' } - }) - await expect(proofs.rewind!.onProved!('wrong')).rejects.toThrow('proof-mismatch') - expect(state.record().rewind?.phase).toBe('prepared') - expect(state.record().providerHandleChain.at(-1)?.handle).toMatchObject({ leafUuid: 'tip' }) - await proofs.rewind!.onProved!('kept') - expect(state.record().rewind).toMatchObject({ - phase: 'provider-succeeded', - hydrationVerified: true - }) - expect(state.record().providerHandleChain.at(-1)?.handle).toMatchObject({ leafUuid: 'kept' }) - expect( - claudeRewindAcquisitionProofs({ - store: state.store, - record: state.record(), - now: () => 3_001, - rewind: undefined - }) - ).toEqual({}) - }) - it('restores prepared recovery through ordinary proof without carrying rewind authorization', async () => { - const state = setup() - const proofs = claudeRewindAcquisitionProofs({ - store: state.store, - record: state.record(), - now: () => 3_000, - rewind: undefined - }) - expect(proofs.rewind).toBeUndefined() - expect(proofs.rewindRecovery?.leafUuid).toBe('tip') - expect(state.record().rewind?.phase).toBe('prepared') - await proofs.rewindRecovery!.onProved() - expect(state.record().rewind).toMatchObject({ phase: 'refused', retained: [] }) - expect(state.record().providerHandleChain.at(-1)?.handle).toMatchObject({ leafUuid: 'tip' }) - }) - it('refuses a proof checkpoint from a superseded acquisition', async () => { - const state = setup() - const proofs = claudeRewindAcquisitionProofs({ - store: state.store, - record: state.record(), - now: () => 3_000, - rewind: { previousLeafUuid: 'tip', targetUuid: 'kept' } - }) - state.setFence() - await expect(proofs.rewind!.onProved!('kept')).rejects.toThrow('checkpoint_stale') - expect(state.record().rewind?.phase).toBe('prepared') - expect(state.record().providerHandleChain.at(-1)?.handle).toMatchObject({ leafUuid: 'tip' }) - }) -}) diff --git a/src/main/native-chat/agent-session-wire/structured-rewind-claude-proof.ts b/src/main/native-chat/agent-session-wire/structured-rewind-claude-proof.ts deleted file mode 100644 index 87dd9dbb4e7..00000000000 --- a/src/main/native-chat/agent-session-wire/structured-rewind-claude-proof.ts +++ /dev/null @@ -1,69 +0,0 @@ -import { agentSessionProviderHandleChainHead } from '../../../shared/agent-session-provider-handle' -import type { AgentSessionRecord } from '../../../shared/agent-session-record' -import { claudeProviderHandleLink } from '../../claude/claude-structured-owner-identity' -import { recordAgentSessionProviderHandle } from '../../runtime/agent-session-provider-handle-transition' -import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' -import type { StructuredAgentSessionAcquireInput } from './structured-agent-session-adapter' - -/** Proof checkpoints survive failures later in acquisition, before an owner can be published. */ -export function claudeRewindAcquisitionProofs(input: { - store: Pick - record: AgentSessionRecord - rewind: StructuredAgentSessionAcquireInput['rewind'] - now: () => number -}): Pick { - const { record, store } = input - const pending = record.rewind - const head = agentSessionProviderHandleChainHead(record.providerHandleChain)?.handle - if ( - record.provider !== 'claude' || - pending?.phase !== 'prepared' || - head?.provider !== 'claude' - ) { - return input.rewind ? { rewind: input.rewind } : {} - } - const checkpoint = async (leafUuid?: string): Promise => { - await store.transitionHandoff(record.sessionId, (current) => { - if ( - current.lease.runtimeFence !== record.lease.runtimeFence || - current.rewind?.operationId !== pending.operationId || - current.rewind.callerKey !== pending.callerKey || - current.rewind.phase !== 'prepared' - ) { - throw new Error('agent_session_checkpoint_stale') - } - if (leafUuid === undefined) { - return { - ...current, - rewind: { ...pending, phase: 'refused', reason: 'outcome-unknown', retained: [] } - } - } - if (leafUuid !== input.rewind?.targetUuid) { - throw new Error('agent_session_rewind:proof-mismatch') - } - const observedAt = input.now() - return { - ...recordAgentSessionProviderHandle({ - record: current, - fence: record.lease.runtimeFence, - link: claudeProviderHandleLink({ - sessionId: head.sessionId, - leafUuid, - resumed: true, - fence: record.lease.runtimeFence, - observedAt - }), - now: observedAt - }), - rewind: { ...pending, phase: 'provider-succeeded', hydrationVerified: true } - } - }) - } - if (input.rewind) { - return { rewind: { ...input.rewind, onProved: checkpoint } } - } - if (!head.leafUuid) { - throw new Error('agent_session_rewind:invalid-target') - } - return { rewindRecovery: { leafUuid: head.leafUuid, onProved: () => checkpoint() } } -} diff --git a/src/main/native-chat/agent-session-wire/structured-rewind-recovery.ts b/src/main/native-chat/agent-session-wire/structured-rewind-recovery.ts index 4916710e9c1..42b6e8dd91d 100644 --- a/src/main/native-chat/agent-session-wire/structured-rewind-recovery.ts +++ b/src/main/native-chat/agent-session-wire/structured-rewind-recovery.ts @@ -10,6 +10,7 @@ import type { AgentSessionRecordStore } from '../../runtime/agent-session-record import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' import { AGENT_SESSION_HISTORY_MAX_PAGE_BYTES } from './agent-session-history-page-bounds' +import { rewindRefusal } from './structured-rewind-refusal' export function persistRewindRecord( store: AgentSessionRecordStore, @@ -25,6 +26,38 @@ export function persistRewindRecord( }) } +/** + * Claude rewind is unsupported, so a pending one (an older build's, or an interrupted one) is + * settled refused rather than proven. Bookkeeping only: the chat is already attached either way. + */ +async function settleUnsupportedClaudeRewind( + store: AgentSessionRecordStore, + sessionId: string, + fence: number, + rewind: AgentSessionRewindRecord +): Promise { + const refusal = rewindRefusal('unsupported').refusal + try { + await persistRewindRecord(store, sessionId, fence, { + ...rewind, + phase: 'refused', + reason: 'unsupported', + retained: [] + }) + await store.recordOperationOutcome({ + callerKey: rewind.callerKey, + operationId: rewind.operationId, + outcome: { status: 'failed', code: refusal.code, rewindReason: 'unsupported' } + }) + } catch (error) { + console.warn('[structured-rewind] pending Claude rewind was not settled:', { + sessionId, + operationId: rewind.operationId, + error + }) + } +} + /** Recovery observes provider state; it never repeats an ambiguous native mutation. */ export async function recoverStructuredRewind( store: AgentSessionRecordStore, @@ -39,6 +72,10 @@ export async function recoverStructuredRewind( return } const target = parseAgentJournalItemKey(rewind.providerItemId ?? rewind.itemId) + if (target?.provider === 'claude') { + await settleUnsupportedClaudeRewind(store, sessionId, fence, rewind) + return + } if (target?.provider === 'codex' && !rewind.hydrationVerified) { const recovered = await adapter?.recoverRewind?.({ sessionId, diff --git a/src/main/native-chat/session-file-resolver.test.ts b/src/main/native-chat/session-file-resolver.test.ts index 04946f5b464..ffc97c9efa1 100644 --- a/src/main/native-chat/session-file-resolver.test.ts +++ b/src/main/native-chat/session-file-resolver.test.ts @@ -1,12 +1,9 @@ import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { dirname, join } from 'node:path' -import { afterEach, describe, expect, it, vi } from 'vitest' +import { afterEach, describe, expect, it } from 'vitest' -import { - ClaudeTranscriptTailIncompleteError, - readClaudeTranscriptLeafWithReproof -} from '../claude/claude-transcript-branch-proof' +import { ClaudeTranscriptTailIncompleteError } from '../claude/claude-transcript-branch-proof' import { readClaudeTranscriptLeafUuid, resolveSessionFilePath } from './session-file-resolver' let tempRoots: string[] = [] @@ -336,69 +333,6 @@ describe('resolveSessionFilePath', () => { ) }) - it('does not re-prove a divergent sibling after the sampled cursor rejects', async () => { - const root = await makeRoot('orca-native-chat-resolve-claude-sibling-reproof-') - const transcript = join(root, 'transcript.jsonl') - await writeFile( - transcript, - [ - { type: 'user', uuid: 'root', parentUuid: null, sessionId: 'session-1' }, - { type: 'assistant', uuid: 'old', parentUuid: 'root', sessionId: 'session-1' }, - { type: 'assistant', uuid: 'new', parentUuid: 'root', sessionId: 'session-1' }, - { type: 'last-prompt', leafUuid: 'new', sessionId: 'session-1' } - ] - .map((record) => JSON.stringify(record)) - .join('\n'), - 'utf8' - ) - const calls: (string | null)[] = [] - const readTranscriptLeaf = async ({ - previousLeafUuid - }: { - previousLeafUuid: string | null - }) => { - calls.push(previousLeafUuid) - return readClaudeTranscriptLeafUuid(transcript, 'session-1', previousLeafUuid) - } - - await expect(readClaudeTranscriptLeafUuid(transcript, 'session-1', 'old')).rejects.toThrow( - 'sibling branch' - ) - - await expect( - readClaudeTranscriptLeafWithReproof({ - readTranscriptLeaf, - claudeConfigDir: '/accounts/claude', - providerSessionId: 'session-1', - previousLeafUuid: 'old' - }) - ).rejects.toThrow('sibling branch') - expect(calls).toEqual(['old']) - }) - - it('does not accept a divergent sibling after a truncated-tail reproof', async () => { - const calls: (string | null)[] = [] - const readTranscriptLeaf = vi.fn( - async ({ previousLeafUuid }: { previousLeafUuid: string | null }) => { - calls.push(previousLeafUuid) - if (calls.length === 1) { - throw new ClaudeTranscriptTailIncompleteError() - } - return 'divergent-sibling' - } - ) - - await expect( - readClaudeTranscriptLeafWithReproof({ - readTranscriptLeaf, - claudeConfigDir: '/accounts/claude', - providerSessionId: 'session-1', - previousLeafUuid: 'old' - }) - ).rejects.toBeInstanceOf(ClaudeTranscriptTailIncompleteError) - expect(calls).toEqual(['old']) - }) - it('globs Claude project subdirs for .jsonl', async () => { const root = await makeRoot('orca-native-chat-resolve-claude-') const claudeProjectsDir = join(root, 'claude-projects') diff --git a/src/main/runtime/agent-session-provider-handle-transition.test.ts b/src/main/runtime/agent-session-provider-handle-transition.test.ts index 19131562366..ddfaa163d73 100644 --- a/src/main/runtime/agent-session-provider-handle-transition.test.ts +++ b/src/main/runtime/agent-session-provider-handle-transition.test.ts @@ -4,7 +4,11 @@ import { agentSessionRecordFixture } from '../../shared/agent-session-record.test-fixture' import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' -import { recordAgentSessionProviderHandle } from './agent-session-provider-handle-transition' +import { isAgentSessionRecord } from '../../shared/agent-session-record' +import { + recordAgentSessionProviderHandle, + reviseAgentSessionClaudeResumePoint +} from './agent-session-provider-handle-transition' function resumedLink(fence: number): AgentSessionProviderHandleLink { return { @@ -46,3 +50,59 @@ describe('recordAgentSessionProviderHandle', () => { expect(next.lease).toMatchObject({ claimStatus: 'reserved', provenHandleLinkId: null }) }) }) + +describe('reviseAgentSessionClaudeResumePoint', () => { + const revise = (record = agentSessionRecordFixture(), leafUuid = 'leaf-2') => + reviseAgentSessionClaudeResumePoint({ + record, + fence: record.lease.runtimeFence, + providerSessionId: 'provider-session-alpha-1', + leafUuid, + now: 5_000 + }) + + it('moves the head leaf in place, turn after turn, without growing the chain', () => { + const record = agentSessionRecordFixture() + const second = revise(revise(record, 'leaf-2'), 'leaf-3') + expect(second.providerHandleChain).toHaveLength(record.providerHandleChain.length) + expect(second.providerHandleChain.at(-1)).toMatchObject({ + linkId: 'link-1', + origin: 'created', + handle: { leafUuid: 'leaf-3' } + }) + expect(second.lease.provenHandleLinkId).toBe('link-1') + expect(isAgentSessionRecord(second)).toBe(true) + }) + + it('refuses a stale owner, a released lease, and a head minted by another owner', () => { + const record = agentSessionRecordFixture() + expect(() => + reviseAgentSessionClaudeResumePoint({ + record, + fence: record.lease.runtimeFence - 1, + providerSessionId: 'provider-session-alpha-1', + leafUuid: 'leaf-2', + now: 5_000 + }) + ).toThrow('agent_session_stale_fence') + expect(() => + revise(agentSessionRecordFixture(agentSessionLeaseFixture({ claimStatus: 'released' }))) + ).toThrow('agent_session_ownership_unknown') + const later = agentSessionRecordFixture(agentSessionLeaseFixture({ runtimeFence: 9 })) + expect(() => + revise({ + ...later, + providerHandleChain: [{ ...later.providerHandleChain[0]!, mintedAtFence: 7 }] + }) + ).toThrow('agent_session_provider_handle_invalid') + expect(() => + reviseAgentSessionClaudeResumePoint({ + record, + fence: record.lease.runtimeFence, + providerSessionId: 'another-provider-session', + leafUuid: 'leaf-2', + now: 5_000 + }) + ).toThrow('agent_session_provider_handle_invalid') + }) +}) diff --git a/src/main/runtime/agent-session-provider-handle-transition.ts b/src/main/runtime/agent-session-provider-handle-transition.ts index 474e4be7135..f0408901868 100644 --- a/src/main/runtime/agent-session-provider-handle-transition.ts +++ b/src/main/runtime/agent-session-provider-handle-transition.ts @@ -1,5 +1,7 @@ import { + agentSessionProviderHandleChainHead, appendAgentSessionProviderHandleLink, + isAgentSessionProviderHandleChain, type AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' import type { AgentSessionRecord } from '../../shared/agent-session-record' @@ -39,3 +41,43 @@ export function recordAgentSessionProviderHandle(args: { updatedAt: args.now } } + +/** + * Advance the live owner's Claude resume point in place. The head link this owner minted keeps + * its id and provenance; only its leaf moves, so a long conversation does not grow the chain. + */ +export function reviseAgentSessionClaudeResumePoint(args: { + record: AgentSessionRecord + fence: number + providerSessionId: string + leafUuid: string + now: number +}): AgentSessionRecord { + const { record } = args + if (record.lease.runtimeFence !== args.fence) { + throw new Error('agent_session_stale_fence') + } + if (record.lease.claimStatus !== 'live') { + throw new Error('agent_session_ownership_unknown') + } + const head = agentSessionProviderHandleChainHead(record.providerHandleChain) + if ( + head?.handle.provider !== 'claude' || + head.handle.sessionId !== args.providerSessionId || + head.mintedAtFence !== args.fence + ) { + throw new Error('agent_session_provider_handle_invalid') + } + if (head.handle.leafUuid === args.leafUuid) { + return record + } + const providerHandleChain = [ + ...record.providerHandleChain.slice(0, -1), + { ...head, handle: { ...head.handle, leafUuid: args.leafUuid }, observedAt: args.now } + ] + // The revised chain must still read back as the same persisted chain. + if (!isAgentSessionProviderHandleChain(providerHandleChain)) { + throw new Error('agent_session_provider_handle_invalid') + } + return { ...record, providerHandleChain, updatedAt: args.now } +} diff --git a/src/main/runtime/claude-structured-session-integration.test.ts b/src/main/runtime/claude-structured-session-integration.test.ts index 15a901b0ea1..fe56b2c0dca 100644 --- a/src/main/runtime/claude-structured-session-integration.test.ts +++ b/src/main/runtime/claude-structured-session-integration.test.ts @@ -760,29 +760,34 @@ describe('a structured Claude session over agentSession.*', () => { } } } + // A completed turn advances the durable resume point in place while the owner is live. expect(host.deps.store.getRecord(SESSION).providerHandleChain.at(-1)?.handle).toMatchObject({ provider: 'claude', - leafUuid: null + leafUuid: 'assistant-leaf' }) + // Claude's marker names a hook row after a turn; close must never adopt it. + readClaudeTranscriptLeafUuid.mockClear().mockResolvedValue('stop-hook-summary-row') const old = claude.live() const resumed = await ok<{ fence: number }>('agentSession.ensure', ensureParams(created.fence)) expect(resumed.fence).toBe(created.fence + 1) expect(old.closed).toBe(true) - expect(resolveSessionFilePath).toHaveBeenCalledWith('claude', PROVIDER_SESSION, { - claudeProjectsDir: join(root, 'claude-home', 'projects') - }) - expect(claude.live().launch.options).toMatchObject({ - resume: PROVIDER_SESSION, - resumeSessionAt: 'provider-opened-assistant' - }) - expect(host.deps.store.getRecord(SESSION).providerHandleChain.at(-1)).toMatchObject({ - handle: { - provider: 'claude', - sessionId: PROVIDER_SESSION, - leafUuid: 'provider-opened-assistant' - }, + // Claude owns where the conversation continues; the stored leaf is the last completed turn. + expect(claude.live().launch.options).toMatchObject({ resume: PROVIDER_SESSION }) + expect(claude.live().launch.options).not.toHaveProperty('resumeSessionAt') + const lastCompletedTurn = { + handle: { provider: 'claude', sessionId: PROVIDER_SESSION, leafUuid: 'assistant-leaf' }, origin: 'resumed' - }) + } + expect(host.deps.store.getRecord(SESSION).providerHandleChain.at(-1)).toMatchObject( + lastCompletedTurn + ) + // Open, close, open with no turn in between keeps that leaf and never reads the transcript. + const reopened = await ok<{ fence: number }>('agentSession.ensure', ensureParams(resumed.fence)) + expect(reopened.fence).toBe(resumed.fence + 1) + expect(host.deps.store.getRecord(SESSION).providerHandleChain.at(-1)).toMatchObject( + lastCompletedTurn + ) + expect(readClaudeTranscriptLeafUuid).not.toHaveBeenCalled() }) it('completes a scripted native to TUI to native cycle with provider-history rehydration', async () => { diff --git a/src/main/runtime/structured-claude-pending-rewind.test.ts b/src/main/runtime/structured-claude-pending-rewind.test.ts new file mode 100644 index 00000000000..c7774ffa7fd --- /dev/null +++ b/src/main/runtime/structured-claude-pending-rewind.test.ts @@ -0,0 +1,194 @@ +// Claude rewind is reported unsupported until it returns through a fork. These pin that a rewind +// RPC leaves nothing behind, and that a pending rewind persisted by an older build never strands +// the chat: the next attach resumes by session id and settles the rewind as refused. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' +import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' +import type { AgentSessionRecord } from '../../shared/agent-session-record' +import { claudeSessionIdForOrcaSession } from '../claude/claude-structured-launch-resolution' +import { fakeClaude } from '../claude/claude-structured-session-test-support' +import { StructuredAgentSessionAdapterRouter } from '../native-chat/agent-session-wire/structured-agent-session-adapter-router' +import { StructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-host' +import { + HOST_TEST_NOW, + HOST_TEST_SESSION, + hostTestAttachParams, + hostTestOperationId, + resetHostTestOperationIds +} from '../native-chat/agent-session-wire/structured-agent-session-host-test-data' +import { AgentSessionRecordStore } from './agent-session-record-store' +import { createStructuredClaudeRuntimeAdapter } from './structured-claude-runtime-adapter' + +const caller = { callerKey: 'desktop' } +const PROVIDER_SESSION_ID = claudeSessionIdForOrcaSession(HOST_TEST_SESSION) +const TARGET = agentJournalItemKey({ + provider: 'claude', + sessionId: PROVIDER_SESSION_ID, + uuid: 'kept' +}) +let directory: string +let store: AgentSessionRecordStore +let claude: ReturnType +let adapter: ReturnType +let host: StructuredAgentSessionHost + +function attachParams(fence: number | null) { + return hostTestAttachParams(fence, { + provider: 'claude', + agent: 'claude', + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: join(directory, 'claude-home') }, + providerHandle: { kind: 'claude', sessionId: PROVIDER_SESSION_ID, leafUuid: 'turn-end' } + }) +} + +function rewindParams(fence: number) { + const fields = { itemId: TARGET, expectedEpoch: 'epoch-before' } + return { + ...fields, + envelope: { + sessionId: HOST_TEST_SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: fence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.rewind', + sessionId: HOST_TEST_SESSION, + fields + }) + } + } +} + +const fence = (): number => store.getRecord(HOST_TEST_SESSION)!.lease.runtimeFence + +/** What an older build left behind: an admitted rewind whose outcome was never recorded. */ +async function seedPendingRewind(phase: 'prepared' | 'provider-succeeded') { + const request = rewindParams(fence()) + await store.admitMutationOperation({ + callerKey: caller.callerKey, + envelope: request.envelope, + hostFingerprint: request.envelope.payloadFingerprint, + now: HOST_TEST_NOW + }) + await store.recordOperationOutcome({ + callerKey: caller.callerKey, + operationId: request.envelope.clientOperationId, + outcome: { status: 'unknown' } + }) + await store.transitionHandoff(HOST_TEST_SESSION, (record) => ({ + ...record, + rewind: { + operationId: request.envelope.clientOperationId, + callerKey: caller.callerKey, + itemId: TARGET, + providerItemId: TARGET, + expectedEpoch: request.expectedEpoch, + phase, + retained: [] + } + })) + return request +} + +async function reattach() { + await host.close(HOST_TEST_SESSION) + expect(await host.attach(caller, attachParams(fence()))).toMatchObject({ ok: true }) +} + +beforeEach(async () => { + resetHostTestOperationIds() + directory = await mkdtemp(join(tmpdir(), 'orca-claude-pending-rewind-')) + store = await AgentSessionRecordStore.open({ + directory: join(directory, 'store'), + hostId: 'local' + }) + claude = fakeClaude({ initSessionId: PROVIDER_SESSION_ID }) + adapter = createStructuredClaudeRuntimeAdapter({ + store, + resolveWorkspacePath: async (id) => `/repos/${id}`, + resolveClaudeCommand: () => '/usr/local/bin/claude', + resolveClaudeAuthPolicy: () => ({ stripAuthEnv: false }), + openClaudeConnection: claude.openConnection, + readProcessStartTime: async () => HOST_TEST_NOW, + onUnexpectedExit: () => {} + }) + host = new StructuredAgentSessionHost({ + store, + // Only Claude sessions are attached here; the router supplies the production create gate. + adapter: new StructuredAgentSessionAdapterRouter({ claude: adapter, codex: adapter }, () => + adapter.closeAll() + ), + journalRoot: directory, + claimKeyId: 'key', + now: () => HOST_TEST_NOW, + probeOwner: async () => ({ outcome: 'exit-observed' }) + }) + expect(await host.attach(caller, attachParams(null))).toMatchObject({ ok: true }) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await host.flushAllStreamedEvents() + await adapter.closeAll() + await rm(directory, { recursive: true, force: true }) +}) + +describe('Claude rewind is unsupported', () => { + it('refuses a rewind RPC before writing any rewind record', async () => { + expect(adapter.rewindSupport(HOST_TEST_SESSION)).toEqual({ + supported: false, + reason: 'unsupported' + }) + expect(await host.rewind(caller, rewindParams(fence()))).toMatchObject({ + ok: false, + refusal: { rewindReason: 'unsupported' } + }) + expect(store.getRecord(HOST_TEST_SESSION)?.rewind).toBeUndefined() + }) + + it.each(['prepared', 'provider-succeeded'] as const)( + 'settles an older build’s %s rewind as refused and resumes by session id', + async (phase) => { + const pending = await seedPendingRewind(phase) + await reattach() + + expect(claude.connections.at(-1)?.launch.options).toMatchObject({ + resume: PROVIDER_SESSION_ID + }) + expect(claude.connections.at(-1)?.launch.options).not.toHaveProperty('resumeSessionAt') + const record: AgentSessionRecord | null = store.getRecord(HOST_TEST_SESSION) + expect(record?.rewind).toMatchObject({ phase: 'refused', reason: 'unsupported' }) + // The operation resolves too: a retry is refused as unsupported, not as an unknown outcome. + expect(await host.rewind(caller, pending)).toMatchObject({ + ok: false, + refusal: { rewindReason: 'unsupported' } + }) + } + ) + + it('still attaches when settling the pending rewind fails, and logs it', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + await seedPendingRewind('prepared') + const transition = store.transitionHandoff.bind(store) + vi.spyOn(store, 'transitionHandoff').mockImplementation((sessionId, apply) => + transition(sessionId, (record) => { + const next = apply(record) + if (next.rewind?.phase === 'refused') { + throw new Error('record write failed') + } + return next + }) + ) + + await reattach() + + expect(store.getRecord(HOST_TEST_SESSION)?.rewind?.phase).toBe('prepared') + expect(warn).toHaveBeenCalledWith( + '[structured-rewind] pending Claude rewind was not settled:', + expect.objectContaining({ sessionId: HOST_TEST_SESSION, error: expect.any(Error) }) + ) + }) +}) diff --git a/src/main/runtime/structured-claude-runtime-adapter.ts b/src/main/runtime/structured-claude-runtime-adapter.ts index 23add8ed6dc..b4aff275ace 100644 --- a/src/main/runtime/structured-claude-runtime-adapter.ts +++ b/src/main/runtime/structured-claude-runtime-adapter.ts @@ -1,8 +1,6 @@ import type { PermissionMode } from '@anthropic-ai/claude-agent-sdk' -import { proveClaudeTranscriptBranch } from '../claude/claude-transcript-branch-proof' import type { AgentSessionRecord } from '../../shared/agent-session-record' import type { AgentSessionBackgroundTaskState } from '../../shared/agent-session-wire' -import { join } from 'node:path' import { resolveClaudeCommand } from '../codex-cli/command' import type { ClaudeStructuredAuthPolicy } from '../claude-accounts/claude-structured-auth-policy' import { createClaudeStructuredLaunchResolver } from '../claude/claude-structured-launch-resolution' @@ -13,10 +11,9 @@ import { import { claudeProviderHandleLink } from '../claude/claude-structured-owner-identity' import type { StructuredAgentSessionLifecycleEvent } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import { - readClaudeTranscriptLeafUuid, - resolveSessionFilePath -} from '../native-chat/session-file-resolver' -import { recordAgentSessionProviderHandle } from './agent-session-provider-handle-transition' + recordAgentSessionProviderHandle, + reviseAgentSessionClaudeResumePoint +} from './agent-session-provider-handle-transition' import type { ClaudeManagedAccountGateSettings } from '../native-chat/claude-structured-managed-account-support' import type { AgentSessionRecordStore } from './agent-session-record-store' @@ -82,28 +79,16 @@ export function createStructuredClaudeRuntimeAdapter( }) ) }, - readTranscriptLeaf: async ({ - providerSessionId, - previousLeafUuid, - intentionalRewindUuid, - claudeConfigDir - }) => { - const transcriptPath = await resolveSessionFilePath('claude', providerSessionId, { - claudeProjectsDir: join(claudeConfigDir, 'projects') - }) - if (transcriptPath && intentionalRewindUuid !== undefined) { - return ( - await proveClaudeTranscriptBranch({ - transcriptPath, - providerSessionId, - previousLeafUuid, - intentionalRewindUuid - }) - ).leafUuid - } - return transcriptPath - ? await readClaudeTranscriptLeafUuid(transcriptPath, providerSessionId, previousLeafUuid) - : null + persistResumePoint: async ({ sessionId, providerSessionId, leafUuid, fence }) => { + await store.transitionHandoff(sessionId, (record: AgentSessionRecord) => + reviseAgentSessionClaudeResumePoint({ + record, + fence, + providerSessionId, + leafUuid, + now: Date.now() + }) + ) }, onEvent: (event) => { if (