From 8ff0056edc1b820a7d9d5d7f2a7e6ece223a18ef Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 28 Sep 2026 03:03:11 -0700 Subject: [PATCH] perf(native-chat): parse a Claude transcript on attach only when a send is in doubt Every Claude chat attach read the whole provider transcript before checking whether any message needed deciding: once for the branch proof, once more to replay the anchored window, and a third time for the id lookup, each parsing every line. Most attaches have nothing in doubt and paid for all of it. The attach now samples only what must be fixed before a new child starts: the running-turn fact, the transcript path, and its size and file identity (one stat). Reconciliation reads the sample only after the journal shows a send in doubt, and that read is one pass over exactly the pinned bytes, parsing each line once for the branch proof, the anchored window and the id lookup alike. Bytes appended after the pin, which the new child may have written, are never read, and a file replaced or truncated since the pin is not read as evidence. Linear: STA-7434 --- ...laude-history-window-source-budget.test.ts | 91 +++--- .../claude-structured-history-window.test.ts | 18 +- .../claude-structured-history-window.ts | 280 ++++++++---------- .../claude-structured-session-adapter.ts | 6 +- .../claude/claude-transcript-branch-graph.ts | 19 +- .../claude/claude-transcript-branch-proof.ts | 230 ++++++-------- ...claude-transcript-branch-streaming.test.ts | 260 +++++++--------- .../journal-restart-reconciliation.test.ts | 39 ++- .../journal-restart-reconciliation.ts | 22 +- .../journal-submission-reconciler.ts | 8 + ...-agent-session-accept-then-deliver.test.ts | 6 +- ...-agent-session-acquisition-options.test.ts | 10 +- ...structured-agent-session-adapter-router.ts | 4 +- .../structured-agent-session-adapter.ts | 12 +- .../structured-agent-session-attach-flow.ts | 35 ++- ...gent-session-attach-reconciliation.test.ts | 15 +- ...nt-session-attach-transcript-reads.test.ts | 194 ++++++++++++ .../structured-agent-session-attach.ts | 22 +- ...ssion-restart-interruption-test-harness.ts | 20 +- 19 files changed, 729 insertions(+), 562 deletions(-) create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-attach-transcript-reads.test.ts 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 90d74b94e373..7f6f6144e4f3 100644 --- a/src/main/claude/claude-history-window-source-budget.test.ts +++ b/src/main/claude/claude-history-window-source-budget.test.ts @@ -1,4 +1,4 @@ -// The streaming contract the history window now reads under: two bounded passes +// The streaming contract the history window now reads under: one bounded pass // over ONE pinned snapshot, and no whole-file buffer at any point. The previous // contract was a single bounded read, which made an oversized transcript report // an inconsistent boundary — that answer left reconciliation permanently @@ -35,9 +35,9 @@ vi.mock('node:fs/promises', async (importOriginal) => { } state.opens += 1 return { - stat: async () => { - const snapshot = await handle.stat() - state.observedStatBytes = snapshot.size + stat: async (options?: Parameters[0]) => { + const snapshot = await handle.stat(options) + state.observedStatBytes = Number(snapshot.size) if (state.growth) { const growth = state.growth state.growth = '' @@ -69,7 +69,9 @@ vi.mock('node:fs/promises', async (importOriginal) => { }) import { appendFile, mkdtemp, rm, writeFile } from 'node:fs/promises' -import { readClaudeProviderHistoryWindow } from './claude-structured-history-window' +import { readClaudeProviderHistory } from './claude-structured-history-window' +import { pinClaudeTranscript } from './claude-transcript-branch-proof' +import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' const LEGACY_LIMIT = 16 * 1024 * 1024 const SOURCE = `${[ @@ -93,9 +95,8 @@ const SOURCE = `${[ .join('\n')}\n` let directory = '' -const read = (previousLeafUuid: string | null = 'anchor') => - readClaudeProviderHistoryWindow({ - transcriptPath: state.path, +const read = async (previousLeafUuid: string | null = 'anchor') => + readClaudeProviderHistory(await pinClaudeTranscript(state.path), { providerSessionId: 'provider', previousLeafUuid, sessionId: 'orca', @@ -155,30 +156,13 @@ describe('Claude provider history source budget', () => { expect(result.items.map((item) => item.providerItemId)).toEqual(['latest']) }) - it('streams two passes over one descriptor rather than one whole-file read', async () => { - await read() - // One open for both passes: pass 2 must see the bytes pass 1 vouched for. - expect(state.opens).toBe(1) - expect(state.streams).toBe(2) - expect(state.bytesRead).toBe(2 * Buffer.byteLength(SOURCE)) - }) - - it('opens no second pass when nothing followed the anchor', async () => { - await writeFile( - state.path, - SOURCE.split('\n') - // Only the record — `"latest"` alone would take the marker row with it, - // leaving an unprovable transcript that satisfies this test vacuously. - .filter((line) => !line.includes('"uuid":"latest"')) - .join('\n') - .replace('"leafUuid":"latest"', '"leafUuid":"anchor"') - ) - + it('streams one pass for the window and the recorded history together', async () => { const result = await read() - - expect(result.boundaryConsistent).toBe(true) - expect(result.items).toEqual([]) + expect(result.recorded?.itemIds.size).toBe(2) + // The proof, the window and the id lookup all come from the same bytes, read once. + expect(state.opens).toBe(1) expect(state.streams).toBe(1) + expect(state.bytesRead).toBe(Buffer.byteLength(SOURCE)) }) it('resolves a source past the legacy whole-file limit instead of refusing it', async () => { @@ -191,7 +175,7 @@ describe('Claude provider history source budget', () => { expect(result.boundaryConsistent).toBe(true) expect(result.items.map((item) => item.providerItemId)).toEqual(['latest']) - expect(state.bytesRead).toBeGreaterThan(2 * LEGACY_LIMIT) + expect(state.bytesRead).toBeGreaterThan(LEGACY_LIMIT) }) it('keeps resident bytes bounded by the chunk size, not the file size', async () => { @@ -209,7 +193,7 @@ describe('Claude provider history source budget', () => { expect((await read()).boundaryConsistent).toBe(false) }) - it('replays a concurrent repair at the grown size, not the pinned one', async () => { + it('never reads what was appended after the pin, even to finish a torn tail', async () => { const grown = `${JSON.stringify({ type: 'user', uuid: 'grown', @@ -218,20 +202,28 @@ describe('Claude provider history source budget', () => { message: { role: 'user', content: 'appended' } })}\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() - - // The retry re-runs BOTH passes at the new size: a pass 2 left at the old - // size would silently drop the record the proof just accepted. - expect(result.boundaryConsistent).toBe(true) - expect(result.items.map((item) => item.providerItemId)).toEqual(['latest', 'grown']) - // Failed graph pass, re-run graph pass, replay pass — all on the one descriptor. - expect(state.streams).toBe(3) - expect(state.opens).toBe(1) + const pinned = await pinClaudeTranscript(state.path) + // A child started after the pin may be the writer, so its bytes are no evidence. + await appendFile(state.path, grown.slice(torn)) + + const result = await readClaudeProviderHistory(pinned, { + providerSessionId: 'provider', + previousLeafUuid: 'anchor', + sessionId: 'orca', + turnInFlight: false + }) + + expect(result.boundaryConsistent).toBe(false) + // The torn line may be the record, so the file does not prove its absence. + const grownKey = agentJournalItemKey({ + provider: 'claude', + sessionId: 'provider', + uuid: 'grown' + }) + expect(result.recorded?.provesAbsenceOf(grownKey)).toBe(false) + expect(state.streams).toBe(1) + expect(state.bytesRead).toBe(pinned.size) }) it('preserves the inconsistent result on a read error', async () => { @@ -239,9 +231,10 @@ describe('Claude provider history source budget', () => { expect((await read()).boundaryConsistent).toBe(false) }) - it('does not open a source without an anchor', async () => { - expect((await read(null)).boundaryConsistent).toBe(false) - expect(state.opens).toBe(0) - expect(state.bytesRead).toBe(0) + it('still reads the recorded history, once, without an anchor', async () => { + const result = await read(null) + expect(result.boundaryConsistent).toBe(false) + expect(result.recorded?.itemIds.size).toBe(2) + expect(state.streams).toBe(1) }) }) diff --git a/src/main/claude/claude-structured-history-window.test.ts b/src/main/claude/claude-structured-history-window.test.ts index c23df513aca8..3f4a550cab46 100644 --- a/src/main/claude/claude-structured-history-window.test.ts +++ b/src/main/claude/claude-structured-history-window.test.ts @@ -13,7 +13,7 @@ import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key import type { AgentJournalSubmission } from '../../shared/agent-session-journal-types' import { claudeProviderHistoryWindowFromJsonl, - resolveClaudeProviderHistoryWindow + sampleClaudeProviderHistory } from './claude-structured-history-window' const PROVIDER_SESSION = 'provider-1' @@ -78,7 +78,7 @@ describe('claudeProviderHistoryWindowFromJsonl', () => { 'utf8' ) - const window = await resolveClaudeProviderHistoryWindow({ + const sample = await sampleClaudeProviderHistory({ identity: { sessionId: ORCA_SESSION, workspaceId: 'workspace-1', @@ -89,6 +89,7 @@ describe('claudeProviderHistoryWindowFromJsonl', () => { accountHomePath: accountHome, hasLiveSession: false }) + const window = await sample?.read() expect(window?.items.map((item) => item.providerItemId)).toEqual(['u-1']) }) @@ -268,8 +269,7 @@ describe('claudeProviderHistoryWindowFromJsonl', () => { }) it('refuses a malformed line rather than skipping past it into the window', () => { - // The branch proof runs first and throws on any unparseable record; the - // replay that follows only tolerates them because that pass already ran. + // An unparseable line ends the proof, though the read goes on for the recorded history. const contents = jsonl([ANCHOR, prompt('u-1', 'anchor', 'ship it')], 'u-1').replace( '{"type":"last-prompt"', 'not json\n{"type":"last-prompt"' @@ -283,6 +283,16 @@ describe('claudeProviderHistoryWindowFromJsonl', () => { expect(read(contents, 'anchor', true).turnInFlight).toBe(true) }) + + it('keeps a running turn running when the boundary cannot be vouched for', () => { + // A send absent from the whole file is still no proof while a turn is running. + const contents = jsonl([ANCHOR], 'anchor') + + expect(read(contents, null, true)).toMatchObject({ + boundaryConsistent: false, + turnInFlight: true + }) + }) }) describe('a crash between Claude saving a prompt and Orca recording its echo', () => { diff --git a/src/main/claude/claude-structured-history-window.ts b/src/main/claude/claude-structured-history-window.ts index 83ab2705d9de..35cdbb71a5d3 100644 --- a/src/main/claude/claude-structured-history-window.ts +++ b/src/main/claude/claude-structured-history-window.ts @@ -14,27 +14,27 @@ // of those makes absence meaningless. Failing it reports an inconsistent // boundary rather than an empty window, because the two decide opposite things. -import { createReadStream } from 'node:fs' import { join } from 'node:path' import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types' import { resolveSessionFilePath } from '../native-chat/session-file-resolver' import type { ProviderHistoryItem, - ProviderHistoryWindow, - ProviderRecordedHistory + ProviderHistorySample, + ProviderHistoryWindow } from '../native-chat/agent-session-journal/journal-submission-reconciler' import { agentJournalItemKey, parseAgentJournalItemKey } from '../../shared/agent-session-journal-item-key' -import { splitTranscriptStreamLines } from '../native-chat/transcript-stream-lines' import { claudeContentBlocks } from '../native-chat/transcript-record-blocks' import { isKnownHarnessInjectedUserTurnText } from '../../shared/harness-injected-user-turns' import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' import type { NativeChatBlock } from '../../shared/native-chat-types' import { - replayClaudeTranscriptBranchAncestry, - replayClaudeTranscriptBranchAncestryFromJsonl + createClaudeBranchAncestryPass, + pinClaudeTranscript, + readPinnedClaudeTranscript, + type ClaudeTranscriptSnapshot } from './claude-transcript-branch-proof' /** The legacy-import bound, now applied PER RECORD rather than per file. The @@ -42,12 +42,6 @@ import { * between a pathological row and the whole file being resident. */ const MAX_HISTORY_WINDOW_RECORD_BYTES = 16 * 1024 * 1024 -const INCONSISTENT: ProviderHistoryWindow = { - items: [], - boundaryConsistent: false, - turnInFlight: false -} - type TranscriptRecord = Record function stringField(source: unknown, key: string): string | null { @@ -134,14 +128,6 @@ function claudePromptBlocks(record: TranscriptRecord): NativeChatBlock[] | null return blocks.map((block, index) => ({ ...block, text: rawTexts[index]! })) } -/** The anchored window's prompts: harness-shaped text is left out, so machinery cannot claim a - * send by content. A send handed over under an id never reaches this filter. */ -function claudeUserPromptBlocks(record: TranscriptRecord): NativeChatBlock[] | null { - const blocks = claudePromptBlocks(record) - const [first] = blocks ?? [] - return first?.type === 'text' && !isKnownHarnessInjectedUserTurnText(first.text) ? blocks : null -} - /** * The digest the submission row is GUARANTEED to carry. `admitAndRunAgentSessionMutation` * recomputes this exact call over the send's own body and refuses the send on a @@ -169,46 +155,37 @@ type HistoryWindowInput = { } /** - * Sink for the ancestry replay. Only the fingerprinted item survives the call, - * so the window holds at most one small row per anchor..leaf record — never the - * prompt bodies it was computed from. - */ -function createWindowCollector(input: HistoryWindowInput) { - const byUuid = new Map() - return { byUuid, onAncestorRecord } - - function onAncestorRecord(record: TranscriptRecord, uuid: string): void { - const blocks = claudeUserPromptBlocks(record) - if (!blocks) { - return - } - byUuid.set(uuid, { - providerItemId: uuid, - // Claude echoes no client message id, so identity matching reduces to the - // fingerprint pass; the reconciler treats that as the weakest evidence. - clientMessageId: null, - payloadFingerprint: promptFingerprint(input.sessionId, blocks), - identity: { - provider: 'claude', - sessionId: stringField(record, 'sessionId') ?? input.providerSessionId, - uuid - } - }) - } -} - -/** - * Every user record in the file, wherever it sits: a send handed over under an id is found by - * that id, so no anchor is needed. Absence counts only when every line of this session's file - * was read; any line that does not parse may be the missing record. + * Every consumer of one transcript read, each line parsed once: + * + * - The anchored window. The branch proof runs over every line, and each prompt + * is kept as one small fingerprinted row under the first line carrying its + * uuid; once the proof has named the chain, the window is those rows on it. + * - The recorded history: every user record in the file, wherever it sits. A + * send handed over under an id is found by that id, so no anchor is needed. + * Absence counts only when every line of this session's file was read; any + * line that does not parse may be the missing record. */ -function createRecordedCollector(input: HistoryWindowInput) { +function createClaudeHistoryPass( + input: HistoryWindowInput, + onUnprovable: (error: unknown) => void = () => {} +) { + const anchorUuid = input.previousLeafUuid + const branch = anchorUuid + ? createClaudeBranchAncestryPass({ + providerSessionId: input.providerSessionId, + previousLeafUuid: anchorUuid, + ancestryAnchorUuid: anchorUuid + }) + : null + const prompts = new Map() const itemIds = new Set() const itemIdsByFingerprint = new Map() let whole = true + let lineIndex = 0 return { add, finish } - function add(line: string): void { + function add(line: string, terminated: boolean): void { + const index = lineIndex++ if (!line.trim()) { return } @@ -217,8 +194,10 @@ function createRecordedCollector(input: HistoryWindowInput) { record = JSON.parse(line) } catch { whole = false + branch?.reject(terminated) return } + const firstSeenUuid = branch?.add(record, index) ?? null const uuid = stringField(record, 'uuid') if (!uuid || !record || typeof record !== 'object' || Array.isArray(record)) { return @@ -232,89 +211,97 @@ function createRecordedCollector(input: HistoryWindowInput) { const itemId = agentJournalItemKey({ provider: 'claude', sessionId, uuid }) itemIds.add(itemId) const blocks = claudePromptBlocks(row) - if (blocks) { - const fingerprint = promptFingerprint(input.sessionId, blocks) - itemIdsByFingerprint.set(fingerprint, [ - ...(itemIdsByFingerprint.get(fingerprint) ?? []), - itemId - ]) + if (!blocks) { + return + } + const payloadFingerprint = promptFingerprint(input.sessionId, blocks) + const sameContent = itemIdsByFingerprint.get(payloadFingerprint) + if (sameContent) { + sameContent.push(itemId) + } else { + itemIdsByFingerprint.set(payloadFingerprint, [itemId]) + } + // Harness-shaped text is left out of the window, so machinery cannot claim a + // send by content. A send handed over under an id never reaches this filter. + const [first] = blocks + if ( + firstSeenUuid && + first?.type === 'text' && + !isKnownHarnessInjectedUserTurnText(first.text) + ) { + prompts.set(firstSeenUuid, { + providerItemId: firstSeenUuid, + // Claude echoes no client message id, so identity matching reduces to the + // fingerprint pass; the reconciler treats that as the weakest evidence. + clientMessageId: null, + payloadFingerprint, + identity: { provider: 'claude', sessionId, uuid: firstSeenUuid } + }) } } - function finish(): ProviderRecordedHistory { + function finish(): ProviderHistoryWindow { const provable = whole return { - itemIds, - itemIdsByFingerprint, - provesAbsenceOf: (itemId) => { - const identity = provable ? parseAgentJournalItemKey(itemId) : null - return identity?.provider === 'claude' && identity.sessionId === input.providerSessionId + ...anchoredWindow(), + recorded: { + itemIds, + itemIdsByFingerprint, + provesAbsenceOf: (itemId) => { + const identity = provable ? parseAgentJournalItemKey(itemId) : null + return identity?.provider === 'claude' && identity.sessionId === input.providerSessionId + } } } } -} -/** The chain is leaf first; a history window is oldest first. */ -function windowFromChain( - chain: readonly string[], - byUuid: Map, - turnInFlight: boolean -): ProviderHistoryWindow { - const items: ProviderHistoryItem[] = [] - for (const uuid of chain.toReversed()) { - const item = byUuid.get(uuid) - if (item) { - items.push(item) + function anchoredWindow(): ProviderHistoryWindow { + if (!branch) { + return inconsistent(input) + } + let chain: string[] + try { + chain = branch.finish().chain + } catch (error) { + // Every failure mode here — missing ancestor, sibling branch, compacted + // cursor, torn tail — is a boundary we cannot vouch for. + onUnprovable(error) + return inconsistent(input) } + // The chain is leaf first; a history window is oldest first. + const items = chain.toReversed().flatMap((uuid) => prompts.get(uuid) ?? []) + return { items, boundaryConsistent: true, turnInFlight: input.turnInFlight } } - return { items, boundaryConsistent: true, turnInFlight } } -export function claudeProviderHistoryWindowFromJsonl( - input: HistoryWindowInput & { contents: string } -): ProviderHistoryWindow { - const recorded = createRecordedCollector(input) - for (const line of input.contents.split('\n')) { - recorded.add(line) - } - return { ...anchoredWindowFromJsonl(input), recorded: recorded.finish() } +/** Unvouched-for is not empty; a running turn stays a running turn either way. */ +function inconsistent(input: HistoryWindowInput): ProviderHistoryWindow { + return { items: [], boundaryConsistent: false, turnInFlight: input.turnInFlight } } -function anchoredWindowFromJsonl( +export function claudeProviderHistoryWindowFromJsonl( input: HistoryWindowInput & { contents: string } ): ProviderHistoryWindow { - const ancestryAnchorUuid = input.previousLeafUuid - if (!ancestryAnchorUuid) { - return INCONSISTENT - } - const collector = createWindowCollector(input) - try { - const { chain } = replayClaudeTranscriptBranchAncestryFromJsonl({ - contents: input.contents, - providerSessionId: input.providerSessionId, - previousLeafUuid: ancestryAnchorUuid, - ancestryAnchorUuid, - onAncestorRecord: collector.onAncestorRecord - }) - return windowFromChain(chain, collector.byUuid, input.turnInFlight) - } catch { - // Every failure mode here — missing ancestor, sibling branch, compacted - // cursor, torn tail — is a boundary we cannot vouch for. - return INCONSISTENT + const pass = createClaudeHistoryPass(input) + const lines = input.contents.split('\n') + for (const [index, line] of lines.entries()) { + pass.add(line, index < lines.length - 1) } + return pass.finish() } /** - * The window for one attached session: resolve the provider's transcript, then - * read it against the handle's durable leaf. A live child means a send queued - * behind its running turn is not in the file yet, so liveness is carried in - * rather than assumed — only the adapter's session map can answer it. + * Sample one attached session's history: resolve the provider's transcript and + * pin its size, and nothing more. A live child means a send queued behind its + * running turn is not in the file yet, so liveness is carried in rather than + * assumed — only the adapter's session map can answer it, and only before a + * new child joins that map. Parsing waits for `read`, over the pinned bytes. */ -export async function resolveClaudeProviderHistoryWindow(input: { +export async function sampleClaudeProviderHistory(input: { identity: AgentSessionJournalIdentity accountHomePath: string hasLiveSession: boolean -}): Promise { +}): Promise { const handle = input.identity.providerHandle if (handle.kind !== 'claude') { return null @@ -325,67 +312,38 @@ export async function resolveClaudeProviderHistoryWindow(input: { if (!transcriptPath) { return null } - const read = { - transcriptPath, + const snapshot = await pinClaudeTranscript(transcriptPath) + const read: HistoryWindowInput = { providerSessionId: handle.sessionId, previousLeafUuid: handle.leafUuid, sessionId: input.identity.sessionId, turnInFlight: input.hasLiveSession } - const window = await readClaudeProviderHistoryWindow(read) - return { ...window, recorded: await readClaudeRecordedHistory(read) } + return { read: () => readClaudeProviderHistory(snapshot, read) } } -async function readClaudeRecordedHistory( - input: HistoryWindowInput & { transcriptPath: string } -): Promise { - const recorded = createRecordedCollector(input) - try { - const stream = createReadStream(input.transcriptPath) - for await (const { line } of splitTranscriptStreamLines( - stream, - MAX_HISTORY_WINDOW_RECORD_BYTES - )) { - recorded.add(line) - } - } catch (error) { - console.warn('[claude-history-window] transcript unreadable; sends stay unconfirmed:', { - transcriptPath: input.transcriptPath, - sessionId: input.sessionId, +/** One pass over the pinned transcript serves the window and the recorded history alike. */ +export async function readClaudeProviderHistory( + snapshot: ClaudeTranscriptSnapshot, + input: HistoryWindowInput +): Promise { + const context = { transcriptPath: snapshot.transcriptPath, sessionId: input.sessionId } + const pass = createClaudeHistoryPass(input, (error) => + console.warn('[claude-history-window] transcript unprovable; boundary inconsistent:', { + ...context, error }) - return null - } - return recorded.finish() -} - -export async function readClaudeProviderHistoryWindow( - input: HistoryWindowInput & { transcriptPath: string } -): Promise { - const ancestryAnchorUuid = input.previousLeafUuid - if (!ancestryAnchorUuid) { - return INCONSISTENT - } - const collector = createWindowCollector(input) + ) try { - const { chain } = await replayClaudeTranscriptBranchAncestry({ - transcriptPath: input.transcriptPath, - providerSessionId: input.providerSessionId, - previousLeafUuid: ancestryAnchorUuid, - ancestryAnchorUuid, - maxRecordBytes: MAX_HISTORY_WINDOW_RECORD_BYTES, - onAncestorRecord: collector.onAncestorRecord - }) - return windowFromChain(chain, collector.byUuid, input.turnInFlight) + await readPinnedClaudeTranscript(snapshot, MAX_HISTORY_WINDOW_RECORD_BYTES, pass.add) } catch (error) { - // Unreadable, unprovable, or a single record too large to frame: all of them - // leave the boundary unvouched for, which is not the same as an empty window. - // Oversize is no longer among them, so only the log separates what is left. - console.warn('[claude-history-window] transcript unprovable; boundary inconsistent:', { - transcriptPath: input.transcriptPath, - sessionId: input.sessionId, + // Unreadable, replaced, or a single record too large to frame: the boundary + // is unvouched for and absence proves nothing, so sends stay unconfirmed. + console.warn('[claude-history-window] transcript unreadable; sends stay unconfirmed:', { + ...context, error }) - return INCONSISTENT + return { ...inconsistent(input), recorded: null } } + return pass.finish() } diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index 80d563159ed0..4caf0c0144ea 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -38,7 +38,7 @@ import { type ClaudeExitLifecycle } from './claude-structured-session-exit-lifecycle' import type { AgentSessionBackgroundTaskState } from '../../shared/agent-session-wire' -import { resolveClaudeProviderHistoryWindow } from './claude-structured-history-window' +import { sampleClaudeProviderHistory } from './claude-structured-history-window' import { drainClaudeChildWork } from './claude-child-work-evidence' import { admitClaudePromptCancellation, @@ -135,10 +135,10 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda } /** Restart reconciliation reads the transcript a resume replays; these maps track liveness. */ - providerHistoryWindow: NonNullable = ( + sampleProviderHistory: NonNullable = ( input ) => - resolveClaudeProviderHistoryWindow({ + sampleClaudeProviderHistory({ identity: input.identity, accountHomePath: input.accountHome.path, hasLiveSession: diff --git a/src/main/claude/claude-transcript-branch-graph.ts b/src/main/claude/claude-transcript-branch-graph.ts index e739da07975d..33d53d6a9686 100644 --- a/src/main/claude/claude-transcript-branch-graph.ts +++ b/src/main/claude/claude-transcript-branch-graph.ts @@ -49,6 +49,11 @@ export class ClaudeTranscriptPreviousCursorMissingError extends Error { } } +/** An unterminated line may still be mid-append; a terminated one is simply malformed. */ +function unparsableTranscriptLine(terminated: boolean): Error { + return terminated ? transcriptError('malformed JSONL') : new ClaudeTranscriptTailIncompleteError() +} + function proveMainLineAncestry( nodes: Map, startUuid: string, @@ -114,7 +119,7 @@ function createBranchProof(input: BranchProofInput) { let leafMarkerLineIndex = -1 let tailUuid: string | null = null let tailLineIndex = -1 - return { add, finish, ancestryChain } + return { add, addParsed, has: (uuid: string) => nodes.has(uuid), finish, ancestryChain } function add(line: string, index: number, terminated: boolean): void { if (!line.trim()) { @@ -124,11 +129,13 @@ function createBranchProof(input: BranchProofInput) { try { record = JSON.parse(line) } catch { - if (!terminated) { - throw new ClaudeTranscriptTailIncompleteError() - } - throw transcriptError('malformed JSONL') + throw unparsableTranscriptLine(terminated) } + addParsed(record, index) + } + + /** `add` for a line the caller already parsed, so a shared read parses each line once. */ + function addParsed(record: unknown, index: number): void { if (typeof record !== 'object' || record === null || Array.isArray(record)) { throw transcriptError('non-object record') } @@ -289,5 +296,5 @@ function createBranchProof(input: BranchProofInput) { } } -export { ClaudeTranscriptMarkerMissingError, createBranchProof } +export { ClaudeTranscriptMarkerMissingError, createBranchProof, unparsableTranscriptLine } export type { BranchProofInput } diff --git a/src/main/claude/claude-transcript-branch-proof.ts b/src/main/claude/claude-transcript-branch-proof.ts index 440fad3a3fce..1f38fa55a780 100644 --- a/src/main/claude/claude-transcript-branch-proof.ts +++ b/src/main/claude/claude-transcript-branch-proof.ts @@ -1,15 +1,13 @@ -// Which bytes the branch graph gets to see. Every read here pins ONE descriptor -// and ONE size before it starts, because a caller that walks the proven branch -// afterwards must walk the same snapshot the proof was computed over — a second -// read at a later size would vouch for rows nothing checked. +// Which bytes the branch graph gets to see. A read pins ONE file and ONE size +// before it starts, and every consumer of that read sees the same lines: a +// caller that walks the proven branch must walk the snapshot the proof was +// computed over, and bytes appended after the pin are never evidence. -import { open } from 'node:fs/promises' +import { open, stat } from 'node:fs/promises' import { splitTranscriptStreamLines } from '../native-chat/transcript-stream-lines' import { - ClaudeTranscriptMarkerMissingError, - ClaudeTranscriptPreviousCursorMissingError, - ClaudeTranscriptTailIncompleteError, createBranchProof, + unparsableTranscriptLine, type BranchProofInput, type ClaudeTranscriptBranchProof } from './claude-transcript-branch-graph' @@ -24,49 +22,62 @@ function nonEmptyString(value: unknown): string | null { return typeof value === 'string' && value.trim().length > 0 ? value.trim() : null } -type AncestryRecord = Record - export type ClaudeTranscriptBranchAncestry = { proof: ClaudeTranscriptBranchProof /** Leaf first, anchor excluded. */ chain: string[] } -/** 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. */ - onAncestorRecord: (record: AncestryRecord, uuid: string) => void +function recordUuid(record: unknown): string | null { + return record && typeof record === 'object' && 'uuid' in record + ? nonEmptyString(record.uuid) + : null } -function createAncestryReplay( - chain: readonly string[], - onRecord: (record: AncestryRecord, uuid: string) => void -): (line: string) => void { - const pending = new Set(chain) - return (line) => { - if (!line.trim()) { - return +/** + * The branch proof as one consumer of a shared line pass: the caller parses each + * line once and hands it to every consumer. A line the proof rejects ends the + * proof, not the read, so the other consumers still see the whole file. + */ +export function createClaudeBranchAncestryPass( + input: Omit & { + /** The ancestry walk stops here; the proof is what established it is reachable. */ + ancestryAnchorUuid: string + } +) { + // Replay always proves from the file tail, where a resume by session id continues. + const builder = createBranchProof({ ...input, tip: 'file-tail' }) + let failure: { error: unknown } | null = null + return { add, reject, finish } + + /** Returns the record's uuid when it is the FIRST line carrying it, which is the + * record the chain reports for that uuid; null otherwise or once the proof failed. */ + function add(record: unknown, index: number): string | null { + if (failure) { + return null } - let parsed: unknown + const uuid = recordUuid(record) + const firstSeen = uuid !== null && !builder.has(uuid) try { - parsed = JSON.parse(line) - } catch { - // Unreachable while the proof runs first over the same bytes — it throws on - // any malformed line. Kept so a future reordering degrades, not corrupts. - return + builder.addParsed(record, index) + } catch (error) { + failure = { error } + return null } - if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { - return - } - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The parsed value is a non-array object checked above. - const record = parsed as AncestryRecord - const uuid = nonEmptyString(record.uuid) - // `delete` is the first-occurrence tie-break, not an optimisation. - if (uuid && pending.delete(uuid)) { - onRecord(record, uuid) + return firstSeen ? uuid : null + } + + function reject(terminated: boolean): void { + failure ??= { error: unparsableTranscriptLine(terminated) } + } + + /** Throws for every branch the proof cannot vouch for. */ + function finish(): ClaudeTranscriptBranchAncestry { + if (failure) { + throw failure.error } + const proof = builder.finish() + return { proof, chain: builder.ancestryChain(proof.leafUuid, input.ancestryAnchorUuid) } } } @@ -81,111 +92,54 @@ export function proveClaudeTranscriptBranchFromJsonl( return proof.finish() } -type PinnedTranscriptLines = () => AsyncGenerator<{ line: string; terminated: boolean }> +/** The transcript as it stood when pinned: this file, up to this size. */ +export type ClaudeTranscriptSnapshot = { + transcriptPath: string + size: number + dev: bigint + ino: bigint +} + +/** One `stat`: cheap enough to take on every attach, before anything can append. */ +export async function pinClaudeTranscript( + transcriptPath: string +): Promise { + const stats = await stat(transcriptPath, { bigint: true }) + return { transcriptPath, size: Number(stats.size), dev: stats.dev, ino: stats.ino } +} /** - * Run an attempt over ONE pinned snapshot: a single fd and a single `size`, so a - * second pass cannot read bytes the first pass did not vouch for. A repair - * appended while we read is finished once, at the larger size, by re-running the - * WHOLE attempt — never by reading the tail at a size the graph pass never saw. + * Stream exactly the pinned bytes, once. Nothing appended after the pin is read, + * because a provider child started since may be what appended it. A file that is + * no longer the pinned one, or no longer holds the pinned bytes, throws rather + * than pass a different transcript off as the pinned one. */ -async function runPinnedTranscriptPasses( - transcriptPath: string, - maxRecordBytes: number | undefined, - attempt: (readLines: PinnedTranscriptLines) => Promise -): Promise { - const handle = await open(transcriptPath, 'r') +export async function readPinnedClaudeTranscript( + snapshot: ClaudeTranscriptSnapshot, + maxRecordBytes: number, + onLine: (line: string, terminated: boolean) => void +): Promise { + const handle = await open(snapshot.transcriptPath, 'r') try { - let size = (await handle.stat()).size - let refreshed = false - while (true) { - const pinned = size - const readLines: PinnedTranscriptLines = async function* () { - if (pinned === 0) { - return - } - const stream = handle.createReadStream({ start: 0, end: pinned - 1, autoClose: false }) - yield* splitTranscriptStreamLines(stream, maxRecordBytes) - } - try { - return await attempt(readLines) - } catch (error) { - if ( - !(error instanceof ClaudeTranscriptMarkerMissingError) && - !(error instanceof ClaudeTranscriptPreviousCursorMissingError) && - !(error instanceof ClaudeTranscriptTailIncompleteError) - ) { - throw error - } - if (refreshed) { - throw new ClaudeTranscriptTailIncompleteError() - } - const nextSize = await handle.stat().then( - (current) => current.size, - () => size - ) - if (nextSize <= size) { - throw error - } - // Finish an already-appended repair without making the caller retry. - size = nextSize - refreshed = true - } + const current = await handle.stat({ bigint: true }) + if ( + current.dev !== snapshot.dev || + current.ino !== snapshot.ino || + current.size < BigInt(snapshot.size) + ) { + throw new Error('Claude transcript was replaced or truncated after it was pinned') + } + if (snapshot.size === 0) { + return + } + const stream = handle.createReadStream({ start: 0, end: snapshot.size - 1, autoClose: false }) + for await (const { line, terminated } of splitTranscriptStreamLines(stream, maxRecordBytes)) { + onLine(line, terminated) + } + if (stream.bytesRead !== snapshot.size) { + throw new Error('Claude transcript was truncated while it was read') } } finally { await handle.close() } } - -/** - * 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. - */ -export async function replayClaudeTranscriptBranchAncestry( - input: AncestryInput & { transcriptPath: string; maxRecordBytes?: number } -): Promise { - return runPinnedTranscriptPasses( - input.transcriptPath, - input.maxRecordBytes, - async (readLines) => { - const builder = createBranchProof({ ...input, tip: 'file-tail' }) - let index = 0 - for await (const record of readLines()) { - builder.add(record.line, index++, record.terminated) - } - const proof = builder.finish() - const chain = builder.ancestryChain(proof.leafUuid, input.ancestryAnchorUuid) - if (chain.length > 0) { - const replay = createAncestryReplay(chain, input.onAncestorRecord) - for await (const record of readLines()) { - replay(record.line) - } - } - return { proof, chain } - } - ) -} - -/** The string-source twin of `replayClaudeTranscriptBranchAncestry`. */ -export function replayClaudeTranscriptBranchAncestryFromJsonl( - input: AncestryInput & { contents: string } -): ClaudeTranscriptBranchAncestry { - 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) - } - const proof = builder.finish() - const chain = builder.ancestryChain(proof.leafUuid, input.ancestryAnchorUuid) - const replay = createAncestryReplay(chain, input.onAncestorRecord) - for (const line of lines) { - replay(line) - } - return { proof, chain } -} diff --git a/src/main/claude/claude-transcript-branch-streaming.test.ts b/src/main/claude/claude-transcript-branch-streaming.test.ts index ce283ce501db..f770fe1cc493 100644 --- a/src/main/claude/claude-transcript-branch-streaming.test.ts +++ b/src/main/claude/claude-transcript-branch-streaming.test.ts @@ -10,9 +10,9 @@ type ReaderState = { closes: number handles: FileHandle[] bytesRead: number - statErrorOn: number + statError: boolean readError: boolean - afterStat: (() => Promise) | null + afterOpen: (() => Promise) | null } const state = vi.hoisted((): ReaderState => ({ @@ -21,9 +21,9 @@ const state = vi.hoisted((): ReaderState => ({ closes: 0, handles: [], bytesRead: 0, - statErrorOn: 0, + statError: false, readError: false, - afterStat: null + afterOpen: null })) vi.mock('node:fs/promises', async (importOriginal) => { @@ -37,17 +37,15 @@ vi.mock('node:fs/promises', async (importOriginal) => { } state.opens++ state.handles.push(handle) - let statsRead = 0 return { - stat: async () => { - statsRead++ - if (statsRead === state.statErrorOn) { + stat: async (options?: Parameters[0]) => { + if (state.statError) { throw new Error('Injected stat failure') } - const snapshot = await handle.stat() - const afterStat = state.afterStat - state.afterStat = null - await afterStat?.() + const snapshot = await handle.stat(options) + const afterOpen = state.afterOpen + state.afterOpen = null + await afterOpen?.() return snapshot }, createReadStream: (options: Parameters[0]) => { @@ -73,8 +71,10 @@ import { appendFile, mkdtemp, rename, rm, truncate, unlink, writeFile } from 'no import { ClaudeTranscriptPreviousCursorMissingError, ClaudeTranscriptTailIncompleteError, - replayClaudeTranscriptBranchAncestry, - replayClaudeTranscriptBranchAncestryFromJsonl + createClaudeBranchAncestryPass, + pinClaudeTranscript, + readPinnedClaudeTranscript, + type ClaudeTranscriptSnapshot } from './claude-transcript-branch-proof' const row = (uuid: string, parentUuid: string | null, extra = {}) => @@ -87,30 +87,52 @@ const ROOT = row('root', null) const CHILD = row('child', 'root') const SOURCE = ROOT + marker('root') let directory = '' -let replayed: string[] = [] -// The live caller always anchors the replay at the cursor it already holds. -const replay = (anchorUuid = 'root') => - replayClaudeTranscriptBranchAncestry({ - transcriptPath: state.path, + +async function readLines(snapshot: ClaudeTranscriptSnapshot): Promise { + const lines: string[] = [] + await readPinnedClaudeTranscript(snapshot, Infinity, (line) => lines.push(line)) + return lines +} + +/** The proof over `contents` as a shared line pass feeds it; `firstSeen` is what the pass reports. */ +function prove(contents: string, anchorUuid = 'root') { + const pass = createClaudeBranchAncestryPass({ providerSessionId: 'provider', previousLeafUuid: anchorUuid, - ancestryAnchorUuid: anchorUuid, - onAncestorRecord: (_record, uuid) => replayed.push(uuid) + ancestryAnchorUuid: anchorUuid }) -const read = async (anchorUuid?: string) => (await replay(anchorUuid)).proof + const firstSeen: string[] = [] + const lines = contents.split('\n') + for (const [index, line] of lines.entries()) { + if (!line.trim()) { + continue + } + let record: unknown + try { + record = JSON.parse(line) + } catch { + pass.reject(index < lines.length - 1) + continue + } + const uuid = pass.add(record, index) + if (uuid) { + firstSeen.push(uuid) + } + } + return { finish: () => pass.finish(), firstSeen } +} beforeEach(async () => { directory = await mkdtemp(join(tmpdir(), 'orca-branch-streaming-')) - replayed = [] Object.assign(state, { path: join(directory, 'transcript.jsonl'), opens: 0, closes: 0, handles: [], bytesRead: 0, - statErrorOn: 0, + statError: false, readError: false, - afterStat: null + afterOpen: null }) await writeFile(state.path, SOURCE) }) @@ -128,7 +150,7 @@ afterEach(async () => { } }) -it('decodes UTF-8 across chunks and replays ancestry beyond large message bodies', async () => { +it('decodes UTF-8 across chunks and frames lines beyond large message bodies', async () => { const prefix = JSON.stringify({ type: 'comment', content: '' }).indexOf('"content":"') + 11 const comment = `${JSON.stringify({ type: 'comment', content: `${'x'.repeat(65535 - prefix)}🙂é漢字` })}\n` const contents = @@ -137,150 +159,98 @@ it('decodes UTF-8 across chunks and replays ancestry beyond large message bodies CHILD + marker('child').trimEnd() await writeFile(state.path, contents) - const fromString: string[] = [] - expect(await replay()).toEqual( - replayClaudeTranscriptBranchAncestryFromJsonl({ - contents, - providerSessionId: 'provider', - previousLeafUuid: 'root', - ancestryAnchorUuid: 'root', - onAncestorRecord: (_record, uuid) => fromString.push(uuid) - }) - ) - expect(replayed).toEqual(fromString) - expect(replayed).toEqual(['child']) -}) - -it.each([ - ['malformed middle', `{"broken":\n${SOURCE}`, false], - ['unterminated malformed tail', `${SOURCE}{"broken":`, true], - ['terminated malformed tail', `${SOURCE}{"broken":\n`, false] -] as const)( - 'preserves error classification for %s and closes the file', - async (_name, contents, incomplete) => { - await writeFile(state.path, contents) - const result = read() - await (incomplete - ? expect(result).rejects.toBeInstanceOf(ClaudeTranscriptTailIncompleteError) - : expect(result).rejects.not.toBeInstanceOf(ClaudeTranscriptTailIncompleteError)) - } -) -it('proves only the original finite prefix while the same file grows', async () => { - state.afterStat = () => appendFile(state.path, CHILD + marker('child') + ' '.repeat(1024 * 1024)) - expect(await read()).toEqual({ leafUuid: 'root', relation: 'same' }) - expect(state.bytesRead).toBe(Buffer.byteLength(SOURCE)) -}) - -it('replays the ancestry off the same pinned prefix the proof read', async () => { - const contents = ROOT + CHILD - await writeFile(state.path, contents) - state.afterStat = () => appendFile(state.path, row('grandchild', 'child')) - expect(await replay()).toEqual({ + expect(await readLines(await pinClaudeTranscript(state.path))).toEqual(contents.split('\n')) + expect(prove(contents).finish()).toEqual({ proof: { leafUuid: 'child', relation: 'descendant' }, chain: ['child'] }) - expect(replayed).toEqual(['child']) - expect(state.bytesRead).toBe(Buffer.byteLength(contents) * 2) }) -it.each([ - ['empty', '', SOURCE, 'root'], - ['missing previous cursor', SOURCE, CHILD + marker('child'), 'child'] -] as const)( - 'finishes a growing %s proof on the same open handle', - async (_name, contents, growth, anchor) => { - await writeFile(state.path, contents) - state.afterStat = () => appendFile(state.path, growth) - expect((await read(anchor)).leafUuid).toBe(anchor) - expect(state.opens).toBe(1) - } -) - -it.each(['', SUMMARY])('keeps a static missing tip fatal', async (contents) => { - await writeFile(state.path, contents) - await expect(read()).rejects.toThrow('missing last-prompt marker') -}) +it('reads only the pinned bytes while the same file grows', async () => { + const snapshot = await pinClaudeTranscript(state.path) + await appendFile(state.path, CHILD + marker('child')) + state.afterOpen = () => appendFile(state.path, ' '.repeat(1024 * 1024)) -it('preserves the typed static missing-cursor error for existing root reproof', async () => { - await expect(read('absent')).rejects.toBeInstanceOf(ClaudeTranscriptPreviousCursorMissingError) + expect((await readLines(snapshot)).join('\n')).toBe(SOURCE) + expect(state.bytesRead).toBe(Buffer.byteLength(SOURCE)) + expect(state.opens).toBe(1) }) it.each([ - ['conflict', ROOT + row('root', 'foreign') + marker('root'), 'root', 'conflicting ancestry'], - ['wrong session', ROOT + marker('root', 'foreign'), 'root', 'invalid last-prompt'], - ['append order', CHILD + ROOT, 'root', 'parent row follows'], - ['missing ancestor', CHILD, 'child', 'missing ancestor'] -] as const)( - 'does not turn %s into a retry when the file grows', - async (_name, contents, anchor, message) => { - await writeFile(state.path, contents) - state.afterStat = () => appendFile(state.path, CHILD) - await expect(read(anchor)).rejects.toThrow(message) - } -) + [ + 'replaced', + async () => { + await rename(state.path, `${state.path}.original`) + await writeFile(state.path, ROOT + CHILD + marker('child')) + } + ], + ['truncated', () => truncate(state.path, 0)] +] as const)('refuses a file %s after the pin', async (_name, change) => { + const snapshot = await pinClaudeTranscript(state.path) + await change() + await expect(readLines(snapshot)).rejects.toThrow('replaced or truncated') +}) -it('does not infer growth from a failed second stat', async () => { - await writeFile(state.path, '') - state.afterStat = () => appendFile(state.path, SOURCE) - state.statErrorOn = 2 - await expect(read()).rejects.toThrow('missing last-prompt marker') +it('refuses a file truncated while it is read', async () => { + const snapshot = await pinClaudeTranscript(state.path) + state.afterOpen = () => truncate(state.path, 3) + await expect(readLines(snapshot)).rejects.toThrow('truncated while it was read') }) -it('keeps reading the original handle after pathname replacement', async () => { - state.afterStat = async () => { - await rename(state.path, `${state.path}.original`) - await writeFile(state.path, ROOT + CHILD + marker('child')) - } - expect((await read()).leafUuid).toBe('root') +it('refuses a file unlinked after the pin', async () => { + const snapshot = await pinClaudeTranscript(state.path) + await unlink(state.path) + await expect(readLines(snapshot)).rejects.toMatchObject({ code: 'ENOENT' }) }) -it('does not use a replacement file to establish growth', async () => { - await writeFile(state.path, '') - state.afterStat = async () => { - await rename(state.path, `${state.path}.original`) - await writeFile(state.path, ROOT + CHILD + marker('child')) - } - await expect(read()).rejects.toThrow('missing last-prompt marker') +it.each(['stat', 'read'] as const)('awaits closure after a %s failure', async (failure) => { + const snapshot = await pinClaudeTranscript(state.path) + state.statError = failure === 'stat' + state.readError = failure === 'read' + await expect(readLines(snapshot)).rejects.toThrow(`Injected ${failure} failure`) }) -it('can finish an opened file after unlink', async () => { - state.afterStat = () => unlink(state.path) - expect((await read()).leafUuid).toBe('root') +it('opens no stream for an empty pin', async () => { + await writeFile(state.path, '') + expect(await readLines(await pinClaudeTranscript(state.path))).toEqual([]) + expect(state.bytesRead).toBe(0) }) -it('does not infer growth from truncation', async () => { - state.afterStat = () => truncate(state.path, 0) - await expect(read()).rejects.toThrow('missing last-prompt marker') +it.each([ + ['malformed middle', `{"broken":\n${SOURCE}`, false], + ['unterminated malformed tail', `${SOURCE}{"broken":`, true], + ['terminated malformed tail', `${SOURCE}{"broken":\n`, false] +] as const)('preserves error classification for %s', (_name, contents, incomplete) => { + let error: unknown = null + try { + prove(contents).finish() + } catch (caught) { + error = caught + } + expect(error).toBeInstanceOf(Error) + expect(error instanceof ClaudeTranscriptTailIncompleteError).toBe(incomplete) }) -it.each(['stat', 'read'] as const)('awaits closure after a %s failure', async (failure) => { - state.statErrorOn = failure === 'stat' ? 1 : 0 - state.readError = failure === 'read' - await expect(read()).rejects.toThrow(`Injected ${failure} failure`) +it('reports each uuid once, at its first line', () => { + const { finish, firstSeen } = prove(ROOT + CHILD + CHILD + marker('child')) + expect(finish().chain).toEqual(['child']) + expect(firstSeen).toEqual(['root', 'child']) }) -it('completes a record appended after the first observed extent without caller retry', async () => { - const contents = ROOT + marker('root').slice(0, -5) - await writeFile(state.path, contents) - state.afterStat = () => appendFile(state.path, marker('root').slice(-5)) - expect((await read()).leafUuid).toBe('root') - expect(state.opens).toBe(1) +it.each(['', SUMMARY])('keeps a missing tip fatal', (contents) => { + expect(prove(contents).finish).toThrow('missing last-prompt marker') }) -it('validates the entire refreshed prefix instead of accepting a malformed repair', async () => { - const conflicting = row('root', 'foreign') - await writeFile(state.path, ROOT + conflicting.slice(0, -5)) - state.afterStat = () => appendFile(state.path, conflicting.slice(-5)) - await expect(read()).rejects.toThrow('conflicting ancestry') - expect(state.opens).toBe(1) +it('preserves the typed missing-cursor error for existing root reproof', () => { + expect(prove(SOURCE, 'absent').finish).toThrow(ClaudeTranscriptPreviousCursorMissingError) }) -it('keeps an unfinished growing repair retryable after one internal refresh', async () => { - const torn = ROOT.slice(0, -5) - await writeFile(state.path, torn) - state.afterStat = () => appendFile(state.path, ' ') - await expect(read()).rejects.toBeInstanceOf(ClaudeTranscriptTailIncompleteError) - expect(state.opens).toBe(1) - expect(state.bytesRead).toBe(Buffer.byteLength(torn) * 2 + 1) +it.each([ + ['conflict', ROOT + row('root', 'foreign') + marker('root'), 'root', 'conflicting ancestry'], + ['wrong session', ROOT + marker('root', 'foreign'), 'root', 'invalid last-prompt'], + ['append order', CHILD + ROOT, 'root', 'parent row follows'], + ['missing ancestor', CHILD, 'child', 'missing ancestor'] +] as const)('fails the proof on %s', (_name, contents, anchor, message) => { + expect(prove(contents, anchor).finish).toThrow(message) }) diff --git a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts index e7e06d800ef4..594bf794922e 100644 --- a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts @@ -4,7 +4,7 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' -import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' import type { AgentJournalItemIdentity, @@ -15,6 +15,7 @@ import { digestPayload } from './journal-payload-bounds' import { reconcileJournalSubmissionsAgainstHistory } from './journal-restart-reconciliation' import type { ProviderHistoryItem, + ProviderHistorySample, ProviderHistoryWindow, ProviderRecordedHistory } from './journal-submission-reconciler' @@ -70,8 +71,10 @@ function history(uuid: string, text: string): ProviderHistoryItem { function window( items: ProviderHistoryItem[], overrides: Partial = {} -): ProviderHistoryWindow { - return { items, boundaryConsistent: true, turnInFlight: false, ...overrides } +): ProviderHistorySample { + return { + read: async () => ({ items, boundaryConsistent: true, turnInFlight: false, ...overrides }) + } } /** The whole transcript, holding these records; each is `[uuid, text]`. */ @@ -388,4 +391,34 @@ describe('reconcileJournalSubmissionsAgainstHistory', () => { expect(settled).toEqual([]) expect(journal.submissions()[0]?.dispatchState).toBe('unknown') }) + + it('reads no history when nothing is in doubt', async () => { + const journal = await open() + await appendSend(journal, 'cm_1', 'uuid-sent') + await journal.resolveDispatch({ + clientMessageId: 'cm_1', + state: 'accepted', + providerIdentity: claudeIdentity('uuid-sent'), + fence: 1 + }) + const read = vi.fn(window([]).read) + + expect( + await reconcileJournalSubmissionsAgainstHistory({ journal, fence: 2, history: { read } }) + ).toEqual([]) + expect(read).not.toHaveBeenCalled() + }) + + it('leaves a send in doubt when reading history fails', async () => { + const journal = await reopenAfterCrash(undefined, undefined, 'uuid-sent') + + const settled = await reconcileJournalSubmissionsAgainstHistory({ + journal, + fence: 2, + history: { read: () => Promise.reject(new Error('transcript unreadable')) } + }) + + expect(settled).toEqual([]) + expect(journal.submissions()[0]?.dispatchState).toBe('unknown') + }) }) diff --git a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts index 48e5f4ef54f0..e8e1b0167cf5 100644 --- a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts +++ b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts @@ -22,7 +22,11 @@ import { } from '../../../shared/agent-session-journal-item-key' import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' import type { AgentSessionJournal } from './journal-store' -import { reconcileSubmissions, type ProviderHistoryWindow } from './journal-submission-reconciler' +import { + reconcileSubmissions, + type ProviderHistorySample, + type ProviderHistoryWindow +} from './journal-submission-reconciler' /** * Only a text-only body can be compared against provider content. A submission @@ -92,20 +96,32 @@ function unseenHistory( /** * Decide what the crash boundary could only doubt. Returns the client message * ids this pass settled, so the attach result stops reporting them unconfirmed. + * History is read only when something is in doubt: most attaches have nothing + * to decide and should not pay for parsing the provider's whole transcript. + * A failed read is not an answer, so it leaves every submission as it was. */ export async function reconcileJournalSubmissionsAgainstHistory(input: { journal: AgentSessionJournal fence: number - history: ProviderHistoryWindow + history: ProviderHistorySample }): Promise { const submissions = comparableSubmissions(input.journal) if (submissions.length === 0) { return [] } + let history: ProviderHistoryWindow | null + try { + history = await input.history.read() + } catch { + return [] + } + if (!history) { + return [] + } const settled: string[] = [] for (const outcome of reconcileSubmissions({ submissions, - history: unseenHistory(input.journal, input.history) + history: unseenHistory(input.journal, history) })) { if (outcome.outcome === 'unknown') { // Narrowing failed: the submission stays unconfirmed, so record why. diff --git a/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts b/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts index 999b7114f6f0..b702b4ca4f3c 100644 --- a/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts +++ b/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts @@ -56,6 +56,14 @@ export type ProviderHistoryWindow = { recorded?: ProviderRecordedHistory | null } +/** + * Provider history whose facts were fixed when it was taken — before a new child + * could start or append — and read only if a submission is actually in doubt. + */ +export type ProviderHistorySample = { + read: () => Promise +} + export type ProviderRecordedHistory = { /** Journal keys of every user item in the history. */ itemIds: ReadonlySet diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts index fb4ee607e38d..0f3e7ac73796 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts @@ -540,8 +540,8 @@ describe('what an earlier host process left behind', () => { }) it('leaves a legacy pending message and a handed-over one in doubt, never re-sent (W4′c)', async () => { - const providerHistoryWindow = vi.fn(async () => null) - adapterExtras = { providerHistoryWindow } + const sampleProviderHistory = vi.fn(async () => null) + adapterExtras = { sampleProviderHistory } await writeAsEarlierProcess(async (journal, fence) => { await journal.appendSubmission({ ...earlierSubmission('legacy', 'l'), fence }) await journal.appendSubmission({ ...earlierSubmission('handed', 'h', true), fence }) @@ -555,7 +555,7 @@ describe('what an earlier host process left behind', () => { expect(await submission('legacy')).toMatchObject({ dispatchState: 'unknown', recovered: true }) expect(await submission('handed')).toMatchObject({ dispatchState: 'unknown', recovered: true }) // Deciding them from provider history waits for a won lease (W4′d). - expect(providerHistoryWindow).not.toHaveBeenCalled() + expect(sampleProviderHistory).not.toHaveBeenCalled() expect(dispatch).not.toHaveBeenCalled() }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts index 29b98c942937..1a46c6f98643 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts @@ -147,9 +147,14 @@ describe('structured session acquisition options', () => { } } }) - sessionAdapter.providerHistoryWindow = vi.fn(async () => historyWindow()) + // The sample fixes liveness when it is taken; parsing is left to `read`. + sessionAdapter.sampleProviderHistory = vi.fn(async () => { + const window = historyWindow() + return { read: () => readHistory(window) } + }) return sessionAdapter } + const readHistory = vi.fn(async (window: ProviderHistoryWindow) => window) let firstJournal: AgentSessionJournal | undefined const first = await performAttach({ @@ -171,6 +176,8 @@ describe('structured session acquisition options', () => { } }) expect(first).toMatchObject({ ok: true }) + // Nothing was in doubt, so the transcript was never parsed. + expect(readHistory).not.toHaveBeenCalled() await firstJournal!.appendSubmission({ clientMessageId: 'crashed-send', payloadFingerprint: digestPayload('deploy the thing'), @@ -217,6 +224,7 @@ describe('structured session acquisition options', () => { }) expect(second).toMatchObject({ ok: true, value: { unconfirmedClientMessageIds: [] } }) + expect(readHistory).toHaveBeenCalledTimes(1) expect(second).toMatchObject({ value: { page: { submissions: [{ clientMessageId: 'crashed-send', dispatchState: 'rejected' }] } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts index c9c17ce961dc..3e153734f009 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts @@ -144,10 +144,10 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi historyFilePath = (input: { identity: AgentSessionJournalIdentity }) => this.requireAgent(input.identity).historyFilePath?.(input) ?? Promise.resolve(null) - providerHistoryWindow = (input: { + sampleProviderHistory = (input: { identity: AgentSessionJournalIdentity accountHome: AgentSessionAccountHome - }) => this.requireAgent(input.identity).providerHistoryWindow?.(input) ?? Promise.resolve(null) + }) => this.requireAgent(input.identity).sampleProviderHistory?.(input) ?? Promise.resolve(null) closeSession = (sessionId: string): Promise => this.stopSession(sessionId, (adapter) => adapter.closeSession) 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 c043e65f20a3..0ab42d854792 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 @@ -36,7 +36,7 @@ import { import type { SubmissionRejectionFact } from '../../../shared/agent-session-failure' import type { AgentJournalDispatchRejection } from '../../../shared/agent-session-failure-words' import type { AgentSessionPromptResponse } from '../../../shared/agent-session-question-answer' -import type { ProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' +import type { ProviderHistorySample } from '../agent-session-journal/journal-submission-reconciler' import type { StructuredAgentSessionEventSink } from './structured-agent-session-event-sink' import type { StructuredSessionCompactionResult } from './structured-session-compaction' import type { AgentSessionCreatePhaseRecorder } from '../../observability/agent-session-instrumentation' @@ -343,12 +343,14 @@ export type StructuredAgentSessionAdapter = { /** Provider history for restart reconciliation, bounded to what the provider * recorded after the journal's last committed item. Only the adapter can say * whether the read has a proven start and whether a turn is still running, so - * it owns both flags. Omit where the provider records no boundary-consistent - * history; an omitted window leaves every unsettled submission `unknown`. */ - providerHistoryWindow?(input: { + * it owns both flags. Sampling fixes those facts and must stay cheap; the + * sample's `read` does the parsing, and runs only when a submission is in doubt. + * Omit where the provider records no boundary-consistent history; an omitted + * sample leaves every unsettled submission `unknown`. */ + sampleProviderHistory?(input: { identity: AgentSessionJournalIdentity accountHome: AgentSessionAccountHome - }): Promise + }): Promise /** Gracefully stops the structured owner after its event stream is drained. */ /** Returns true only after the provider child exit is proven. A root-exit or processless verdict * is thrown only once the session is finalized; read it through `stopAgentSessionProviderRoot`. */ 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 8fdee5c62bdb..034ea08336cd 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 @@ -42,7 +42,7 @@ import { withAgentSessionCreatePhase, type AgentSessionCreatePhaseRecorder } from '../../observability/agent-session-instrumentation' -import type { ProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' +import type { ProviderHistorySample } from '../agent-session-journal/journal-submission-reconciler' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' export type AttachFlowInput = { @@ -108,7 +108,7 @@ export async function performAttach( let reservedRecord: AgentSessionRecord | null = null let unsupportedReservationSettlementAttempted = false let replayed = false - let providerHistoryWindow: ProviderHistoryWindow | null = null + let providerHistory: ProviderHistorySample | null = null const preparedTranscript = store.getRecord(sessionId) ? { ok: true as const, items: null } : await prepareAdoptedTranscript(params) @@ -161,8 +161,10 @@ export async function performAttach( } // Sample provider history before a new child is acquired. Once acquireOwner // starts the child, the adapter's liveness signal intentionally becomes - // conservative and an absent prompt can no longer prove non-delivery. - providerHistoryWindow = await readProviderHistoryWindow({ + // conservative and an absent prompt can no longer prove non-delivery, and the + // child may append to the history. The sample fixes both; parsing waits until + // the journal shows a submission in doubt. + providerHistory = await sampleProviderHistory({ adapter: input.adapter, identity: journalIdentityFor(record, params), accountHome: record.accountHome, @@ -230,7 +232,7 @@ export async function performAttach( journalRoot: input.journalRoot, adapter: input.adapter, openConversation: input.openConversation, - providerHistoryWindow + providerHistory }) await importAdoptedTranscript(params, attached, record, preparedTranscript.items) await input.onAttached(attached, acquisitionGeneration, acquiredOwner, providerChildPhase) @@ -260,25 +262,34 @@ export async function performAttach( } } -async function readProviderHistoryWindow(input: { +async function sampleProviderHistory(input: { adapter: StructuredAgentSessionAdapter identity: AgentSessionJournalIdentity accountHome: AgentSessionRecord['accountHome'] ownerAlreadyAdmitted: boolean -}): Promise { - const read = input.adapter.providerHistoryWindow - if (!read) { +}): Promise { + const sample = input.adapter.sampleProviderHistory + if (!sample) { return null } - let history: ProviderHistoryWindow | null + let history: ProviderHistorySample | null try { - history = await read({ identity: input.identity, accountHome: input.accountHome }) + history = await sample({ identity: input.identity, accountHome: input.accountHome }) } catch { return null } + if (!history || !input.ownerAlreadyAdmitted) { + return history + } // A lease that was already live may belong to a provider child this process // has not indexed yet. Preserve the safe unknown outcome in that case. - return history && input.ownerAlreadyAdmitted ? { ...history, turnInFlight: true } : history + const { read } = history + return { + read: async () => { + const window = await read() + return window && { ...window, turnInFlight: true } + } + } } async function settleUnsupportedReservation( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-reconciliation.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-reconciliation.test.ts index 554696c87c76..4db834b35e32 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-reconciliation.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-reconciliation.test.ts @@ -57,17 +57,20 @@ function window(overrides: Partial = {}): ProviderHistory } } -/** Only the surface `attachJournal` touches; every send-shaped method is a spy - * so a re-delivery would be visible rather than silent. */ -function adapterWith(providerHistoryWindow?: () => Promise): { +/** Every send-shaped method is a spy, so a re-delivery would be visible rather than silent. */ +function adapterWith(read?: () => Promise): { adapter: StructuredAgentSessionAdapter dispatch: ReturnType } { const dispatch = vi.fn() - const adapter = { + const adapter: StructuredAgentSessionAdapter = { + acquire: vi.fn(), dispatch, - ...(providerHistoryWindow ? { providerHistoryWindow } : {}) - } as unknown as StructuredAgentSessionAdapter + cancelTurn: vi.fn(), + answerPrompt: vi.fn(), + setOption: vi.fn(), + ...(read ? { sampleProviderHistory: async () => ({ read }) } : {}) + } return { adapter, dispatch } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-transcript-reads.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-transcript-reads.test.ts new file mode 100644 index 000000000000..d702a1bcb246 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-transcript-reads.test.ts @@ -0,0 +1,194 @@ +// What a Claude attach costs in transcript I/O. Most attaches have nothing in +// doubt and must not parse the provider's transcript at all; one that does must +// read it once for every question it asks of it. + +import type { FileHandle } from 'node:fs/promises' +import type * as FsPromises from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const reads = vi.hoisted(() => ({ path: '', opens: 0, streams: 0 })) + +vi.mock('node:fs/promises', async (importOriginal) => { + const fs = await importOriginal() + return { + ...fs, + open: async (path: string, flags: string) => { + const handle = await fs.open(path, flags) + if (path !== reads.path) { + return handle + } + reads.opens += 1 + return { + stat: (options?: Parameters[0]) => handle.stat(options), + createReadStream: (options: Parameters[0]) => { + reads.streams += 1 + return handle.createReadStream(options) + }, + close: () => handle.close() + } + } + } +}) + +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' +import { agentSessionRecordFixture } from '../../../shared/agent-session-record.test-fixture' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import { structuredAgentSessionPayloadFingerprint } from '../../../shared/structured-agent-session-mutation' +import { structuredAgentSessionSendBody } from '../../../shared/structured-agent-session-outbox' +import { sampleClaudeProviderHistory } from '../../claude/claude-structured-history-window' +import { journalDirectoryFor } from '../agent-session-journal/journal-paths' +import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import { openTestAttachConversation } from './structured-agent-session-attach-test-conversation' +import { + attachJournal, + journalIdentityFor, + type AgentSessionAttachParams +} from './structured-agent-session-attach' + +const PROVIDER_SESSION = 'provider-1' +const BASE = agentSessionRecordFixture() +let root = '' +let record: AgentSessionRecord +const journals = createTrackedJournalOpener() + +function params(): AgentSessionAttachParams { + return { + envelope: { + sessionId: record.sessionId, + clientOperationId: 'op-1', + expectedRuntimeFence: record.lease.runtimeFence, + payloadFingerprint: 'fp' + }, + location: record.location, + provider: 'claude', + agent: 'claude', + accountHome: record.accountHome, + runtimeKind: 'native', + providerHandle: { kind: 'claude', sessionId: PROVIDER_SESSION, leafUuid: 'anchor' } + } +} + +/** Mirrors the Claude adapter: the real sampler, over this test's account home. */ +const adapter: StructuredAgentSessionAdapter = { + acquire: vi.fn(), + dispatch: vi.fn(), + cancelTurn: vi.fn(), + answerPrompt: vi.fn(), + setOption: vi.fn(), + sampleProviderHistory: (input) => + sampleClaudeProviderHistory({ + identity: input.identity, + accountHomePath: input.accountHome.path, + hasLiveSession: false + }) +} + +function fingerprint(text: string): string { + return structuredAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId: record.sessionId, + fields: { body: structuredAgentSessionSendBody(text, []) } + }) +} + +/** A previous process handed `text` over under `frameUuid` and died before its outcome. */ +async function crashAfterHandover(text: string, frameUuid: string): Promise { + const identity = journalIdentityFor(record, params()) + const journal = await journals.open({ + identity, + journalDir: journalDirectoryFor(root, { + workspaceId: identity.workspaceId, + sessionId: identity.sessionId + }) + }) + const body: AgentJournalMessageItem = { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text }] + } + const fence = record.lease.runtimeFence + await journal.appendSubmission({ + clientMessageId: 'cm_1', + payloadFingerprint: fingerprint(text), + body, + fence + }) + await journal.resolveDispatch({ + clientMessageId: 'cm_1', + state: 'pending', + providerIdentity: { provider: 'claude', sessionId: PROVIDER_SESSION, uuid: frameUuid }, + fence + }) + await journal.close() +} + +async function attach() { + const attached = await attachJournal({ + record, + params: params(), + journalRoot: root, + openConversation: openTestAttachConversation(root), + adapter + }) + journals.track(attached.journal) + return attached +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-attach-transcript-reads-')) + const accountHome = join(root, 'claude-home') + const projectDir = join(accountHome, 'projects', 'work') + await mkdir(projectDir, { recursive: true }) + record = { + ...BASE, + providerHandleChain: [ + { + ...BASE.providerHandleChain[0]!, + handle: { provider: 'claude', sessionId: PROVIDER_SESSION, leafUuid: 'anchor' } + } + ], + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: accountHome } + } + reads.path = join(projectDir, `${PROVIDER_SESSION}.jsonl`) + reads.opens = 0 + reads.streams = 0 + const rows = [ + { type: 'user', uuid: 'anchor', parentUuid: null, message: { content: 'earlier turn' } }, + { type: 'user', uuid: 'u-sent', parentUuid: 'anchor', message: { content: 'ship it' } }, + { type: 'last-prompt', leafUuid: 'u-sent' } + ] + await writeFile( + reads.path, + `${rows.map((row) => JSON.stringify({ ...row, sessionId: PROVIDER_SESSION })).join('\n')}\n` + ) +}) + +afterEach(async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +describe('Claude attach transcript reads', () => { + it('parses no transcript when nothing is in doubt', async () => { + const attached = await attach() + + expect(attached.unconfirmedClientMessageIds).toEqual([]) + expect(reads.opens).toBe(0) + expect(reads.streams).toBe(0) + }) + + it('reads the transcript once to decide a send in doubt', async () => { + await crashAfterHandover('ship it', 'u-sent') + + const attached = await attach() + + expect(attached.journal.submissions()[0]).toMatchObject({ dispatchState: 'accepted' }) + // The branch proof, the anchored window and the id lookup share one pass. + expect(reads.opens).toBe(1) + expect(reads.streams).toBe(1) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts index 10c4866cea3f..f76ff9e0e9ae 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts @@ -41,7 +41,7 @@ import type { AgentSessionRecordStore } from '../../runtime/agent-session-record import { agentSessionProviderHandleChainHead } from '../../../shared/agent-session-provider-handle' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { reconcileJournalSubmissionsAgainstHistory } from '../agent-session-journal/journal-restart-reconciliation' -import type { ProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' +import type { ProviderHistorySample } from '../agent-session-journal/journal-submission-reconciler' import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' import { structuredAgentSessionRefusalMessage } from './structured-agent-session-refusal-message' @@ -173,9 +173,9 @@ export type AttachedJournal = { * open — with provider history deciding the submissions that boundary could only doubt. * * Why the reconciliation belongs HERE and nowhere else: this runs after the - * record store handed this host the lease and before `onAttached` starts a - * provider child, so nothing can be appending to the provider's history while it - * is read, and the window stays valid until the resume consumes it. Every other + * record store handed this host the lease and before `onAttached` lets anyone + * send, and the history it reads was pinned before this attach started a child, + * so what that child appends is never taken as evidence. Every other * settlement site — a proven child exit — runs while the host * may still start another child, and a read there could be overtaken before it * is acted on. Orca still never re-sends: this decides state only. A queued @@ -189,8 +189,8 @@ export async function attachJournal(input: { /** The host's open conversation, whose journal the attach adopts. */ openConversation: (record: AgentSessionRecord) => Promise /** Provider history sampled before a new child is acquired. `null` means the - * adapter had no usable history; omit to read lazily for direct callers. */ - providerHistoryWindow?: ProviderHistoryWindow | null + * adapter had no usable history; omit to sample here for direct callers. */ + providerHistory?: ProviderHistorySample | null }): Promise { const identity = journalIdentityFor(input.record, input.params) const fence = input.record.lease.runtimeFence @@ -201,9 +201,7 @@ export async function attachJournal(input: { journal, fence, accountHome: input.record.accountHome, - ...(Object.hasOwn(input, 'providerHistoryWindow') - ? { history: input.providerHistoryWindow } - : {}) + ...(Object.hasOwn(input, 'providerHistory') ? { history: input.providerHistory } : {}) }) return { journal, @@ -229,15 +227,15 @@ async function reconcileAgainstProviderHistory(input: { journal: AgentSessionJournal fence: number accountHome: AgentSessionAccountHome - history?: ProviderHistoryWindow | null + history?: ProviderHistorySample | null }): Promise { let history = input.history if (history === undefined) { - if (!input.adapter.providerHistoryWindow) { + if (!input.adapter.sampleProviderHistory) { return [] } try { - history = await input.adapter.providerHistoryWindow({ + history = await input.adapter.sampleProviderHistory({ identity: input.identity, accountHome: input.accountHome }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-interruption-test-harness.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-interruption-test-harness.ts index 5223763214e5..25e4fd2363e9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-interruption-test-harness.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-interruption-test-harness.ts @@ -130,15 +130,17 @@ export async function interruptedRestart( closeSession, ...(work === 'submission' || work === 'send-after-reply' ? { - providerHistoryWindow: async () => ({ - items: [], - boundaryConsistent: historyBoundaryConsistent, - turnInFlight: false, - recorded: { - itemIds: new Set(), - itemIdsByFingerprint: new Map(), - provesAbsenceOf: () => historyBoundaryConsistent - } + sampleProviderHistory: async () => ({ + read: async () => ({ + items: [], + boundaryConsistent: historyBoundaryConsistent, + turnInFlight: false, + recorded: { + itemIds: new Set(), + itemIdsByFingerprint: new Map(), + provesAbsenceOf: () => historyBoundaryConsistent + } + }) }) } : {})