From 4deba3a68b95329513d628e9a7189df69726a7af Mon Sep 17 00:00:00 2001 From: brother7 <7brother7@gmail.com> Date: Sat, 22 Aug 2026 19:08:26 +0800 Subject: [PATCH] feat: add conversation runtime contracts --- electron/coding-runtime/contracts.ts | 362 +++++++++++ .../coding-runtime/conversation-reducer.ts | 569 ++++++++++++++++++ .../in-memory-conversation-runtime.ts | 451 ++++++++++++++ .../coding-conversation-product-fixtures.ts | 344 +++++++++++ .../coding-conversation-contracts.test.ts | 397 ++++++++++++ 5 files changed, 2123 insertions(+) create mode 100644 electron/coding-runtime/contracts.ts create mode 100644 electron/coding-runtime/conversation-reducer.ts create mode 100644 electron/coding-runtime/in-memory-conversation-runtime.ts create mode 100644 tests/fixtures/coding-conversation-product-fixtures.ts create mode 100644 tests/unit/coding-conversation-contracts.test.ts diff --git a/electron/coding-runtime/contracts.ts b/electron/coding-runtime/contracts.ts new file mode 100644 index 0000000..7ac944e --- /dev/null +++ b/electron/coding-runtime/contracts.ts @@ -0,0 +1,362 @@ +export type ConversationThinkingLevel = 'off' | 'minimal' | 'low' | 'medium' | 'high'; + +export interface ProductModelRef { + accountId: string; + modelId: string; + thinkingLevel: ConversationThinkingLevel; +} + +export interface ConversationModelState { + model: ProductModelRef | null; + modelResolution: 'resolved' | 'required'; +} + +export interface PublicUsage { + inputTokens: number; + outputTokens: number; + cacheReadTokens?: number; + cacheWriteTokens?: number; +} + +export type CodingRuntimeErrorCode = + | 'CODING_RUNTIME_START_FAILED' + | 'CODING_RUNTIME_READY_TIMEOUT' + | 'CODING_RUNTIME_PROTOCOL_ERROR' + | 'CODING_PROVIDER_AUTH_REQUIRED' + | 'CODING_MODEL_UNAVAILABLE' + | 'CODING_SESSION_UNREADABLE' + | 'CODING_STORAGE_WRITE_FAILED' + | 'CODING_REQUEST_UNCERTAIN' + | 'CODING_MIGRATION_MODEL_REQUIRED' + | 'CODING_CONVERSATION_NOT_FOUND'; + +export interface CodingRuntimePublicError { + code: CodingRuntimeErrorCode; + message: string; + recoverable: boolean; +} + +export type ConversationRunStatus = + | 'idle' + | 'preparing' + | 'queued' + | 'running' + | 'retrying' + | 'compacting' + | 'aborting' + | 'error'; + +export type PromptMode = 'prompt' | 'steer' | 'follow-up'; + +export interface ConversationRunState { + status: ConversationRunStatus; + runId?: string; + mode?: PromptMode; + startedAt?: number; + settledAt?: number; + terminalReason?: 'completed' | 'aborted' | 'failed'; + retry?: { + attempt: number; + delayMs: number; + }; + error?: CodingRuntimePublicError; +} + +export interface ConversationQueueItem { + id: string; + clientRequestId: string; + mode: 'steer' | 'follow-up'; + text: string; + attachmentIds: string[]; +} + +export interface ConversationQueueState { + items: ConversationQueueItem[]; +} + +export interface ConversationContextState { + usedTokens: number; + contextWindow: number; + outputLimit?: number; + compaction: 'idle' | 'running'; + lastCompactionId?: string; +} + +export interface ConversationInteractionOption { + id: string; + label: string; + description?: string; +} + +export interface ConversationInteraction { + id: string; + conversationId: string; + runId: string; + kind: 'select' | 'confirm' | 'input' | 'editor'; + title: string; + message?: string; + options?: ConversationInteractionOption[]; + status: 'pending' | 'answered' | 'rejected' | 'cancelled'; +} + +export interface PublicWorkerState { + status: 'stopped' | 'starting' | 'ready' | 'recovering' | 'error'; + generation: number; + error?: CodingRuntimePublicError; +} + +export type ConversationContentBlock = + | { + kind: 'text'; + id: string; + text: string; + status: 'streaming' | 'complete'; + } + | { + kind: 'thinking'; + id: string; + text: string; + status: 'streaming' | 'complete'; + } + | { + kind: 'image'; + id: string; + attachmentId: string; + mime: string; + }; + +export interface ConversationMessageNode { + kind: 'message'; + id: string; + sourceEntryId?: string; + clientRequestId?: string; + role: 'user' | 'assistant'; + status: 'optimistic' | 'streaming' | 'complete' | 'error' | 'aborted'; + blocks: ConversationContentBlock[]; + usage?: PublicUsage; + stopReason?: 'stop' | 'length' | 'tool-use' | 'error' | 'aborted'; +} + +export interface ChangedFileDetailsV1 { + schema: 'changed-file.v1'; + paths: string[]; +} + +export interface TaskStateDetailsV1 { + schema: 'task-state.v1'; + tasks: Array<{ + id: string; + title: string; + status: 'pending' | 'running' | 'complete' | 'error'; + }>; +} + +export interface WriteLeaseDetailsV1 { + schema: 'write-lease.v1'; + status: 'waiting' | 'held' | 'released'; +} + +export interface SubagentDetailsV1 { + schema: 'subagent.v1'; + dispatchId: string; + mode: 'single' | 'parallel' | 'chain'; + tasks: Array<{ + taskId: string; + agentId: string; + toolProfile: 'read-only' | 'coding'; + status: 'queued' | 'running' | 'complete' | 'error' | 'aborted' | 'skipped'; + summary?: string; + errorCode?: string; + usage?: PublicUsage; + }>; +} + +export type KnownToolDetails = + | ChangedFileDetailsV1 + | TaskStateDetailsV1 + | WriteLeaseDetailsV1 + | SubagentDetailsV1; + +export interface ConversationToolNode { + kind: 'tool'; + id: string; + toolCallId: string; + toolName: string; + title: string; + inputText: string; + status: 'declared' | 'waiting' | 'running' | 'complete' | 'error' | 'aborted'; + output: ConversationContentBlock[]; + details?: KnownToolDetails; +} + +export interface ConversationCompactionNode { + kind: 'compaction'; + id: string; + runId: string; + source: 'manual' | 'automatic'; + status: 'running' | 'complete' | 'error'; + willRetry: boolean; + summary?: string; +} + +export interface ConversationBoundaryNode { + kind: 'boundary'; + id: string; + runId: string; + boundary: 'turn-start' | 'turn-end' | 'retry'; + attempt?: number; + delayMs?: number; +} + +export interface ConversationSubagentNode { + kind: 'subagent'; + id: string; + runId: string; + details: SubagentDetailsV1; +} + +export interface ConversationNoticeNode { + kind: 'notice'; + id: string; + code: string; + level: 'info' | 'warning' | 'error'; + message: string; +} + +export type ConversationNode = + | ConversationMessageNode + | ConversationToolNode + | ConversationCompactionNode + | ConversationBoundaryNode + | ConversationSubagentNode + | ConversationNoticeNode; + +export interface ConversationSnapshot { + schemaVersion: 1; + conversation: { + id: string; + projectId: string; + agentId: string; + title: string; + model: ConversationModelState; + }; + nodes: ConversationNode[]; + run: ConversationRunState; + queue: ConversationQueueState; + context: ConversationContextState; + pendingInteractions: ConversationInteraction[]; + worker: PublicWorkerState; + cursor: { + workerGeneration: number; + seq: number; + leafEntryId?: string; + }; +} + +export type ConversationPatch = + | { op: 'worker.state'; state: PublicWorkerState } + | { op: 'run.state'; run: ConversationRunState } + | { op: 'message.upsert'; node: ConversationMessageNode } + | { op: 'message.block-delta'; messageId: string; blockId: string; delta: string } + | { op: 'tool.upsert'; node: ConversationToolNode } + | { op: 'compaction.upsert'; node: ConversationCompactionNode } + | { op: 'subagent.upsert'; node: ConversationSubagentNode } + | { op: 'queue.replace'; queue: ConversationQueueState } + | { op: 'interaction.upsert'; interaction: ConversationInteraction } + | { op: 'interaction.remove'; interactionId: string } + | { op: 'context.replace'; context: ConversationContextState } + | { op: 'snapshot.invalidated'; reason: string }; + +export interface ConversationPatchEnvelope { + conversationId: string; + workerGeneration: number; + runId?: string; + seq: number; + at: number; + patch: ConversationPatch; +} + +export interface PrepareConversationInput { + conversationId: string; + projectId: string; + agentId: string; + title: string; + model: ConversationModelState; +} + +export interface ConversationRuntimeState { + conversationId: string; + status: PublicWorkerState['status']; + workerGeneration: number; + error?: CodingRuntimePublicError; +} + +export interface PromptConversationInput { + clientRequestId: string; + conversationId: string; + mode: PromptMode; + text: string; + attachments: Array<{ attachmentId: string }>; +} + +export interface QueueMessageInput { + clientRequestId: string; + conversationId: string; + text: string; + attachments: Array<{ attachmentId: string }>; +} + +export interface PromptAcceptance { + accepted: true; + conversationId: string; + clientRequestId: string; + runId: string; + mode: PromptMode; + queuePosition?: number; +} + +export interface QueueAcceptance { + accepted: true; + conversationId: string; + clientRequestId: string; + mode: 'steer' | 'follow-up'; + queuePosition: number; +} + +export interface SetConversationModelInput { + conversationId: string; + accountId: string; + modelId: string; +} + +export interface SetThinkingLevelInput { + conversationId: string; + thinkingLevel: ConversationThinkingLevel; +} + +export interface ForkConversationInput { + sourceConversationId: string; + sourceEntryId?: string; + conversation: PrepareConversationInput; +} + +export interface ForkResult { + conversationId: string; + snapshot: ConversationSnapshot; +} + +export interface CodingConversationRuntime { + prepare(input: PrepareConversationInput): Promise; + getSnapshot(conversationId: string): Promise; + prompt(input: PromptConversationInput): Promise; + steer(input: QueueMessageInput): Promise; + followUp(input: QueueMessageInput): Promise; + abort(conversationId: string): Promise; + setModel(input: SetConversationModelInput): Promise; + setThinking(input: SetThinkingLevelInput): Promise; + compact(conversationId: string): Promise; + fork(input: ForkConversationInput): Promise; + recover(conversationId: string): Promise; + dispose(conversationId: string): Promise; + subscribe(listener: (patch: ConversationPatchEnvelope) => void): () => void; +} diff --git a/electron/coding-runtime/conversation-reducer.ts b/electron/coding-runtime/conversation-reducer.ts new file mode 100644 index 0000000..20a1c69 --- /dev/null +++ b/electron/coding-runtime/conversation-reducer.ts @@ -0,0 +1,569 @@ +import type { + CodingRuntimePublicError, + ConversationContentBlock, + ConversationInteraction, + ConversationMessageNode, + ConversationNode, + ConversationPatch, + ConversationPatchEnvelope, + ConversationQueueState, + ConversationRunState, + ConversationSnapshot, + ConversationSubagentNode, + ConversationToolNode, + KnownToolDetails, + PublicWorkerState, +} from './contracts'; + +export type ConversationInvalidationCode = + | 'snapshot-required' + | 'unsupported-schema' + | 'malformed-snapshot' + | 'malformed-envelope' + | 'unknown-patch' + | 'generation-gap' + | 'sequence-gap' + | 'patch-target-missing' + | 'explicit-invalidation'; + +export interface ConversationInvalidation { + code: ConversationInvalidationCode; + reason: string; + expectedGeneration?: number; + actualGeneration?: number; + expectedSeq?: number; + actualSeq?: number; +} + +export interface ConversationReducerState { + snapshot: ConversationSnapshot | null; + invalidation: ConversationInvalidation | null; +} + +const RUN_STATUSES = new Set([ + 'idle', + 'preparing', + 'queued', + 'running', + 'retrying', + 'compacting', + 'aborting', + 'error', +]); + +const WORKER_STATUSES = new Set(['stopped', 'starting', 'ready', 'recovering', 'error']); +const ERROR_CODES = new Set([ + 'CODING_RUNTIME_START_FAILED', + 'CODING_RUNTIME_READY_TIMEOUT', + 'CODING_RUNTIME_PROTOCOL_ERROR', + 'CODING_PROVIDER_AUTH_REQUIRED', + 'CODING_MODEL_UNAVAILABLE', + 'CODING_SESSION_UNREADABLE', + 'CODING_STORAGE_WRITE_FAILED', + 'CODING_REQUEST_UNCERTAIN', + 'CODING_MIGRATION_MODEL_REQUIRED', + 'CODING_CONVERSATION_NOT_FOUND', +]); +const PATCH_OPS = new Set([ + 'worker.state', + 'run.state', + 'message.upsert', + 'message.block-delta', + 'tool.upsert', + 'compaction.upsert', + 'subagent.upsert', + 'queue.replace', + 'interaction.upsert', + 'interaction.remove', + 'context.replace', + 'snapshot.invalidated', +]); + +function clone(value: T): T { + return structuredClone(value); +} + +function asRecord(value: unknown): Record | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : null; +} + +function isNonEmptyString(value: unknown): value is string { + return typeof value === 'string' && value.length > 0; +} + +function isNonNegativeInteger(value: unknown): value is number { + return Number.isInteger(value) && Number(value) >= 0; +} + +function isOptionalNumber(value: unknown): boolean { + return value === undefined || typeof value === 'number'; +} + +function isPublicError(value: unknown): value is CodingRuntimePublicError { + const record = asRecord(value); + return record !== null + && ERROR_CODES.has(String(record.code)) + && isNonEmptyString(record.message) + && typeof record.recoverable === 'boolean'; +} + +function isModelState(value: unknown): boolean { + const record = asRecord(value); + if (!record || !['resolved', 'required'].includes(String(record.modelResolution))) return false; + if (record.model === null) return record.modelResolution === 'required'; + const model = asRecord(record.model); + return model !== null + && isNonEmptyString(model.accountId) + && isNonEmptyString(model.modelId) + && ['off', 'minimal', 'low', 'medium', 'high'].includes(String(model.thinkingLevel)) + && record.modelResolution === 'resolved'; +} + +function isUsage(value: unknown): boolean { + const record = asRecord(value); + return record !== null + && typeof record.inputTokens === 'number' + && typeof record.outputTokens === 'number' + && isOptionalNumber(record.cacheReadTokens) + && isOptionalNumber(record.cacheWriteTokens); +} + +function isContentBlock(value: unknown): value is ConversationContentBlock { + const record = asRecord(value); + if (!record || !isNonEmptyString(record.id) || !isNonEmptyString(record.kind)) return false; + if (record.kind === 'image') { + return isNonEmptyString(record.attachmentId) && isNonEmptyString(record.mime); + } + return (record.kind === 'text' || record.kind === 'thinking') + && typeof record.text === 'string' + && (record.status === 'streaming' || record.status === 'complete'); +} + +function isKnownToolDetails(value: unknown): value is KnownToolDetails { + const record = asRecord(value); + if (!record || !isNonEmptyString(record.schema)) return false; + if (record.schema === 'changed-file.v1') { + return Array.isArray(record.paths) && record.paths.every(isNonEmptyString); + } + if (record.schema === 'write-lease.v1') { + return record.status === 'waiting' || record.status === 'held' || record.status === 'released'; + } + if (record.schema === 'task-state.v1') { + return Array.isArray(record.tasks) && record.tasks.every((task) => { + const item = asRecord(task); + return item !== null + && isNonEmptyString(item.id) + && isNonEmptyString(item.title) + && ['pending', 'running', 'complete', 'error'].includes(String(item.status)); + }); + } + if (record.schema === 'subagent.v1') { + return isNonEmptyString(record.dispatchId) + && ['single', 'parallel', 'chain'].includes(String(record.mode)) + && Array.isArray(record.tasks) + && record.tasks.every((task) => { + const item = asRecord(task); + return item !== null + && isNonEmptyString(item.taskId) + && isNonEmptyString(item.agentId) + && ['read-only', 'coding'].includes(String(item.toolProfile)) + && ['queued', 'running', 'complete', 'error', 'aborted', 'skipped'] + .includes(String(item.status)) + && (item.usage === undefined || isUsage(item.usage)); + }); + } + return false; +} + +function isMessageNode(value: unknown): value is ConversationMessageNode { + const record = asRecord(value); + return record !== null + && record.kind === 'message' + && isNonEmptyString(record.id) + && (record.role === 'user' || record.role === 'assistant') + && ['optimistic', 'streaming', 'complete', 'error', 'aborted'].includes(String(record.status)) + && Array.isArray(record.blocks) + && record.blocks.every(isContentBlock) + && (record.usage === undefined || isUsage(record.usage)); +} + +function isToolNode(value: unknown): value is ConversationToolNode { + const record = asRecord(value); + return record !== null + && record.kind === 'tool' + && isNonEmptyString(record.id) + && isNonEmptyString(record.toolCallId) + && isNonEmptyString(record.toolName) + && typeof record.title === 'string' + && typeof record.inputText === 'string' + && ['declared', 'waiting', 'running', 'complete', 'error', 'aborted'] + .includes(String(record.status)) + && Array.isArray(record.output) + && record.output.every(isContentBlock) + && (record.details === undefined || isKnownToolDetails(record.details)); +} + +function isSubagentNode(value: unknown): value is ConversationSubagentNode { + const record = asRecord(value); + return record !== null + && record.kind === 'subagent' + && isNonEmptyString(record.id) + && isNonEmptyString(record.runId) + && isKnownToolDetails(record.details) + && record.details.schema === 'subagent.v1'; +} + +function isConversationNode(value: unknown): value is ConversationNode { + const record = asRecord(value); + if (!record || !isNonEmptyString(record.kind) || !isNonEmptyString(record.id)) return false; + if (record.kind === 'message') return isMessageNode(record); + if (record.kind === 'tool') return isToolNode(record); + if (record.kind === 'subagent') return isSubagentNode(record); + if (record.kind === 'compaction') { + return isNonEmptyString(record.runId) + && ['manual', 'automatic'].includes(String(record.source)) + && ['running', 'complete', 'error'].includes(String(record.status)) + && typeof record.willRetry === 'boolean'; + } + if (record.kind === 'boundary') { + return isNonEmptyString(record.runId) + && ['turn-start', 'turn-end', 'retry'].includes(String(record.boundary)); + } + if (record.kind === 'notice') { + return isNonEmptyString(record.code) + && ['info', 'warning', 'error'].includes(String(record.level)) + && typeof record.message === 'string'; + } + return false; +} + +function isInteraction(value: unknown): value is ConversationInteraction { + const record = asRecord(value); + return record !== null + && isNonEmptyString(record.id) + && isNonEmptyString(record.conversationId) + && isNonEmptyString(record.runId) + && ['select', 'confirm', 'input', 'editor'].includes(String(record.kind)) + && typeof record.title === 'string' + && ['pending', 'answered', 'rejected', 'cancelled'].includes(String(record.status)); +} + +function isRunState(value: unknown): value is ConversationRunState { + const record = asRecord(value); + return record !== null + && RUN_STATUSES.has(String(record.status)) + && (record.error === undefined || isPublicError(record.error)); +} + +function isQueueState(value: unknown): value is ConversationQueueState { + const record = asRecord(value); + return record !== null && Array.isArray(record.items) && record.items.every((item) => { + const queueItem = asRecord(item); + return queueItem !== null + && isNonEmptyString(queueItem.id) + && isNonEmptyString(queueItem.clientRequestId) + && ['steer', 'follow-up'].includes(String(queueItem.mode)) + && typeof queueItem.text === 'string' + && Array.isArray(queueItem.attachmentIds) + && queueItem.attachmentIds.every(isNonEmptyString); + }); +} + +function isWorkerState(value: unknown): value is PublicWorkerState { + const record = asRecord(value); + return record !== null + && WORKER_STATUSES.has(String(record.status)) + && isNonNegativeInteger(record.generation) + && (record.error === undefined || isPublicError(record.error)); +} + +export function isConversationSnapshot(value: unknown): value is ConversationSnapshot { + const record = asRecord(value); + const conversation = asRecord(record?.conversation); + const cursor = asRecord(record?.cursor); + const context = asRecord(record?.context); + const worker = record?.worker; + return record !== null + && record.schemaVersion === 1 + && conversation !== null + && isNonEmptyString(conversation.id) + && isNonEmptyString(conversation.projectId) + && isNonEmptyString(conversation.agentId) + && typeof conversation.title === 'string' + && isModelState(conversation.model) + && Array.isArray(record.nodes) + && record.nodes.every(isConversationNode) + && isRunState(record.run) + && isQueueState(record.queue) + && context !== null + && typeof context.usedTokens === 'number' + && typeof context.contextWindow === 'number' + && ['idle', 'running'].includes(String(context.compaction)) + && Array.isArray(record.pendingInteractions) + && record.pendingInteractions.every(isInteraction) + && isWorkerState(worker) + && cursor !== null + && isNonNegativeInteger(cursor.workerGeneration) + && isNonNegativeInteger(cursor.seq) + && asRecord(worker)?.generation === cursor.workerGeneration; +} + +function isPatch(value: unknown): value is ConversationPatch { + const record = asRecord(value); + if (!record || !PATCH_OPS.has(String(record.op))) return false; + switch (record.op) { + case 'worker.state': + return isWorkerState(record.state); + case 'run.state': + return isRunState(record.run); + case 'message.upsert': + return isMessageNode(record.node); + case 'message.block-delta': + return isNonEmptyString(record.messageId) + && isNonEmptyString(record.blockId) + && typeof record.delta === 'string'; + case 'tool.upsert': + return isToolNode(record.node); + case 'compaction.upsert': + return isConversationNode(record.node) && record.node.kind === 'compaction'; + case 'subagent.upsert': + return isSubagentNode(record.node); + case 'queue.replace': + return isQueueState(record.queue); + case 'interaction.upsert': + return isInteraction(record.interaction); + case 'interaction.remove': + return isNonEmptyString(record.interactionId); + case 'context.replace': { + const context = asRecord(record.context); + return context !== null + && typeof context.usedTokens === 'number' + && typeof context.contextWindow === 'number' + && ['idle', 'running'].includes(String(context.compaction)); + } + case 'snapshot.invalidated': + return isNonEmptyString(record.reason); + default: + return false; + } +} + +function parseEnvelopeShell(value: unknown): Omit & { patch: unknown } | null { + const record = asRecord(value); + if (!record + || !isNonEmptyString(record.conversationId) + || !isNonNegativeInteger(record.workerGeneration) + || !isNonNegativeInteger(record.seq) + || typeof record.at !== 'number') { + return null; + } + return record as unknown as Omit & { patch: unknown }; +} + +function invalidate( + state: ConversationReducerState, + invalidation: ConversationInvalidation, +): ConversationReducerState { + return { snapshot: state.snapshot, invalidation }; +} + +export function createConversationReducerState(snapshot?: unknown): ConversationReducerState { + if (snapshot === undefined) return { snapshot: null, invalidation: null }; + return replaceConversationSnapshot({ snapshot: null, invalidation: null }, snapshot); +} + +export function replaceConversationSnapshot( + state: ConversationReducerState, + snapshot: unknown, +): ConversationReducerState { + const record = asRecord(snapshot); + if (record?.schemaVersion !== 1) { + return invalidate(state, { + code: 'unsupported-schema', + reason: 'Conversation snapshot schemaVersion must equal 1', + }); + } + if (!isConversationSnapshot(snapshot)) { + return invalidate(state, { + code: 'malformed-snapshot', + reason: 'Conversation snapshot does not match the product contract', + }); + } + return { snapshot: clone(snapshot), invalidation: null }; +} + +function upsertMessage(nodes: ConversationNode[], incoming: ConversationMessageNode): ConversationNode[] { + const index = nodes.findIndex((node) => node.kind === 'message' && ( + node.id === incoming.id + || (incoming.sourceEntryId !== undefined && node.sourceEntryId === incoming.sourceEntryId) + || (incoming.clientRequestId !== undefined && node.clientRequestId === incoming.clientRequestId) + )); + if (index < 0) return [...nodes, clone(incoming)]; + const existing = nodes[index] as ConversationMessageNode; + const next = [...nodes]; + next[index] = { ...clone(incoming), id: existing.id }; + return next; +} + +function upsertTool(nodes: ConversationNode[], incoming: ConversationToolNode): ConversationNode[] { + const index = nodes.findIndex((node) => node.kind === 'tool' && ( + node.id === incoming.id || node.toolCallId === incoming.toolCallId + )); + if (index < 0) return [...nodes, clone(incoming)]; + const existing = nodes[index] as ConversationToolNode; + const next = [...nodes]; + next[index] = { ...clone(incoming), id: existing.id }; + return next; +} + +function upsertById(nodes: ConversationNode[], incoming: ConversationNode): ConversationNode[] { + const index = nodes.findIndex((node) => node.kind === incoming.kind && node.id === incoming.id); + if (index < 0) return [...nodes, clone(incoming)]; + const next = [...nodes]; + next[index] = clone(incoming); + return next; +} + +function applyPatch( + snapshot: ConversationSnapshot, + patch: ConversationPatch, +): { snapshot: ConversationSnapshot } | { missingTarget: string } { + switch (patch.op) { + case 'worker.state': + return { snapshot: { ...snapshot, worker: clone(patch.state) } }; + case 'run.state': + return { snapshot: { ...snapshot, run: clone(patch.run) } }; + case 'message.upsert': + return { snapshot: { ...snapshot, nodes: upsertMessage(snapshot.nodes, patch.node) } }; + case 'message.block-delta': { + const messageIndex = snapshot.nodes.findIndex( + (node) => node.kind === 'message' && node.id === patch.messageId, + ); + if (messageIndex < 0) return { missingTarget: `message:${patch.messageId}` }; + const message = snapshot.nodes[messageIndex] as ConversationMessageNode; + const blockIndex = message.blocks.findIndex((block) => block.id === patch.blockId); + if (blockIndex < 0) return { missingTarget: `block:${patch.blockId}` }; + const block = message.blocks[blockIndex]; + if (block.kind === 'image') return { missingTarget: `text-block:${patch.blockId}` }; + const blocks = [...message.blocks]; + blocks[blockIndex] = { ...block, text: `${block.text}${patch.delta}` }; + const nodes = [...snapshot.nodes]; + nodes[messageIndex] = { ...message, blocks }; + return { snapshot: { ...snapshot, nodes } }; + } + case 'tool.upsert': + return { snapshot: { ...snapshot, nodes: upsertTool(snapshot.nodes, patch.node) } }; + case 'compaction.upsert': + case 'subagent.upsert': + return { snapshot: { ...snapshot, nodes: upsertById(snapshot.nodes, patch.node) } }; + case 'queue.replace': + return { snapshot: { ...snapshot, queue: clone(patch.queue) } }; + case 'interaction.upsert': { + const index = snapshot.pendingInteractions.findIndex((item) => item.id === patch.interaction.id); + const pendingInteractions = [...snapshot.pendingInteractions]; + if (index < 0) pendingInteractions.push(clone(patch.interaction)); + else pendingInteractions[index] = clone(patch.interaction); + return { snapshot: { ...snapshot, pendingInteractions } }; + } + case 'interaction.remove': + return { + snapshot: { + ...snapshot, + pendingInteractions: snapshot.pendingInteractions.filter( + (item) => item.id !== patch.interactionId, + ), + }, + }; + case 'context.replace': + return { snapshot: { ...snapshot, context: clone(patch.context) } }; + case 'snapshot.invalidated': + return { missingTarget: patch.reason }; + default: + return { missingTarget: 'unknown patch' }; + } +} + +export function reduceConversationPatch( + state: ConversationReducerState, + value: unknown, +): ConversationReducerState { + if (state.invalidation) return state; + if (!state.snapshot) { + return invalidate(state, { + code: 'snapshot-required', + reason: 'A Conversation snapshot is required before live patches', + }); + } + + const envelope = parseEnvelopeShell(value); + if (!envelope) { + return invalidate(state, { + code: 'malformed-envelope', + reason: 'Conversation patch envelope is malformed', + }); + } + if (envelope.conversationId !== state.snapshot.conversation.id) return state; + + const currentGeneration = state.snapshot.cursor.workerGeneration; + if (envelope.workerGeneration < currentGeneration) return state; + if (envelope.workerGeneration > currentGeneration) { + return invalidate(state, { + code: 'generation-gap', + reason: 'A newer worker generation requires a fresh snapshot', + expectedGeneration: currentGeneration, + actualGeneration: envelope.workerGeneration, + }); + } + + const currentSeq = state.snapshot.cursor.seq; + if (envelope.seq <= currentSeq) return state; + if (envelope.seq !== currentSeq + 1) { + return invalidate(state, { + code: 'sequence-gap', + reason: 'Conversation patch sequence has a gap', + expectedSeq: currentSeq + 1, + actualSeq: envelope.seq, + }); + } + + const patchRecord = asRecord(envelope.patch); + if (!patchRecord || !PATCH_OPS.has(String(patchRecord.op))) { + return invalidate(state, { + code: 'unknown-patch', + reason: 'Conversation patch operation is not supported', + }); + } + if (!isPatch(envelope.patch)) { + return invalidate(state, { + code: 'unknown-patch', + reason: 'Conversation patch payload does not match its operation', + }); + } + if (envelope.patch.op === 'snapshot.invalidated') { + return invalidate(state, { + code: 'explicit-invalidation', + reason: envelope.patch.reason, + }); + } + + const applied = applyPatch(state.snapshot, envelope.patch); + if ('missingTarget' in applied) { + return invalidate(state, { + code: 'patch-target-missing', + reason: `Conversation patch target is missing: ${applied.missingTarget}`, + }); + } + return { + snapshot: { + ...applied.snapshot, + cursor: { + ...applied.snapshot.cursor, + workerGeneration: envelope.workerGeneration, + seq: envelope.seq, + }, + }, + invalidation: null, + }; +} diff --git a/electron/coding-runtime/in-memory-conversation-runtime.ts b/electron/coding-runtime/in-memory-conversation-runtime.ts new file mode 100644 index 0000000..9ed22f4 --- /dev/null +++ b/electron/coding-runtime/in-memory-conversation-runtime.ts @@ -0,0 +1,451 @@ +import type { + CodingConversationRuntime, + CodingRuntimeErrorCode, + CodingRuntimePublicError, + ConversationModelState, + ConversationPatch, + ConversationPatchEnvelope, + ConversationRuntimeState, + ConversationSnapshot, + ForkConversationInput, + ForkResult, + PrepareConversationInput, + PromptAcceptance, + PromptConversationInput, + QueueAcceptance, + QueueMessageInput, + SetConversationModelInput, + SetThinkingLevelInput, +} from './contracts'; +import { + createConversationReducerState, + reduceConversationPatch, + type ConversationReducerState, +} from './conversation-reducer'; + +export class CodingRuntimeContractError extends Error { + readonly publicError: CodingRuntimePublicError; + + constructor(code: CodingRuntimeErrorCode, message: string, recoverable: boolean) { + super(message); + this.name = 'CodingRuntimeContractError'; + this.publicError = { code, message, recoverable }; + } +} + +export interface InMemoryConversationRuntimeOptions { + snapshots?: ConversationSnapshot[]; + now?: () => number; + createId?: (kind: 'run' | 'node' | 'queue' | 'compaction') => string; +} + +function clone(value: T): T { + return structuredClone(value); +} + +function emptySnapshot(input: PrepareConversationInput): ConversationSnapshot { + return { + schemaVersion: 1, + conversation: { + id: input.conversationId, + projectId: input.projectId, + agentId: input.agentId, + title: input.title, + model: clone(input.model), + }, + nodes: [], + run: { status: 'idle' }, + queue: { items: [] }, + context: { + usedTokens: 0, + contextWindow: 0, + compaction: 'idle', + }, + pendingInteractions: [], + worker: { + status: 'ready', + generation: 0, + }, + cursor: { + workerGeneration: 0, + seq: 0, + }, + }; +} + +export class InMemoryConversationRuntime implements CodingConversationRuntime { + private readonly states = new Map(); + private readonly listeners = new Set<(patch: ConversationPatchEnvelope) => void>(); + private readonly promptAcceptances = new Map(); + private readonly queueAcceptances = new Map(); + private readonly now: () => number; + private readonly createId: InMemoryConversationRuntimeOptions['createId']; + private nextId = 0; + + constructor(options: InMemoryConversationRuntimeOptions = {}) { + this.now = options.now ?? Date.now; + this.createId = options.createId; + for (const snapshot of options.snapshots ?? []) { + const state = createConversationReducerState(snapshot); + if (!state.snapshot || state.invalidation) { + throw new CodingRuntimeContractError( + 'CODING_SESSION_UNREADABLE', + 'Initial Conversation snapshot is invalid', + true, + ); + } + this.states.set(snapshot.conversation.id, state); + } + } + + private id(kind: 'run' | 'node' | 'queue' | 'compaction'): string { + if (this.createId) return this.createId(kind); + this.nextId += 1; + return `${kind}-${this.nextId}`; + } + + private state(conversationId: string): ConversationReducerState { + const state = this.states.get(conversationId); + if (!state?.snapshot) { + throw new CodingRuntimeContractError( + 'CODING_CONVERSATION_NOT_FOUND', + 'Conversation is not prepared', + true, + ); + } + return state; + } + + private snapshot(conversationId: string): ConversationSnapshot { + return this.state(conversationId).snapshot as ConversationSnapshot; + } + + private replaceSnapshot(snapshot: ConversationSnapshot): void { + const state = createConversationReducerState(snapshot); + if (!state.snapshot || state.invalidation) { + throw new CodingRuntimeContractError( + 'CODING_SESSION_UNREADABLE', + 'Conversation snapshot replacement is invalid', + true, + ); + } + this.states.set(snapshot.conversation.id, state); + } + + private emit(conversationId: string, patch: ConversationPatch, runId?: string): void { + const state = this.state(conversationId); + const snapshot = state.snapshot as ConversationSnapshot; + const envelope: ConversationPatchEnvelope = { + conversationId, + workerGeneration: snapshot.cursor.workerGeneration, + ...(runId ? { runId } : {}), + seq: snapshot.cursor.seq + 1, + at: this.now(), + patch: clone(patch), + }; + const next = reduceConversationPatch(state, envelope); + if (next.invalidation) { + throw new CodingRuntimeContractError( + 'CODING_RUNTIME_PROTOCOL_ERROR', + next.invalidation.reason, + true, + ); + } + this.states.set(conversationId, next); + for (const listener of this.listeners) listener(clone(envelope)); + } + + private runtimeState(conversationId: string): ConversationRuntimeState { + const snapshot = this.snapshot(conversationId); + return { + conversationId, + status: snapshot.worker.status, + workerGeneration: snapshot.worker.generation, + ...(snapshot.worker.error ? { error: clone(snapshot.worker.error) } : {}), + }; + } + + async prepare(input: PrepareConversationInput): Promise { + const existing = this.states.get(input.conversationId)?.snapshot; + if (!existing) this.replaceSnapshot(emptySnapshot(input)); + return this.runtimeState(input.conversationId); + } + + async getSnapshot(conversationId: string): Promise { + return clone(this.snapshot(conversationId)); + } + + async prompt(input: PromptConversationInput): Promise { + const key = `${input.conversationId}:${input.clientRequestId}`; + const prior = this.promptAcceptances.get(key); + if (prior) return clone(prior); + + const snapshot = this.snapshot(input.conversationId); + if (snapshot.worker.status !== 'ready') { + throw new CodingRuntimeContractError( + 'CODING_RUNTIME_START_FAILED', + 'Conversation worker is not ready', + true, + ); + } + + const runId = this.id('run'); + const messageId = this.id('node'); + this.emit(input.conversationId, { + op: 'message.upsert', + node: { + kind: 'message', + id: messageId, + clientRequestId: input.clientRequestId, + role: 'user', + status: 'optimistic', + blocks: [ + ...(input.text ? [{ + kind: 'text' as const, + id: `${messageId}:text`, + text: input.text, + status: 'complete' as const, + }] : []), + ...input.attachments.map(({ attachmentId }, index) => ({ + kind: 'image' as const, + id: `${messageId}:attachment:${index}`, + attachmentId, + mime: 'application/octet-stream', + })), + ], + }, + }, runId); + this.emit(input.conversationId, { + op: 'run.state', + run: { + status: 'running', + runId, + mode: input.mode, + startedAt: this.now(), + }, + }, runId); + + const acceptance: PromptAcceptance = { + accepted: true, + conversationId: input.conversationId, + clientRequestId: input.clientRequestId, + runId, + mode: input.mode, + }; + this.promptAcceptances.set(key, acceptance); + return clone(acceptance); + } + + private async queue( + mode: 'steer' | 'follow-up', + input: QueueMessageInput, + ): Promise { + const key = `${input.conversationId}:${mode}:${input.clientRequestId}`; + const prior = this.queueAcceptances.get(key); + if (prior) return clone(prior); + + const snapshot = this.snapshot(input.conversationId); + const queuePosition = snapshot.queue.items.length + 1; + this.emit(input.conversationId, { + op: 'queue.replace', + queue: { + items: [ + ...snapshot.queue.items, + { + id: this.id('queue'), + clientRequestId: input.clientRequestId, + mode, + text: input.text, + attachmentIds: input.attachments.map(({ attachmentId }) => attachmentId), + }, + ], + }, + }); + + const acceptance: QueueAcceptance = { + accepted: true, + conversationId: input.conversationId, + clientRequestId: input.clientRequestId, + mode, + queuePosition, + }; + this.queueAcceptances.set(key, acceptance); + return clone(acceptance); + } + + async steer(input: QueueMessageInput): Promise { + return this.queue('steer', input); + } + + async followUp(input: QueueMessageInput): Promise { + return this.queue('follow-up', input); + } + + async abort(conversationId: string): Promise { + const current = this.snapshot(conversationId).run; + this.emit(conversationId, { + op: 'run.state', + run: { + status: 'idle', + ...(current.runId ? { runId: current.runId } : {}), + settledAt: this.now(), + terminalReason: 'aborted', + }, + }, current.runId); + } + + async setModel(input: SetConversationModelInput): Promise { + const snapshot = this.snapshot(input.conversationId); + const thinkingLevel = snapshot.conversation.model.model?.thinkingLevel ?? 'off'; + const model: ConversationModelState = { + model: { + accountId: input.accountId, + modelId: input.modelId, + thinkingLevel, + }, + modelResolution: 'resolved', + }; + this.replaceSnapshot({ + ...snapshot, + conversation: { ...snapshot.conversation, model }, + }); + return clone(model); + } + + async setThinking(input: SetThinkingLevelInput): Promise { + const snapshot = this.snapshot(input.conversationId); + if (!snapshot.conversation.model.model) { + throw new CodingRuntimeContractError( + 'CODING_MIGRATION_MODEL_REQUIRED', + 'Conversation model must be selected first', + true, + ); + } + const model: ConversationModelState = { + model: { + ...snapshot.conversation.model.model, + thinkingLevel: input.thinkingLevel, + }, + modelResolution: 'resolved', + }; + this.replaceSnapshot({ + ...snapshot, + conversation: { ...snapshot.conversation, model }, + }); + return clone(model); + } + + async compact(conversationId: string): Promise { + const before = this.snapshot(conversationId); + const compactionId = this.id('compaction'); + const runId = before.run.runId ?? this.id('run'); + const willRetry = before.run.status !== 'idle'; + this.emit(conversationId, { + op: 'compaction.upsert', + node: { + kind: 'compaction', + id: compactionId, + runId, + source: 'manual', + status: 'running', + willRetry, + }, + }, runId); + this.emit(conversationId, { + op: 'context.replace', + context: { + ...before.context, + compaction: 'running', + lastCompactionId: compactionId, + }, + }, runId); + this.emit(conversationId, { + op: 'compaction.upsert', + node: { + kind: 'compaction', + id: compactionId, + runId, + source: 'manual', + status: 'complete', + willRetry, + }, + }, runId); + this.emit(conversationId, { + op: 'context.replace', + context: { + ...before.context, + compaction: 'idle', + lastCompactionId: compactionId, + }, + }, runId); + } + + async fork(input: ForkConversationInput): Promise { + const source = this.snapshot(input.sourceConversationId); + let nodes = source.nodes; + if (input.sourceEntryId) { + const index = source.nodes.findIndex( + (node) => node.kind === 'message' && node.sourceEntryId === input.sourceEntryId, + ); + if (index < 0) { + throw new CodingRuntimeContractError( + 'CODING_SESSION_UNREADABLE', + 'Fork source entry is not on the active product path', + true, + ); + } + nodes = source.nodes.slice(0, index + 1); + } + const snapshot: ConversationSnapshot = { + ...emptySnapshot(input.conversation), + nodes: clone(nodes), + context: clone(source.context), + cursor: { + workerGeneration: 0, + seq: 0, + ...(input.sourceEntryId ? { leafEntryId: input.sourceEntryId } : {}), + }, + worker: { + status: 'stopped', + generation: 0, + }, + }; + this.replaceSnapshot(snapshot); + return { conversationId: input.conversation.conversationId, snapshot: clone(snapshot) }; + } + + async recover(conversationId: string): Promise { + const snapshot = this.snapshot(conversationId); + const generation = snapshot.cursor.workerGeneration + 1; + this.replaceSnapshot({ + ...snapshot, + worker: { status: 'ready', generation }, + run: { status: 'idle' }, + pendingInteractions: [], + cursor: { + ...snapshot.cursor, + workerGeneration: generation, + seq: 0, + }, + }); + return this.runtimeState(conversationId); + } + + async dispose(conversationId: string): Promise { + const snapshot = this.snapshot(conversationId); + this.replaceSnapshot({ + ...snapshot, + worker: { + status: 'stopped', + generation: snapshot.cursor.workerGeneration, + }, + run: { status: 'idle' }, + pendingInteractions: [], + }); + } + + subscribe(listener: (patch: ConversationPatchEnvelope) => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } +} diff --git a/tests/fixtures/coding-conversation-product-fixtures.ts b/tests/fixtures/coding-conversation-product-fixtures.ts new file mode 100644 index 0000000..d99b3e9 --- /dev/null +++ b/tests/fixtures/coding-conversation-product-fixtures.ts @@ -0,0 +1,344 @@ +import type { + ConversationPatch, + ConversationPatchEnvelope, + ConversationSnapshot, +} from '../../electron/coding-runtime/contracts'; + +export const PI_PRODUCT_FIXTURE_SOURCE = Object.freeze({ + package: '@earendil-works/pi-coding-agent', + version: '0.84.2', + rule: 'Wire labels are fixture provenance only; projected values use Makelore product types.', +}); + +export function createProductSnapshot( + conversationId = 'conversation-a', + workerGeneration = 1, +): ConversationSnapshot { + return { + schemaVersion: 1, + conversation: { + id: conversationId, + projectId: 'project-a', + agentId: 'agent-a', + title: 'Fixture Conversation', + model: { + model: { + accountId: 'account-a', + modelId: 'model-a', + thinkingLevel: 'medium', + }, + modelResolution: 'resolved', + }, + }, + nodes: [], + run: { status: 'idle' }, + queue: { items: [] }, + context: { + usedTokens: 0, + contextWindow: 128_000, + outputLimit: 8_192, + compaction: 'idle', + }, + pendingInteractions: [], + worker: { status: 'ready', generation: workerGeneration }, + cursor: { workerGeneration, seq: 0 }, + }; +} + +function envelope(seq: number, patch: ConversationPatch): ConversationPatchEnvelope { + return { + conversationId: 'conversation-a', + workerGeneration: 1, + runId: 'run-a', + seq, + at: 1_000 + seq, + patch, + }; +} + +export const MIXED_PRODUCT_PATCHES: ConversationPatchEnvelope[] = [ + envelope(1, { + op: 'run.state', + run: { status: 'running', runId: 'run-a', mode: 'prompt', startedAt: 1_001 }, + }), + envelope(2, { + op: 'message.upsert', + node: { + kind: 'message', + id: 'ui-user-a', + clientRequestId: 'request-a', + role: 'user', + status: 'optimistic', + blocks: [{ kind: 'text', id: 'user-text-a', text: 'Build it', status: 'complete' }], + }, + }), + envelope(3, { + op: 'message.upsert', + node: { + kind: 'message', + id: 'durable-user-a', + sourceEntryId: 'entry-user-a', + clientRequestId: 'request-a', + role: 'user', + status: 'complete', + blocks: [{ kind: 'text', id: 'user-text-a', text: 'Build it', status: 'complete' }], + }, + }), + envelope(4, { + op: 'message.upsert', + node: { + kind: 'message', + id: 'ui-assistant-a', + sourceEntryId: 'entry-assistant-a', + role: 'assistant', + status: 'streaming', + blocks: [ + { kind: 'thinking', id: 'thinking-a', text: 'Plan', status: 'streaming' }, + { kind: 'text', id: 'assistant-text-a', text: 'Working', status: 'streaming' }, + ], + }, + }), + envelope(5, { + op: 'message.block-delta', + messageId: 'ui-assistant-a', + blockId: 'assistant-text-a', + delta: ' now', + }), + envelope(6, { + op: 'tool.upsert', + node: { + kind: 'tool', + id: 'tool-ui-a', + toolCallId: 'tool-call-a', + toolName: 'write', + title: 'Write file', + inputText: '{"path":"src/a.ts"}', + status: 'running', + output: [], + details: { schema: 'write-lease.v1', status: 'held' }, + }, + }), + envelope(7, { + op: 'tool.upsert', + node: { + kind: 'tool', + id: 'tool-ui-b', + toolCallId: 'tool-call-b', + toolName: 'read', + title: 'Read file', + inputText: '{"path":"src/b.ts"}', + status: 'running', + output: [], + }, + }), + envelope(8, { + op: 'tool.upsert', + node: { + kind: 'tool', + id: 'tool-wire-a', + toolCallId: 'tool-call-a', + toolName: 'write', + title: 'Write file', + inputText: '{"path":"src/a.ts"}', + status: 'running', + output: [{ kind: 'text', id: 'tool-output-a', text: 'first', status: 'streaming' }], + details: { schema: 'changed-file.v1', paths: ['src/a.ts'] }, + }, + }), + envelope(9, { + op: 'tool.upsert', + node: { + kind: 'tool', + id: 'tool-wire-a', + toolCallId: 'tool-call-a', + toolName: 'write', + title: 'Write file', + inputText: '{"path":"src/a.ts"}', + status: 'complete', + output: [{ kind: 'text', id: 'tool-output-a', text: 'first final', status: 'complete' }], + details: { schema: 'changed-file.v1', paths: ['src/a.ts'] }, + }, + }), + envelope(10, { + op: 'tool.upsert', + node: { + kind: 'tool', + id: 'tool-wire-b', + toolCallId: 'tool-call-b', + toolName: 'read', + title: 'Read file', + inputText: '{"path":"src/b.ts"}', + status: 'error', + output: [{ kind: 'text', id: 'tool-output-b', text: 'not found', status: 'complete' }], + }, + }), + envelope(11, { + op: 'queue.replace', + queue: { + items: [{ + id: 'queue-a', + clientRequestId: 'request-follow-up', + mode: 'follow-up', + text: 'Add tests', + attachmentIds: [], + }], + }, + }), + envelope(12, { + op: 'interaction.upsert', + interaction: { + id: 'interaction-a', + conversationId: 'conversation-a', + runId: 'run-a', + kind: 'select', + title: 'Choose target', + options: [{ id: 'web', label: 'Web' }, { id: 'desktop', label: 'Desktop' }], + status: 'pending', + }, + }), + envelope(13, { + op: 'compaction.upsert', + node: { + kind: 'compaction', + id: 'compaction-a', + runId: 'run-a', + source: 'automatic', + status: 'running', + willRetry: true, + }, + }), + envelope(14, { + op: 'context.replace', + context: { + usedTokens: 96_000, + contextWindow: 128_000, + outputLimit: 8_192, + compaction: 'running', + lastCompactionId: 'compaction-a', + }, + }), + envelope(15, { + op: 'subagent.upsert', + node: { + kind: 'subagent', + id: 'subagent-a', + runId: 'run-a', + details: { + schema: 'subagent.v1', + dispatchId: 'dispatch-a', + mode: 'parallel', + tasks: [ + { + taskId: 'child-a', + agentId: 'reviewer', + toolProfile: 'read-only', + status: 'complete', + summary: 'Reviewed', + }, + { + taskId: 'child-b', + agentId: 'tester', + toolProfile: 'coding', + status: 'running', + }, + ], + }, + }, + }), + envelope(16, { + op: 'compaction.upsert', + node: { + kind: 'compaction', + id: 'compaction-a', + runId: 'run-a', + source: 'automatic', + status: 'complete', + willRetry: true, + }, + }), + envelope(17, { + op: 'context.replace', + context: { + usedTokens: 24_000, + contextWindow: 128_000, + outputLimit: 8_192, + compaction: 'idle', + lastCompactionId: 'compaction-a', + }, + }), + envelope(18, { op: 'interaction.remove', interactionId: 'interaction-a' }), + envelope(19, { + op: 'message.upsert', + node: { + kind: 'message', + id: 'durable-assistant-a', + sourceEntryId: 'entry-assistant-a', + role: 'assistant', + status: 'complete', + blocks: [ + { kind: 'thinking', id: 'thinking-a', text: 'Plan complete', status: 'complete' }, + { kind: 'text', id: 'assistant-text-a', text: 'Done', status: 'complete' }, + ], + usage: { inputTokens: 120, outputTokens: 40 }, + stopReason: 'stop', + }, + }), + envelope(20, { + op: 'run.state', + run: { + status: 'idle', + runId: 'run-a', + settledAt: 1_020, + terminalReason: 'completed', + }, + }), +]; + +export const PI_WIRE_TO_PRODUCT_BOUNDARY = Object.freeze([ + { + wireEvent: 'agent_end', + projectedPatches: [] as ConversationPatch[], + }, + { + wireEvent: 'agent_settled', + projectedPatches: [{ + op: 'run.state', + run: { + status: 'idle', + runId: 'run-a', + terminalReason: 'completed', + }, + }] satisfies ConversationPatch[], + }, + { + wireEvent: 'retry', + projectedPatches: [{ + op: 'run.state', + run: { + status: 'retrying', + runId: 'run-a', + retry: { attempt: 2, delayMs: 750 }, + }, + }] satisfies ConversationPatch[], + }, + { + wireEvent: 'toolResult.message_end', + projectedPatches: [{ + op: 'tool.upsert', + node: { + kind: 'tool', + id: 'tool-ui-a', + toolCallId: 'tool-call-a', + toolName: 'write', + title: 'Write file', + inputText: '{}', + status: 'complete', + output: [{ kind: 'text', id: 'tool-output-a', text: 'done', status: 'complete' }], + }, + }] satisfies ConversationPatch[], + }, + { + wireEvent: 'custom.display-false', + projectedPatches: [] as ConversationPatch[], + }, +]); diff --git a/tests/unit/coding-conversation-contracts.test.ts b/tests/unit/coding-conversation-contracts.test.ts new file mode 100644 index 0000000..bd3e15c --- /dev/null +++ b/tests/unit/coding-conversation-contracts.test.ts @@ -0,0 +1,397 @@ +import { readFile } from 'node:fs/promises'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +import type { + ConversationPatch, + ConversationPatchEnvelope, + ConversationSnapshot, +} from '../../electron/coding-runtime/contracts'; +import { + createConversationReducerState, + isConversationSnapshot, + reduceConversationPatch, + replaceConversationSnapshot, + type ConversationReducerState, +} from '../../electron/coding-runtime/conversation-reducer'; +import { + CodingRuntimeContractError, + InMemoryConversationRuntime, +} from '../../electron/coding-runtime/in-memory-conversation-runtime'; +import { + createProductSnapshot, + MIXED_PRODUCT_PATCHES, + PI_PRODUCT_FIXTURE_SOURCE, + PI_WIRE_TO_PRODUCT_BOUNDARY, +} from '../fixtures/coding-conversation-product-fixtures'; + +function reduceAll( + snapshot: ConversationSnapshot, + envelopes: readonly ConversationPatchEnvelope[], +): ConversationReducerState { + return envelopes.reduce( + (state, envelope) => reduceConversationPatch(state, envelope), + createConversationReducerState(snapshot), + ); +} + +function envelope( + snapshot: ConversationSnapshot, + seq: number, + patch: ConversationPatch, +): ConversationPatchEnvelope { + return { + conversationId: snapshot.conversation.id, + workerGeneration: snapshot.cursor.workerGeneration, + seq, + at: 2_000 + seq, + patch, + }; +} + +describe('Conversation product contracts', () => { + it('accepts schema v1 snapshots and fail-closes unknown schemas until replacement', () => { + const snapshot = createProductSnapshot(); + expect(isConversationSnapshot(snapshot)).toBe(true); + + const invalid = createConversationReducerState({ ...snapshot, schemaVersion: 2 }); + expect(invalid.snapshot).toBeNull(); + expect(invalid.invalidation?.code).toBe('unsupported-schema'); + + const recovered = replaceConversationSnapshot(invalid, snapshot); + expect(recovered.invalidation).toBeNull(); + expect(recovered.snapshot).toEqual(snapshot); + expect(recovered.snapshot).not.toBe(snapshot); + }); + + it('reduces the locked-version mixed product fixture without vendor-shaped state', () => { + expect(PI_PRODUCT_FIXTURE_SOURCE.version).toBe('0.84.2'); + const state = reduceAll(createProductSnapshot(), MIXED_PRODUCT_PATCHES); + expect(state.invalidation).toBeNull(); + expect(state.snapshot?.cursor.seq).toBe(20); + expect(state.snapshot?.run).toMatchObject({ + status: 'idle', + terminalReason: 'completed', + }); + expect(state.snapshot?.queue.items).toHaveLength(1); + expect(state.snapshot?.pendingInteractions).toEqual([]); + expect(state.snapshot?.context).toMatchObject({ + usedTokens: 24_000, + compaction: 'idle', + lastCompactionId: 'compaction-a', + }); + + const messages = state.snapshot?.nodes.filter((node) => node.kind === 'message') ?? []; + const tools = state.snapshot?.nodes.filter((node) => node.kind === 'tool') ?? []; + expect(messages).toHaveLength(2); + expect(messages[0]).toMatchObject({ + id: 'ui-user-a', + sourceEntryId: 'entry-user-a', + clientRequestId: 'request-a', + status: 'complete', + }); + expect(messages[1]).toMatchObject({ + id: 'ui-assistant-a', + sourceEntryId: 'entry-assistant-a', + status: 'complete', + }); + expect(tools).toHaveLength(2); + expect(tools[0]).toMatchObject({ + id: 'tool-ui-a', + toolCallId: 'tool-call-a', + status: 'complete', + output: [{ text: 'first final' }], + }); + expect(tools[1]).toMatchObject({ status: 'error', output: [{ text: 'not found' }] }); + expect(state.snapshot?.nodes.some((node) => ( + node.kind === 'message' && (node as { role?: string }).role === 'toolResult' + ))).toBe(false); + expect(state.snapshot?.nodes).toEqual(expect.arrayContaining([ + expect.objectContaining({ kind: 'compaction', status: 'complete', willRetry: true }), + expect.objectContaining({ kind: 'subagent' }), + ])); + }); + + it('replaces cumulative tool output instead of appending duplicate chunks', () => { + const state = reduceAll(createProductSnapshot(), MIXED_PRODUCT_PATCHES.slice(0, 9)); + const tool = state.snapshot?.nodes.find( + (node) => node.kind === 'tool' && node.toolCallId === 'tool-call-a', + ); + expect(tool).toMatchObject({ + kind: 'tool', + output: [{ id: 'tool-output-a', text: 'first final', status: 'complete' }], + }); + if (tool?.kind !== 'tool') throw new Error('tool fixture missing'); + expect(tool.output).toHaveLength(1); + }); + + it('invalidates only the target Conversation on a sequence gap', () => { + const snapshotA = createProductSnapshot('conversation-a'); + const snapshotB = createProductSnapshot('conversation-b'); + const stateA = reduceConversationPatch( + createConversationReducerState(snapshotA), + envelope(snapshotA, 2, { op: 'queue.replace', queue: { items: [] } }), + ); + const stateB = reduceConversationPatch( + createConversationReducerState(snapshotB), + envelope(snapshotA, 2, { op: 'queue.replace', queue: { items: [] } }), + ); + + expect(stateA.invalidation).toMatchObject({ + code: 'sequence-gap', + expectedSeq: 1, + actualSeq: 2, + }); + expect(stateB.invalidation).toBeNull(); + expect(stateB.snapshot).toEqual(snapshotB); + }); + + it('drops stale generation patches and invalidates a future generation', () => { + const snapshot = createProductSnapshot('conversation-a', 4); + const base = createConversationReducerState(snapshot); + const stale = reduceConversationPatch(base, { + ...envelope(snapshot, 1, { op: 'queue.replace', queue: { items: [] } }), + workerGeneration: 3, + }); + expect(stale).toBe(base); + + const future = reduceConversationPatch(base, { + ...envelope(snapshot, 1, { op: 'queue.replace', queue: { items: [] } }), + workerGeneration: 5, + }); + expect(future.invalidation).toMatchObject({ + code: 'generation-gap', + expectedGeneration: 4, + actualGeneration: 5, + }); + }); + + it('fail-closes unknown product operations and missing delta targets', () => { + const snapshot = createProductSnapshot(); + const unknown = reduceConversationPatch(createConversationReducerState(snapshot), { + ...envelope(snapshot, 1, { op: 'queue.replace', queue: { items: [] } }), + patch: { op: 'custom.vendor-event', raw: { secret: 'not projected' } }, + }); + expect(unknown.invalidation?.code).toBe('unknown-patch'); + + const missing = reduceConversationPatch(createConversationReducerState(snapshot), envelope( + snapshot, + 1, + { op: 'message.block-delta', messageId: 'missing', blockId: 'missing', delta: 'x' }, + )); + expect(missing.invalidation?.code).toBe('patch-target-missing'); + }); + + it('defines agent_end as a non-idle checkpoint and agent_settled as authoritative idle', () => { + const agentEnd = PI_WIRE_TO_PRODUCT_BOUNDARY.find(({ wireEvent }) => wireEvent === 'agent_end'); + const settled = PI_WIRE_TO_PRODUCT_BOUNDARY.find( + ({ wireEvent }) => wireEvent === 'agent_settled', + ); + const custom = PI_WIRE_TO_PRODUCT_BOUNDARY.find( + ({ wireEvent }) => wireEvent === 'custom.display-false', + ); + const retry = PI_WIRE_TO_PRODUCT_BOUNDARY.find(({ wireEvent }) => wireEvent === 'retry'); + expect(agentEnd?.projectedPatches).toEqual([]); + expect(custom?.projectedPatches).toEqual([]); + expect(settled?.projectedPatches).toEqual([ + expect.objectContaining({ op: 'run.state', run: expect.objectContaining({ status: 'idle' }) }), + ]); + expect(retry?.projectedPatches).toEqual([ + expect.objectContaining({ + op: 'run.state', + run: expect.objectContaining({ status: 'retrying', retry: { attempt: 2, delayMs: 750 } }), + }), + ]); + + const running = createProductSnapshot(); + running.run = { status: 'running', runId: 'run-a' }; + const afterAgentEnd = reduceAll(running, []); + expect(afterAgentEnd.snapshot?.run.status).toBe('running'); + const afterRetry = reduceConversationPatch( + afterAgentEnd, + envelope(running, 1, retry?.projectedPatches[0] as ConversationPatch), + ); + expect(afterRetry.snapshot?.run).toMatchObject({ + status: 'retrying', + retry: { attempt: 2, delayMs: 750 }, + }); + const afterSettled = reduceConversationPatch( + afterRetry, + envelope(running, 2, settled?.projectedPatches[0] as ConversationPatch), + ); + expect(afterSettled.snapshot?.run.status).toBe('idle'); + }); + + it('applies worker state through the same ordered envelope contract', () => { + const snapshot = createProductSnapshot(); + const next = reduceConversationPatch( + createConversationReducerState(snapshot), + envelope(snapshot, 1, { + op: 'worker.state', + state: { status: 'recovering', generation: 1 }, + }), + ); + expect(next.invalidation).toBeNull(); + expect(next.snapshot?.worker).toEqual({ status: 'recovering', generation: 1 }); + }); + + it('rehydrates the same normalized Snapshot produced by live patches', () => { + const live = reduceAll(createProductSnapshot(), MIXED_PRODUCT_PATCHES); + if (!live.snapshot) throw new Error('live fixture did not produce a snapshot'); + const hydrated = createConversationReducerState(live.snapshot); + expect(hydrated.invalidation).toBeNull(); + expect(hydrated.snapshot).toEqual(live.snapshot); + expect(hydrated.snapshot).not.toBe(live.snapshot); + }); +}); + +describe('InMemoryConversationRuntime', () => { + const input = { + conversationId: 'conversation-memory', + projectId: 'project-memory', + agentId: 'agent-memory', + title: 'Memory', + model: { + model: { + accountId: 'account-memory', + modelId: 'model-memory', + thinkingLevel: 'low' as const, + }, + modelResolution: 'resolved' as const, + }, + }; + + it('supports idempotent prompt acceptance, queueing, subscription, and target abort', async () => { + let tick = 10_000; + let id = 0; + const runtime = new InMemoryConversationRuntime({ + now: () => ++tick, + createId: (kind) => `${kind}-${++id}`, + }); + await runtime.prepare(input); + const events: ConversationPatchEnvelope[] = []; + const unsubscribe = runtime.subscribe((event) => events.push(event)); + + const promptInput = { + clientRequestId: 'request-memory', + conversationId: input.conversationId, + mode: 'prompt' as const, + text: 'Hello', + attachments: [], + }; + const first = await runtime.prompt(promptInput); + const duplicate = await runtime.prompt(promptInput); + expect(duplicate).toEqual(first); + expect(events).toHaveLength(2); + + const queued = await runtime.followUp({ + clientRequestId: 'follow-up-memory', + conversationId: input.conversationId, + text: 'Next', + attachments: [], + }); + expect(queued).toMatchObject({ mode: 'follow-up', queuePosition: 1 }); + expect((await runtime.getSnapshot(input.conversationId)).queue.items).toHaveLength(1); + + await runtime.abort(input.conversationId); + expect((await runtime.getSnapshot(input.conversationId)).run).toMatchObject({ + status: 'idle', + terminalReason: 'aborted', + }); + unsubscribe(); + }); + + it('keeps active run state through willRetry compaction and isolates recovery generation', async () => { + const runtime = new InMemoryConversationRuntime(); + await runtime.prepare(input); + await runtime.prompt({ + clientRequestId: 'request-compaction', + conversationId: input.conversationId, + mode: 'prompt', + text: 'Compact while running', + attachments: [], + }); + await runtime.compact(input.conversationId); + const compacted = await runtime.getSnapshot(input.conversationId); + expect(compacted.run.status).toBe('running'); + expect(compacted.context.compaction).toBe('idle'); + expect(compacted.nodes).toEqual(expect.arrayContaining([ + expect.objectContaining({ kind: 'compaction', status: 'complete', willRetry: true }), + ])); + + const recovered = await runtime.recover(input.conversationId); + expect(recovered).toMatchObject({ status: 'ready', workerGeneration: 1 }); + const snapshot = await runtime.getSnapshot(input.conversationId); + expect(snapshot.cursor).toMatchObject({ workerGeneration: 1, seq: 0 }); + expect(snapshot.run.status).toBe('idle'); + }); + + it('updates only the target model and forks a durable active-path prefix', async () => { + const source = createProductSnapshot(input.conversationId); + source.nodes = [ + { + kind: 'message', + id: 'ui-user-1', + sourceEntryId: 'entry-user-1', + role: 'user', + status: 'complete', + blocks: [], + }, + { + kind: 'message', + id: 'ui-assistant-1', + sourceEntryId: 'entry-assistant-1', + role: 'assistant', + status: 'complete', + blocks: [], + }, + ]; + const runtime = new InMemoryConversationRuntime({ snapshots: [source] }); + await runtime.setModel({ + conversationId: input.conversationId, + accountId: 'account-b', + modelId: 'model-b', + }); + const model = await runtime.setThinking({ + conversationId: input.conversationId, + thinkingLevel: 'high', + }); + expect(model.model).toMatchObject({ + accountId: 'account-b', + modelId: 'model-b', + thinkingLevel: 'high', + }); + + const fork = await runtime.fork({ + sourceConversationId: input.conversationId, + sourceEntryId: 'entry-user-1', + conversation: { ...input, conversationId: 'conversation-fork', title: 'Fork' }, + }); + expect(fork.snapshot.nodes).toEqual([source.nodes[0]]); + expect(fork.snapshot.worker.status).toBe('stopped'); + expect((await runtime.getSnapshot(input.conversationId)).conversation.id) + .toBe(input.conversationId); + }); + + it('returns a stable public error when a Conversation is absent', async () => { + const runtime = new InMemoryConversationRuntime(); + await expect(runtime.getSnapshot('missing')).rejects.toMatchObject({ + name: 'CodingRuntimeContractError', + publicError: { + code: 'CODING_CONVERSATION_NOT_FOUND', + recoverable: true, + }, + } satisfies Partial); + }); + + it('keeps the contracts, reducer, and in-memory implementation free of vendor imports', async () => { + const files = [ + '../../electron/coding-runtime/contracts.ts', + '../../electron/coding-runtime/conversation-reducer.ts', + '../../electron/coding-runtime/in-memory-conversation-runtime.ts', + ]; + for (const relativePath of files) { + const source = await readFile(fileURLToPath(new URL(relativePath, import.meta.url)), 'utf8'); + expect(source).not.toMatch(/(?:import|export)[\s\S]*?from\s+['"][^'"]*(?:opencode|pi(?:-|\/))/i); + } + }); +});