/** * Stateful, incremental event processor that transforms DebugEvents into chat Turns. * * Extracted from the ChatPanel useMemo so the logic can be unit-tested without * React rendering. The ChatPanel hooks into this via process() on each render. */ import type { DebugEvent } from '@/lib/stores/types'; import { INCOMPLETE_PREFIX } from '@/lib/interview/completion'; export interface ToolCall { id: string; name: string; parameters?: any; status?: 'pending' | 'executing' | 'completed' | 'failed'; result?: any; error?: string; } export interface TurnItem { id: string; type: 'waiting' | 'reasoning' | 'plan' | 'agent' | 'progress' | 'tool' | 'text' | 'error' | 'error_paused' | 'user' | 'synthetic_error' | 'project_context' | 'compaction' | 'ask' | 'interview_gate'; timestamp: number; data: any; eventId?: string; complete?: boolean; focusContext?: { domPath: string; snippet: string }; semanticBlocks?: Array<{ name: string; domPath: string; position: string; description: string }>; attachedFiles?: Array<{ name: string }>; } export interface Turn { id: string; items: TurnItem[]; usage?: any; iteration?: number; checkpointId?: string; taskStartTime?: number; } interface DeltaAccum { text: string; fragmentCount: number; } interface ProcessorState { result: Turn[]; currentTurn: Turn; currentIterationTools: ToolCall[]; itemIdCounter: number; taskStartTime: number; prevTaskCumulativeTokens: number; prevTaskCumulativeCost: number; textAccum: Map; toolParamAccum: Map; toolCmdCache: Map; } function extractPartialCmd(raw: string): string | null { const match = raw.match(/^\s*\{\s*"(?:command|cmd)"\s*:\s*"((?:\\.|[^"\\])*)/); if (!match) return null; try { return JSON.parse('"' + match[1] + '"'); } catch { return match[1] || null; } } export function classifyBashCommand(cmd: string | string[] | undefined): 'bash' | 'write' | 'status' | 'agent' { if (!cmd) return 'bash'; const s = (Array.isArray(cmd) ? cmd.join(' ') : String(cmd)).trimStart(); if (/^(?:agent|delegate)\b/.test(s)) return 'agent'; if (/^status\b/.test(s)) return 'status'; if (/^build\b/.test(s)) return 'status'; if (/^cat\s*>/.test(s)) return 'write'; if (/^cat\b/.test(s) && /(?>?\s*\//.test(s)) return 'write'; if (/^sed\s+-i\b/.test(s)) return 'write'; if (/^ss\b/.test(s)) return 'write'; if (/^(mkdir|touch|rm|mv|cp)\b/.test(s)) return 'write'; if (/^echo\b/.test(s) && /(?>?\s*\//.test(s)) return 'write'; if (/<<-?\s*['"]?\w+/.test(s)) return 'write'; return 'bash'; } function freshState(): ProcessorState { return { result: [], currentTurn: { id: `turn-${Date.now()}`, items: [] }, currentIterationTools: [], itemIdCounter: 0, taskStartTime: 0, prevTaskCumulativeTokens: 0, prevTaskCumulativeCost: 0, textAccum: new Map(), toolParamAccum: new Map(), toolCmdCache: new Map(), }; } export class EventProcessor { private lastProcessedEventId: string | null = null; private lastEventVersions = new Map(); private state: ProcessorState = freshState(); process(events: DebugEvent[]): Turn[] { let state = this.state; if (events.length === 0) { this.lastProcessedEventId = null; this.lastEventVersions = new Map(); this.state = freshState(); return []; } // Find start index (pruning-safe) let startIndex = 0; if (this.lastProcessedEventId) { const idx = events.findIndex(e => e.id === this.lastProcessedEventId); if (idx !== -1) { startIndex = idx + 1; } else { this.lastEventVersions = new Map(); state = freshState(); this.state = state; } } // Check if any conversation_message was updated (e.g. projectContext merged) let needsFullReparse = false; for (let i = 0; i < startIndex; i++) { const evt = events[i]; if (evt.event === 'conversation_message' && evt.version) { const storedVersion = this.lastEventVersions.get(evt.id); if (storedVersion !== undefined && storedVersion !== evt.version) { needsFullReparse = true; break; } } } if (needsFullReparse) { this.lastEventVersions = new Map(); state = freshState(); this.state = state; startIndex = 0; } const newEventsCount = events.length - startIndex; // Re-process coalesced events in lookback window const coalescedEvents: DebugEvent[] = []; const lookbackStart = Math.max(0, startIndex - 4); for (let i = lookbackStart; i < startIndex; i++) { const evt = events[i]; if (evt.event === 'assistant_delta' || evt.event === 'tool_param_delta' || evt.event === 'reasoning_delta') { const storedVersion = this.lastEventVersions.get(evt.id); if (evt.version && storedVersion !== evt.version) { coalescedEvents.push(evt); this.lastEventVersions.set(evt.id, evt.version); } } } if (newEventsCount === 0 && coalescedEvents.length === 0) { return [...state.result, ...(state.currentTurn.items.length > 0 ? [state.currentTurn] : [])]; } const eventsToProcess = [ ...coalescedEvents, ...events.slice(startIndex), ]; for (const event of eventsToProcess) { switch (event.event) { case 'waiting': state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'waiting', timestamp: event.timestamp, data: null, }); break; case 'reasoning_start': state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); break; case 'reasoning_delta': { const reasoningItems = event.data?.all || [event.data]; const accum = state.textAccum.get(event.id) || { text: '', fragmentCount: 0 }; for (let i = accum.fragmentCount; i < reasoningItems.length; i++) { accum.text += reasoningItems[i]?.text || ''; } accum.fragmentCount = reasoningItems.length; state.textAccum.set(event.id, accum); if (!accum.text.trim()) { state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); break; } const matchingReasoningItem = state.currentTurn.items.find( item => item.type === 'reasoning' && item.eventId === event.id ); if (matchingReasoningItem) { matchingReasoningItem.data = accum.text; } else { state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'reasoning', timestamp: event.timestamp, data: accum.text, eventId: event.id, }); } state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); break; } case 'reasoning_complete': state.currentTurn.items.forEach(item => { if (item.type === 'reasoning') item.complete = true; }); state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); break; case 'toolCalls': { state.currentTurn.items.forEach(item => { if (item.type === 'reasoning') item.complete = true; }); const calls = event.data?.toolCalls || []; for (const call of calls) { let parameters: Record = {}; try { parameters = call.function?.arguments ? JSON.parse(call.function.arguments) : {}; } catch { const raw = call.function?.arguments || ''; const partialCmd = extractPartialCmd(raw); parameters = partialCmd !== null ? { command: partialCmd, _raw: raw } : { _raw: raw }; } const tool: ToolCall = { id: call.id || `tool-${state.currentIterationTools.length}`, name: call.function?.name || 'unknown', parameters, status: 'pending', }; state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'tool', timestamp: event.timestamp, data: tool, }); state.currentIterationTools.push(tool); } state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); break; } case 'tool_status': { const { toolCallId, toolIndex, status, result: toolStatusResult, error, args } = event.data || {}; // Match by toolCallId first (reliable), then by index, then by last pending/executing let tool: ToolCall | undefined; if (toolCallId) { tool = state.currentIterationTools.find(t => t.id === toolCallId); } if (!tool && toolIndex != null) { tool = state.currentIterationTools[toolIndex]; } if (!tool) { for (let i = state.currentIterationTools.length - 1; i >= 0; i--) { const t = state.currentIterationTools[i]; if (t.status === 'pending' || t.status === 'executing') { tool = t; break; } } } if (tool) { tool.status = status; if (toolStatusResult) tool.result = toolStatusResult; if (error) tool.error = error; if (status === 'executing') { if (tool.parameters?._raw && typeof tool.parameters._raw === 'string') { try { tool.parameters = JSON.parse(tool.parameters._raw); } catch { /* leave _raw */ } } else if (args && typeof args === 'string') { try { tool.parameters = JSON.parse(args); } catch { /* ignore */ } } } } break; } case 'tool_healed': { const healedTool = state.currentIterationTools[event.data?.toolIndex]; if (healedTool) { healedTool.name = event.data.name || 'bash'; if (event.data.parameters) healedTool.parameters = event.data.parameters; } break; } case 'tool_result': { let toolResult: ToolCall | undefined; if (event.data?.toolCallId) { toolResult = state.currentIterationTools.find(t => t.id === event.data.toolCallId); } if (!toolResult) { toolResult = state.currentIterationTools[event.data?.toolIndex]; } if (toolResult && event.data?.result) { toolResult.result = event.data.result; } break; } case 'tool_param_delta': { const paramDeltaItems = event.data?.all || [event.data]; const globalAccum = state.textAccum.get(event.id) || { text: '', fragmentCount: 0 }; const seenTools = new Set(); for (let i = globalAccum.fragmentCount; i < paramDeltaItems.length; i++) { const { toolId, fragment, partialArguments } = paramDeltaItems[i] || {}; if (!toolId) continue; const accum = state.toolParamAccum.get(toolId) || { text: '', fragmentCount: 0 }; accum.text += fragment ?? partialArguments ?? ''; accum.fragmentCount++; seenTools.add(toolId); state.toolParamAccum.set(toolId, accum); } globalAccum.fragmentCount = paramDeltaItems.length; state.textAccum.set(event.id, globalAccum); for (const toolId of seenTools) { const accum = state.toolParamAccum.get(toolId)!; const toolItem = state.currentTurn.items.find( item => item.type === 'tool' && (item.data as ToolCall)?.id === toolId ); if (toolItem) { const tool = toolItem.data as ToolCall; if (!state.toolCmdCache.has(toolId) || state.toolCmdCache.get(toolId) === null) { state.toolCmdCache.set(toolId, extractPartialCmd(accum.text)); } const cachedCmd = state.toolCmdCache.get(toolId); tool.parameters = cachedCmd !== null && cachedCmd !== undefined ? { command: cachedCmd, _raw: accum.text } : { _raw: accum.text }; } } break; } case 'assistant_delta': { state.currentTurn.items.forEach(item => { if (item.type === 'reasoning') item.complete = true; }); const deltaItems = event.data?.all || [event.data]; const accum = state.textAccum.get(event.id) || { text: '', fragmentCount: 0 }; for (let i = accum.fragmentCount; i < deltaItems.length; i++) { accum.text += deltaItems[i]?.text || ''; } accum.fragmentCount = deltaItems.length; state.textAccum.set(event.id, accum); if (accum.text.trim()) { const matchingTextItem = state.currentTurn.items.find( item => item.type === 'text' && item.eventId === event.id ); if (matchingTextItem) { matchingTextItem.data = accum.text; } else { state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'text', timestamp: event.timestamp, data: accum.text, eventId: event.id, }); } } state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); break; } case 'plan_message': state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'plan', timestamp: event.timestamp, data: event.data?.content || '', }); break; case 'ask': state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'ask', timestamp: event.timestamp, data: { prompt: event.data?.prompt, options: Array.isArray(event.data?.options) ? event.data.options : [], }, }); break; case 'agent_message': state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'agent', timestamp: event.timestamp, data: event.data?.content || '', }); break; case 'task_progress': state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'progress', timestamp: event.timestamp, data: event.data?.content || '', }); break; case 'interview_gate': state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'interview_gate', timestamp: event.timestamp, data: event.data, }); break; case 'conversation_message': { if (event.version) this.lastEventVersions.set(event.id, event.version); const message = event.data?.message; if (message?.role === 'user') { // Harness-injected reminders/nudges/retries are wrapped in // ; they steer the model but are not user input, // so never render them as user messages. if (message.content?.includes('')) break; if (message.content?.includes('Before finishing, run the status command')) break; // The interview gate's incomplete feedback is surfaced via the // interview_gate item — don't also render it as a user message. if (message.content?.includes(INCOMPLETE_PREFIX)) break; const isSyntheticError = message.ui_metadata?.isSyntheticError === true; if (!isSyntheticError && state.currentTurn.items.length > 0) { state.result.push(state.currentTurn); state.currentTurn = { id: `turn-${Date.now()}-${state.result.length}`, items: [], }; } if (!isSyntheticError) { state.taskStartTime = event.timestamp; const lastUsageTurn = [...state.result].reverse().find(t => t.usage); if (lastUsageTurn?.usage) { state.prevTaskCumulativeTokens = lastUsageTurn.usage.totalUsage?.totalTokens || lastUsageTurn.usage.usage?.totalTokens || 0; state.prevTaskCumulativeCost = lastUsageTurn.usage.totalCost ?? 0; } } const projectContext = message.ui_metadata?.projectContext; if (projectContext && !isSyntheticError) { state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'project_context', timestamp: event.timestamp, data: projectContext, }); } // Use ?? so an intentionally-empty displayContent (e.g. an audio-only // message) stays empty instead of falling back to the raw content, // which carries the project-context prefix. const displayContent = message.ui_metadata?.displayContent ?? message.content ?? ''; state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: isSyntheticError ? 'synthetic_error' : 'user', timestamp: event.timestamp, data: displayContent, focusContext: message.ui_metadata?.focusContext, semanticBlocks: message.ui_metadata?.semanticBlocks, attachedFiles: message.ui_metadata?.attachedFiles, }); } else if (message?.role === 'assistant') { // Reconstruct reasoning and tool calls from buffered conversation_message // (delta events aren't buffered, so on replay this is the only source) if (message.reasoning_details?.length) { for (const detail of message.reasoning_details) { if (detail.text?.trim()) { const existing = state.currentTurn.items.find( item => item.type === 'reasoning' && item.data === detail.text ); if (!existing) { state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'reasoning', timestamp: event.timestamp, data: detail.text, complete: true, }); } } } } if (message.tool_calls?.length) { for (const call of message.tool_calls) { const existingTool = state.currentIterationTools.find(t => t.id === call.id); if (existingTool) { // Tool already exists from toolCalls event — update parameters if empty if (!existingTool.parameters?.command || existingTool.parameters.command === '') { try { existingTool.parameters = JSON.parse(call.function.arguments); } catch { /* ignore */ } } } else { let parameters: Record = {}; try { parameters = JSON.parse(call.function.arguments); } catch { parameters = { _raw: call.function.arguments }; } const tool: ToolCall = { id: call.id, name: call.function?.name || 'unknown', parameters, status: 'pending', }; state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'tool', timestamp: event.timestamp, data: tool, }); state.currentIterationTools.push(tool); } } } if (message.content?.trim()) { // Dedup against the streamed assistant_delta text item (same // content already shown live). On replay there is no streamed // item, so this pushes fresh. const existingText = state.currentTurn.items.find( item => item.type === 'text' && item.data === message.content ); if (!existingText) { state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'text', timestamp: event.timestamp, data: message.content, }); } } state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); } else if (message?.role === 'tool') { // Match tool result to its tool call const matchedTool = state.currentIterationTools.find(t => t.id === message.tool_call_id); if (matchedTool) { if (message.content) matchedTool.result = message.content; if (matchedTool.status === 'pending' || matchedTool.status === 'executing') { matchedTool.status = message.content?.startsWith('Error:') ? 'failed' : 'completed'; if (matchedTool.status === 'failed') matchedTool.error = message.content; } } } break; } case 'user_message': state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'user', timestamp: event.timestamp, data: event.data?.content || '', }); break; case 'error': state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'error', timestamp: event.timestamp, data: event.data, }); state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); break; case 'error_paused': state.currentTurn.items.push({ id: `item-${state.itemIdCounter++}`, type: 'error_paused', timestamp: event.timestamp, data: event.data, }); state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); break; case 'usage': state.currentTurn.usage = { ...event.data, timestamp: event.timestamp, taskTokenOffset: state.prevTaskCumulativeTokens, taskCostOffset: state.prevTaskCumulativeCost, taskTokens: event.data.taskTokens, taskCost: event.data.taskCost, }; state.currentTurn.taskStartTime = state.taskStartTime; state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); break; case 'checkpoint_created': state.currentTurn.checkpointId = event.data?.checkpointId; break; case 'iteration': state.currentTurn.iteration = event.data?.iteration; state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); if (state.currentTurn.items.length > 0) { state.result.push(state.currentTurn); state.currentTurn = { id: `turn-${Date.now()}-${state.result.length}`, items: [], }; } state.currentIterationTools = []; break; case 'compaction': state.currentTurn.items.push({ id: `compaction-${event.id}`, type: 'compaction', timestamp: event.timestamp, data: event.data, }); break; case 'agent_progress': { const { event: innerEvent, data: innerData, agentIndex, parentToolIndex: pti } = event.data || {}; const label = `subagent ${agentIndex || 1}`; let agentTool: ToolCall | undefined; if (typeof pti === 'number') { agentTool = state.currentIterationTools[pti]; } if (!agentTool || classifyBashCommand(agentTool.parameters?.command ?? agentTool.parameters?.cmd) !== 'agent') { agentTool = state.currentIterationTools.find( t => t.status === 'executing' && classifyBashCommand(t.parameters?.command ?? t.parameters?.cmd) === 'agent' ); } if (!agentTool) break; if (innerEvent === 'agent_start') { agentTool.result = `[${label}] starting...`; } else if (innerEvent === 'agent_done') { agentTool.result = `[${label}] done (${innerData?.elapsed || '?'}s)`; } else if (innerEvent === 'tool_status' && innerData?.status === 'executing') { let cmd = ''; try { const a = JSON.parse(innerData.args || '{}'); cmd = a?.command || a?.cmd || ''; } catch { cmd = innerData.args || ''; } agentTool.result = `[${label}] ${cmd.length > 100 ? cmd.slice(0, 100) + '...' : cmd}`; } break; } case 'stopped': case 'task_complete': state.currentTurn.items = state.currentTurn.items.filter(item => item.type !== 'waiting'); break; } } if (newEventsCount > 0) { this.lastProcessedEventId = events[events.length - 1].id; } return [...state.result, ...(state.currentTurn.items.length > 0 ? [state.currentTurn] : [])]; } }