From 199fc7b2c10c609b02075d8cdc26774d0ecc34c3 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 22 Oct 2026 10:00:11 -0700 Subject: [PATCH] A Pi chat shows the turns made outside it (#31618) * feat(native-chat): a Pi chat shows the turns made outside it When a Pi chat's agent opens its conversation, Orca asks Pi for the entries saved after the point the chat last accounted for (get_entries since that entry id) and adds the terminal's turns once, in Pi's order, under the "Continued outside this chat." divider. That point is saved on the conversation handle after each chat turn and at each open. When the chat's own unsaved work and the terminal's share a suffix that can't be split (Orca died mid-turn), that suffix is skipped rather than duplicated, and later terminal turns show normally. A Pi reply over the 8 MiB line limit now fails only the optional read that asked for it instead of ending the chat. * refactor(native-chat): share the catch-up open runner across agents The Codex catch-up's agent-neutral parts move to one module: the open-time envelope (its own deadline and cancellation, never gating the open), the bounded staged batch, the single settlement commit that re-checks the journal before writing, and the "Continued outside this chat." divider. Codex keeps only how it reads its saved turns and what the chat already holds. No behavior change; the other agents' catch-up builds on this instead of copying it. * refactor(pi): catch up through the shared catch-up runner instead of a copy * chore: remove unrelated pnpm executable metadata * refactor(native-chat): the catch-up runner reads the chat's view only when the agent asks * refactor(pi): read the chat's view through the runner's view() * chore: remove regenerated pnpm executable metadata * fix(native-chat): the catch-up deadline also bounds waiting for its commit to be written A journal write barrier that never settled after the import was admitted held the provider's open past the catch-up deadline. The wait now ends with the deadline or the open's cancellation, like every other step. * test(pi): check Pi's session entry ids and get_entries since against the pinned real Pi * test: type the history catch-up sink fixture without assertions * fix(native-chat): bound and revoke history preparation * test: the runner's journal fake answers the cursor Pi's view reads * feat(native-chat): validate expected saved items during history catch-up * fix(pi): preserve outside history and optional read recovery * test(pi): fail RPC smoke promptly when the child stops --- .github/workflows/pi-provider-runtime.yml | 3 +- src/main/daemon/ndjson.test.ts | 41 +++ src/main/jsonl-rpc/agent-connection.ts | 3 +- src/main/jsonl-rpc/peer-oversized-reply.ts | 38 ++ src/main/jsonl-rpc/peer.test.ts | 143 ++++++++ src/main/jsonl-rpc/peer.ts | 31 +- .../journal-legacy-import.ts | 14 +- .../journal-pi-active-branch.ts | 32 +- ...red-agent-session-history-catch-up.test.ts | 1 + .../structured-agent-session-journal-views.ts | 1 + .../transcript-line-decoders-omp.ts | 15 +- src/main/pi/rpc-history-capture.ts | 137 +++++++ src/main/pi/rpc-history-catch-up.test.ts | 343 ++++++++++++++++++ src/main/pi/rpc-history-catch-up.ts | 161 ++++++++ src/main/pi/rpc-history-import.ts | 146 ++++++++ src/main/pi/rpc-history-journal-view.ts | 109 ++++++ src/main/pi/rpc-launch-resolution.ts | 22 ++ src/main/pi/rpc-runtime-registration.ts | 6 + src/main/pi/rpc-session-file-tail.ts | 62 ++++ src/main/pi/rpc-session-startup.ts | 30 +- src/main/pi/rpc-session.ts | 6 +- src/main/pi/rpc-turns.ts | 10 +- src/shared/main-process-ndjson-framer.ts | 85 +++-- tests/tools/pi-provider-runtime-smoke.mjs | 88 +++++ 24 files changed, 1451 insertions(+), 76 deletions(-) create mode 100644 src/main/jsonl-rpc/peer-oversized-reply.ts create mode 100644 src/main/pi/rpc-history-capture.ts create mode 100644 src/main/pi/rpc-history-catch-up.test.ts create mode 100644 src/main/pi/rpc-history-catch-up.ts create mode 100644 src/main/pi/rpc-history-import.ts create mode 100644 src/main/pi/rpc-history-journal-view.ts create mode 100644 src/main/pi/rpc-session-file-tail.ts diff --git a/.github/workflows/pi-provider-runtime.yml b/.github/workflows/pi-provider-runtime.yml index 4cde4ab5f69..ac652d64e63 100644 --- a/.github/workflows/pi-provider-runtime.yml +++ b/.github/workflows/pi-provider-runtime.yml @@ -5,6 +5,7 @@ on: paths: - 'src/shared/commit-message-agent-specs-primary.ts' - 'tests/tools/pi-provider-runtime-smoke.mjs' + - 'src/main/pi/rpc-history-*.ts' - '.github/workflows/pi-provider-runtime.yml' permissions: contents: read @@ -29,5 +30,5 @@ jobs: - uses: ./.github/actions/install-node-dependencies - name: Install pinned Pi runtime run: npm install --prefix .cache/pi-provider --ignore-scripts --no-audit --no-fund @earendil-works/pi-coding-agent@0.84.2 - - name: Verify extension model generation before and after + - name: Verify extension model generation and session entry ids run: node tests/tools/pi-provider-runtime-smoke.mjs .cache/pi-provider/node_modules/@earendil-works/pi-coding-agent/dist/cli.js diff --git a/src/main/daemon/ndjson.test.ts b/src/main/daemon/ndjson.test.ts index 9c7c481eac3..e075539e139 100644 --- a/src/main/daemon/ndjson.test.ts +++ b/src/main/daemon/ndjson.test.ts @@ -266,6 +266,45 @@ describe('createNdjsonParser', () => { expect(records).toEqual([{ index: 0 }, { index: 1 }, { index: 2 }, { index: 3 }]) }) + it('discards only the paused oversized tail after delivering older records across re-pauses', () => { + const records: unknown[] = [] + const rejected: unknown[] = [] + let paused = false + const framer = createIncrementalNdjsonFramer( + (record) => { + records.push(record) + if (records.length <= 2) { + paused = true + } + }, + (error) => rejected.push(error), + { maxLineBytes: 32, shouldPause: () => paused } + ) + framer.feed('{"a":1}\n{"b":2}\n{"c":3}\n{"oversized":"') + framer.feed('x'.repeat(64 * 1024)) + expect(rejected).toEqual([ + expect.objectContaining({ + prefix: expect.stringMatching(/^\{"oversized":/), + maxLineBytes: 64 * 1024 + }) + ]) + expect(framer.takeTail()).toBe('') + paused = false + framer.resume() + expect(records).toEqual([{ a: 1 }, { b: 2 }]) + paused = false + framer.feed('still discarded') + expect(records).toEqual([{ a: 1 }, { b: 2 }, { c: 3 }]) + paused = true + framer.feed('"}\n{"d":4}\n{"e":') + expect(records).toHaveLength(3) + paused = false + framer.resume() + framer.feed('5}\n') + expect(records).toEqual([{ a: 1 }, { b: 2 }, { c: 3 }, { d: 4 }, { e: 5 }]) + expect(rejected).toHaveLength(1) + }) + it('caps an actually incomplete paused suffix and recovers at the next delimiter', () => { const records: unknown[] = [] const rejected: unknown[] = [] @@ -278,6 +317,8 @@ describe('createNdjsonParser', () => { framer.feed('{}\nx') framer.feed('y'.repeat(70_000)) + expect(framer.takeTail()).toBe('') + framer.feed('more discarded bytes') paused = false framer.resume() framer.feed('\n{"recovered":true}\n') diff --git a/src/main/jsonl-rpc/agent-connection.ts b/src/main/jsonl-rpc/agent-connection.ts index d9c5667d961..bf44c9034c1 100644 --- a/src/main/jsonl-rpc/agent-connection.ts +++ b/src/main/jsonl-rpc/agent-connection.ts @@ -14,6 +14,7 @@ import { type JsonlRpcPeerHandlers, type JsonlRpcRecord } from './peer' +import type { JsonlRpcRequestOptions } from './peer-oversized-reply' import { resolveJsonlRpcPeerOptions, type JsonlRpcPeerOptions } from './peer-limits' export type JsonlRpcAgentConnectionOptions = JsonlRpcPeerHandlers & { @@ -151,7 +152,7 @@ export class JsonlRpcAgentConnection { request( command: string, params: Record = {}, - options: { timeoutMs?: number | null } = {} + options: JsonlRpcRequestOptions = {} ): Promise { return this.peer.request(command, params, options) } diff --git a/src/main/jsonl-rpc/peer-oversized-reply.ts b/src/main/jsonl-rpc/peer-oversized-reply.ts new file mode 100644 index 00000000000..fa8bb629211 --- /dev/null +++ b/src/main/jsonl-rpc/peer-oversized-reply.ts @@ -0,0 +1,38 @@ +import type { NdjsonRejectedRecord } from '../../shared/main-process-ndjson-framer' + +/** A reply over the line cap, refused for the one request that asked to survive it. */ +export class JsonlRpcReplyTooLargeError extends Error { + constructor( + readonly command: string, + readonly observedBytes: number, + readonly maxBytes: number + ) { + super(`Agent reply to ${command} exceeded ${maxBytes} bytes`) + this.name = 'JsonlRpcReplyTooLargeError' + } +} + +export type JsonlRpcRequestOptions = { + timeoutMs?: number | null + /** An optional read whose oversized reply fails only itself instead of the stream. */ + oversizedReply?: 'reject' +} + +// Pi echoes opaque ids before the response payload; the category survives its waiter's timeout. +const OVERSIZED_REPLY_HEAD = /^\{"id":"(orca-optional-([1-9]\d*))","type":"response"(?=[,}])/ + +/** + * Only an issued optional reply can be dropped alone. Complete-queue loss may also omit + * lifecycle records, even when its prefix identifies an optional reply. + */ +export function oversizedReplyRequest( + rejected: Extract, + highestOptionalAssigned: number +): string | null { + const head = OVERSIZED_REPLY_HEAD.exec(rejected.prefix) + if (!head || rejected.droppedCompleteRecords) { + return null + } + const number = Number(head[2]) + return Number.isSafeInteger(number) && number <= highestOptionalAssigned ? head[1] : null +} diff --git a/src/main/jsonl-rpc/peer.test.ts b/src/main/jsonl-rpc/peer.test.ts index dc4b8e816b5..40711ed9795 100644 --- a/src/main/jsonl-rpc/peer.test.ts +++ b/src/main/jsonl-rpc/peer.test.ts @@ -159,6 +159,149 @@ describe('JSON-lines RPC peer', () => { expect(onRecord).not.toHaveBeenCalled() }) + it('fails only an opted-in read whose reply is oversized, then keeps the stream', async () => { + const onRecord = vi.fn() + const { peer, input, reply } = fixture({ onRecord }, { maxLineBytes: 128 }) + const big = { id: 'orca-optional-1', type: 'response', command: 'get_entries', success: true } + const entries = peer.request('get_entries', {}, { oversizedReply: 'reject' }) + input.write(`${JSON.stringify({ ...big, data: { entries: ['x'.repeat(300)] } })}\n`) + await expect(entries).rejects.toThrow('exceeded 128 bytes') + reply({ type: 'agent_settled' }) + const state = peer.request('get_state') + reply({ type: 'response', id: 'orca-1', command: 'get_state', success: true, data: {} }) + expect(await state).toEqual({}) + expect(onRecord).toHaveBeenCalledExactlyOnceWith({ type: 'agent_settled' }) + // A read that did not opt in still ends the stream. + const plain = expect(peer.request('get_entries')).rejects.toThrow('size limit') + input.write(`${JSON.stringify({ ...big, id: 'orca-2', data: 'y'.repeat(300) })}\n`) + await plain + expect(peer.closed).toBe(true) + }) + + it('survives a late oversized optional reply after timeout and permits another read', async () => { + vi.useFakeTimers() + const onRecord = vi.fn() + const { peer, input, output, reply } = fixture({ onRecord }, { maxLineBytes: 128 }) + const expired = expect( + peer.request('get_entries', {}, { oversizedReply: 'reject', timeoutMs: 10 }) + ).rejects.toThrow('in time') + expect(JSON.parse(output.read().toString()).id).toBe('orca-optional-1') + await vi.advanceTimersByTimeAsync(10) + await expired + const state = peer.request('get_state') + input.write( + '{"id":"orca-optional-1","type":"response","command":"get_entries","success":true,"data":"' + ) + input.write('x'.repeat(300)) + input.write('"}\n') + reply({ id: 'orca-1', type: 'response', command: 'get_state', success: true, data: {} }) + reply({ type: 'agent_settled' }) + await expect(state).resolves.toEqual({}) + const next = peer.request('get_entries', {}, { oversizedReply: 'reject' }) + reply({ + id: 'orca-optional-2', + type: 'response', + command: 'get_entries', + success: true, + data: [] + }) + await expect(next).resolves.toEqual([]) + expect(onRecord).toHaveBeenCalledExactlyOnceWith({ type: 'agent_settled' }) + expect(peer.closed).toBe(false) + }) + + it.each([undefined, 'different_command'])( + 'uses local context for oversized optional replies with command %s', + async (command) => { + const { peer, input } = fixture({}, { maxLineBytes: 128 }) + const pending = expect( + peer.request('get_entries', {}, { oversizedReply: 'reject' }) + ).rejects.toThrow('reply to get_entries exceeded 128 bytes') + input.write( + `${JSON.stringify({ id: 'orca-optional-1', type: 'response', command, data: 'x'.repeat(300) })}\n` + ) + await pending + expect(peer.closed).toBe(false) + } + ) + + it.each([ + ['orca-optional-2', 'response'], + ['orca-optional-0', 'response'], + ['orca-optional-01', 'response'], + ['orca-optional-9007199254740992', 'response'], + ['orca-1', 'response'], + ['foreign', 'response'], + ['orca-optional-1', 'agent_end'] + ])('closes on an oversized record without issued optional ownership: %s %s', async (id, type) => { + const { peer, input } = fixture({}, { maxLineBytes: 128 }) + const pending = expect( + peer.request('get_entries', {}, { oversizedReply: 'reject' }) + ).rejects.toThrow('size limit') + input.write(`${JSON.stringify({ id, type, data: 'x'.repeat(300) })}\n`) + await pending + expect(peer.closed).toBe(true) + }) + + it('closes when the response type has no structural field boundary', async () => { + const { peer, input } = fixture({}, { maxLineBytes: 128 }) + const pending = expect( + peer.request('get_entries', {}, { oversizedReply: 'reject' }) + ).rejects.toThrow('size limit') + input.write(`{"id":"orca-optional-1","type":"response"garbage${'x'.repeat(300)}`) + await pending + expect(peer.closed).toBe(true) + }) + + it('keeps required replies and queued events around an oversized paused optional suffix', async () => { + const records: unknown[] = [] + const { peer, input } = fixture( + { + onRecord: (record) => { + records.push(record) + if (record.type === 'first' || record.type === 'second') { + peer.pauseReading() + } + } + }, + { maxLineBytes: 256 } + ) + const optional = expect( + peer.request('get_entries', {}, { oversizedReply: 'reject' }) + ).rejects.toThrow('exceeded 256 bytes') + const state = peer.request('get_state') + // Already-read bytes can arrive within a burst after the consumer pauses. + input.emit( + 'data', + '{"type":"first"}\n{"type":"second"}\n{"id":"orca-1","type":"response","command":"get_state","success":true,"data":{}}\n{"id":"orca-optional-1","type":"response","data":"' + ) + input.emit('data', 'x'.repeat(70_000)) + await optional + peer.resumeReading() + expect(records).toEqual([{ type: 'first' }, { type: 'second' }]) + peer.resumeReading() + await expect(state).resolves.toEqual({}) + input.emit('data', 'remaining optional bytes\n') + input.emit('data', '{"type":"agent_settled"}\n') + expect(records).toEqual([{ type: 'first' }, { type: 'second' }, { type: 'agent_settled' }]) + expect(peer.closed).toBe(false) + }) + + it('closes on complete-queue loss even if the omitted batch begins with an optional reply', async () => { + const { peer, input } = fixture({ onRecord: () => peer.pauseReading() }, { maxLineBytes: 256 }) + const pending = expect( + peer.request('get_entries', {}, { oversizedReply: 'reject' }) + ).rejects.toThrow('size limit') + input.emit('data', '{"type":"first"}\n{"type":"queued"}\n') + const optional = '{"id":"orca-optional-1","type":"response","data":"' + input.emit( + 'data', + `${optional}${'x'.repeat(64 * 1024 * 1024)}"}\n{"type":"tool_execution_end"}\n` + ) + await pending + expect(peer.closed).toBe(true) + }) + it('enforces pending capacity, and releases it after timeout and invalid response', async () => { vi.useFakeTimers() const { peer, reply } = fixture({}, { maxPendingRequests: 1, requestTimeoutMs: 10 }) diff --git a/src/main/jsonl-rpc/peer.ts b/src/main/jsonl-rpc/peer.ts index 992d7f987e2..7e934b9a278 100644 --- a/src/main/jsonl-rpc/peer.ts +++ b/src/main/jsonl-rpc/peer.ts @@ -11,6 +11,11 @@ import { resolveJsonlRpcPeerOptions, type JsonlRpcPeerOptions } from './peer-limits' +import { + JsonlRpcReplyTooLargeError, + oversizedReplyRequest, + type JsonlRpcRequestOptions +} from './peer-oversized-reply' const recordSchema = z.looseObject({ type: z.string() }) const responseSchema = z.looseObject({ @@ -59,6 +64,7 @@ export class JsonlRpcPeer { private readonly writer: ProviderStdioWriteQueue private readonly framer: ReturnType private nextId = 0 + private nextOptionalId = 0 private terminalError?: Error private inputEndError?: Error private readingPaused = false @@ -83,7 +89,20 @@ export class JsonlRpcPeer { (record) => this.dispatch(record), (rejected) => { if (rejected.kind === 'line-too-long') { - this.close(new Error('JSON-lines RPC record exceeded the size limit')) + const id = oversizedReplyRequest(rejected, this.nextOptionalId) + if (id) { + const request = this.pending.get(id) + if (request) { + const error = new JsonlRpcReplyTooLargeError( + request.command, + rejected.observedBytes, + this.limits.maxLineBytes + ) + this.rejectPending(id, error) + } + } else { + this.close(new Error('JSON-lines RPC record exceeded the size limit')) + } } else { this.diagnose('Ignored non-JSON agent output') } @@ -138,7 +157,7 @@ export class JsonlRpcPeer { request( command: string, params: Record = {}, - options: { timeoutMs?: number | null } = {} + options: JsonlRpcRequestOptions = {} ): Promise { if (this.terminalError) { return Promise.reject(this.terminalError) @@ -155,7 +174,13 @@ export class JsonlRpcPeer { } catch (error) { return Promise.reject(error) } - const id = `orca-${++this.nextId}` + if (options.oversizedReply === 'reject' && this.nextOptionalId >= Number.MAX_SAFE_INTEGER) { + return Promise.reject(new Error('JSON-lines RPC optional request identity exhausted')) + } + const id = + options.oversizedReply === 'reject' + ? `orca-optional-${++this.nextOptionalId}` + : `orca-${++this.nextId}` const controller = new AbortController() return new Promise((resolve, reject) => { const timer = diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts index 01e1d246628..b49cf2bacce 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts @@ -292,21 +292,19 @@ async function decodeReplacementItems(input: { /** * A message whose only content is a tool invocation becomes a tool-call item so * the reducer renders it as one. Everything else stays a message item with its - * blocks bounded in place. + * adoption limits applied only when supplied. */ -function legacyItemBody( +export function legacyItemBody( message: NativeChatMessage, - limits: JournalPayloadLimits + limits?: JournalPayloadLimits ): AgentJournalItemBody { const only = message.blocks.length === 1 ? message.blocks[0] : undefined if (only?.type === 'tool-call') { - // Legacy transcripts are untrusted and can contain arbitrarily large tool - // arguments. Keep them on the same bounded path as live events before the - // replacement epoch is published. + // Adoption bounds bulk before publishing its replacement epoch; catch-up admits whole batches. return { kind: 'tool-call', name: only.name, - input: boundToolInput(only.input, limits), + input: limits ? boundToolInput(only.input, limits) : only.input, ...(only.callId !== undefined ? { callId: only.callId } : {}), state: 'completed' } @@ -314,7 +312,7 @@ function legacyItemBody( return { kind: 'message', role: message.role, - blocks: message.blocks.map((block) => boundBlock(block, limits)) + blocks: limits ? message.blocks.map((block) => boundBlock(block, limits)) : message.blocks } } diff --git a/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts b/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts index fb0e723f41e..85ac3f6b126 100644 --- a/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts +++ b/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts @@ -164,17 +164,9 @@ export function createPiTranscriptCollector( } processedRows += 1 processedBytes += Buffer.byteLength(row.raw_line, 'utf8') - const record = parseJsonObject(row.raw_line) - if (record) { - const message = - record.type === 'compaction' || record.type === 'branch_summary' - ? piSummaryMessage(record, row.fallback_id) - : asRecord(record.message)?.role === 'system' - ? null - : decodeOmpTranscriptLine(row.raw_line, row.fallback_id) - if (message) { - keep(toItem(message, identify(row.raw_line, row.line_index))) - } + const message = decodePiSessionEntryLine(row.raw_line, row.fallback_id) + if (message) { + keep(toItem(message, identify(row.raw_line, row.line_index))) } if ( processedBytes >= 1024 * 1024 || @@ -211,6 +203,24 @@ export function createPiTranscriptCollector( } } +/** One saved Pi entry as a chat shows it: summaries for compaction and abandoned branches, nothing + * for a system prompt, else the decoder Pi shares with omp. */ +export function decodePiSessionEntryLine( + line: string, + fallbackId: string +): NativeChatMessage | null { + const record = parseJsonObject(line) + if (!record) { + return null + } + if (record.type === 'compaction' || record.type === 'branch_summary') { + return piSummaryMessage(record, fallbackId) + } + return asRecord(record.message)?.role === 'system' + ? null + : decodeOmpTranscriptLine(line, fallbackId) +} + function piSummaryMessage( record: Record, fallbackId: string diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts index 166990cfca7..a7d01fe8167 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts @@ -20,6 +20,7 @@ describe('history catch-up runner', () => { let barriers = 0 const journal = { epoch: 'epoch-1', + cursor: () => ({ epoch: 'epoch-1', sequence: 0 }), item: () => null, itemBody: () => null, canonicalItemId: (key: string) => key, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-views.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-views.ts index 54db9e893e2..81c417061dc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-views.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-views.ts @@ -25,6 +25,7 @@ export type StructuredAgentSessionRevisionJournal = Pick< export type StructuredAgentSessionHistoryJournal = Pick< AgentSessionJournal, | 'epoch' + | 'cursor' | 'visitItems' | 'visitItemsWithLinkage' | 'item' diff --git a/src/main/native-chat/transcript-line-decoders-omp.ts b/src/main/native-chat/transcript-line-decoders-omp.ts index bfe44fd87bf..7f0519feb00 100644 --- a/src/main/native-chat/transcript-line-decoders-omp.ts +++ b/src/main/native-chat/transcript-line-decoders-omp.ts @@ -13,6 +13,7 @@ import { type NativeChatMessage } from '../../shared/native-chat-types' import type { AgentSessionTokenUsage } from '../../shared/agent-session-context-usage' +import { buildImageDataUri } from '../../shared/image-data-uri' import { asRecord, extractString, @@ -268,12 +269,14 @@ function ompContentBlock(record: Record | null): NativeChatBloc const name = extractString(record.name) ?? 'tool' return { type: 'tool-call', name, input: record.arguments } } - case 'image': - // Why: omp stores images as content-addressed blob handles - // (`blob:sha256:…`) rather than a path or URL, so there is nothing the - // renderer can load. Dropping the block matches how the Claude mapper - // treats an image record with neither source. - return null + case 'image': { + const data = extractString(record.data) + const url = + data && !data.startsWith('blob:') + ? buildImageDataUri(extractString(record.mimeType) ?? undefined, data) + : null + return url ? { type: 'image-ref', url } : null + } default: return null } diff --git a/src/main/pi/rpc-history-capture.ts b/src/main/pi/rpc-history-capture.ts new file mode 100644 index 00000000000..115c6d36731 --- /dev/null +++ b/src/main/pi/rpc-history-capture.ts @@ -0,0 +1,137 @@ +import type { AgentSessionProviderHandle } from '../../shared/agent-session-provider-handle' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' +import { readPiEntries } from './rpc-history-catch-up' +import { piBranchInTree, piBranchSince } from './rpc-history-import' +import { encodePiHistoryClaim } from './rpc-history-journal-view' +import type { PiRpcConnection, PiRpcSession } from './rpc-session' + +export type PiHistoryResumePointWrite = { + sessionId: string + fence: number + handle: AgentSessionProviderHandle +} + +/** Where the chat's journal and Pi's entries were last known to meet; null is a session with no + * entries yet. */ +export type PiHistoryAnchor = { entry: string | null } + +export type PiHistoryCaptureDeps = { + /** Revises the open's handle in place with where the chat's journal meets Pi's entries. */ + persistResumePoint?: (write: PiHistoryResumePointWrite) => Promise +} + +export type PiHistoryCaptureParts = { + sessionId: string + fence: number + connection: Pick + sink: StructuredAgentSessionEventSink | undefined + /** Resolves once the timeline holds no events the sink refused. */ + drained: () => Promise + /** Changes whenever the chat's Pi does anything after a turn. */ + activity: () => number + handle: () => AgentSessionProviderHandle | null + persist?: PiHistoryCaptureDeps['persistResumePoint'] + logger: StructuredAgentSessionLogger +} + +/** + * After each turn the chat's Pi ran, saves the claim that its entries are shown: Pi's leaf, and the + * journal position its rows reached. Bookkeeping only: never awaited by the turn, never failing a + * send. A capture the chat moved past, or one that fails, saves nothing, so the next open treats + * that turn as not accounted for and skips rather than duplicates. + */ +export class PiHistoryCapture { + private anchor: PiHistoryAnchor | undefined + private queue: Promise = Promise.resolve() + + constructor(private readonly parts: () => PiHistoryCaptureParts) {} + + /** Built with its session, before the parts it reads when a turn ends exist. */ + static forSession( + session: Parameters[0], + deps: () => Parameters[1] + ): PiHistoryCapture { + return new PiHistoryCapture(() => piRpcHistoryCaptureParts(session, deps())) + } + + /** Only an open that left the claim at Pi's current leaf lets captures continue from it. */ + start(anchor: PiHistoryAnchor | undefined): void { + this.anchor = anchor + } + + turnEnded(): void { + const deps = this.parts() + const anchor = this.anchor + if (!anchor || !deps.persist) { + return + } + const activity = deps.activity() + this.queue = this.queue + .then(() => this.capture(deps, anchor, activity)) + .catch((error: unknown) => + deps.logger.warn('Pi turn was not recorded for catch-up', { + scope: 'pi-history-capture', + sessionId: deps.sessionId, + error + }) + ) + } + + private async capture( + deps: PiHistoryCaptureParts, + anchor: PiHistoryAnchor, + activity: number + ): Promise { + const moved = () => this.anchor !== anchor || deps.activity() !== activity + await deps.drained() + const sink = deps.sink + const written = await sink?.written?.() + const history = sink?.journalHistory?.() + if (!written?.ok || !history || moved()) { + return + } + // Read before Pi's leaf: a row landing after this is not claimed. + const through = history.cursor() + const answer = await readPiEntries(deps.connection, anchor.entry) + const branch = + anchor.entry === null ? piBranchInTree(answer, null) : piBranchSince(answer, anchor.entry) + const handle = deps.handle() + if (moved() || !branch || answer.leafId === null || !handle) { + return + } + const entry = answer.leafId + // The anchor only keeps the next read small; the saved claim is what an open trusts. + this.anchor = { entry } + await deps.persist?.({ + sessionId: deps.sessionId, + fence: deps.fence, + handle: { + ...handle, + resumeCursor: encodePiHistoryClaim({ + entry, + epoch: through.epoch, + sequence: through.sequence + }) + } + }) + } +} + +/** A session's parts, read when a turn ends: the capture is built before they exist. */ +function piRpcHistoryCaptureParts( + session: Pick, + deps: PiHistoryCaptureDeps & { logger: StructuredAgentSessionLogger } +): PiHistoryCaptureParts { + return { + sessionId: session.input.identity.sessionId, + fence: session.input.fence, + connection: session.connection, + sink: session.input.events, + drained: () => session.lane.drained(), + activity: () => session.turns.activity, + handle: () => session.openedLink?.handle ?? null, + ...(deps.persistResumePoint ? { persist: deps.persistResumePoint } : {}), + logger: deps.logger + } +} diff --git a/src/main/pi/rpc-history-catch-up.test.ts b/src/main/pi/rpc-history-catch-up.test.ts new file mode 100644 index 00000000000..a0bbd2bd97f --- /dev/null +++ b/src/main/pi/rpc-history-catch-up.test.ts @@ -0,0 +1,343 @@ +// Turns made outside a Pi chat, through the path they take: Pi's `get_entries` → walk → staged +// rows → on-disk journal, with the claim the handle would carry between opens. + +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../shared/agent-session-journal-types' +import { createTrackedJournalOpener } from '../native-chat/agent-session-journal/journal-host-database-test-support' +import { createDeferredStructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support' +import { projectStructuredAgentSessionMessages } from '../../shared/structured-agent-session-message-projection' +import { JsonlRpcResponseError } from '../jsonl-rpc/peer' +import { catchUpPiSession } from './rpc-history-catch-up' +import { PiHistoryCapture } from './rpc-history-capture' +import { decodePiHistoryClaim, type PiHistoryClaim } from './rpc-history-journal-view' + +const SESSION = 'session-pi-catch-up' +const logger = { warn: vi.fn(), error: vi.fn() } + +type Entry = { type: 'message'; id: string; parentId: string | null; message: unknown } +function entry( + id: string, + parentId: string | null, + role: 'user' | 'assistant', + text: string +): Entry { + return { type: 'message', id, parentId, message: { role, content: [{ type: 'text', text }] } } +} + +/** Answers `get_entries` the way Pi 1.3.0 does (`rpc-mode.js:503-514`), from what it loaded. */ +function piWith(entries: Entry[], leafId = entries.at(-1)?.id ?? null) { + return { + request: async (command: string, params: Record = {}) => { + expect(command).toBe('get_entries') + if (params.since === undefined) { + return { entries, leafId } + } + const index = entries.findIndex((candidate) => candidate.id === params.since) + if (index === -1) { + throw new JsonlRpcResponseError('get_entries', `Entry not found: ${String(params.since)}`) + } + return { entries: entries.slice(index + 1), leafId } + } + } +} + +const journals = createTrackedJournalOpener() +const cleanups: (() => Promise)[] = [] +afterEach(async () => { + for (const cleanup of cleanups.splice(0)) { + await cleanup() + } + logger.warn.mockClear() +}) + +async function chat() { + const root = await mkdtemp(join(tmpdir(), 'orca-pi-catch-up-')) + const file = join(root, 'session.jsonl') + const journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'pi', + providerHandle: null + }, + stateDirectory: root, + now: () => 1_000 + }) + const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging()) + deferred.bind({ journal, fence: 1, publish: () => {} }) + cleanups.push(async () => { + deferred.close() + await journals.closeAll() + await rm(root, { recursive: true, force: true }) + }) + let claim: PiHistoryClaim | null = null + let ordinal = 0 + const live = (recordId: string): AgentJournalItemIdentity => ({ + provider: 'legacy', + agent: 'pi', + sessionId: SESSION, + recordId + }) + const save = async (entries: Entry[]) => + writeFile( + file, + `${[{ type: 'session', id: 'pi-1' }, ...entries].map((e) => JSON.stringify(e)).join('\n')}\n` + ) + return { + /** One open of the chat's Pi, which has just loaded `entries` from its file. */ + open: async (entries: Entry[], leafId?: string) => { + await save(entries) + const result = await catchUpPiSession({ + connection: piWith(entries, leafId), + sessionId: SESSION, + sessionFile: file, + lineage: new Set([file]), + claim, + sink: deferred.sink, + signal: new AbortController().signal, + assertCurrent: () => {}, + logger + }) + claim = result.claim + return result + }, + /** A turn the chat ran: its prompt and answer rows, then the capture its end starts. */ + turn: async (text: string, entries: Entry[], anchor: string | null) => { + ordinal += 1 + deferred.sink.appendItem( + { provider: 'orca', clientMessageId: `m${ordinal}` }, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + deferred.sink.appendItem( + live(`run:g:${ordinal}`), + { kind: 'turn', turnId: `run:g:${ordinal}`, state: 'completed' }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + deferred.sink.appendItem( + live(`message:g:${ordinal}`), + { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: `answer ${text}` }] }, + { turnScope: { kind: 'turn', turnItemId: `legacy/pi/${SESSION}/run:g:${ordinal}` } } + ) + const persist = vi.fn(async ({ handle }: { handle: { resumeCursor?: string } }) => { + claim = decodePiHistoryClaim(handle.resumeCursor) + }) + const capture = new PiHistoryCapture(() => ({ + sessionId: SESSION, + fence: 1, + connection: piWith(entries), + sink: deferred.sink, + drained: async () => {}, + activity: () => 0, + handle: () => ({ transport: 'jsonl-rpc', agent: 'pi', nativeId: file }), + persist, + logger + })) + capture.start({ entry: anchor }) + capture.turnEnded() + await vi.waitFor(() => expect(persist).toHaveBeenCalledOnce()) + }, + /** A prompt handed to the chat's Pi whose turn never reached the journal (Orca died). */ + crashedPrompt: async (text: string) => { + ordinal += 1 + deferred.sink.appendItem( + { provider: 'orca', clientMessageId: `m${ordinal}` }, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + await deferred.sink.written?.() + }, + queuedPrompt: async (text: string) => { + await journal.appendSubmission({ + clientMessageId: 'queued', + payloadFingerprint: 'queued', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] }, + fence: 1, + handoverRecorded: true + }) + }, + projected: () => + projectStructuredAgentSessionMessages(journal.snapshot().items, [], journal.submissions()), + items: () => journal.snapshot().items, + texts: () => + journal.snapshot().items.flatMap((item) => { + if (item.body.kind === 'message') { + const text = item.body.blocks.map((block) => (block.type === 'text' ? block.text : '')) + return [`${item.body.role}:${text.join('')}`] + } + return item.body.kind === 'status' && item.body.presentation + ? [`(${item.body.presentation})`] + : [] + }) + } +} + +describe('Pi chat catch-up', () => { + it('adds terminal turns once, in Pi order, under one divider, and places a rewind', async () => { + const c = await chat() + expect(await c.open([])).toEqual({ claim: null, current: true }) + const own = [entry('u1', null, 'user', 'ask 1'), entry('a1', 'u1', 'assistant', 'answer ask 1')] + await c.turn('ask 1', own, null) + + const fullAnswer = `${'long answer '.repeat(2000)}TAIL` + const outside = [ + ...own, + entry('u2', 'a1', 'user', 'ask 2'), + entry('a2', 'u2', 'assistant', fullAnswer), + entry('u3', 'a2', 'user', 'ask 3'), + entry('a3', 'u3', 'assistant', 'answer 3') + ] + await c.open(outside) + const caughtUp = [ + 'user:ask 1', + 'assistant:answer ask 1', + '(continued-outside)', + 'user:ask 2', + `assistant:${fullAnswer}`, + 'user:ask 3', + 'assistant:answer 3' + ] + expect(c.texts()).toEqual(caughtUp) + + // Reopening adds nothing. + await c.open(outside) + expect(c.texts()).toEqual(caughtUp) + + // The terminal went back to ask 1's answer and continued from there. + const rewound = [ + ...outside, + entry('u4', 'a1', 'user', 'ask 4'), + entry('a4', 'u4', 'assistant', 'answer 4') + ] + await c.open(rewound) + const afterRewind = [ + ...caughtUp, + '(continued-outside-rewound)', + 'user:ask 4', + 'assistant:answer 4' + ] + expect(c.texts()).toEqual(afterRewind) + + // Re-editing the first prompt starts a new root in the same session tree. + const rootRewind = [ + ...rewound, + entry('u5', null, 'user', 'ask 5'), + entry('a5', 'u5', 'assistant', 'answer 5') + ] + expect(await c.open(rootRewind)).toMatchObject({ claim: { entry: 'a5' }, current: true }) + const afterRootRewind = [ + ...afterRewind, + '(continued-outside-rewound)', + 'user:ask 5', + 'assistant:answer 5' + ] + expect(c.texts()).toEqual(afterRootRewind) + await c.open(rootRewind) + expect(c.texts()).toEqual(afterRootRewind) + expect(logger.warn).not.toHaveBeenCalled() + }) + + it('keeps an image-only terminal prompt and its reply across repeated opens', async () => { + const c = await chat() + await c.open([]) + const own = [entry('u1', null, 'user', 'ask 1'), entry('a1', 'u1', 'assistant', 'answer ask 1')] + await c.turn('ask 1', own, null) + const data = + 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+/lXcAAAAASUVORK5CYII=' + const outside = [ + ...own, + { + ...entry('u2', 'a1', 'user', ''), + message: { + role: 'user', + content: [ + { type: 'text', text: '' }, + { type: 'image', data, mimeType: 'image/png' } + ] + } + }, + entry('a2', 'u2', 'assistant', 'A tiny image') + ] + expect(await c.open(outside)).toMatchObject({ claim: { entry: 'a2' }, current: true }) + const imported = c.items() + expect(imported.slice(-3).map((item) => item.body)).toEqual([ + expect.objectContaining({ kind: 'status', presentation: 'continued-outside' }), + { + kind: 'message', + role: 'user', + blocks: [{ type: 'image-ref', url: `data:image/png;base64,${data}` }] + }, + { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'A tiny image' }] } + ]) + await c.open(outside) + expect(c.items()).toEqual(imported) + expect(logger.warn).not.toHaveBeenCalled() + }) + + it('imports outside work above the queued prompt that starts a resting Pi', async () => { + const c = await chat() + await c.open([]) + const own = [entry('u1', null, 'user', 'ask 1'), entry('a1', 'u1', 'assistant', 'answer ask 1')] + await c.turn('ask 1', own, null) + await c.queuedPrompt('ask next') + const outside = [ + ...own, + entry('u2', 'a1', 'user', 'outside'), + entry('a2', 'u2', 'assistant', 'outside answer') + ] + expect(await c.open(outside)).toMatchObject({ claim: { entry: 'a2' }, current: true }) + const projected = c.projected() + expect( + projected + .filter((message) => message.role === 'user' || message.role === 'assistant') + .map((message) => + message.blocks + .filter((block) => block.type === 'text') + .map((block) => block.text) + .join('') + ) + ).toEqual(['ask 1', 'answer ask 1', 'outside', 'outside answer', 'ask next']) + await c.open(outside) + expect(c.projected()).toEqual(projected) + expect(logger.warn).not.toHaveBeenCalled() + }) + + it('skips a suffix it cannot split after a crash, then shows later terminal turns', async () => { + const c = await chat() + await c.open([]) + const own = [entry('u1', null, 'user', 'ask 1'), entry('a1', 'u1', 'assistant', 'answer ask 1')] + await c.turn('ask 1', own, null) + + // The chat's Pi saved ask 2 before Orca died; then a terminal wrote ask 3. + await c.crashedPrompt('ask 2') + const mixed = [ + ...own, + entry('u2', 'a1', 'user', 'ask 2'), + entry('u3', 'u2', 'user', 'ask 3'), + entry('a3', 'u3', 'assistant', 'answer 3') + ] + await c.open(mixed) + expect(c.texts()).toEqual(['user:ask 1', 'assistant:answer ask 1', 'user:ask 2']) + + // The skip is saved: the next terminal turn shows, nothing before it does. + await c.open([ + ...mixed, + entry('u4', 'a3', 'user', 'ask 4'), + entry('a4', 'u4', 'assistant', 'answer 4') + ]) + expect(c.texts()).toEqual([ + 'user:ask 1', + 'assistant:answer ask 1', + 'user:ask 2', + '(continued-outside)', + 'user:ask 4', + 'assistant:answer 4' + ]) + }) +}) diff --git a/src/main/pi/rpc-history-catch-up.ts b/src/main/pi/rpc-history-catch-up.ts new file mode 100644 index 00000000000..3a1499fc042 --- /dev/null +++ b/src/main/pi/rpc-history-catch-up.ts @@ -0,0 +1,161 @@ +import { waitForPromiseWithSignal } from '../../shared/abort-signal-reason' +import { JsonlRpcResponseError } from '../jsonl-rpc/peer' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' +import { runHistoryCatchUp } from '../native-chat/agent-session-wire/structured-agent-session-history-catch-up' +import { + createPiHistoryImportPlan, + piBranchInTree, + piBranchSince, + piEntriesAnswerSchema, + type PiEntriesAnswer, + type PiSessionEntry +} from './rpc-history-import' +import { + readPiHistoryJournalView, + samePiHistoryJournalView, + usablePiHistoryClaim, + type PiHistoryClaim +} from './rpc-history-journal-view' +import type { PiRpcConnection } from './rpc-session' +import { readPiSessionLastEntryId } from './rpc-session-file-tail' + +export const PI_HISTORY_CATCH_UP_TIMEOUT_MS = 30_000 + +/** What a catch-up leaves the chat with: the claim to save on this open's link, and whether it + * names Pi's current leaf (a capture may then continue from it) or was only carried forward. */ +export type PiHistoryCatchUp = { claim: PiHistoryClaim | null; current: boolean } + +/** `get_entries`, whose oversized answer fails only this read. */ +export async function readPiEntries( + connection: Pick, + since: string | null, + timeoutMs?: number +): Promise { + return piEntriesAnswerSchema.parse( + await connection.request('get_entries', since === null ? {} : { since }, { + oversizedReply: 'reject', + ...(timeoutMs === undefined ? {} : { timeoutMs }) + }) + ) +} + +// Pi has no writer lock: a terminal can write the conversation only while the chat's Pi isn't +// running, and every start of the chat's Pi opens here, with Pi's view freshly loaded from disk. + +/** Optional reconciliation never gates the open; a failure keeps the claim the chat already had. */ +export async function catchUpPiSession(input: { + connection: Pick + sessionId: string + sessionFile: string + /** Every file of this conversation: entries keep their ids across a directory fork. */ + lineage: ReadonlySet + claim: PiHistoryClaim | null + sink: StructuredAgentSessionEventSink | undefined + signal: AbortSignal + assertCurrent: () => void + timeoutMs?: number + logger: StructuredAgentSessionLogger +}): Promise { + const carried: PiHistoryCatchUp = { claim: input.claim, current: false } + if (!input.sink) { + return carried + } + // Set once Pi's current leaf is known; it stands only if nothing after it fails. + let leaf: { entry: string | null } | undefined + const outcome = await runHistoryCatchUp({ + agentName: 'Pi', + scope: 'pi-history-catch-up', + sessionId: input.sessionId, + sink: input.sink, + signal: input.signal, + assertCurrent: input.assertCurrent, + timeoutMs: input.timeoutMs ?? PI_HISTORY_CATCH_UP_TIMEOUT_MS, + logger: input.logger, + readView: (history) => readPiHistoryJournalView(history, input.lineage), + sameView: samePiHistoryJournalView, + prepare: async ({ view: readView, journal, signal, deadline, check, warn }) => { + const view = readView() + const read = async (since: string | null) => { + check() + const answer = await waitForPromiseWithSignal( + readPiEntries(input.connection, since, Math.max(1, deadline - Date.now())), + signal + ) + check() + return answer + } + const claim = usablePiHistoryClaim(input.claim, view) + // Without a claim, any Pi history the journal already shows (adopted, or an older build's + // turns) leaves nothing to split it by; with one, neither does the chat's own work after it. + const known = view.ownWorkSequence > 0 || view.accounted.size > 0 + if (claim ? view.ownWorkSequence > claim.sequence : known) { + leaf = { entry: await currentLeaf(input.sessionFile, read) } + return null + } + let answer = claim ? await readSince(read, claim.entry) : null + let branch: { path: PiSessionEntry[]; rewound: boolean } | null = + answer && claim ? piBranchSince(answer, claim.entry) : null + if (!answer || !branch) { + answer = await read(null) + const inTree = piBranchInTree(answer, claim?.entry ?? null) + if (inTree === 'unknown-claim') { + leaf = { entry: answer.leafId } + return null + } + if (!inTree) { + warn('incomplete-history') + return null + } + branch = inTree + } + leaf = { entry: answer.leafId } + return createPiHistoryImportPlan({ + sessionFile: input.sessionFile, + journal, + path: branch.path.filter((entry) => !view.accounted.has(entry.id)), + rewound: branch.rewound, + divider: view.hasRows, + check + }) + } + }) + const history = input.sink.journalHistory?.() + if (outcome === 'skipped' || !leaf || !history) { + return carried + } + const cursor = history.cursor() + return { + claim: + leaf.entry === null + ? null + : { entry: leaf.entry, epoch: cursor.epoch, sequence: cursor.sequence }, + current: true + } +} + +/** A claim Pi no longer has (a replaced file) answers as if the whole tree had to be read. */ +async function readSince( + read: (since: string | null) => Promise, + entry: string +): Promise { + try { + return await read(entry) + } catch (error) { + if (error instanceof JsonlRpcResponseError && error.message.startsWith('Entry not found')) { + return null + } + throw error + } +} + +/** Pi's leaf, asked cheaply after the file's last entry when the file still matches what Pi + * loaded, else from the whole tree. */ +async function currentLeaf( + sessionFile: string, + read: (since: string | null) => Promise +): Promise { + const last = await readPiSessionLastEntryId(sessionFile).catch(() => null) + const hinted = last === null ? null : await readSince(read, last) + return (hinted ?? (await read(null))).leafId +} diff --git a/src/main/pi/rpc-history-import.ts b/src/main/pi/rpc-history-import.ts new file mode 100644 index 00000000000..d2f1bf7deee --- /dev/null +++ b/src/main/pi/rpc-history-import.ts @@ -0,0 +1,146 @@ +import { z } from 'zod' +import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalItemIdentity +} from '../../shared/agent-session-journal-types' +import { legacyItemBody } from '../native-chat/agent-session-journal/journal-legacy-import' +import { decodePiSessionEntryLine } from '../native-chat/agent-session-journal/journal-pi-active-branch' +import { journalLifecycleItemMutation } from '../native-chat/agent-session-journal/journal-row-builders' +import type { StructuredAgentSessionTransitionJournal } from '../native-chat/agent-session-wire/structured-agent-session-transition' +import { + continuedOutsideNotice, + continuedOutsideRecordId, + createHistoryCatchUpBatch, + type HistoryCatchUpPlan +} from '../native-chat/agent-session-wire/structured-agent-session-history-catch-up' + +const piEntrySchema = z.looseObject({ + id: z.string().min(1), + parentId: z.string().min(1).nullable(), + type: z.string() +}) +export type PiSessionEntry = z.infer + +/** `get_entries`: the entries after `since` in the order Pi appended them, and its leaf. */ +export const piEntriesAnswerSchema = z.object({ + entries: z.array(piEntrySchema), + leafId: z.string().min(1).nullable() +}) +export type PiEntriesAnswer = z.infer + +/** The answer to `get_entries since: claim`: the active branch after the claim, oldest first, when + * the leaf's `parentId` chain reaches the claim inside it; null when the terminal went back to an + * earlier point, which only the whole tree can place. The answer lists in append order and keeps + * abandoned branches, so the branch is always walked from the leaf. */ +export function piBranchSince( + answer: PiEntriesAnswer, + claim: string +): { path: PiSessionEntry[]; rewound: false } | null { + const walked = walkFromLeaf(answer, (id) => id === claim) + return walked?.stop === claim ? { path: walked.path, rewound: false } : null +} + +/** + * The whole tree's active branch after the newest entry the chat accounts for: the claim, or the + * nearest of its ancestors where the terminal branched off (rewound). Without a claim, the whole + * branch. 'unknown-claim' when the claim isn't in this conversation at all. + */ +export function piBranchInTree( + answer: PiEntriesAnswer, + claim: string | null +): { path: PiSessionEntry[]; rewound: boolean } | 'unknown-claim' | null { + const byId = new Map(answer.entries.map((entry) => [entry.id, entry])) + const accounted = new Set() + for (let id = claim; id !== null && !accounted.has(id); id = byId.get(id)?.parentId ?? null) { + if (!byId.has(id)) { + return 'unknown-claim' + } + accounted.add(id) + } + const walked = walkFromLeaf(answer, (id) => accounted.has(id)) + if (!walked) { + return null + } + return { path: walked.path, rewound: walked.stop !== claim } +} + +function walkFromLeaf( + answer: PiEntriesAnswer, + stopsAt: (id: string) => boolean +): { path: PiSessionEntry[]; stop: string | null } | null { + const byId = new Map(answer.entries.map((entry) => [entry.id, entry])) + const path: PiSessionEntry[] = [] + let current = answer.leafId + while (current !== null && !stopsAt(current)) { + const entry = byId.get(current) + if (!entry) { + return null + } + // Each id is visited once: a cycle ends the walk as incomplete history. + byId.delete(current) + path.push(entry) + current = entry.parentId + } + return { path: path.toReversed(), stop: current } +} + +/** Rows keyed as adoption keys them (`legacy/pi//`), so both dedupe. */ +export function piHistoryEntryIdentity( + sessionFile: string, + entryId: string +): AgentJournalItemIdentity { + return { provider: 'legacy', agent: 'pi', sessionId: sessionFile, recordId: entryId } +} + +/** Outside entries as journal rows, staged whole before anything is written; entries the journal + * already holds are left out first, so an earlier import never vetoes newer entries. Null when + * nothing on the path shows as a row. */ +export function createPiHistoryImportPlan(input: { + sessionFile: string + journal: StructuredAgentSessionTransitionJournal + path: readonly PiSessionEntry[] + rewound: boolean + /** Only a chat that already shows something needs the divider. */ + divider: boolean + check: () => void +}): HistoryCatchUpPlan | null { + const batch = createHistoryCatchUpBatch() + const entryIds: string[] = [] + for (const entry of input.path) { + input.check() + const identity = piHistoryEntryIdentity(input.sessionFile, entry.id) + if (input.journal.item(input.journal.canonicalItemId(agentJournalItemKey(identity))) !== null) { + continue + } + const message = decodePiSessionEntryLine(JSON.stringify(entry), entry.id) + if (!message) { + continue + } + const body = legacyItemBody(message) + if (entryIds.length === 0 && input.divider) { + batch.push( + journalLifecycleItemMutation( + { turnScope: AGENT_JOURNAL_THREAD_SCOPE }, + piHistoryEntryIdentity(input.sessionFile, continuedOutsideRecordId(entry.id)), + continuedOutsideNotice(input.rewound) + ) + ) + } + // Imported prompts are history, never sends delivery recovery could claim. + batch.push( + journalLifecycleItemMutation( + { + turnScope: AGENT_JOURNAL_THREAD_SCOPE, + ...(body.kind === 'message' && body.role === 'user' ? { echoOf: null } : {}) + }, + identity, + body + ) + ) + entryIds.push(entry.id) + } + return batch.size === 0 + ? null + : batch.finish('pi-history-import', [input.journal.epoch, input.sessionFile, entryIds]) +} diff --git a/src/main/pi/rpc-history-journal-view.ts b/src/main/pi/rpc-history-journal-view.ts new file mode 100644 index 00000000000..1101041bfd2 --- /dev/null +++ b/src/main/pi/rpc-history-journal-view.ts @@ -0,0 +1,109 @@ +import { z } from 'zod' +import { + agentJournalSubmissionKey, + parseAgentJournalItemKey +} from '../../shared/agent-session-journal-item-key' +import { isQueuedAgentJournalSubmission } from '../../shared/agent-session-queued-submission' +import type { StructuredAgentSessionHistoryJournal } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' + +/** + * Pi's entries up to `entry` are accounted for (shown, or deliberately skipped) by the chat's journal + * up to `sequence` of `epoch`. Saved as the Pi handle's resume cursor, which only Pi's adapter reads. + */ +export type PiHistoryClaim = { entry: string; epoch: string; sequence: number } + +const claimSchema = z.object({ + entry: z.string().min(1), + epoch: z.string().min(1), + sequence: z.number().int().nonnegative() +}) + +export function encodePiHistoryClaim(claim: PiHistoryClaim): string { + return JSON.stringify(claim) +} + +/** A cursor an older build left, or one this build can't read, is no claim at all. */ +export function decodePiHistoryClaim(value: string | undefined): PiHistoryClaim | null { + if (value === undefined) { + return null + } + try { + const parsed = claimSchema.safeParse(JSON.parse(value)) + return parsed.success ? parsed.data : null + } catch { + return null + } +} + +export type PiHistoryJournalView = { + epoch: string + sequence: number + hasRows: boolean + /** Pi entries the journal already holds as rows, from any file of this conversation. */ + accounted: ReadonlySet + /** Where the chat's own newest work was created: a turn its Pi ran, or a prompt handed to it. */ + ownWorkSequence: number +} + +export function readPiHistoryJournalView( + journal: StructuredAgentSessionHistoryJournal, + lineage: ReadonlySet +): PiHistoryJournalView { + const accounted = new Set() + const submissions = journal.submissions() + const queued = new Set( + submissions + .filter(isQueuedAgentJournalSubmission) + .map((submission) => agentJournalSubmissionKey(submission.clientMessageId)) + ) + let hasRows = false + let ownWorkSequence = 0 + journal.visitItems((itemId, sequence, body) => { + hasRows = true + const identity = parseAgentJournalItemKey(itemId) + if ( + identity?.provider === 'legacy' && + identity.agent === 'pi' && + lineage.has(identity.sessionId) + ) { + accounted.add(identity.recordId) + return + } + if ( + body.kind === 'turn' || + (identity?.provider === 'orca' && body.kind === 'message' && !queued.has(itemId)) + ) { + ownWorkSequence = Math.max(ownWorkSequence, sequence) + } + }) + // A prompt Pi accepted before its turn row landed still counts from its acceptance. + for (const submission of submissions) { + const receipt = journal.receiptFor(submission.clientMessageId) + if (receipt?.cursor.epoch === journal.epoch) { + ownWorkSequence = Math.max(ownWorkSequence, receipt.cursor.sequence) + } + } + const cursor = journal.cursor() + return { epoch: cursor.epoch, sequence: cursor.sequence, hasRows, accounted, ownWorkSequence } +} + +/** Usable only in the journal it describes, and never past what that journal holds. */ +export function usablePiHistoryClaim( + claim: PiHistoryClaim | null, + view: PiHistoryJournalView +): PiHistoryClaim | null { + return claim && claim.epoch === view.epoch && claim.sequence <= view.sequence ? claim : null +} + +export function samePiHistoryJournalView( + a: PiHistoryJournalView, + b: PiHistoryJournalView +): boolean { + return ( + a.epoch === b.epoch && + a.hasRows === b.hasRows && + a.ownWorkSequence === b.ownWorkSequence && + a.accounted.size === b.accounted.size && + [...a.accounted].every((id) => b.accounted.has(id)) + ) +} diff --git a/src/main/pi/rpc-launch-resolution.ts b/src/main/pi/rpc-launch-resolution.ts index f4c5ec61c00..d2195fea53d 100644 --- a/src/main/pi/rpc-launch-resolution.ts +++ b/src/main/pi/rpc-launch-resolution.ts @@ -5,6 +5,7 @@ import { agentSessionRefusalError } from '../../shared/agent-session-wire-refusa import { agentSessionProviderHandleChainHead, agentSessionProviderHandleKey, + type AgentSessionProviderHandleChain, type AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' import { requireLegacyAgentSessionAccountHome } from '../../shared/agent-session-account-home' @@ -32,6 +33,8 @@ import { export type PiRpcResolvedLaunch = PiRpcLaunchOptions & { previous: AgentSessionProviderHandleLink | null replacement?: 'unsaved' | 'restore-failed' + /** The files of the conversation this launch continues, newest first; empty for a new one. */ + lineage?: readonly string[] } export type PiRpcLaunchPartDeps = { resolveCommand?: typeof resolveCliCommand @@ -197,6 +200,9 @@ export function piRpcLaunchPart( command, fullAccess: deps.resolveFullAccess?.() ?? true, previous, + ...(sessionFile || forkFile + ? { lineage: piConversationFiles(record.providerHandleChain) } + : {}), ...(sessionFile ? { sessionFile } : {}), ...(forkFile ? { forkFile } : {}), ...(replacement ? { replacement } : {}) @@ -207,6 +213,22 @@ export function piRpcLaunchPart( } } +/** The head back through resumes and directory forks, which copy every entry with its id, to the + * session the conversation started as. */ +function piConversationFiles(chain: AgentSessionProviderHandleChain): string[] { + const files: string[] = [] + for (const link of chain.toReversed()) { + if (link.handle.transport !== 'jsonl-rpc' || link.handle.agent !== 'pi') { + break + } + files.push(link.handle.nativeId) + if (link.origin !== 'resumed' && link.origin !== 'forked') { + break + } + } + return files +} + export function piRpcProviderLink( launch: PiRpcResolvedLaunch, file: string, diff --git a/src/main/pi/rpc-runtime-registration.ts b/src/main/pi/rpc-runtime-registration.ts index aa20e80f636..346665bece1 100644 --- a/src/main/pi/rpc-runtime-registration.ts +++ b/src/main/pi/rpc-runtime-registration.ts @@ -11,6 +11,7 @@ import { PI_RPC_AGENT } from './rpc-agent-definition' import { piRpcLaunchPart } from './rpc-launch-resolution' import { PiRpcSessionAdapter } from './rpc-session-adapter' import { PI_RPC_RUNTIME_METADATA } from './rpc-runtime-metadata' +import { reviseAgentSessionProviderResumePoint } from '../runtime/agent-session-provider-handle-transition' function createPiRpcAdapter(context: StructuredAgentAdapterContext): StructuredAgentRuntimeAdapter { const { deps } = context @@ -35,6 +36,11 @@ function createPiRpcAdapter(context: StructuredAgentAdapterContext): StructuredA ...(outcome.state === 'accepted' ? { providerIdentity: outcome.providerIdentity } : outcome) }) }, + persistResumePoint: async ({ sessionId, fence, handle }) => { + await context.store.transitionHandoff(sessionId, (record) => + reviseAgentSessionProviderResumePoint({ record, fence, handle, now: Date.now() }) + ) + }, logger: deps.logger }) } diff --git a/src/main/pi/rpc-session-file-tail.ts b/src/main/pi/rpc-session-file-tail.ts new file mode 100644 index 00000000000..f09a79425fc --- /dev/null +++ b/src/main/pi/rpc-session-file-tail.ts @@ -0,0 +1,62 @@ +import { open } from 'node:fs/promises' +import { z } from 'zod' + +const BLOCK_BYTES = 64 * 1024 +/** One entry line is never longer than a reply the peer accepts. */ +const MAX_TAIL_BYTES = 8 * 1024 * 1024 + +/** + * The id of the newest entry Pi saved, read backwards from the file's end. Only a hint for a cheap + * `get_entries since` that asks Pi for its leaf: Pi answers from what it loaded, which the file + * may no longer match, so the answer, not this, is what counts. + */ +export async function readPiSessionLastEntryId(file: string): Promise { + const handle = await open(file, 'r') + try { + let position = (await handle.stat()).size + // The newest line not yet ended by an earlier newline, oldest bytes first. + let partial: Buffer[] = [] + let partialBytes = 0 + // A last line without its newline is still being appended; it is not an entry yet. + let complete = false + while (position > 0 && partialBytes <= MAX_TAIL_BYTES) { + const length = Math.min(BLOCK_BYTES, position) + position -= length + const block = Buffer.alloc(length) + await handle.read(block, 0, length, position) + let end = length + for (let index = length - 1; index >= 0; index--) { + if (block[index] !== 0x0a) { + continue + } + const id = complete ? entryId([block.subarray(index + 1, end), ...partial]) : null + if (id !== null) { + return id + } + complete = true + partial = [] + partialBytes = 0 + end = index + } + partial.unshift(block.subarray(0, end)) + partialBytes += end + } + return position === 0 && complete ? entryId(partial) : null + } finally { + await handle.close() + } +} + +const entryHeadSchema = z.looseObject({ type: z.unknown(), id: z.string().min(1) }) + +function entryId(parts: Buffer[]): string | null { + let record: unknown + try { + record = JSON.parse(Buffer.concat(parts).toString('utf8')) + } catch { + return null + } + const head = entryHeadSchema.safeParse(record) + // The header names the session, not an entry. + return head.success && head.data.type !== 'session' ? head.data.id : null +} diff --git a/src/main/pi/rpc-session-startup.ts b/src/main/pi/rpc-session-startup.ts index 518ebe4650b..1b5a458f33a 100644 --- a/src/main/pi/rpc-session-startup.ts +++ b/src/main/pi/rpc-session-startup.ts @@ -18,9 +18,14 @@ import { piRpcProviderLink, type PiRpcResolvedLaunch } from './rpc-launch-resolu import { piModelOptionId, readPiRpcCommands, readPiRpcSessionOptions } from './rpc-options' import { piRpcStateSchema, type PiRpcState } from './rpc-protocol' import type { PiRpcConnection, PiRpcSessionDeps } from './rpc-session' +import { catchUpPiSession } from './rpc-history-catch-up' +import type { PiHistoryAnchor } from './rpc-history-capture' +import { decodePiHistoryClaim, encodePiHistoryClaim } from './rpc-history-journal-view' export class PiRpcSessionStartup { commands?: AgentSessionSlashCommand[] + /** Set by an open that left its claim at Pi's current leaf. */ + historyAnchor?: PiHistoryAnchor constructor( private readonly input: StructuredAgentSessionAcquireInput, @@ -51,13 +56,36 @@ export class PiRpcSessionStartup { Date.now() ) this.setState(state) + // Only the conversation this open continues can carry the claim; a replacement starts over. + const continued = launch.sessionFile !== undefined || launch.forkFile !== undefined + const history = await this.wait( + catchUpPiSession({ + connection: this.connection, + sessionId: this.input.identity.sessionId, + sessionFile: state.sessionFile, + lineage: new Set([...(launch.lineage ?? []), state.sessionFile]), + claim: continued ? decodePiHistoryClaim(launch.previous?.handle.resumeCursor) : null, + sink: this.input.events, + signal, + assertCurrent: () => this.assertLive(signal), + logger: this.deps.logger + }), + signal + ) + this.assertLive(signal) + if (history.current) { + this.historyAnchor = { entry: history.claim?.entry ?? null } + } const current: AgentSessionOptionsResult['current'] = { model: state.model ? piModelOptionId(state.model.provider, state.model.id) : '', ...(state.thinkingLevel ? { effort: state.thinkingLevel } : {}) } this.assertLive(signal) current.confirmed = [...(current.model ? ['model'] : []), ...(current.effort ? ['effort'] : [])] - return { link, reportedOptions: current } + const claimed = history.claim + ? { ...link, handle: { ...link.handle, resumeCursor: encodePiHistoryClaim(history.claim) } } + : link + return { link: claimed, reportedOptions: current } } /** After the open: the listing and the `/` commands reach the host as later reports, and a read diff --git a/src/main/pi/rpc-session.ts b/src/main/pi/rpc-session.ts index 4f934fe9857..f241bd50e1d 100644 --- a/src/main/pi/rpc-session.ts +++ b/src/main/pi/rpc-session.ts @@ -25,6 +25,7 @@ import type { PiRpcResolvedLaunch } from './rpc-launch-resolution' import type { StructuredAgentSessionProviderOpenResult } from '../native-chat/agent-session-wire/structured-agent-session-provider-open-contract' import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' import { PiRpcSessionStartup } from './rpc-session-startup' +import { PiHistoryCapture, type PiHistoryCaptureDeps } from './rpc-history-capture' export type PiRpcConnection = Pick< JsonlRpcAgentConnection, @@ -40,7 +41,7 @@ export type PiRpcConnection = Pick< | 'lastCloseResult' | 'onExit' > -export type PiRpcSessionDeps = { +export type PiRpcSessionDeps = PiHistoryCaptureDeps & { openConnection?: ( launch: ProviderProcessLaunch, handlers: JsonlRpcAgentConnectionOptions @@ -66,6 +67,7 @@ export class PiRpcSession { readonly heldSends = new Set() readonly optionProof = { current: new Map(), writes: new Map() } readonly startup: PiRpcSessionStartup + readonly history = PiHistoryCapture.forSession(this, () => this.deps) initialOptions: Readonly> = {} openedLink: AgentSessionProviderHandleLink | null = null requestedClose = false @@ -109,6 +111,7 @@ export class PiRpcSession { request: (command, params, options) => this.connection.request(command, params, options), settled: (id, outcome) => this.settle(id, outcome), idle: () => this.dialogs.cancelAll(), + turnEnded: () => this.history.turnEnded(), failed: (error) => this.fail(error), diagnostic: (error) => deps.logger.warn('Pi context usage could not be read', { @@ -178,6 +181,7 @@ export class PiRpcSession { ...(result.reportedOptions.effort ? { effort: result.reportedOptions.effort } : {}) } this.openedLink = result.link ?? null + this.history.start(this.startup.historyAnchor) this.optionProof.current = new Map(Object.entries(this.initialOptions)) void this.startup .readAfterOpen(this.generation, result.reportedOptions) diff --git a/src/main/pi/rpc-turns.ts b/src/main/pi/rpc-turns.ts index 34cd3e7bd23..e04f5a518ce 100644 --- a/src/main/pi/rpc-turns.ts +++ b/src/main/pi/rpc-turns.ts @@ -5,6 +5,7 @@ import type { } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { ProviderTimelineEvent } from '../native-chat/agent-session-timeline/provider-timeline-event' import type { JsonlRpcRecord } from '../jsonl-rpc/peer' +import type { JsonlRpcRequestOptions } from '../jsonl-rpc/peer-oversized-reply' import type { JsonlRpcTimelineLane } from '../jsonl-rpc/timeline-lane' import { PiRpcContextUsage } from './rpc-context-usage' import { PiRpcMessages } from './rpc-messages' @@ -20,10 +21,12 @@ export type PiRpcTurnDeps = { request: ( command: string, params?: Record, - options?: { timeoutMs?: number | null } + options?: JsonlRpcRequestOptions ) => Promise settled: (id: string, outcome: AgentSessionDispatchOutcome) => void idle: () => void + /** A turn's end reached the timeline. */ + turnEnded?: () => void failed: (error: Error) => void diagnostic?: (error: unknown) => void } @@ -89,6 +92,10 @@ export class PiRpcTurns { get working(): boolean { return this.active !== undefined } + /** Moves whenever Pi or a send does anything. */ + get activity(): number { + return this.revision + } async submit( id: string, @@ -284,6 +291,7 @@ export class PiRpcTurns { this.settling = false this.messages.reset() this.deps.idle() + this.deps.turnEnded?.() } private refreshUsage(): void { diff --git a/src/shared/main-process-ndjson-framer.ts b/src/shared/main-process-ndjson-framer.ts index abd66cf97b1..5d21814ddb0 100644 --- a/src/shared/main-process-ndjson-framer.ts +++ b/src/shared/main-process-ndjson-framer.ts @@ -23,6 +23,7 @@ export type NdjsonRejectedRecord = maxLineBytes: number observedBytes: number prefix: string + droppedCompleteRecords?: true } | { kind: 'invalid-json'; line: string; error: Error } @@ -39,10 +40,6 @@ export type IncrementalNdjsonFramerOptions = { shouldPause?: () => boolean } -function errorFrom(value: unknown): Error { - return value instanceof Error ? value : new Error(String(value)) -} - function boundedUtf8Prefix(value: string, maxBytes: number): string { if (Buffer.byteLength(value, 'utf8') <= maxBytes) { return value @@ -80,6 +77,7 @@ export function createIncrementalNdjsonFramer( let prefixSegments: string[] = [] let prefixBytes = 0 let discardingOversizedLine = false + let discardingPausedSuffix = false let pendingInput: string | null = null let pausedCompleteInput: string[] = [] let pausedCompleteInputBytes = 0 @@ -118,6 +116,7 @@ export function createIncrementalNdjsonFramer( kind: 'line-too-long', maxLineBytes: maxQueuedCompleteInputBytes, observedBytes: pausedCompleteInputBytes + completeBytes, + droppedCompleteRecords: true, prefix: ownRetainedString(boundedUtf8Prefix(complete, REJECTED_LINE_PREFIX_MAX_BYTES)) }) return @@ -133,19 +132,25 @@ export function createIncrementalNdjsonFramer( } const suffixBytes = Buffer.byteLength(suffix, 'utf8') if (suffixBytes > maxPendingInputBytes) { + pendingInput = null + discardingPausedSuffix = true onRejected({ kind: 'line-too-long', maxLineBytes: maxPendingInputBytes, observedBytes: suffixBytes, prefix: ownRetainedString(boundedUtf8Prefix(suffix, REJECTED_LINE_PREFIX_MAX_BYTES)) }) - pendingInput = null - discardingOversizedLine = true return } pendingInput = detachFrom(suffix, source) } + const retainPausedInput = (input: string, source = input): void => { + const newlineIndex = input.lastIndexOf('\n') + queuePausedCompleteInput(newlineIndex === -1 ? '' : input.slice(0, newlineIndex + 1), source) + retainPendingSuffix(newlineIndex === -1 ? input : input.slice(newlineIndex + 1), source) + } + const process = (input: string): void => { let cursor = 0 while (cursor < input.length) { @@ -193,7 +198,11 @@ export function createIncrementalNdjsonFramer( try { parsed = JSON.parse(line) } catch (error) { - onRejected({ kind: 'invalid-json', line, error: errorFrom(error) }) + onRejected({ + kind: 'invalid-json', + line, + error: error instanceof Error ? error : new Error(String(error)) + }) continue } onRecord(parsed, line) @@ -202,12 +211,7 @@ export function createIncrementalNdjsonFramer( } if (options.shouldPause?.() && cursor < input.length) { - const remainder = input.slice(cursor) - const newlineIndex = remainder.lastIndexOf('\n') - const complete = newlineIndex === -1 ? '' : remainder.slice(0, newlineIndex + 1) - const suffix = newlineIndex === -1 ? remainder : remainder.slice(newlineIndex + 1) - queuePausedCompleteInput(complete, input) - retainPendingSuffix(suffix, input) + retainPausedInput(input.slice(cursor), input) return } } @@ -215,36 +219,32 @@ export function createIncrementalNdjsonFramer( return { feed(chunk): void { - if (chunk.length === 0) { - return - } if (pausedCompleteInputBytes > 0 && !options.shouldPause?.()) { const queued = pausedCompleteInput pausedCompleteInput = [] pausedCompleteInputBytes = 0 process(queued.join('')) } - if (pendingInput !== null || pausedCompleteInputBytes > 0) { - // Complete records are retained as a separate bounded queue. The pending - // input limit applies only to the final, actually incomplete record. - if (pendingInput === null) { - if (options.shouldPause?.()) { - const newlineIndex = chunk.lastIndexOf('\n') - const complete = newlineIndex === -1 ? '' : chunk.slice(0, newlineIndex + 1) - const suffix = newlineIndex === -1 ? chunk : chunk.slice(newlineIndex + 1) - queuePausedCompleteInput(complete, chunk) - retainPendingSuffix(suffix, chunk) - return - } - process(chunk) + const endedPausedSuffix = discardingPausedSuffix + // The dropped tail follows the queued records; only new bytes can finish it. + if (discardingPausedSuffix) { + const newlineIndex = chunk.indexOf('\n') + if (newlineIndex === -1) { return } - const combined = pendingInput + chunk - const newlineIndex = combined.lastIndexOf('\n') - const complete = newlineIndex === -1 ? '' : combined.slice(0, newlineIndex + 1) - const suffix = newlineIndex === -1 ? combined : combined.slice(newlineIndex + 1) - queuePausedCompleteInput(complete, combined) - retainPendingSuffix(suffix, combined) + discardingPausedSuffix = false + chunk = chunk.slice(newlineIndex + 1) + } + if (chunk.length === 0) { + return + } + + if (pendingInput !== null) { + retainPausedInput(pendingInput + chunk) + return + } + if ((pausedCompleteInputBytes > 0 || endedPausedSuffix) && options.shouldPause?.()) { + retainPausedInput(chunk) return } process(chunk) @@ -275,7 +275,10 @@ export function createIncrementalNdjsonFramer( } }, takeTail(): string { - const tail = discardingOversizedLine ? '' : `${lineSegments.join('')}${pendingInput ?? ''}` + const tail = + discardingOversizedLine || discardingPausedSuffix + ? '' + : `${lineSegments.join('')}${pendingInput ?? ''}` clearLine() pendingInput = null return tail @@ -283,6 +286,7 @@ export function createIncrementalNdjsonFramer( reset(): void { clearLine() discardingOversizedLine = false + discardingPausedSuffix = false pendingInput = null pausedCompleteInput = [] pausedCompleteInputBytes = 0 @@ -300,14 +304,9 @@ export function encodeNdjson(msg: unknown, maxLineBytes = NDJSON_MAX_LINE_BYTES) return `${line}\n` } -export type NdjsonParser = { - feed(chunk: string): void - reset(): void -} +export type NdjsonParser = Pick -export type NdjsonParserOptions = { - maxLineBytes?: number -} +export type NdjsonParserOptions = Pick export function createNdjsonParser( onMessage: (msg: unknown) => void, diff --git a/tests/tools/pi-provider-runtime-smoke.mjs b/tests/tools/pi-provider-runtime-smoke.mjs index d095cb72446..04309165098 100644 --- a/tests/tools/pi-provider-runtime-smoke.mjs +++ b/tests/tools/pi-provider-runtime-smoke.mjs @@ -2,6 +2,8 @@ import assert from 'node:assert/strict' import { once } from 'node:events' import { mkdtemp, mkdir, writeFile, rm } from 'node:fs/promises' import { createServer } from 'node:http' +import { spawn } from 'node:child_process' +import { createInterface } from 'node:readline' import { createRequire } from 'node:module' import { tmpdir } from 'node:os' import { join, resolve } from 'node:path' @@ -111,6 +113,92 @@ try { assert.equal(results[1].code, 0, results[1].stderr) assert.match(results[1].stdout, /fixture-generated-commit/) assert.equal(results[1].requests, 1) + // The Pi catch-up's contract: entries keep stable ids, and `since` returns what was appended after + // one of them, or refuses an id the session doesn't have. + const sessionFile = join(scratch, 'catch-up.jsonl') + const rpc = spawn(process.execPath, [piCli, '--mode', 'rpc', '--session', sessionFile], { + cwd: scratch, + env: { + PATH: process.env.PATH, + SystemRoot: process.env.SystemRoot, + WINDIR: process.env.WINDIR, + HOME: scratch, + USERPROFILE: scratch, + ORCA_BACKGROUND_LAUNCH: '1', + PI_CODING_AGENT_DIR: dir + }, + stdio: ['pipe', 'pipe', 'inherit'] + }) + const replies = new Map() + let settled + const settledOnce = new Promise((resolve) => (settled = resolve)) + let fail + const failed = new Promise((_, reject) => (fail = reject)) + const deadline = setTimeout( + () => fail(new Error('Pi RPC smoke timed out after 30 seconds')), + 30000 + ) + rpc.on('error', fail) + rpc.on('close', (code, signal) => + fail(new Error(`Pi RPC closed (code ${code}, signal ${signal})`)) + ) + rpc.stdin.on('error', fail) + rpc.stdout.on('error', fail) + const lines = createInterface({ input: rpc.stdout }) + lines.on('line', (line) => { + try { + const record = JSON.parse(line) + if (record.type === 'response' && replies.has(record.id)) { + const reply = replies.get(record.id) + replies.delete(record.id) + reply(record, line) + } else if (record.type === 'agent_settled' || record.type === 'agent_end') { + settled() + } + } catch (error) { + fail(error) + } + }) + let next = 0 + const ask = (type, params = {}, id = `smoke-${++next}`) => + new Promise((resolve) => { + replies.set(id, (record, line) => { + if (id === 'orca-optional-1') { + assert.ok(line.startsWith('{"id":"orca-optional-1","type":"response",')) + assert.equal(record.id, id) + } + resolve(record) + }) + rpc.stdin.write(`${JSON.stringify({ ...params, type, id })}\n`) + }) + try { + await Promise.race([ + failed, + (async () => { + assert.equal((await ask('prompt', { message: 'hello' })).success, true) + await settledOnce + const all = await ask('get_entries', {}, 'orca-optional-1') + assert.equal((await ask('get_state')).success, true) + assert.equal(all.success, true, all.error) + const ids = all.data.entries.map((entry) => entry.id) + assert.ok(ids.length >= 2 && ids.every((id) => typeof id === 'string' && id.length > 0)) + assert.equal(all.data.leafId, ids.at(-1)) + const since = await ask('get_entries', { since: ids[0] }) + assert.deepEqual( + since.data.entries.map((entry) => entry.id), + ids.slice(1) + ) + const unknown = await ask('get_entries', { since: 'not-an-entry' }) + assert.equal(unknown.success, false) + assert.match(unknown.error, /Entry not found/) + results.push({ variant: 'rpc-get-entries', entries: ids.length }) + })() + ]) + } finally { + clearTimeout(deadline) + lines.close() + rpc.kill() + } console.log( JSON.stringify( {