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 + } + }) }) } : {})