import type { CodingRuntimePublicError, ConversationContentBlock, ConversationInteraction, ConversationMessageNode, ConversationNode, ConversationPatch, ConversationPatchEnvelope, ConversationQueueState, ConversationRunState, ConversationSnapshot, ConversationSubagentNode, ConversationToolNode, KnownToolDetails, PublicWorkerState, } from './coding-conversation-contracts'; import { productToolDetails } from './coding-conversation-product-tool-protocol'; import { projectSubagentDetailsV1 } from './coding-conversation-subagent-protocol'; 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 THINKING_LEVELS = new Set(['off', 'minimal', 'low', 'medium', 'high', 'max']); const ERROR_CODES = new Set([ 'CODING_RUNTIME_START_FAILED', 'CODING_RUNTIME_READY_TIMEOUT', 'CODING_RUNTIME_PROTOCOL_ERROR', 'CODING_PROVIDER_AUTH_REQUIRED', 'CODING_PROVIDER_QUOTA_EXHAUSTED', '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', 'boundary.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'; } import { parseManagedReasoningChoice } from './managed-model-capabilities'; 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' && record.availableThinkingLevels === undefined; } const model = asRecord(record.model); return model !== null && isNonEmptyString(model.accountId) && isNonEmptyString(model.modelId) && THINKING_LEVELS.has(String(model.thinkingLevel)) && (model.reasoningChoice === undefined || parseManagedReasoningChoice(model.reasoningChoice) !== null) && (record.availableThinkingLevels === undefined || (Array.isArray(record.availableThinkingLevels) && record.availableThinkingLevels.length > 0 && record.availableThinkingLevels.every((level) => THINKING_LEVELS.has(String(level))))) && 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 (productToolDetails(record)) return true; if (record.schema === 'write-lease.v1') { return record.status === 'waiting' || record.status === 'held' || record.status === 'released'; } if (record.schema === 'subagent.v1') { return projectSubagentDetailsV1(record) !== undefined; } 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)) && (context.recalculating === undefined || typeof context.recalculating === 'boolean') && 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 'boundary.upsert': return isConversationNode(record.node) && record.node.kind === 'boundary'; 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)) && (context.recalculating === undefined || typeof context.recalculating === 'boolean'); } 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 'boundary.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, }; }