diff --git a/.project-docs/30-worklog/tasks/20260823-pi-event-projector-91c6a7e4.md b/.project-docs/30-worklog/tasks/20260823-pi-event-projector-91c6a7e4.md new file mode 100644 index 0000000..be175e8 --- /dev/null +++ b/.project-docs/30-worklog/tasks/20260823-pi-event-projector-91c6a7e4.md @@ -0,0 +1,132 @@ +# Task: Implement PI-060 event projector and recovery + +## Identity + +- Task ID: 20260823-pi-event-projector-91c6a7e4 +- Mode: Feature +- Branch: codex/20260823-pi-event-projector-91c6a7e4-pi-event-projector +- Worktree: D:\Datas\OthersProjects\makelore-pi-event-projector-91c6a7e4 +- Base commit: e79aeffffab16c4c5570971e24d1a20997be681d +- Owner: codex +- Status: Completed + +## Scope + +- Implement `PI-060 — Event projector, hydration, and recovery` on top of + PI-050 commit `e79aeffffab16c4c5570971e24d1a20997be681d`. +- Own Main-private Pi live-event projection, persisted active-branch hydration, + durable reconciliation, generation/sequence envelopes, and recovery snapshot + integration. +- Add exact-version fixtures and focused tests for the event/session projector + seam without wiring Host API or Renderer consumers owned by later tickets. + +## Intent And Constraints + +- Follow planner task `20260822-pi-runtime-spec-b6e2c9a4`, ticket `PI-060`, and + Spec sections 10, 11, and 18. +- Keep Pi wire event, message, and entry types inside Electron Main. Product + callers consume only the PI-010 `ConversationSnapshot` / patch contracts. +- Preserve stable live UI IDs during durable entry reconciliation; never use a + timestamp or array position as identity. +- Treat only `agent_settled` as authoritative idle. `agent_end`, compaction + retry, and queue continuation must not release the run early. +- Recover only the target Conversation, hydrate the new generation, and never + replay an uncertain prompt or alter an unreadable session file. +- Real external Provider verification remains Explicitly Waived / Accepted + Risk with `realTurnVerified=false`; macOS remains deferred to `PI-150`. +- Keep the implementation surgical: no Host API/Renderer wiring, OpenCode + compatibility layer, feature flag, or unrelated refactor. + +## Project Context Loaded + +- Concurrent Task Gate: Passed. The task owns feature worktree + `D:\Datas\OthersProjects\makelore-pi-event-projector-91c6a7e4`, branch + `codex/20260823-pi-event-projector-91c6a7e4-pi-event-projector`, at base + `e79aeffffab16c4c5570971e24d1a20997be681d`. +- Planning Gate: Passed after reading the required project positioning, + integrated state, decisions, architecture, domain, evidence, reflection, + commitments, stale-items, planner, PI-050, and all registered peer scopes. +- The canonical positioning file is still a placeholder and integrated state + predates the Pi feature chain, so the task-scoped cutover Spec/tickets and + completed PI task records are authoritative for this implementation. +- No peer owns the PI event/session projector. Older undefined OpenCode partner + model tasks do not own projector behavior and this task does not change model + switching semantics; no semantic conflict was found. +- Likely modules: `electron/coding-runtime/contracts.ts`, + `electron/coding-runtime/conversation-reducer.ts`, + `electron/coding-runtime/pi/runtime.ts`, new Main-private projector modules, + and focused tests/fixtures. +- Unknowns to resolve from source before implementation: exact Pi 0.84.2 live + event and persisted-entry shapes, and the smallest existing PI-050 generation + replacement seam. These are code/fixture questions, not product decisions. + +## Plan + +1. Inspect PI-010/PI-050 contracts and exact Pi event/entry shapes; fix the + projector seam. +2. Use vertical red-green slices for live event projection. +3. Use vertical red-green slices for active-branch hydration, reconciliation, + generation/gap handling, and recover integration. +4. Run focused and full repository verification, then synchronize evidence to + the planner task for independent review. + +## Outcome + +- Added a Main-private `PiEventProjector` for exact Pi 0.84.2 live RPC + shapes. It projects stable message/content-index streams, cumulative tool + output, authoritative `toolResult`, retry/turn/compaction boundaries, queue, + interaction, usage, context, and authoritative `agent_settled` idle patches. +- Added a pure `projectPiSessionSnapshot` seam for authoritative + `entries + leafId` traversal, latest-compaction retained-tail semantics, + abandoned-branch exclusion, tool/result reconstruction, large-image + attachment projection, context recalculation state, and stable live-to- + durable ID reconciliation. +- Integrated per-Conversation serial projection into `PiConversationRuntime`. + Initial prepare/fork hydrates persisted state, `agent_end` creates a + non-idle checkpoint, `agent_settled` creates the final idle snapshot, and + worker replacement/recovery reopens and hydrates only the target generation. +- Added recoverable fail-closed handling for unreadable session trees. The + last good nodes remain in memory, no source-session write occurs, sibling + Conversations remain unchanged, and a later corrected recovery succeeds. +- Extended the product contract/reducer only for the already-modeled turn + boundary patch and `context.recalculating`; no Pi wire type escaped Main. +- Added a named exact-version fixture for + `@earendil-works/pi-coding-agent@0.84.2` and focused regression coverage for + live/cold equivalence, compaction/retry/queue settling, tool partials, + attachment references, target-only recovery, and corrupt sessions. + +## Verification + +- `corepack pnpm run lint:check` — passed with 0 errors; the repository's 6 + pre-existing React warnings remain outside this task. +- `corepack pnpm run typecheck` — passed. +- `corepack pnpm test` — passed, 195 files / 2181 tests. +- `corepack pnpm run build:vite` — passed for Renderer, Electron Main, + Preload, and release utility outputs; existing chunk-size/dynamic-import + warnings remain. +- `corepack pnpm run test:electron:windows` — passed, 1 file / 3 tests. +- `corepack pnpm run probe:pi-runtime` — command passed against staged Pi + 0.84.2 production closure, local session/reopen/isolation probes, and ready + budgets. Probe result remains `partial-pass / incomplete` because + `realTurnVerified=false`; real Provider validation is Explicitly Waived / + Accepted Risk. macOS execution remains deferred to PI-150 and is not Pass. +- Focused projector/runtime suite — passed, including all 15 `pi-*.test.ts` + files / 78 tests after an isolated Windows temporary-file `rename EPERM` + was rerun successfully; the later full suite passed without recurrence. +- `git diff --check` — passed. + +## Follow-ups + +- PI-070 owns registered extension tool-detail schemas and interaction response + behavior; PI-060 deliberately keeps unknown/custom payloads hidden or in + bounded metadata-only diagnostics. +- Later Host/runtime assembly must supply the `projectImage` persistence hook + so live/session image bytes become product attachment IDs in production. +- PI-150 owns the deferred macOS matrix. Real Provider compatibility, + concurrency, credential isolation, and protocol behavior remain accepted + unverified risks under the explicit user waiver. + +## Promotion Candidates + +- None. The durable rules implemented here are already canonical in the Pi + cutover Spec and PI-060 ticket; no duplicate architectural prose is needed. diff --git a/electron/coding-runtime/contracts.ts b/electron/coding-runtime/contracts.ts index 7ac944e..aedcad5 100644 --- a/electron/coding-runtime/contracts.ts +++ b/electron/coding-runtime/contracts.ts @@ -80,6 +80,7 @@ export interface ConversationContextState { outputLimit?: number; compaction: 'idle' | 'running'; lastCompactionId?: string; + recalculating?: boolean; } export interface ConversationInteractionOption { @@ -260,6 +261,7 @@ export type ConversationPatch = | { op: 'message.block-delta'; messageId: string; blockId: string; delta: string } | { op: 'tool.upsert'; node: ConversationToolNode } | { op: 'compaction.upsert'; node: ConversationCompactionNode } + | { op: 'boundary.upsert'; node: ConversationBoundaryNode } | { op: 'subagent.upsert'; node: ConversationSubagentNode } | { op: 'queue.replace'; queue: ConversationQueueState } | { op: 'interaction.upsert'; interaction: ConversationInteraction } diff --git a/electron/coding-runtime/conversation-reducer.ts b/electron/coding-runtime/conversation-reducer.ts index 20a1c69..3531e0f 100644 --- a/electron/coding-runtime/conversation-reducer.ts +++ b/electron/coding-runtime/conversation-reducer.ts @@ -71,6 +71,7 @@ const PATCH_OPS = new Set([ 'message.block-delta', 'tool.upsert', 'compaction.upsert', + 'boundary.upsert', 'subagent.upsert', 'queue.replace', 'interaction.upsert', @@ -301,6 +302,7 @@ export function isConversationSnapshot(value: unknown): value is ConversationSna && typeof context.usedTokens === 'number' && typeof context.contextWindow === 'number' && ['idle', 'running'].includes(String(context.compaction)) + && (context.recalculating === undefined || typeof context.recalculating === 'boolean') && Array.isArray(record.pendingInteractions) && record.pendingInteractions.every(isInteraction) && isWorkerState(worker) @@ -328,6 +330,8 @@ function isPatch(value: unknown): value is ConversationPatch { return isToolNode(record.node); case 'compaction.upsert': return isConversationNode(record.node) && record.node.kind === 'compaction'; + case 'boundary.upsert': + return isConversationNode(record.node) && record.node.kind === 'boundary'; case 'subagent.upsert': return isSubagentNode(record.node); case 'queue.replace': @@ -341,7 +345,8 @@ function isPatch(value: unknown): value is ConversationPatch { return context !== null && typeof context.usedTokens === 'number' && typeof context.contextWindow === 'number' - && ['idle', 'running'].includes(String(context.compaction)); + && ['idle', 'running'].includes(String(context.compaction)) + && (context.recalculating === undefined || typeof context.recalculating === 'boolean'); } case 'snapshot.invalidated': return isNonEmptyString(record.reason); @@ -456,6 +461,7 @@ function applyPatch( case 'tool.upsert': return { snapshot: { ...snapshot, nodes: upsertTool(snapshot.nodes, patch.node) } }; case 'compaction.upsert': + case 'boundary.upsert': case 'subagent.upsert': return { snapshot: { ...snapshot, nodes: upsertById(snapshot.nodes, patch.node) } }; case 'queue.replace': diff --git a/electron/coding-runtime/pi/event-projector.ts b/electron/coding-runtime/pi/event-projector.ts new file mode 100644 index 0000000..72bf2ba --- /dev/null +++ b/electron/coding-runtime/pi/event-projector.ts @@ -0,0 +1,589 @@ +import type { + ConversationContentBlock, + ConversationMessageNode, + ConversationPatch, + ConversationSnapshot, + ConversationToolNode, + PublicUsage, +} from '../contracts'; +import type { PiRpcEvent } from './rpc-client'; + +export interface PiEventProjectorOptions { + createId(): string; + now?: () => number; + projectImage?(image: PiImageProjectionInput): Promise; +} + +export interface PiImageProjectionInput { + conversationId: string; + data: string; + mime: string; + source: 'live' | 'session'; +} + +export interface PiProjectedAttachment { + attachmentId: string; + mime: string; +} + +export interface PiProjectionDiagnostic { + eventType: string; + reason: 'unsupported-event'; +} + +function asRecord(value: unknown): Record | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : null; +} + +function usageOf(value: unknown): PublicUsage | undefined { + const usage = asRecord(value); + if (!usage + || typeof usage.input !== 'number' + || typeof usage.output !== 'number') return undefined; + return { + inputTokens: usage.input, + outputTokens: usage.output, + ...(typeof usage.cacheRead === 'number' ? { cacheReadTokens: usage.cacheRead } : {}), + ...(typeof usage.cacheWrite === 'number' ? { cacheWriteTokens: usage.cacheWrite } : {}), + }; +} + +function stopReasonOf(value: unknown): ConversationMessageNode['stopReason'] | undefined { + if (value === 'toolUse') return 'tool-use'; + if (value === 'stop' || value === 'length' || value === 'error' || value === 'aborted') { + return value; + } + return undefined; +} + +function currentAssistant(snapshot: ConversationSnapshot): ConversationMessageNode | undefined { + return snapshot.nodes.findLast( + (node): node is ConversationMessageNode => node.kind === 'message' + && node.role === 'assistant' + && node.status === 'streaming', + ); +} + +function outputBlocks(toolId: string, value: unknown): ConversationContentBlock[] { + const result = asRecord(value); + if (!result || !Array.isArray(result.content)) return []; + return result.content.flatMap((item, contentIndex) => { + const block = asRecord(item); + if (!block || block.type !== 'text' || typeof block.text !== 'string') return []; + return [{ + kind: 'text' as const, + id: `${toolId}:output:${contentIndex}`, + text: block.text, + status: 'complete' as const, + }]; + }); +} + +function currentTool(snapshot: ConversationSnapshot, toolCallId: string): ConversationToolNode | undefined { + return snapshot.nodes.find( + (node): node is ConversationToolNode => node.kind === 'tool' + && node.toolCallId === toolCallId, + ); +} + +async function liveContentBlocks( + messageId: string, + content: unknown, + conversationId: string, + projectImage: PiEventProjectorOptions['projectImage'], +): Promise { + const values = typeof content === 'string' + ? [{ type: 'text', text: content }] + : Array.isArray(content) ? content : []; + const blocks: ConversationContentBlock[] = []; + for (let contentIndex = 0; contentIndex < values.length; contentIndex += 1) { + const block = asRecord(values[contentIndex]); + if (block?.type === 'text' && typeof block.text === 'string') { + blocks.push({ + kind: 'text', + id: `${messageId}:content:${contentIndex}`, + text: block.text, + status: 'complete', + }); + } else if (block?.type === 'thinking' && typeof block.thinking === 'string') { + blocks.push({ + kind: 'thinking', + id: `${messageId}:content:${contentIndex}`, + text: block.thinking, + status: 'complete', + }); + } else if (block?.type === 'image' + && typeof block.data === 'string' + && typeof block.mimeType === 'string' + && projectImage) { + const attachment = await projectImage({ + conversationId, + data: block.data, + mime: block.mimeType, + source: 'live', + }); + blocks.push({ + kind: 'image', + id: `${messageId}:content:${contentIndex}`, + attachmentId: attachment.attachmentId, + mime: attachment.mime, + }); + } + } + return blocks; +} + +export class PiEventProjector { + private readonly createId: () => string; + private readonly now: () => number; + private readonly projectImage: PiEventProjectorOptions['projectImage']; + private assistantMessageId: string | null = null; + private readonly toolDrafts = new Map(); + private compactionId: string | null = null; + private readonly diagnostics: PiProjectionDiagnostic[] = []; + + constructor(options: PiEventProjectorOptions) { + this.createId = options.createId; + this.now = options.now ?? Date.now; + this.projectImage = options.projectImage; + } + + getDiagnostics(): PiProjectionDiagnostic[] { + return structuredClone(this.diagnostics); + } + + async project( + snapshot: ConversationSnapshot, + event: PiRpcEvent, + ): Promise { + if (event.type === 'agent_start') { + return [{ + op: 'run.state', + run: { + ...snapshot.run, + status: 'running', + runId: snapshot.run.runId ?? this.createId(), + startedAt: snapshot.run.startedAt ?? this.now(), + }, + }]; + } + + if (event.type === 'turn_start' || event.type === 'turn_end') { + const runId = snapshot.run.runId ?? this.createId(); + return [{ + op: 'boundary.upsert', + node: { + kind: 'boundary', + id: this.createId(), + runId, + boundary: event.type === 'turn_start' ? 'turn-start' : 'turn-end', + }, + }]; + } + + if (event.type === 'auto_retry_start' + && typeof event.attempt === 'number' + && typeof event.delayMs === 'number') { + const runId = snapshot.run.runId ?? this.createId(); + return [ + { + op: 'boundary.upsert', + node: { + kind: 'boundary', + id: this.createId(), + runId, + boundary: 'retry', + attempt: event.attempt, + delayMs: event.delayMs, + }, + }, + { + op: 'run.state', + run: { + ...snapshot.run, + status: 'retrying', + runId, + retry: { attempt: event.attempt, delayMs: event.delayMs }, + }, + }, + ]; + } + + if (event.type === 'auto_retry_end' && event.success === true) { + const { retry: _retry, ...run } = snapshot.run; + return [{ op: 'run.state', run: { ...run, status: 'running' } }]; + } + + if (event.type === 'compaction_start') { + const runId = snapshot.run.runId ?? this.createId(); + const id = this.createId(); + this.compactionId = id; + return [ + { + op: 'compaction.upsert', + node: { + kind: 'compaction', + id, + runId, + source: event.reason === 'manual' ? 'manual' : 'automatic', + status: 'running', + willRetry: false, + }, + }, + { + op: 'context.replace', + context: { ...snapshot.context, compaction: 'running', lastCompactionId: id }, + }, + { op: 'run.state', run: { ...snapshot.run, status: 'compacting', runId } }, + ]; + } + + if (event.type === 'compaction_end') { + const id = this.compactionId ?? this.createId(); + const runId = snapshot.run.runId ?? this.createId(); + this.compactionId = null; + return [ + { + op: 'compaction.upsert', + node: { + kind: 'compaction', + id, + runId, + source: event.reason === 'manual' ? 'manual' : 'automatic', + status: event.aborted === true || typeof event.errorMessage === 'string' + ? 'error' + : 'complete', + willRetry: event.willRetry === true, + }, + }, + { + op: 'context.replace', + context: { ...snapshot.context, compaction: 'idle', lastCompactionId: id }, + }, + { + op: 'run.state', + run: { + ...snapshot.run, + status: event.willRetry === true ? 'running' : snapshot.run.status, + runId, + }, + }, + ]; + } + + if (event.type === 'queue_update' + && Array.isArray(event.steering) + && Array.isArray(event.followUp)) { + const existing = [...snapshot.queue.items]; + const item = (mode: 'steer' | 'follow-up', text: string) => { + const matchIndex = existing.findIndex((candidate) => candidate.mode === mode + && candidate.text === text); + if (matchIndex >= 0) return existing.splice(matchIndex, 1)[0]!; + const id = this.createId(); + return { id, clientRequestId: id, mode, text, attachmentIds: [] }; + }; + return [{ + op: 'queue.replace', + queue: { + items: [ + ...event.steering.filter((value): value is string => typeof value === 'string') + .map((text) => item('steer', text)), + ...event.followUp.filter((value): value is string => typeof value === 'string') + .map((text) => item('follow-up', text)), + ], + }, + }]; + } + + if (event.type === 'agent_settled') { + return [ + { + op: 'run.state', + run: { + status: 'idle', + ...(snapshot.run.runId ? { runId: snapshot.run.runId } : {}), + settledAt: this.now(), + terminalReason: snapshot.run.status === 'aborting' ? 'aborted' : 'completed', + }, + }, + { op: 'queue.replace', queue: { items: [] } }, + ]; + } + + if (event.type === 'extension_ui_request' + && typeof event.id === 'string' + && typeof event.method === 'string' + && typeof event.title === 'string' + && ['select', 'confirm', 'input', 'editor'].includes(event.method)) { + const kind = event.method as 'select' | 'confirm' | 'input' | 'editor'; + return [{ + op: 'interaction.upsert', + interaction: { + id: event.id, + conversationId: snapshot.conversation.id, + runId: snapshot.run.runId ?? `interaction:${event.id}`, + kind, + title: event.title, + ...(typeof event.message === 'string' ? { message: event.message } : {}), + ...(kind === 'select' && Array.isArray(event.options) + ? { + options: event.options + .filter((option): option is string => typeof option === 'string') + .map((label, index) => ({ + id: `${event.id}:option:${index}`, + label, + })), + } + : {}), + status: 'pending', + }, + }]; + } + + if (event.type === 'message_start') { + const message = asRecord(event.message); + if (message?.role === 'user') { + const optimistic = snapshot.nodes.findLast( + (node): node is ConversationMessageNode => node.kind === 'message' + && node.role === 'user' + && node.status === 'optimistic', + ); + const id = optimistic?.id ?? this.createId(); + return [{ + op: 'message.upsert', + node: { + kind: 'message', + id, + ...(optimistic?.clientRequestId + ? { clientRequestId: optimistic.clientRequestId } + : {}), + role: 'user', + status: 'complete', + blocks: await liveContentBlocks( + id, + message.content, + snapshot.conversation.id, + this.projectImage, + ), + }, + }]; + } + if (message?.role !== 'assistant') return []; + const id = this.createId(); + this.assistantMessageId = id; + return [{ + op: 'message.upsert', + node: { + kind: 'message', + id, + role: 'assistant', + status: 'streaming', + blocks: [], + ...(usageOf(message.usage) ? { usage: usageOf(message.usage) } : {}), + }, + }]; + } + + if (event.type === 'message_update') { + const update = asRecord(event.assistantMessageEvent); + const message = currentAssistant(snapshot); + if (!update || !message || typeof update.contentIndex !== 'number') return []; + const blockId = `${message.id}:content:${update.contentIndex}`; + if (update.type === 'text_start') { + return [{ + op: 'message.upsert', + node: { + ...message, + blocks: [ + ...message.blocks.filter((block) => block.id !== blockId), + { kind: 'text', id: blockId, text: '', status: 'streaming' }, + ], + ...(usageOf(event.usage) ? { usage: usageOf(event.usage) } : {}), + }, + }]; + } + if (update.type === 'text_delta' && typeof update.delta === 'string') { + return [{ + op: 'message.block-delta', + messageId: message.id, + blockId, + delta: update.delta, + }]; + } + if (update.type === 'text_end' && typeof update.content === 'string') { + return [{ + op: 'message.upsert', + node: { + ...message, + blocks: message.blocks.map((block) => block.id === blockId && block.kind === 'text' + ? { ...block, text: update.content as string, status: 'complete' } + : block), + ...(usageOf(event.usage) ? { usage: usageOf(event.usage) } : {}), + }, + }]; + } + if (update.type === 'thinking_start') { + return [{ + op: 'message.upsert', + node: { + ...message, + blocks: [ + ...message.blocks.filter((block) => block.id !== blockId), + { kind: 'thinking', id: blockId, text: '', status: 'streaming' }, + ], + ...(usageOf(event.usage) ? { usage: usageOf(event.usage) } : {}), + }, + }]; + } + if (update.type === 'thinking_delta' && typeof update.delta === 'string') { + return [{ + op: 'message.block-delta', + messageId: message.id, + blockId, + delta: update.delta, + }]; + } + if (update.type === 'thinking_end' && typeof update.content === 'string') { + return [{ + op: 'message.upsert', + node: { + ...message, + blocks: message.blocks.map((block) => block.id === blockId && block.kind === 'thinking' + ? { ...block, text: update.content as string, status: 'complete' } + : block), + ...(usageOf(event.usage) ? { usage: usageOf(event.usage) } : {}), + }, + }]; + } + if (update.type === 'toolcall_start') { + const id = this.createId(); + this.toolDrafts.set(update.contentIndex, { id, argumentsText: '' }); + return [{ + op: 'tool.upsert', + node: { + kind: 'tool', + id, + toolCallId: `pending:${id}`, + toolName: 'tool', + title: 'tool', + inputText: '', + status: 'declared', + output: [], + }, + }]; + } + if (update.type === 'toolcall_delta' && typeof update.delta === 'string') { + const draft = this.toolDrafts.get(update.contentIndex); + if (draft) draft.argumentsText += update.delta; + return []; + } + if (update.type === 'toolcall_end') { + const toolCall = asRecord(update.toolCall); + if (!toolCall || typeof toolCall.id !== 'string' || typeof toolCall.name !== 'string') return []; + const draft = this.toolDrafts.get(update.contentIndex) ?? { + id: this.createId(), + argumentsText: '', + }; + this.toolDrafts.delete(update.contentIndex); + return [{ + op: 'tool.upsert', + node: { + kind: 'tool', + id: draft.id, + toolCallId: toolCall.id, + toolName: toolCall.name, + title: toolCall.name, + inputText: asRecord(toolCall.arguments) + ? JSON.stringify(toolCall.arguments) + : draft.argumentsText, + status: 'declared', + output: [], + }, + }]; + } + return []; + } + + if (event.type === 'tool_execution_start') { + if (typeof event.toolCallId !== 'string') return []; + const tool = currentTool(snapshot, event.toolCallId); + if (!tool) return []; + return [{ op: 'tool.upsert', node: { ...tool, status: 'running' } }]; + } + + if (event.type === 'tool_execution_update') { + if (typeof event.toolCallId !== 'string') return []; + const tool = currentTool(snapshot, event.toolCallId); + if (!tool) return []; + return [{ + op: 'tool.upsert', + node: { + ...tool, + status: 'running', + output: outputBlocks(tool.id, event.partialResult), + }, + }]; + } + + if (event.type === 'tool_execution_end') { + if (typeof event.toolCallId !== 'string') return []; + const tool = currentTool(snapshot, event.toolCallId); + if (!tool) return []; + return [{ + op: 'tool.upsert', + node: { + ...tool, + status: event.isError === true ? 'error' : 'complete', + output: outputBlocks(tool.id, event.result), + }, + }]; + } + + if (event.type === 'message_end') { + const message = asRecord(event.message); + if (message?.role === 'toolResult' && typeof message.toolCallId === 'string') { + const tool = currentTool(snapshot, message.toolCallId); + if (!tool) return []; + return [{ + op: 'tool.upsert', + node: { + ...tool, + status: message.isError === true ? 'error' : 'complete', + output: outputBlocks(tool.id, message), + }, + }]; + } + if (message?.role !== 'assistant') return []; + const current = currentAssistant(snapshot); + const id = current?.id ?? this.assistantMessageId ?? this.createId(); + this.assistantMessageId = null; + return [{ + op: 'message.upsert', + node: { + kind: 'message', + id, + role: 'assistant', + status: message.stopReason === 'aborted' + ? 'aborted' + : message.stopReason === 'error' ? 'error' : 'complete', + blocks: await liveContentBlocks( + id, + message.content, + snapshot.conversation.id, + this.projectImage, + ), + ...(usageOf(message.usage) ? { usage: usageOf(message.usage) } : {}), + ...(stopReasonOf(message.stopReason) ? { stopReason: stopReasonOf(message.stopReason) } : {}), + }, + }]; + } + + if (event.type !== 'agent_end') { + this.diagnostics.push({ eventType: event.type, reason: 'unsupported-event' }); + if (this.diagnostics.length > 32) this.diagnostics.shift(); + } + return []; + } +} diff --git a/electron/coding-runtime/pi/runtime.ts b/electron/coding-runtime/pi/runtime.ts index a12cd32..a4e0374 100644 --- a/electron/coding-runtime/pi/runtime.ts +++ b/electron/coding-runtime/pi/runtime.ts @@ -35,6 +35,10 @@ import { type ConversationReducerState, } from '../conversation-reducer'; import { CodingRuntimeContractError } from '../in-memory-conversation-runtime'; +import { + PiEventProjector, + type PiEventProjectorOptions, +} from './event-projector'; import { PiProcessError } from './process-errors'; import type { PiSessionRegistry } from './session-registry'; import { @@ -66,6 +70,10 @@ import { type PiWorkerPoolState, PiWorkerPool, } from './worker-pool'; +import { + PiSessionProjectionError, + projectPiSessionSnapshot, +} from './session-projector'; type RuntimeIdKind = 'run' | 'queue'; @@ -74,6 +82,7 @@ export interface PiConversationRuntimeOptions { registry: PiSessionRegistry; resolveModel(model: ProductModelRef): Promise; resolveImages?(attachments: Array<{ attachmentId: string }>): Promise; + projectImage?: PiEventProjectorOptions['projectImage']; createId?(kind: RuntimeIdKind): string; now?: () => number; providerRefreshCoordinator?: PiProviderRefreshCoordinator; @@ -332,6 +341,9 @@ function publicWorkerState(state: PiWorkerPoolState): ConversationSnapshot['work function runtimeFailure(error: unknown): CodingRuntimePublicError { if (error instanceof CodingRuntimeContractError) return clone(error.publicError); + if (error instanceof PiSessionProjectionError) { + return { code: error.code, message: error.message, recoverable: error.recoverable }; + } if (error instanceof PiProcessError) { if (error.code === 'PI_RPC_PROTOCOL_ERROR') { return { @@ -383,6 +395,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { private readonly registry: PiSessionRegistry; private readonly resolveModel: PiConversationRuntimeOptions['resolveModel']; private readonly resolveImages: NonNullable; + private readonly projectImage: PiEventProjectorOptions['projectImage']; private readonly createRuntimeId: (kind: RuntimeIdKind) => string; private readonly now: () => number; private readonly providerRefresh: PiProviderRefreshCoordinator; @@ -391,6 +404,12 @@ export class PiConversationRuntime implements CodingConversationRuntime { private readonly states = new Map(); private readonly inputs = new Map(); private readonly listeners = new Set<(patch: ConversationPatchEnvelope) => void>(); + private readonly projectors = new Map(); + private readonly projectionChains = new Map>(); + private readonly hydrationFlights = new Map< + string, + { generation: number; flight: Promise } + >(); private readonly unsubscribePool: () => void; constructor(options: PiConversationRuntimeOptions) { @@ -407,6 +426,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { } return []; }); + this.projectImage = options.projectImage; this.createRuntimeId = options.createId ?? (() => randomUUID()); this.now = options.now ?? Date.now; this.providerRefresh = options.providerRefreshCoordinator ?? new PiProviderRefreshCoordinator(); @@ -433,21 +453,33 @@ export class PiConversationRuntime implements CodingConversationRuntime { const worker = await this.prepareWorker(canonicalInput); await this.registry.ensureBinding(canonicalInput, async () => clone(worker.session)); this.inputs.set(input.conversationId, clone(canonicalInput)); - if (!this.states.has(input.conversationId)) { + const isNewState = !this.states.has(input.conversationId); + if (isNewState) { this.states.set(input.conversationId, createConversationReducerState( emptySnapshot(canonicalInput, worker), )); + this.resetProjector(input.conversationId); } else { this.emit(input.conversationId, { op: 'worker.state', state: publicWorkerState(worker) }); } + if (isNewState) { + try { + await this.requestHydration(input.conversationId, worker, false); + } catch (error) { + this.recordProjectionFailure(input.conversationId, worker.generation, error); + throw error; + } + } return this.runtimeState(input.conversationId); } async getSnapshot(conversationId: string): Promise { + await this.waitForProjection(conversationId); return clone(this.snapshot(conversationId)); } async prompt(input: PromptConversationInput): Promise { + await this.waitForProjection(input.conversationId); if (input.mode === 'steer') { const acceptance = await this.steer(input); return { @@ -474,6 +506,20 @@ export class PiConversationRuntime implements CodingConversationRuntime { this.snapshot(input.conversationId); const images = await this.resolveImages(input.attachments); const runId = this.id('run'); + const messageId = `client:${input.clientRequestId}`; + this.emit(input.conversationId, { + op: 'message.upsert', + node: { + kind: 'message', + id: messageId, + clientRequestId: input.clientRequestId, + role: 'user', + status: 'optimistic', + blocks: input.text.length > 0 + ? [{ kind: 'text', id: `${messageId}:content:0`, text: input.text, status: 'complete' }] + : [], + }, + }); const command: PiRpcCommand = { type: 'prompt', message: input.text, @@ -515,6 +561,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { } async abort(conversationId: string): Promise { + await this.waitForProjection(conversationId); const current = this.snapshot(conversationId).run; this.emit(conversationId, { op: 'run.state', @@ -532,6 +579,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { } async setModel(input: SetConversationModelInput): Promise { + await this.waitForProjection(input.conversationId); const snapshot = this.snapshot(input.conversationId); const thinkingLevel = snapshot.conversation.model.model?.thinkingLevel ?? 'off'; const selection = await this.resolveModel({ @@ -552,10 +600,15 @@ export class PiConversationRuntime implements CodingConversationRuntime { this.replaceModel(input.conversationId, persisted); try { const worker = await this.pool.reconfigureConversationModel(input.conversationId, persisted); - if (worker) await this.hydrateRecoveredGeneration(input.conversationId, worker, true); + if (worker) await this.requestHydration(input.conversationId, worker, true); } catch (error) { const worker = this.pool.getState(input.conversationId); - if (worker) this.replaceWorkerGeneration(input.conversationId, worker, true); + if (worker) { + if (this.snapshot(input.conversationId).cursor.workerGeneration !== worker.generation) { + this.replaceWorkerGeneration(input.conversationId, worker, true); + } + this.recordProjectionFailure(input.conversationId, worker.generation, error); + } throw error; } return clone(persisted); @@ -572,6 +625,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { } async setThinking(input: SetThinkingLevelInput): Promise { + await this.waitForProjection(input.conversationId); const current = this.snapshot(input.conversationId).conversation.model; if (!current.model) { throw new CodingRuntimeContractError( @@ -595,6 +649,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { } async compact(conversationId: string): Promise { + await this.waitForProjection(conversationId); const runId = this.id('run'); const ticket = this.pool.startTopLevel({ conversationId, @@ -614,6 +669,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { } async fork(input: ForkConversationInput): Promise { + await this.waitForProjection(input.sourceConversationId); this.snapshot(input.sourceConversationId); const registered = await this.registry.prepare(input.conversation); const canonicalInput: PrepareConversationInput = { @@ -635,12 +691,24 @@ export class PiConversationRuntime implements CodingConversationRuntime { this.inputs.set(canonicalInput.conversationId, clone(canonicalInput)); const snapshot = emptySnapshot(canonicalInput, worker); this.states.set(canonicalInput.conversationId, createConversationReducerState(snapshot)); - return { conversationId: canonicalInput.conversationId, snapshot: clone(snapshot) }; + this.resetProjector(canonicalInput.conversationId); + try { + await this.requestHydration(canonicalInput.conversationId, worker, false); + } catch (error) { + this.recordProjectionFailure(canonicalInput.conversationId, worker.generation, error); + throw error; + } + return { + conversationId: canonicalInput.conversationId, + snapshot: clone(this.snapshot(canonicalInput.conversationId)), + }; } async recover(conversationId: string): Promise { + await this.waitForProjection(conversationId); const state = await this.pool.recover(conversationId); - await this.hydrateRecoveredGeneration(conversationId, state, false); + await this.requestHydration(conversationId, state, false); + this.settleRecoveredRun(conversationId); return this.runtimeState(conversationId); } @@ -649,6 +717,9 @@ export class PiConversationRuntime implements CodingConversationRuntime { this.registry.forget(conversationId); this.inputs.delete(conversationId); this.states.delete(conversationId); + this.projectors.delete(conversationId); + this.projectionChains.delete(conversationId); + this.hydrationFlights.delete(conversationId); } subscribe(listener: (patch: ConversationPatchEnvelope) => void): () => void { @@ -659,12 +730,16 @@ export class PiConversationRuntime implements CodingConversationRuntime { async shutdown(): Promise { this.unsubscribePool(); await this.pool.shutdown(); + this.projectors.clear(); + this.projectionChains.clear(); + this.hydrationFlights.clear(); } private async queue( mode: 'steer' | 'follow-up', input: QueueMessageInput, ): Promise { + await this.waitForProjection(input.conversationId); const snapshot = this.snapshot(input.conversationId); const images = await this.resolveImages(input.attachments); const queuePosition = snapshot.queue.items.length + 1; @@ -720,7 +795,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { if (!this.pool.getState(input.conversationId)) return; const recovered = await this.pool.recover(input.conversationId); if (this.states.has(input.conversationId)) { - await this.hydrateRecoveredGeneration(input.conversationId, recovered, false); + await this.requestHydration(input.conversationId, recovered, false); } }, }); @@ -748,7 +823,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { refreshCredential: async () => await this.refreshCredential!(accountId), reopenWorker: async () => { const recovered = await this.pool.recover(conversationId); - await this.hydrateRecoveredGeneration(conversationId, recovered, true); + await this.requestHydration(conversationId, recovered, true); }, }); } else { @@ -822,7 +897,12 @@ export class PiConversationRuntime implements CodingConversationRuntime { private onPoolEvent(event: PiWorkerPoolEvent): void { if (event.type === 'worker.replaced') { + if (!this.states.has(event.conversationId)) return; this.replaceWorkerGeneration(event.conversationId, event.state, true); + this.resetProjector(event.conversationId); + void this.requestHydration(event.conversationId, event.state, true).catch((error) => { + this.recordProjectionFailure(event.conversationId, event.generation, error); + }); return; } if (event.type === 'worker.crashed') { @@ -830,19 +910,27 @@ export class PiConversationRuntime implements CodingConversationRuntime { if (state) this.emit(event.conversationId, { op: 'worker.state', state: publicWorkerState(state) }); return; } - if (event.event.type !== 'agent_settled') return; - const current = this.states.get(event.conversationId)?.snapshot?.run; - if (!current || current.status === 'idle') return; - this.emit(event.conversationId, { - op: 'run.state', - run: { - status: 'idle', - ...(current.runId ? { runId: current.runId } : {}), - settledAt: this.now(), - terminalReason: current.status === 'aborting' ? 'aborted' : 'completed', - }, - }, current.runId); - this.emit(event.conversationId, { op: 'queue.replace', queue: { items: [] } }, current.runId); + void this.enqueueProjection(event.conversationId, async () => { + const snapshot = this.states.get(event.conversationId)?.snapshot; + if (!snapshot || snapshot.cursor.workerGeneration !== event.generation) return; + const projector = this.projector(event.conversationId); + const patches = await projector.project(snapshot, event.event); + for (const patch of patches) { + this.emit(event.conversationId, patch, this.snapshot(event.conversationId).run.runId); + } + if (event.event.type === 'agent_end' || event.event.type === 'agent_settled') { + const worker = this.pool.getState(event.conversationId); + if (worker?.generation === event.generation) { + await this.hydrateGenerationNow( + event.conversationId, + worker, + event.event.type === 'agent_end', + ); + } + } + }).catch((error) => { + this.recordProjectionFailure(event.conversationId, event.generation, error); + }); } private failRun(conversationId: string, runId: string, error: unknown): void { @@ -860,16 +948,121 @@ export class PiConversationRuntime implements CodingConversationRuntime { }, runId); } - private async hydrateRecoveredGeneration( + private requestHydration( conversationId: string, workerState: PiWorkerPoolState, preserveRun: boolean, ): Promise { - await Promise.all([ + const existing = this.hydrationFlights.get(conversationId); + if (existing?.generation === workerState.generation) return existing.flight; + const hydration = this.enqueueProjection(conversationId, async () => { + await this.hydrateGenerationNow(conversationId, workerState, preserveRun); + }); + const flight = hydration.finally(() => { + if (this.hydrationFlights.get(conversationId)?.flight === flight) { + this.hydrationFlights.delete(conversationId); + } + }); + this.hydrationFlights.set(conversationId, { generation: workerState.generation, flight }); + return flight; + } + + private async hydrateGenerationNow( + conversationId: string, + workerState: PiWorkerPoolState, + preserveRun: boolean, + ): Promise { + const currentWorker = this.pool.getState(conversationId); + if (!currentWorker || currentWorker.generation !== workerState.generation) return; + if (this.snapshot(conversationId).cursor.workerGeneration !== workerState.generation) { + this.replaceWorkerGeneration(conversationId, workerState, preserveRun); + this.resetProjector(conversationId); + } + const [stateResponse, entriesResponse, statsResponse] = await Promise.all([ this.pool.request(conversationId, { type: 'get_state' }, { retry: 'read-only-once' }), this.pool.request(conversationId, { type: 'get_entries' }, { retry: 'read-only-once' }), + this.pool.request(conversationId, { type: 'get_session_stats' }, { retry: 'read-only-once' }), ]); - this.replaceWorkerGeneration(conversationId, workerState, preserveRun); + const before = this.snapshot(conversationId); + let projected = await projectPiSessionSnapshot({ + snapshot: before, + workerGeneration: workerState.generation, + state: stateResponse.data, + entries: entriesResponse.data, + stats: statsResponse.data, + ...(this.projectImage ? { projectImage: this.projectImage } : {}), + }); + if (preserveRun) { + projected = { + ...projected, + run: clone(before.run), + queue: clone(before.queue), + }; + } + this.states.set(conversationId, createConversationReducerState(projected)); + } + + private enqueueProjection(conversationId: string, action: () => Promise): Promise { + const previous = this.projectionChains.get(conversationId) ?? Promise.resolve(); + const flight = previous.then(action); + const tail = flight.catch(() => undefined); + this.projectionChains.set(conversationId, tail); + void tail.finally(() => { + if (this.projectionChains.get(conversationId) === tail) { + this.projectionChains.delete(conversationId); + } + }); + return flight; + } + + private async waitForProjection(conversationId: string): Promise { + await this.projectionChains.get(conversationId); + } + + private projector(conversationId: string): PiEventProjector { + const existing = this.projectors.get(conversationId); + if (existing) return existing; + return this.resetProjector(conversationId); + } + + private resetProjector(conversationId: string): PiEventProjector { + const projector = new PiEventProjector({ + createId: randomUUID, + now: this.now, + ...(this.projectImage ? { projectImage: this.projectImage } : {}), + }); + this.projectors.set(conversationId, projector); + return projector; + } + + private recordProjectionFailure( + conversationId: string, + generation: number, + error: unknown, + ): void { + const state = this.states.get(conversationId); + const snapshot = state?.snapshot; + if (!state || !snapshot || snapshot.cursor.workerGeneration !== generation) return; + const publicError = runtimeFailure(error); + this.states.set(conversationId, createConversationReducerState({ + ...clone(snapshot), + worker: { status: 'error', generation, error: publicError }, + run: { + ...clone(snapshot.run), + status: 'error', + error: publicError, + }, + })); + } + + private settleRecoveredRun(conversationId: string): void { + const current = this.snapshot(conversationId); + this.states.set(conversationId, createConversationReducerState({ + ...clone(current), + run: { status: 'idle' }, + queue: { items: [] }, + pendingInteractions: [], + })); } private replaceWorkerGeneration( diff --git a/electron/coding-runtime/pi/session-projector.ts b/electron/coding-runtime/pi/session-projector.ts new file mode 100644 index 0000000..7f2b7b5 --- /dev/null +++ b/electron/coding-runtime/pi/session-projector.ts @@ -0,0 +1,368 @@ +import type { + CodingRuntimeErrorCode, + ConversationContentBlock, + ConversationMessageNode, + ConversationNode, + ConversationSnapshot, + ConversationToolNode, + PublicUsage, +} from '../contracts'; +import type { + PiEventProjectorOptions, + PiImageProjectionInput, + PiProjectedAttachment, +} from './event-projector'; + +export interface PiSessionSnapshotInput { + snapshot: ConversationSnapshot; + workerGeneration: number; + state: unknown; + entries: unknown; + stats?: unknown; + projectImage?(image: PiImageProjectionInput): Promise; +} + +interface SessionEntryRecord extends Record { + id: string; + parentId: string | null; + type: string; +} + +export class PiSessionProjectionError extends Error { + readonly code: CodingRuntimeErrorCode = 'CODING_SESSION_UNREADABLE'; + readonly recoverable = true; + + constructor() { + super('Pi session data is unreadable; the source session was left unchanged.'); + this.name = 'PiSessionProjectionError'; + } +} + +function unreadable(): never { + throw new PiSessionProjectionError(); +} + +function asRecord(value: unknown): Record | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : null; +} + +function sessionEntry(value: unknown): SessionEntryRecord | null { + const entry = asRecord(value); + if (!entry + || typeof entry.id !== 'string' + || (entry.parentId !== null && typeof entry.parentId !== 'string') + || typeof entry.type !== 'string') return null; + return entry as SessionEntryRecord; +} + +function usageOf(value: unknown): PublicUsage | undefined { + const usage = asRecord(value); + if (!usage + || typeof usage.input !== 'number' + || typeof usage.output !== 'number') return undefined; + return { + inputTokens: usage.input, + outputTokens: usage.output, + ...(typeof usage.cacheRead === 'number' ? { cacheReadTokens: usage.cacheRead } : {}), + ...(typeof usage.cacheWrite === 'number' ? { cacheWriteTokens: usage.cacheWrite } : {}), + }; +} + +function stopReasonOf(value: unknown): ConversationMessageNode['stopReason'] | undefined { + if (value === 'toolUse') return 'tool-use'; + if (value === 'stop' || value === 'length' || value === 'error' || value === 'aborted') { + return value; + } + return undefined; +} + +function activePath(entriesValue: unknown, leafId: unknown): SessionEntryRecord[] { + if (!Array.isArray(entriesValue)) unreadable(); + if (leafId !== null && typeof leafId !== 'string') unreadable(); + const index = new Map(); + for (const value of entriesValue) { + const entry = sessionEntry(value); + if (!entry || index.has(entry.id)) unreadable(); + index.set(entry.id, entry); + } + if (leafId === null) return []; + const path: SessionEntryRecord[] = []; + const visited = new Set(); + let current = index.get(leafId); + if (!current) unreadable(); + while (current) { + if (visited.has(current.id)) unreadable(); + visited.add(current.id); + path.push(current); + if (current.parentId === null) break; + const parent = index.get(current.parentId); + if (!parent) unreadable(); + current = parent; + } + return path.reverse(); +} + +function retainedTail(path: SessionEntryRecord[]): SessionEntryRecord[] { + const compactionIndex = path.findLastIndex((entry) => entry.type === 'compaction'); + if (compactionIndex < 0) return path; + const compaction = path[compactionIndex]; + const firstKeptEntryId = compaction.firstKeptEntryId; + if (typeof firstKeptEntryId !== 'string') unreadable(); + const firstKeptIndex = path.findIndex( + (entry, index) => index < compactionIndex && entry.id === firstKeptEntryId, + ); + if (firstKeptIndex < 0) unreadable(); + return [ + compaction, + ...path.slice(firstKeptIndex, compactionIndex), + ...path.slice(compactionIndex + 1), + ]; +} + +async function contentBlocks( + messageId: string, + content: unknown, + conversationId: string, + projectImage: PiEventProjectorOptions['projectImage'], + idPart: 'content' | 'output', +): Promise { + const values = typeof content === 'string' + ? [{ type: 'text', text: content }] + : Array.isArray(content) ? content : []; + const blocks: ConversationContentBlock[] = []; + for (let contentIndex = 0; contentIndex < values.length; contentIndex += 1) { + const block = asRecord(values[contentIndex]); + if (block?.type === 'thinking' && typeof block.thinking === 'string') { + blocks.push({ + kind: 'thinking', + id: `${messageId}:${idPart}:${contentIndex}`, + text: block.thinking, + status: 'complete', + }); + } else if (block?.type === 'text' && typeof block.text === 'string') { + blocks.push({ + kind: 'text', + id: `${messageId}:${idPart}:${contentIndex}`, + text: block.text, + status: 'complete', + }); + } else if (block?.type === 'image' + && typeof block.data === 'string' + && typeof block.mimeType === 'string' + && projectImage) { + const attachment = await projectImage({ + conversationId, + data: block.data, + mime: block.mimeType, + source: 'session', + }); + blocks.push({ + kind: 'image', + id: `${messageId}:${idPart}:${contentIndex}`, + attachmentId: attachment.attachmentId, + mime: attachment.mime, + }); + } + } + return blocks; +} + +function inputTextOf(value: unknown): string { + if (typeof value === 'string') return value; + try { + return JSON.stringify(value ?? {}) ?? '{}'; + } catch { + return '{}'; + } +} + +async function projectEntries( + path: SessionEntryRecord[], + input: PiSessionSnapshotInput, +): Promise { + const nodes: ConversationNode[] = []; + const tools = new Map(); + for (const entry of path) { + if (entry.type === 'compaction') { + nodes.push({ + kind: 'compaction', + id: `entry:${entry.id}`, + runId: `session:${entry.id}`, + source: 'automatic', + status: 'complete', + willRetry: false, + }); + continue; + } + if (entry.type !== 'message') continue; + const message = asRecord(entry.message); + if (!message) unreadable(); + if (message.role === 'toolResult') { + if (typeof message.toolCallId !== 'string') unreadable(); + const tool = tools.get(message.toolCallId); + if (!tool) continue; + tool.output = await contentBlocks( + tool.id, + message.content, + input.snapshot.conversation.id, + input.projectImage, + 'output', + ); + tool.status = message.isError === true ? 'error' : 'complete'; + continue; + } + if (message.role !== 'user' && message.role !== 'assistant') continue; + const messageId = `entry:${entry.id}`; + const usage = usageOf(message.usage); + const stopReason = stopReasonOf(message.stopReason); + const node: ConversationMessageNode = { + kind: 'message', + id: messageId, + sourceEntryId: entry.id, + role: message.role, + status: message.stopReason === 'aborted' + ? 'aborted' + : message.stopReason === 'error' ? 'error' : 'complete', + blocks: await contentBlocks( + messageId, + message.content, + input.snapshot.conversation.id, + input.projectImage, + 'content', + ), + ...(usage ? { usage } : {}), + ...(stopReason ? { stopReason } : {}), + }; + nodes.push(node); + if (!Array.isArray(message.content)) continue; + for (const value of message.content) { + const block = asRecord(value); + if (block?.type !== 'toolCall' + || typeof block.id !== 'string' + || typeof block.name !== 'string') continue; + const tool: ConversationToolNode = { + kind: 'tool', + id: `tool:${block.id}`, + toolCallId: block.id, + toolName: block.name, + title: block.name, + inputText: inputTextOf(block.arguments), + status: 'declared', + output: [], + }; + nodes.push(tool); + tools.set(block.id, tool); + } + } + return nodes; +} + +function blockSignature(block: ConversationContentBlock): string { + if (block.kind === 'image') return `image:${block.mime}`; + return `${block.kind}:${block.text}`; +} + +function messageSignature(node: ConversationMessageNode): string { + return `${node.role}|${node.blocks.map(blockSignature).join('|')}`; +} + +function rebaseBlocks( + blocks: ConversationContentBlock[], + nodeId: string, + idPart: 'content' | 'output', +): ConversationContentBlock[] { + return blocks.map((block, index) => ({ ...block, id: `${nodeId}:${idPart}:${index}` })); +} + +function reconcileLiveIds( + durableNodes: ConversationNode[], + liveNodes: ConversationNode[], +): ConversationNode[] { + const used = new Set(); + return durableNodes.map((node) => { + if (node.kind === 'message') { + const live = liveNodes.find((candidate) => candidate.kind === 'message' + && !used.has(candidate.id) + && (candidate.sourceEntryId === node.sourceEntryId + || messageSignature(candidate) === messageSignature(node))); + if (!live || live.kind !== 'message') return node; + used.add(live.id); + return { + ...node, + id: live.id, + blocks: rebaseBlocks(node.blocks, live.id, 'content'), + ...(live.clientRequestId ? { clientRequestId: live.clientRequestId } : {}), + }; + } + if (node.kind === 'tool') { + const live = liveNodes.find((candidate) => candidate.kind === 'tool' + && !used.has(candidate.id) + && candidate.toolCallId === node.toolCallId); + if (!live || live.kind !== 'tool') return node; + used.add(live.id); + return { + ...node, + id: live.id, + output: rebaseBlocks(node.output, live.id, 'output'), + }; + } + if (node.kind === 'compaction') { + const live = liveNodes.findLast( + (candidate) => candidate.kind === 'compaction' && !used.has(candidate.id), + ); + if (!live || live.kind !== 'compaction') return node; + used.add(live.id); + return { ...node, id: live.id, runId: live.runId }; + } + return node; + }); +} + +function projectedContext(input: PiSessionSnapshotInput, state: Record) { + const stats = asRecord(input.stats); + const contextUsage = asRecord(stats?.contextUsage); + if (!contextUsage) { + return { + ...structuredClone(input.snapshot.context), + compaction: state.isCompacting === true ? 'running' as const : 'idle' as const, + }; + } + const tokens = contextUsage.tokens; + const recalculating = tokens === null; + return { + usedTokens: typeof tokens === 'number' ? tokens : 0, + contextWindow: typeof contextUsage.contextWindow === 'number' ? contextUsage.contextWindow : 0, + compaction: state.isCompacting === true ? 'running' as const : 'idle' as const, + ...(recalculating ? { recalculating: true } : {}), + }; +} + +export async function projectPiSessionSnapshot( + input: PiSessionSnapshotInput, +): Promise { + if (!Number.isSafeInteger(input.workerGeneration) || input.workerGeneration <= 0) unreadable(); + const response = asRecord(input.entries); + const state = asRecord(input.state); + if (!response || !state) unreadable(); + const path = retainedTail(activePath(response.entries, response.leafId)); + const durableNodes = await projectEntries(path, input); + const nodes = reconcileLiveIds(durableNodes, input.snapshot.nodes); + return { + ...structuredClone(input.snapshot), + nodes, + run: state.isStreaming === true + ? { ...structuredClone(input.snapshot.run), status: 'running' } + : { status: 'idle' }, + queue: { items: [] }, + context: projectedContext(input, state), + pendingInteractions: [], + worker: { status: 'ready', generation: input.workerGeneration }, + cursor: { + workerGeneration: input.workerGeneration, + seq: 0, + ...(typeof response.leafId === 'string' ? { leafEntryId: response.leafId } : {}), + }, + }; +} diff --git a/tests/fixtures/pi-0.84.2-projector-fixtures.ts b/tests/fixtures/pi-0.84.2-projector-fixtures.ts new file mode 100644 index 0000000..f9a5bd0 --- /dev/null +++ b/tests/fixtures/pi-0.84.2-projector-fixtures.ts @@ -0,0 +1,123 @@ +// Captured against @earendil-works/pi-coding-agent@0.84.2 RPC/session shapes. +export const PI_084_TEXT_TURN = { + usage: { + input: 12, + output: 4, + cacheRead: 2, + cacheWrite: 0, + totalTokens: 18, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + events: [ + { + type: 'message_start', + message: { role: 'user', content: 'Implement the change', timestamp: 1 }, + }, + { + type: 'message_start', + message: { + role: 'assistant', + content: [], + usage: { + input: 12, + output: 4, + cacheRead: 2, + cacheWrite: 0, + totalTokens: 18, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: 'stop', + timestamp: 2, + }, + }, + { + type: 'message_update', + assistantMessageEvent: { type: 'text_start', contentIndex: 0 }, + usage: { + input: 12, + output: 4, + cacheRead: 2, + cacheWrite: 0, + totalTokens: 18, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + }, + { + type: 'message_update', + assistantMessageEvent: { type: 'text_delta', contentIndex: 0, delta: 'Implemented' }, + usage: { + input: 12, + output: 4, + cacheRead: 2, + cacheWrite: 0, + totalTokens: 18, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + }, + { + type: 'message_update', + assistantMessageEvent: { type: 'text_end', contentIndex: 0, content: 'Implemented' }, + usage: { + input: 12, + output: 4, + cacheRead: 2, + cacheWrite: 0, + totalTokens: 18, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + }, + { + type: 'message_end', + message: { + role: 'assistant', + content: [{ type: 'text', text: 'Implemented' }], + usage: { + input: 12, + output: 4, + cacheRead: 2, + cacheWrite: 0, + totalTokens: 18, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: 'stop', + timestamp: 2, + }, + }, + ], + entries: { + leafId: 'entry-assistant-a', + entries: [ + { + type: 'message', + id: 'entry-user-a', + parentId: null, + timestamp: '2026-08-23T00:00:00.000Z', + message: { role: 'user', content: 'Implement the change', timestamp: 1 }, + }, + { + type: 'message', + id: 'entry-assistant-a', + parentId: 'entry-user-a', + timestamp: '2026-08-23T00:00:01.000Z', + message: { + role: 'assistant', + content: [{ type: 'text', text: 'Implemented' }], + usage: { + input: 12, + output: 4, + cacheRead: 2, + cacheWrite: 0, + totalTokens: 18, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: 'stop', + timestamp: 2, + }, + }, + ], + }, + stats: { + contextUsage: { tokens: 18, contextWindow: 100_000, percent: 0.018 }, + tokens: { input: 12, output: 4, cacheRead: 2, cacheWrite: 0, total: 18 }, + }, +} as const; diff --git a/tests/unit/pi-conversation-runtime.test.ts b/tests/unit/pi-conversation-runtime.test.ts index 70aeb94..794e32d 100644 --- a/tests/unit/pi-conversation-runtime.test.ts +++ b/tests/unit/pi-conversation-runtime.test.ts @@ -12,6 +12,7 @@ import { createMemoryCodingProjectStorage, } from '../../electron/coding-projects/project-store'; import { PiConversationRuntime } from '../../electron/coding-runtime/pi/runtime'; +import { PiSessionProjectionError } from '../../electron/coding-runtime/pi/session-projector'; import { PiSessionRegistry } from '../../electron/coding-runtime/pi/session-registry'; import { PiWorkerPool, @@ -24,18 +25,33 @@ import type { PiRpcRequestOptions, PiRpcResponse, } from '../../electron/coding-runtime/pi/rpc-client'; +import { PI_084_TEXT_TURN } from '../fixtures/pi-0.84.2-projector-fixtures'; const roots: string[] = []; const NOW = '2026-08-22T15:00:00.000Z'; class RuntimeFakeWorker implements PiConversationWorker { readonly requests: PiRpcCommand[] = []; + private stateData: unknown; + private entriesData: unknown = { entries: [], leafId: null }; + private statsData: unknown = { + contextUsage: { tokens: 0, contextWindow: 100_000, percent: 0 }, + tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }; private failType: string | null = null; private readonly responseGates = new Map>(); private readonly events = new Set<(event: PiRpcEvent) => void>(); private readonly invalidations = new Set<(error: PiProcessError) => void>(); - constructor(readonly id: string, readonly generation: number) {} + constructor(readonly id: string, readonly generation: number) { + this.stateData = { + sessionId: `session-${id}`, + thinkingLevel: 'medium', + isStreaming: false, + isCompacting: false, + pendingMessageCount: 0, + }; + } async request( command: PiRpcCommand, @@ -48,7 +64,23 @@ class RuntimeFakeWorker implements PiConversationWorker { this.failType = null; throw new Error(`fake ${command.type} rejection`); } - return { type: 'response', id: `${this.id}-${this.requests.length}`, success: true }; + const data = command.type === 'get_state' + ? this.stateData + : command.type === 'get_entries' + ? this.entriesData + : command.type === 'get_session_stats' ? this.statsData : undefined; + return { + type: 'response', + id: `${this.id}-${this.requests.length}`, + success: true, + ...(data === undefined ? {} : { data: structuredClone(data) as T }), + }; + } + + setSessionData(input: { state?: unknown; entries?: unknown; stats?: unknown }): void { + if (input.state !== undefined) this.stateData = structuredClone(input.state); + if (input.entries !== undefined) this.entriesData = structuredClone(input.entries); + if (input.stats !== undefined) this.statsData = structuredClone(input.stats); } failNext(type: string): void { this.failType = type; } @@ -124,6 +156,7 @@ describe('Pi Conversation runtime', () => { })); const workers = new Map(); const workerHistory = new Map(); + const durableSessions = new Map[0]>(); const openInputs: Array<{ conversationId: string; forkSource?: string; sourceEntryId?: string }> = []; const pool = new PiWorkerPool({ maxIdle: 4, @@ -136,6 +169,8 @@ describe('Pi Conversation runtime', () => { } : {}), }); const worker = new RuntimeFakeWorker(`worker-${conversation.conversationId}-${generation}`, generation); + const durable = durableSessions.get(conversation.conversationId); + if (durable) worker.setSessionData(durable); workers.set(conversation.conversationId, worker); workerHistory.set(conversation.conversationId, [ ...(workerHistory.get(conversation.conversationId) ?? []), @@ -181,16 +216,70 @@ describe('Pi Conversation runtime', () => { acceptanceResolved = true; return value; }); - await expect.poll(() => workers.get(left.id)!.requests.length).toBe(1); + await expect.poll(() => workers.get(left.id)!.requests.at(-1)?.type).toBe('prompt'); + expect(workers.get(left.id)!.requests).toHaveLength(4); expect(acceptanceResolved).toBe(false); releasePromptAcceptance(); const accepted = await acceptance; expect(accepted).toMatchObject({ accepted: true, runId: 'run-fixed', mode: 'prompt' }); expect((await runtime.getSnapshot(left.id)).run.status).toBe('running'); + for (const event of PI_084_TEXT_TURN.events) { + workers.get(left.id)!.emit(structuredClone(event)); + } + const streamed = await runtime.getSnapshot(left.id); + expect(streamed.nodes).toContainEqual(expect.objectContaining({ + kind: 'message', + id: 'client:request-left', + clientRequestId: 'request-left', + status: 'complete', + })); + expect(JSON.stringify(streamed.nodes)).toContain('Implemented'); + const durable = { + state: { + sessionId: `session-${left.id}`, + thinkingLevel: 'medium', + isStreaming: false, + isCompacting: false, + pendingMessageCount: 0, + }, + entries: PI_084_TEXT_TURN.entries, + stats: PI_084_TEXT_TURN.stats, + }; + durableSessions.set(left.id, durable); + workers.get(left.id)!.setSessionData(durable); workers.get(left.id)!.emit({ type: 'agent_end' }); expect((await runtime.getSnapshot(left.id)).run.status).toBe('running'); + const checkpoint = await runtime.getSnapshot(left.id); + expect(checkpoint.cursor.leafEntryId).toBe('entry-assistant-a'); + expect(checkpoint.nodes).toContainEqual(expect.objectContaining({ + kind: 'message', + id: 'client:request-left', + sourceEntryId: 'entry-user-a', + })); + const releaseSettledHydration = workers.get(left.id)!.holdNext('get_entries'); workers.get(left.id)!.emit({ type: 'agent_settled' }); - await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('idle'); + let continuationAccepted = false; + const continuation = runtime.followUp({ + clientRequestId: 'request-after-settled', + conversationId: left.id, + text: 'Continue only after the checkpoint', + attachments: [], + }).then((result) => { + continuationAccepted = true; + return result; + }); + await expect.poll(() => workers.get(left.id)!.requests.filter( + ({ type }) => type === 'get_entries', + ).length).toBe(3); + expect(workers.get(left.id)!.requests.some(({ type }) => type === 'follow_up')).toBe(false); + expect(continuationAccepted).toBe(false); + releaseSettledHydration(); + await expect(continuation).resolves.toMatchObject({ mode: 'follow-up', queuePosition: 1 }); + workers.get(left.id)!.emit({ type: 'agent_settled' }); + await expect.poll(async () => (await runtime.getSnapshot(left.id)).queue.items).toEqual([]); + expect((await runtime.getSnapshot(left.id)).run.status).toBe('idle'); + const settledNodes = (await runtime.getSnapshot(left.id)).nodes; + expect(settledNodes).toEqual(checkpoint.nodes); const changed = await runtime.setModel({ conversationId: left.id, @@ -204,11 +293,13 @@ describe('Pi Conversation runtime', () => { })); expect(workers.get(left.id)!.requests.map(({ type }) => type).sort()).toEqual([ 'get_entries', + 'get_session_stats', 'get_state', ]); expect(pool.getState(left.id)).toMatchObject({ generation: 2, state: 'ready' }); expect(pool.getState(right.id)).toMatchObject({ generation: 1, state: 'ready' }); expect((await runtime.getSnapshot(right.id)).conversation.model.model).toEqual(model); + expect((await runtime.getSnapshot(left.id)).nodes).toEqual(settledNodes); await expect(runtime.setModel({ conversationId: left.id, @@ -264,7 +355,11 @@ describe('Pi Conversation runtime', () => { 'abort', 'abort', ]); - expect(workers.get(right.id)!.requests).toHaveLength(0); + expect(workers.get(right.id)!.requests.map(({ type }) => type)).toEqual([ + 'get_state', + 'get_entries', + 'get_session_stats', + ]); workers.get(left.id)!.emit({ type: 'agent_settled' }); await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('idle'); expect((await runtime.getSnapshot(left.id)).queue.items).toEqual([]); @@ -276,22 +371,62 @@ describe('Pi Conversation runtime', () => { expect((await runtime.getSnapshot(right.id)).run.status).toBe('compacting'); expect(workers.get(right.id)!.requests.at(-1)).toEqual({ type: 'compact' }); expect((await runtime.getSnapshot(left.id)).run.status).toBe('idle'); + const rightDurable = { + entries: { + leafId: 'entry-right-user', + entries: [{ + type: 'message', + id: 'entry-right-user', + parentId: null, + timestamp: NOW, + message: { role: 'user', content: 'Right history', timestamp: 1 }, + }], + }, + }; + durableSessions.set(right.id, rightDurable); + workers.get(right.id)!.setSessionData(rightDurable); workers.get(right.id)!.emit({ type: 'agent_end' }); expect((await runtime.getSnapshot(right.id)).run.status).toBe('compacting'); workers.get(right.id)!.emit({ type: 'agent_settled' }); await expect.poll(async () => (await runtime.getSnapshot(right.id)).run.status).toBe('idle'); + const leftBeforeFailedRecovery = await runtime.getSnapshot(left.id); + const rightBeforeFailedRecovery = await runtime.getSnapshot(right.id); + const corruptRightSession = { + entries: { + leafId: 'entry-corrupt', + entries: [{ + type: 'message', + id: 'entry-corrupt', + parentId: 'missing-parent', + timestamp: NOW, + message: { role: 'user', content: 'must not replace history', timestamp: 1 }, + }], + }, + }; + durableSessions.set(right.id, corruptRightSession); + await expect(runtime.recover(right.id)).rejects.toBeInstanceOf(PiSessionProjectionError); + expect((await runtime.getSnapshot(right.id)).nodes).toEqual(rightBeforeFailedRecovery.nodes); + expect((await runtime.getSnapshot(right.id)).worker).toMatchObject({ + status: 'error', + generation: 2, + error: { code: 'CODING_SESSION_UNREADABLE', recoverable: true }, + }); + expect(await runtime.getSnapshot(left.id)).toEqual(leftBeforeFailedRecovery); + + durableSessions.set(right.id, rightDurable); await expect(runtime.recover(right.id)).resolves.toMatchObject({ conversationId: right.id, status: 'ready', - workerGeneration: 2, + workerGeneration: 3, }); expect((await runtime.getSnapshot(right.id)).cursor).toMatchObject({ - workerGeneration: 2, + workerGeneration: 3, seq: 0, }); expect(workers.get(right.id)!.requests.map(({ type }) => type).sort()).toEqual([ 'get_entries', + 'get_session_stats', 'get_state', ]); expect(pool.getState(left.id)).toMatchObject({ generation: 2, state: 'idle' }); diff --git a/tests/unit/pi-event-projector.test.ts b/tests/unit/pi-event-projector.test.ts new file mode 100644 index 0000000..50c35a3 --- /dev/null +++ b/tests/unit/pi-event-projector.test.ts @@ -0,0 +1,470 @@ +// @vitest-environment node + +import { describe, expect, it } from 'vitest'; +import type { + ConversationPatch, + ConversationSnapshot, +} from '../../electron/coding-runtime/contracts'; +import { + createConversationReducerState, + reduceConversationPatch, +} from '../../electron/coding-runtime/conversation-reducer'; +import { PiEventProjector } from '../../electron/coding-runtime/pi/event-projector'; + +function emptySnapshot(): ConversationSnapshot { + return { + schemaVersion: 1, + conversation: { + id: 'conversation-a', + projectId: 'project-a', + agentId: 'agent-a', + title: 'Conversation A', + model: { + model: { accountId: 'account-a', modelId: 'model-a', thinkingLevel: 'medium' }, + modelResolution: 'resolved', + }, + }, + nodes: [], + run: { status: 'running', runId: 'run-a', mode: 'prompt', startedAt: 10 }, + queue: { items: [] }, + context: { usedTokens: 0, contextWindow: 0, compaction: 'idle' }, + pendingInteractions: [], + worker: { status: 'ready', generation: 1 }, + cursor: { workerGeneration: 1, seq: 0 }, + }; +} + +function apply( + snapshot: ConversationSnapshot, + patches: ConversationPatch[], +): ConversationSnapshot { + let state = createConversationReducerState(snapshot); + for (const patch of patches) { + state = reduceConversationPatch(state, { + conversationId: 'conversation-a', + workerGeneration: 1, + runId: 'run-a', + seq: (state.snapshot?.cursor.seq ?? 0) + 1, + at: 20, + patch, + }); + } + if (!state.snapshot || state.invalidation) throw new Error(state.invalidation?.reason); + return state.snapshot; +} + +describe('Pi event projector', () => { + it('keeps one assistant UI identity while content-index deltas become an authoritative message', async () => { + const projector = new PiEventProjector({ + createId: () => 'assistant-ui-a', + }); + let snapshot = emptySnapshot(); + const initialAssistant = { + role: 'assistant', + content: [], + api: 'openai-completions', + provider: 'managed-account-a', + model: 'model-a', + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: 'pending', + timestamp: 10, + }; + + snapshot = apply(snapshot, await projector.project(snapshot, { + type: 'message_start', + message: initialAssistant, + })); + snapshot = apply(snapshot, await projector.project(snapshot, { + type: 'message_update', + usage: { input: 11, output: 1, cacheRead: 2, cacheWrite: 3, totalTokens: 17 }, + assistantMessageEvent: { type: 'text_start', contentIndex: 0 }, + })); + snapshot = apply(snapshot, await projector.project(snapshot, { + type: 'message_update', + usage: { input: 11, output: 2, cacheRead: 2, cacheWrite: 3, totalTokens: 18 }, + assistantMessageEvent: { type: 'text_delta', contentIndex: 0, delta: 'Hel' }, + })); + snapshot = apply(snapshot, await projector.project(snapshot, { + type: 'message_update', + usage: { input: 11, output: 3, cacheRead: 2, cacheWrite: 3, totalTokens: 19 }, + assistantMessageEvent: { type: 'text_end', contentIndex: 0, content: 'Hello' }, + })); + snapshot = apply(snapshot, await projector.project(snapshot, { + type: 'message_end', + message: { + ...initialAssistant, + content: [{ type: 'text', text: 'Hello, world.' }], + usage: { + input: 11, + output: 4, + cacheRead: 2, + cacheWrite: 3, + totalTokens: 20, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: 'stop', + }, + })); + + expect(snapshot.nodes).toEqual([{ + kind: 'message', + id: 'assistant-ui-a', + role: 'assistant', + status: 'complete', + blocks: [{ + kind: 'text', + id: 'assistant-ui-a:content:0', + text: 'Hello, world.', + status: 'complete', + }], + usage: { + inputTokens: 11, + outputTokens: 4, + cacheReadTokens: 2, + cacheWriteTokens: 3, + }, + stopReason: 'stop', + }]); + }); + + it('replaces cumulative tool output and attaches camel-case toolResult to one tool card', async () => { + const ids = ['assistant-ui-a', 'tool-ui-a']; + const projector = new PiEventProjector({ + createId: () => ids.shift() as string, + }); + let snapshot = emptySnapshot(); + const project = async (event: Record) => { + snapshot = apply(snapshot, await projector.project(snapshot, event as { type: string })); + }; + const assistant = { + role: 'assistant', + content: [], + usage: { input: 5, output: 1, cacheRead: 0, cacheWrite: 0 }, + stopReason: 'pending', + timestamp: 10, + }; + + await project({ type: 'message_start', message: assistant }); + await project({ + type: 'message_update', + usage: assistant.usage, + assistantMessageEvent: { type: 'toolcall_start', contentIndex: 0 }, + }); + await project({ + type: 'message_update', + usage: assistant.usage, + assistantMessageEvent: { type: 'toolcall_delta', contentIndex: 0, delta: '{"path":' }, + }); + await project({ + type: 'message_update', + usage: assistant.usage, + assistantMessageEvent: { type: 'toolcall_delta', contentIndex: 0, delta: '"a.txt"}' }, + }); + await project({ + type: 'message_update', + usage: assistant.usage, + assistantMessageEvent: { + type: 'toolcall_end', + contentIndex: 0, + toolCall: { type: 'toolCall', id: 'call-a', name: 'read', arguments: { path: 'a.txt' } }, + }, + }); + await project({ + type: 'tool_execution_start', + toolCallId: 'call-a', + toolName: 'read', + args: { path: 'a.txt' }, + }); + await project({ + type: 'tool_execution_update', + toolCallId: 'call-a', + toolName: 'read', + args: { path: 'a.txt' }, + partialResult: { content: [{ type: 'text', text: 'one' }], details: {} }, + }); + await project({ + type: 'tool_execution_update', + toolCallId: 'call-a', + toolName: 'read', + args: { path: 'a.txt' }, + partialResult: { content: [{ type: 'text', text: 'one two' }], details: {} }, + }); + await project({ + type: 'tool_execution_end', + toolCallId: 'call-a', + toolName: 'read', + result: { content: [{ type: 'text', text: 'final' }], details: {} }, + isError: false, + }); + await project({ + type: 'message_end', + message: { + role: 'toolResult', + toolCallId: 'call-a', + toolName: 'read', + content: [{ type: 'text', text: 'final authoritative' }], + details: {}, + isError: false, + timestamp: 20, + }, + }); + + expect(snapshot.nodes.filter(({ kind }) => kind === 'tool')).toEqual([{ + kind: 'tool', + id: 'tool-ui-a', + toolCallId: 'call-a', + toolName: 'read', + title: 'read', + inputText: '{"path":"a.txt"}', + status: 'complete', + output: [{ + kind: 'text', + id: 'tool-ui-a:output:0', + text: 'final authoritative', + status: 'complete', + }], + }]); + expect(snapshot.nodes.filter(({ kind }) => kind === 'message')).toEqual([{ + kind: 'message', + id: 'assistant-ui-a', + role: 'assistant', + status: 'streaming', + blocks: [], + usage: { + inputTokens: 5, + outputTokens: 1, + cacheReadTokens: 0, + cacheWriteTokens: 0, + }, + }]); + }); + + it('keeps retries and compaction active until agent_settled is authoritative idle', async () => { + const ids = ['turn-start-a', 'retry-a', 'compaction-a', 'queue-a', 'queue-b']; + const projector = new PiEventProjector({ createId: () => ids.shift() as string }); + let snapshot = emptySnapshot(); + const project = async (event: Record) => { + snapshot = apply(snapshot, await projector.project(snapshot, event as { type: string })); + }; + + await project({ type: 'agent_start' }); + await project({ type: 'turn_start', turnIndex: 0, timestamp: 10 }); + await project({ + type: 'auto_retry_start', + attempt: 1, + maxAttempts: 3, + delayMs: 250, + errorMessage: 'sensitive upstream detail', + }); + expect(snapshot.run).toMatchObject({ + status: 'retrying', + runId: 'run-a', + retry: { attempt: 1, delayMs: 250 }, + }); + expect(JSON.stringify(snapshot)).not.toContain('sensitive upstream detail'); + + await project({ type: 'auto_retry_end', success: true, attempt: 1 }); + await project({ type: 'compaction_start', reason: 'threshold' }); + expect(snapshot.run.status).toBe('compacting'); + expect(snapshot.context.compaction).toBe('running'); + await project({ + type: 'compaction_end', + reason: 'threshold', + result: { summary: 'hidden summary', firstKeptEntryId: 'entry-kept', tokensBefore: 5000 }, + aborted: false, + willRetry: true, + }); + expect(snapshot.run.status).toBe('running'); + expect(snapshot.context.compaction).toBe('idle'); + expect(snapshot.nodes).toContainEqual(expect.objectContaining({ + kind: 'compaction', + id: 'compaction-a', + source: 'automatic', + status: 'complete', + willRetry: true, + })); + expect(JSON.stringify(snapshot)).not.toContain('hidden summary'); + + await project({ type: 'agent_end', messages: [], willRetry: true }); + expect(snapshot.run.status).toBe('running'); + await project({ + type: 'queue_update', + steering: ['Steer once'], + followUp: ['Follow later'], + }); + expect(snapshot.queue.items.map(({ mode, text }) => ({ mode, text }))).toEqual([ + { mode: 'steer', text: 'Steer once' }, + { mode: 'follow-up', text: 'Follow later' }, + ]); + await project({ type: 'agent_settled' }); + expect(snapshot.run).toMatchObject({ + status: 'idle', + runId: 'run-a', + terminalReason: 'completed', + }); + expect(snapshot.queue.items).toEqual([]); + }); + + it('reconciles the optimistic user node and projects image bytes through an attachment hook', async () => { + const projectedImages: unknown[] = []; + const projector = new PiEventProjector({ + createId: () => 'unexpected-id', + projectImage: async (image) => { + projectedImages.push(image); + return { attachmentId: 'attachment-a', mime: image.mime }; + }, + }); + let snapshot: ConversationSnapshot = { + ...emptySnapshot(), + nodes: [{ + kind: 'message', + id: 'user-ui-a', + clientRequestId: 'request-a', + role: 'user', + status: 'optimistic', + blocks: [{ + kind: 'text', + id: 'user-ui-a:content:0', + text: 'Inspect this', + status: 'complete', + }], + }], + }; + const rawImage = 'base64-image-bytes-that-must-not-enter-product-state'; + + snapshot = apply(snapshot, await projector.project(snapshot, { + type: 'message_start', + message: { + role: 'user', + content: [ + { type: 'text', text: 'Inspect this' }, + { type: 'image', data: rawImage, mimeType: 'image/png' }, + ], + timestamp: 20, + }, + })); + + expect(projectedImages).toEqual([{ + conversationId: 'conversation-a', + data: rawImage, + mime: 'image/png', + source: 'live', + }]); + expect(snapshot.nodes).toEqual([{ + kind: 'message', + id: 'user-ui-a', + clientRequestId: 'request-a', + role: 'user', + status: 'complete', + blocks: [ + { + kind: 'text', + id: 'user-ui-a:content:0', + text: 'Inspect this', + status: 'complete', + }, + { + kind: 'image', + id: 'user-ui-a:content:1', + attachmentId: 'attachment-a', + mime: 'image/png', + }, + ], + }]); + expect(JSON.stringify(snapshot)).not.toContain(rawImage); + }); + + it('assembles thinking by contentIndex and lets message_end replace the draft', async () => { + const projector = new PiEventProjector({ createId: () => 'assistant-thinking-a' }); + let snapshot = emptySnapshot(); + const assistant = { + role: 'assistant', + content: [], + usage: { input: 7, output: 1, cacheRead: 0, cacheWrite: 0 }, + stopReason: 'pending', + timestamp: 10, + }; + const project = async (event: Record) => { + snapshot = apply(snapshot, await projector.project(snapshot, event as { type: string })); + }; + + await project({ type: 'message_start', message: assistant }); + await project({ + type: 'message_update', + usage: assistant.usage, + assistantMessageEvent: { type: 'thinking_start', contentIndex: 0 }, + }); + await project({ + type: 'message_update', + usage: assistant.usage, + assistantMessageEvent: { type: 'thinking_delta', contentIndex: 0, delta: 'draft' }, + }); + await project({ + type: 'message_update', + usage: assistant.usage, + assistantMessageEvent: { type: 'thinking_end', contentIndex: 0, content: 'draft done' }, + }); + await project({ + type: 'message_end', + message: { + ...assistant, + content: [{ type: 'thinking', thinking: 'authoritative thinking' }], + stopReason: 'stop', + }, + }); + + expect(snapshot.nodes).toContainEqual(expect.objectContaining({ + kind: 'message', + id: 'assistant-thinking-a', + blocks: [{ + kind: 'thinking', + id: 'assistant-thinking-a:content:0', + text: 'authoritative thinking', + status: 'complete', + }], + })); + }); + + it('projects extension dialogs and bounds unknown-event diagnostics without raw details', async () => { + const projector = new PiEventProjector({ createId: () => 'unused-id' }); + let snapshot = emptySnapshot(); + + snapshot = apply(snapshot, await projector.project(snapshot, { + type: 'extension_ui_request', + id: 'interaction-a', + method: 'select', + title: 'Choose an action', + options: ['Keep', 'Discard'], + })); + expect(snapshot.pendingInteractions).toEqual([{ + id: 'interaction-a', + conversationId: 'conversation-a', + runId: 'run-a', + kind: 'select', + title: 'Choose an action', + options: [ + { id: 'interaction-a:option:0', label: 'Keep' }, + { id: 'interaction-a:option:1', label: 'Discard' }, + ], + status: 'pending', + }]); + + for (let index = 0; index < 40; index += 1) { + expect(await projector.project(snapshot, { + type: `unknown_${index}`, + rawSecret: `secret-${index}`, + })).toEqual([]); + } + const diagnostics = projector.getDiagnostics(); + expect(diagnostics).toHaveLength(32); + expect(diagnostics[0]).toEqual({ eventType: 'unknown_8', reason: 'unsupported-event' }); + expect(JSON.stringify(diagnostics)).not.toContain('secret-'); + }); +}); diff --git a/tests/unit/pi-runtime-auth-recovery.test.ts b/tests/unit/pi-runtime-auth-recovery.test.ts index 708f26a..38e2e9c 100644 --- a/tests/unit/pi-runtime-auth-recovery.test.ts +++ b/tests/unit/pi-runtime-auth-recovery.test.ts @@ -35,7 +35,22 @@ class AuthFailureWorker implements PiConversationWorker { if (command.type === 'prompt') { throw new PiProcessError('PI_RPC_RESPONSE_ERROR', '401 provider authentication failed'); } - return { type: 'response', id: 'fake', success: true }; + const data = command.type === 'get_state' + ? { sessionId: `session-${this.id}`, isStreaming: false, isCompacting: false } + : command.type === 'get_entries' + ? { entries: [], leafId: null } + : command.type === 'get_session_stats' + ? { + contextUsage: { tokens: 0, contextWindow: 100_000, percent: 0 }, + tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + } + : undefined; + return { + type: 'response', + id: 'fake', + success: true, + ...(data === undefined ? {} : { data: data as T }), + }; } subscribe(_listener: (event: PiRpcEvent) => void): () => void { return () => undefined; } diff --git a/tests/unit/pi-session-projector.test.ts b/tests/unit/pi-session-projector.test.ts new file mode 100644 index 0000000..72ba73c --- /dev/null +++ b/tests/unit/pi-session-projector.test.ts @@ -0,0 +1,363 @@ +// @vitest-environment node + +import { describe, expect, it } from 'vitest'; +import type { ConversationSnapshot } from '../../electron/coding-runtime/contracts'; +import { + PiSessionProjectionError, + projectPiSessionSnapshot, +} from '../../electron/coding-runtime/pi/session-projector'; + +function baseSnapshot(): ConversationSnapshot { + return { + schemaVersion: 1, + conversation: { + id: 'conversation-a', + projectId: 'project-a', + agentId: 'agent-a', + title: 'Conversation A', + model: { + model: { accountId: 'account-a', modelId: 'model-a', thinkingLevel: 'medium' }, + modelResolution: 'resolved', + }, + }, + nodes: [], + run: { status: 'idle' }, + queue: { items: [] }, + context: { usedTokens: 0, contextWindow: 0, compaction: 'idle' }, + pendingInteractions: [], + worker: { status: 'ready', generation: 1 }, + cursor: { workerGeneration: 1, seq: 0 }, + }; +} + +const zeroCost = { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }; + +describe('Pi session projector', () => { + it('hydrates only the authoritative active leaf path', async () => { + const snapshot = await projectPiSessionSnapshot({ + snapshot: baseSnapshot(), + workerGeneration: 2, + state: { + sessionId: 'pi-session-a', + thinkingLevel: 'medium', + isStreaming: false, + isCompacting: false, + pendingMessageCount: 0, + }, + entries: { + leafId: 'entry-right-user', + entries: [ + { + type: 'message', + id: 'entry-root-user', + parentId: null, + timestamp: '2026-08-23T00:00:00.000Z', + message: { role: 'user', content: 'Root question', timestamp: 1 }, + }, + { + type: 'message', + id: 'entry-left-assistant', + parentId: 'entry-root-user', + timestamp: '2026-08-23T00:00:01.000Z', + message: { + role: 'assistant', + content: [{ type: 'text', text: 'Active answer' }], + usage: { + input: 4, + output: 2, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 6, + cost: zeroCost, + }, + stopReason: 'stop', + timestamp: 2, + }, + }, + { + type: 'message', + id: 'entry-abandoned-assistant', + parentId: 'entry-root-user', + timestamp: '2026-08-23T00:00:02.000Z', + message: { + role: 'assistant', + content: [{ type: 'text', text: 'Abandoned answer' }], + usage: { + input: 4, + output: 2, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 6, + cost: zeroCost, + }, + stopReason: 'stop', + timestamp: 3, + }, + }, + { + type: 'message', + id: 'entry-right-user', + parentId: 'entry-left-assistant', + timestamp: '2026-08-23T00:00:03.000Z', + message: { role: 'user', content: 'Continue active branch', timestamp: 4 }, + }, + ], + }, + }); + + expect(snapshot.cursor).toEqual({ + workerGeneration: 2, + seq: 0, + leafEntryId: 'entry-right-user', + }); + expect(snapshot.nodes.filter(({ kind }) => kind === 'message')).toEqual([ + expect.objectContaining({ + id: 'entry:entry-root-user', + sourceEntryId: 'entry-root-user', + role: 'user', + }), + expect.objectContaining({ + id: 'entry:entry-left-assistant', + sourceEntryId: 'entry-left-assistant', + role: 'assistant', + }), + expect.objectContaining({ + id: 'entry:entry-right-user', + sourceEntryId: 'entry-right-user', + role: 'user', + }), + ]); + expect(JSON.stringify(snapshot)).not.toContain('Abandoned answer'); + }); + + it('applies retained-tail compaction and reconciles durable entries without replacing live IDs', async () => { + const live: ConversationSnapshot = { + ...baseSnapshot(), + nodes: [ + { + kind: 'message', + id: 'live-user-a', + clientRequestId: 'request-a', + role: 'user', + status: 'complete', + blocks: [{ kind: 'text', id: 'live-user-a:content:0', text: 'Kept question', status: 'complete' }], + }, + { + kind: 'message', + id: 'live-assistant-a', + role: 'assistant', + status: 'complete', + blocks: [{ kind: 'text', id: 'live-assistant-a:content:0', text: 'Kept answer', status: 'complete' }], + }, + { + kind: 'tool', + id: 'live-tool-a', + toolCallId: 'call-a', + toolName: 'read', + title: 'read', + inputText: '{"path":"a.txt"}', + status: 'complete', + output: [{ kind: 'text', id: 'live-tool-a:output:0', text: 'final', status: 'complete' }], + }, + { + kind: 'compaction', + id: 'live-compaction-a', + runId: 'run-a', + source: 'automatic', + status: 'complete', + willRetry: false, + }, + ], + }; + const snapshot = await projectPiSessionSnapshot({ + snapshot: live, + workerGeneration: 2, + state: { + sessionId: 'pi-session-a', + thinkingLevel: 'medium', + isStreaming: false, + isCompacting: false, + pendingMessageCount: 0, + }, + stats: { + contextUsage: { tokens: null, contextWindow: 100_000, percent: null }, + tokens: { input: 20, output: 10, cacheRead: 5, cacheWrite: 0, total: 35 }, + }, + entries: { + leafId: 'entry-after-user', + entries: [ + { + type: 'message', + id: 'entry-old-user', + parentId: null, + timestamp: '2026-08-23T00:00:00.000Z', + message: { role: 'user', content: 'Summarized old question', timestamp: 1 }, + }, + { + type: 'message', + id: 'entry-kept-user', + parentId: 'entry-old-user', + timestamp: '2026-08-23T00:00:01.000Z', + message: { role: 'user', content: 'Kept question', timestamp: 2 }, + }, + { + type: 'message', + id: 'entry-kept-assistant', + parentId: 'entry-kept-user', + timestamp: '2026-08-23T00:00:02.000Z', + message: { + role: 'assistant', + content: [ + { type: 'text', text: 'Kept answer' }, + { type: 'toolCall', id: 'call-a', name: 'read', arguments: { path: 'a.txt' } }, + ], + usage: { + input: 12, + output: 4, + cacheRead: 2, + cacheWrite: 0, + totalTokens: 18, + cost: zeroCost, + }, + stopReason: 'toolUse', + timestamp: 3, + }, + }, + { + type: 'message', + id: 'entry-tool-result', + parentId: 'entry-kept-assistant', + timestamp: '2026-08-23T00:00:03.000Z', + message: { + role: 'toolResult', + toolCallId: 'call-a', + toolName: 'read', + content: [{ type: 'text', text: 'final authoritative' }], + details: {}, + isError: false, + timestamp: 4, + }, + }, + { + type: 'compaction', + id: 'entry-compaction', + parentId: 'entry-tool-result', + timestamp: '2026-08-23T00:00:04.000Z', + summary: 'summary must stay hidden', + firstKeptEntryId: 'entry-kept-user', + tokensBefore: 5000, + }, + { + type: 'message', + id: 'entry-after-user', + parentId: 'entry-compaction', + timestamp: '2026-08-23T00:00:05.000Z', + message: { role: 'user', content: 'After compaction', timestamp: 5 }, + }, + ], + }, + }); + + expect(snapshot.nodes.map(({ kind, id }) => ({ kind, id }))).toEqual([ + { kind: 'compaction', id: 'live-compaction-a' }, + { kind: 'message', id: 'live-user-a' }, + { kind: 'message', id: 'live-assistant-a' }, + { kind: 'tool', id: 'live-tool-a' }, + { kind: 'message', id: 'entry:entry-after-user' }, + ]); + expect(snapshot.nodes).toContainEqual(expect.objectContaining({ + kind: 'message', + id: 'live-user-a', + sourceEntryId: 'entry-kept-user', + clientRequestId: 'request-a', + })); + expect(snapshot.nodes).toContainEqual(expect.objectContaining({ + kind: 'message', + id: 'live-assistant-a', + sourceEntryId: 'entry-kept-assistant', + })); + expect(snapshot.nodes).toContainEqual(expect.objectContaining({ + kind: 'tool', + id: 'live-tool-a', + toolCallId: 'call-a', + status: 'complete', + output: [expect.objectContaining({ text: 'final authoritative' })], + })); + expect(snapshot.context).toEqual({ + usedTokens: 0, + contextWindow: 100_000, + compaction: 'idle', + recalculating: true, + }); + expect(JSON.stringify(snapshot)).not.toContain('Summarized old question'); + expect(JSON.stringify(snapshot)).not.toContain('summary must stay hidden'); + }); + + it('projects persisted images through attachment storage without retaining base64', async () => { + const rawImage = 'A'.repeat(1024 * 1024); + const projected = await projectPiSessionSnapshot({ + snapshot: baseSnapshot(), + workerGeneration: 2, + state: { + sessionId: 'pi-session-a', + isStreaming: false, + isCompacting: false, + }, + entries: { + leafId: 'entry-image', + entries: [{ + type: 'message', + id: 'entry-image', + parentId: null, + timestamp: '2026-08-23T00:00:00.000Z', + message: { + role: 'user', + content: [{ type: 'image', data: rawImage, mimeType: 'image/png' }], + timestamp: 1, + }, + }], + }, + projectImage: async (image) => { + expect(image).toMatchObject({ + conversationId: 'conversation-a', + mime: 'image/png', + source: 'session', + }); + expect(image.data).toBe(rawImage); + return { attachmentId: 'attachment-session-a', mime: image.mime }; + }, + }); + + expect(projected.nodes).toContainEqual(expect.objectContaining({ + kind: 'message', + blocks: [{ + kind: 'image', + id: 'entry:entry-image:content:0', + attachmentId: 'attachment-session-a', + mime: 'image/png', + }], + })); + expect(JSON.stringify(projected)).not.toContain(rawImage); + }); + + it('fails closed on an unreadable active path without mutating the last good snapshot', async () => { + const lastGood = baseSnapshot(); + const before = structuredClone(lastGood); + await expect(projectPiSessionSnapshot({ + snapshot: lastGood, + workerGeneration: 2, + state: { sessionId: 'pi-session-a', isStreaming: false, isCompacting: false }, + entries: { + leafId: 'entry-missing-parent', + entries: [{ + type: 'message', + id: 'entry-missing-parent', + parentId: 'not-present', + timestamp: '2026-08-23T00:00:00.000Z', + message: { role: 'user', content: 'Unreadable', timestamp: 1 }, + }], + }, + })).rejects.toBeInstanceOf(PiSessionProjectionError); + expect(lastGood).toEqual(before); + }); +});