diff --git a/app/src/providers/ChatRuntimeProvider.tsx b/app/src/providers/ChatRuntimeProvider.tsx index 0dfeadb5a..fceb293f4 100644 --- a/app/src/providers/ChatRuntimeProvider.tsx +++ b/app/src/providers/ChatRuntimeProvider.tsx @@ -29,7 +29,6 @@ import { } from '../services/chatService'; import { store } from '../store'; import { - appendProcessingProse, appendSubagentStreamDelta, bumpInferenceHeartbeatForThread, clearInferenceStatusForThread, @@ -42,18 +41,18 @@ import { markInferenceTurnStreaming, parseToolFailure, recordChatTurnUsage, - recordProcessingTool, recordSubagentTranscriptTool, resolveSubagentTranscriptTool, setInferenceStatusForThread, - setParallelStream, setPendingApprovalForThread, setPendingPlanReviewForThread, setStreamingAssistantForThread, setTaskBoardForThread, setToolTimelineForThread, setWorkflowProposalForThread, - type StreamingAssistantState, + streamDeltaReceived, + toolCallReceived, + toolResultReceived, type ToolTimelineEntry, type ToolTimelineEntryStatus, upsertArtifactFailedForThread, @@ -622,44 +621,17 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { // forked turns that share a thread_id keep independent chains (#4288). skillLatencyRef.current.noteToolCall(segmentDeliveryKey(event.thread_id, event.request_id)); - const existing = store.getState().chatRuntime.toolTimelineByThread[event.thread_id] ?? []; - const existingIdx = event.tool_call_id - ? existing.findIndex(entry => entry.id === event.tool_call_id) - : -1; - - // Stable row id, shared with the processing-transcript tool pointer so - // the panel can resolve the row by `callId`. - const rowId = - event.tool_call_id ?? - `${event.thread_id}:${event.round}:${existing.length}:${event.tool_name}`; - - let entries: ToolTimelineEntry[]; - if (existingIdx >= 0) { - entries = [...existing]; - entries[existingIdx] = decorateEntry({ - ...entries[existingIdx], - name: event.tool_name, - round: event.round, - status: 'running', - displayName: event.tool_display_label ?? entries[existingIdx].displayName, - detail: event.tool_display_detail ?? entries[existingIdx].detail, - }); - } else { - entries = [ - ...existing, - decorateEntry({ - id: rowId, - name: event.tool_name, - round: event.round, - status: 'running', - displayName: event.tool_display_label, - detail: event.tool_display_detail, - }), - ]; - } - dispatch(setToolTimelineForThread({ threadId: event.thread_id, entries })); + // Merge + processing-pointer are now a single reducer (Phase 3) — no + // getState()/full-array rebuild in the provider. dispatch( - recordProcessingTool({ threadId: event.thread_id, round: event.round, callId: rowId }) + toolCallReceived({ + threadId: event.thread_id, + round: event.round, + toolName: event.tool_name, + toolCallId: event.tool_call_id, + displayLabel: event.tool_display_label, + displayDetail: event.tool_display_detail, + }) ); }, onToolResult: (event: ChatToolResultEvent) => { @@ -669,59 +641,19 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { ) return; - // On failure, parse the optional structured explanation (#4254) once so - // both the id-match and name/round-fallback paths can attach it. A - // successful result clears any stale failure carried on the row. - const failure = event.success ? undefined : parseToolFailure(event.failure); - - const existing = store.getState().chatRuntime.toolTimelineByThread[event.thread_id] ?? []; - if (existing.length > 0) { - const nextEntries = [...existing]; - let changed = false; - - // The core forwards the (size-capped) tool result text on `output`; - // keep it on the row so the timeline can show what the tool - // returned. Older cores sent a metadata stub here — accept only - // non-empty payloads so a stub-less row stays `undefined`. - const result = event.output && event.output.length > 0 ? event.output : undefined; - - if (event.tool_call_id) { - const idx = nextEntries.findIndex(entry => entry.id === event.tool_call_id); - if (idx >= 0) { - nextEntries[idx] = { - ...nextEntries[idx], - status: event.success ? 'success' : 'error', - failure, - result, - }; - changed = true; - } - } - - if (!changed) { - for (let i = nextEntries.length - 1; i >= 0; i -= 1) { - const entry = nextEntries[i]; - if ( - entry.status === 'running' && - entry.name === event.tool_name && - entry.round === event.round - ) { - nextEntries[i] = { - ...entry, - status: event.success ? 'success' : 'error', - failure, - result, - }; - changed = true; - break; - } - } - } - - if (changed) { - dispatch(setToolTimelineForThread({ threadId: event.thread_id, entries: nextEntries })); - } - } + // Settle the matching row in the reducer (Phase 3) — no getState() / + // full-array rebuild. A no-op when no row matches. + dispatch( + toolResultReceived({ + threadId: event.thread_id, + round: event.round, + toolName: event.tool_name, + toolCallId: event.tool_call_id, + success: event.success, + output: event.output, + failure: event.failure, + }) + ); // Agent-first Workflow authoring (issue B4): a completed // `propose_workflow` call carries a `workflow_proposal` JSON payload @@ -1095,80 +1027,26 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { } }, onTextDelta: event => { - const cr = store.getState().chatRuntime; - // A parallel (forked) turn streams into its own lane so it doesn't - // clobber the primary turn's stream on the same thread. - if (cr.parallelRequestThreads[event.request_id] !== undefined) { - const prev = cr.parallelStreamsByThread[event.thread_id]?.[event.request_id]; - dispatch( - setParallelStream({ - threadId: event.thread_id, - streaming: { - requestId: event.request_id, - content: `${prev?.content ?? ''}${event.delta}`, - thinking: prev?.thinking ?? '', - }, - }) - ); - return; - } - const existing = cr.streamingAssistantByThread[event.thread_id]; - let streaming: StreamingAssistantState; - if (existing && existing.requestId !== event.request_id) { - streaming = { requestId: event.request_id, content: event.delta, thinking: '' }; - } else { - streaming = { - requestId: event.request_id, - content: `${existing?.content ?? ''}${event.delta}`, - thinking: existing?.thinking ?? '', - }; - } - dispatch(setStreamingAssistantForThread({ threadId: event.thread_id, streaming })); - // Build the live interleaved processing transcript so a mid-turn - // "View processing" isn't empty (the persisted one lands on settle). + // Parallel-vs-primary routing + processing transcript now live in the + // reducer (Phase 3) — no getState() in the provider. dispatch( - appendProcessingProse({ + streamDeltaReceived({ threadId: event.thread_id, - kind: 'narration', + requestId: event.request_id, round: event.round, delta: event.delta, + channel: 'content', }) ); }, onThinkingDelta: event => { - const cr = store.getState().chatRuntime; - if (cr.parallelRequestThreads[event.request_id] !== undefined) { - const prev = cr.parallelStreamsByThread[event.thread_id]?.[event.request_id]; - dispatch( - setParallelStream({ - threadId: event.thread_id, - streaming: { - requestId: event.request_id, - content: prev?.content ?? '', - thinking: `${prev?.thinking ?? ''}${event.delta}`, - }, - }) - ); - return; - } - const existing = cr.streamingAssistantByThread[event.thread_id]; - let streaming: StreamingAssistantState; - if (existing && existing.requestId !== event.request_id) { - streaming = { requestId: event.request_id, content: '', thinking: event.delta }; - } else { - streaming = { - requestId: event.request_id, - content: existing?.content ?? '', - thinking: `${existing?.thinking ?? ''}${event.delta}`, - }; - } - dispatch(setStreamingAssistantForThread({ threadId: event.thread_id, streaming })); dispatch( - appendProcessingProse({ + streamDeltaReceived({ threadId: event.thread_id, - kind: 'thinking', + requestId: event.request_id, round: event.round, delta: event.delta, + channel: 'thinking', }) ); }, diff --git a/app/src/store/__tests__/chatRuntimeSlice.test.ts b/app/src/store/__tests__/chatRuntimeSlice.test.ts index c3bec69ed..4e1ab4830 100644 --- a/app/src/store/__tests__/chatRuntimeSlice.test.ts +++ b/app/src/store/__tests__/chatRuntimeSlice.test.ts @@ -25,6 +25,9 @@ import reducer, { setStreamingAssistantForThread, setTaskBoardForThread, setToolTimelineForThread, + streamDeltaReceived, + toolCallReceived, + toolResultReceived, upsertArtifactFailedForThread, upsertArtifactInProgressForThread, upsertArtifactReadyForThread, @@ -905,3 +908,167 @@ describe('chatRuntimeSlice', () => { }); }); }); + +describe('toolCallReceived (Phase 3 reducer-side merge)', () => { + it('appends a new running row with a generated id and records the processing pointer', () => { + const state = reducer( + undefined, + toolCallReceived({ threadId: 't1', round: 0, toolName: 'shell' }) + ); + const rows = state.toolTimelineByThread['t1']; + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ id: 't1:0:0:shell', name: 'shell', status: 'running' }); + // Fold-in of the processing-transcript pointer (was a second dispatch). + expect(state.processingByThread['t1']).toEqual([ + { kind: 'toolCall', round: 0, seq: 0, callId: 't1:0:0:shell' }, + ]); + }); + + it('upserts an existing row by toolCallId instead of duplicating', () => { + let state = reducer( + undefined, + toolCallReceived({ threadId: 't1', round: 0, toolName: 'shell', toolCallId: 'call-1' }) + ); + state = reducer( + state, + toolCallReceived({ threadId: 't1', round: 1, toolName: 'shell', toolCallId: 'call-1' }) + ); + const rows = state.toolTimelineByThread['t1']; + expect(rows).toHaveLength(1); + expect(rows[0].round).toBe(1); + // Processing pointer is recorded once for the stable callId. + expect(state.processingByThread['t1']).toHaveLength(1); + }); +}); + +describe('toolResultReceived (Phase 3 reducer-side merge)', () => { + const withRunningRow = () => + reducer( + undefined, + setToolTimelineForThread({ + threadId: 't1', + entries: [{ id: 'call-1', name: 'shell', round: 0, status: 'running' }], + }) + ); + + it('settles the row matched by toolCallId, attaching output', () => { + const state = reducer( + withRunningRow(), + toolResultReceived({ + threadId: 't1', + round: 0, + toolName: 'shell', + toolCallId: 'call-1', + success: true, + output: 'done', + }) + ); + expect(state.toolTimelineByThread['t1'][0]).toMatchObject({ + status: 'success', + result: 'done', + }); + }); + + it('falls back to the newest running row of the same name+round when no id matches', () => { + const state = reducer( + withRunningRow(), + toolResultReceived({ threadId: 't1', round: 0, toolName: 'shell', success: false }) + ); + expect(state.toolTimelineByThread['t1'][0].status).toBe('error'); + }); + + it('is a no-op when nothing matches (mirrors the provider changed-guard)', () => { + const before = withRunningRow(); + const after = reducer( + before, + toolResultReceived({ threadId: 't1', round: 9, toolName: 'other', success: true }) + ); + expect(after.toolTimelineByThread['t1']).toEqual(before.toolTimelineByThread['t1']); + }); +}); + +describe('streamDeltaReceived (Phase 3 reducer-side merge)', () => { + it('appends a content delta to the primary stream and coalesces processing narration', () => { + let state = reducer( + undefined, + streamDeltaReceived({ + threadId: 't1', + requestId: 'r1', + round: 0, + delta: 'Hel', + channel: 'content', + }) + ); + state = reducer( + state, + streamDeltaReceived({ + threadId: 't1', + requestId: 'r1', + round: 0, + delta: 'lo', + channel: 'content', + }) + ); + expect(state.streamingAssistantByThread['t1']).toEqual({ + requestId: 'r1', + content: 'Hello', + thinking: '', + }); + // Two deltas coalesce into one narration block. + expect(state.processingByThread['t1']).toEqual([ + { kind: 'narration', round: 0, seq: 0, text: 'Hello' }, + ]); + }); + + it('starts a fresh preview when the requestId changes (drops the prior tail)', () => { + let state = reducer( + undefined, + streamDeltaReceived({ + threadId: 't1', + requestId: 'r1', + round: 0, + delta: 'old', + channel: 'content', + }) + ); + state = reducer( + state, + streamDeltaReceived({ + threadId: 't1', + requestId: 'r2', + round: 0, + delta: 'new', + channel: 'thinking', + }) + ); + expect(state.streamingAssistantByThread['t1']).toEqual({ + requestId: 'r2', + content: '', + thinking: 'new', + }); + }); + + it('routes a forked (parallel) turn into its own lane without touching the primary or processing', () => { + let state = reducer( + undefined, + registerParallelRequest({ threadId: 't1', requestId: 'branch' }) + ); + state = reducer( + state, + streamDeltaReceived({ + threadId: 't1', + requestId: 'branch', + round: 0, + delta: 'B', + channel: 'content', + }) + ); + expect(state.parallelStreamsByThread['t1']['branch']).toEqual({ + requestId: 'branch', + content: 'B', + thinking: '', + }); + expect(state.streamingAssistantByThread['t1']).toBeUndefined(); + expect(state.processingByThread['t1']).toBeUndefined(); + }); +}); diff --git a/app/src/store/chatRuntimeSlice.ts b/app/src/store/chatRuntimeSlice.ts index 7872e9dfe..2b33f59dd 100644 --- a/app/src/store/chatRuntimeSlice.ts +++ b/app/src/store/chatRuntimeSlice.ts @@ -13,6 +13,7 @@ import type { PersistedTurnState, TaskBoard, } from '../types/turnState'; +import { formatTimelineEntry, isKnownClientTool } from '../utils/toolTimelineFormatting'; import { resetUserScopedState } from './resetActions'; const turnStateLog = debug('chatRuntime.turnState'); @@ -236,6 +237,21 @@ export function parseToolFailure(raw: unknown): ToolFailureExplanation | undefin }; } +/** + * Attach a human label/detail to a tool-timeline row. The server supplies a + * label/detail for dynamic Composio/MCP/integration tools the client can't know + * — trust it for those; for the fixed set of built-ins the client formatter + * (args-aware) stays authoritative. Pure — shared by the live reducers and any + * caller that materialises a row. + */ +function decorateEntry(entry: ToolTimelineEntry): ToolTimelineEntry { + const formatted = formatTimelineEntry(entry); + if (entry.displayName && !isKnownClientTool(entry.name)) { + return { ...entry, displayName: entry.displayName, detail: entry.detail ?? formatted.detail }; + } + return { ...entry, displayName: formatted.title, detail: formatted.detail ?? entry.detail }; +} + export interface ToolTimelineEntry { id: string; name: string; @@ -1041,6 +1057,157 @@ const chatRuntimeSlice = createSlice({ if (list.some(i => i.kind === 'toolCall' && i.callId === callId)) return; list.push({ kind: 'toolCall', round, seq: list.length, callId }); }, + /** + * Reducer-side merge for a `tool_call` socket event (Phase 3 — replaces the + * provider's `getState()` + find-row + full-array-rebuild). Upserts the row + * by `toolCallId` (falling back to a generated stable id), decorates its + * label/detail, and records the processing-transcript pointer in one pass. + */ + toolCallReceived: ( + state, + action: PayloadAction<{ + threadId: string; + round: number; + toolName: string; + toolCallId?: string; + displayLabel?: string; + displayDetail?: string; + }> + ) => { + const { threadId, round, toolName, toolCallId, displayLabel, displayDetail } = action.payload; + const entries = (state.toolTimelineByThread[threadId] ??= []); + const existingIdx = toolCallId ? entries.findIndex(e => e.id === toolCallId) : -1; + // Stable row id, shared with the processing-transcript tool pointer so the + // panel can resolve the row by `callId`. + const rowId = toolCallId ?? `${threadId}:${round}:${entries.length}:${toolName}`; + if (existingIdx >= 0) { + const prev = entries[existingIdx]; + entries[existingIdx] = decorateEntry({ + ...prev, + name: toolName, + round, + status: 'running', + displayName: displayLabel ?? prev.displayName, + detail: displayDetail ?? prev.detail, + }); + } else { + entries.push( + decorateEntry({ + id: rowId, + name: toolName, + round, + status: 'running', + displayName: displayLabel, + detail: displayDetail, + }) + ); + } + // Fold the processing-transcript pointer (was a second dispatch). + const list = (state.processingByThread[threadId] ??= []); + if (!list.some(i => i.kind === 'toolCall' && i.callId === rowId)) { + list.push({ kind: 'toolCall', round, seq: list.length, callId: rowId }); + } + }, + /** + * Reducer-side merge for a `tool_result` socket event (Phase 3). Settles the + * matching row by `toolCallId`, else the newest still-running row with the + * same name+round. A no-op when no row matches (mirrors the provider's + * `changed` guard). `failure` is the raw socket payload, parsed here. + */ + toolResultReceived: ( + state, + action: PayloadAction<{ + threadId: string; + round: number; + toolName: string; + toolCallId?: string; + success: boolean; + output?: string; + failure?: unknown; + }> + ) => { + const { threadId, round, toolName, toolCallId, success, output, failure } = action.payload; + const entries = state.toolTimelineByThread[threadId]; + if (!entries || entries.length === 0) return; + const status: ToolTimelineEntryStatus = success ? 'success' : 'error'; + // On failure, parse the optional structured explanation (#4254); a + // successful result clears any stale failure carried on the row. + const parsedFailure = success ? undefined : parseToolFailure(failure); + // The core forwards the (size-capped) tool result text on `output`; accept + // only non-empty payloads so a stub-less row stays `undefined`. + const result = output && output.length > 0 ? output : undefined; + if (toolCallId) { + const entry = entries.find(e => e.id === toolCallId); + if (entry) { + entry.status = status; + entry.failure = parsedFailure; + entry.result = result; + return; + } + } + for (let i = entries.length - 1; i >= 0; i -= 1) { + const entry = entries[i]; + if (entry.status === 'running' && entry.name === toolName && entry.round === round) { + entry.status = status; + entry.failure = parsedFailure; + entry.result = result; + return; + } + } + }, + /** + * Reducer-side merge for a `text_delta` / `thinking_delta` socket event + * (Phase 3 — replaces the provider's `getState()` + parallel-vs-primary + * routing). Forked (parallel) turns append into their own lane and skip the + * processing transcript; the primary turn appends to the streaming preview + * and coalesces a narration/thinking block into the live processing panel. + * A `requestId` change starts a fresh preview (drops the prior turn's tail). + */ + streamDeltaReceived: ( + state, + action: PayloadAction<{ + threadId: string; + requestId: string; + round: number; + delta: string; + channel: 'content' | 'thinking'; + }> + ) => { + const { threadId, requestId, round, delta, channel } = action.payload; + // A parallel (forked) turn streams into its own lane so it doesn't clobber + // the primary turn's stream on the same thread. + if (state.parallelRequestThreads[requestId] !== undefined) { + const lane = (state.parallelStreamsByThread[threadId] ??= {}); + const prev = lane[requestId]; + lane[requestId] = { + requestId, + content: channel === 'content' ? `${prev?.content ?? ''}${delta}` : (prev?.content ?? ''), + thinking: + channel === 'thinking' ? `${prev?.thinking ?? ''}${delta}` : (prev?.thinking ?? ''), + }; + return; + } + const existing = state.streamingAssistantByThread[threadId]; + const sameTurn = existing != null && existing.requestId === requestId; + const carryContent = sameTurn ? existing.content : ''; + const carryThinking = sameTurn ? existing.thinking : ''; + state.streamingAssistantByThread[threadId] = { + requestId, + content: channel === 'content' ? `${carryContent}${delta}` : carryContent, + thinking: channel === 'thinking' ? `${carryThinking}${delta}` : carryThinking, + }; + // Live interleaved processing transcript so a mid-turn "View processing" + // isn't empty — coalesce into the trailing same-kind, same-round block. + if (!delta) return; + const kind = channel === 'content' ? 'narration' : 'thinking'; + const list = (state.processingByThread[threadId] ??= []); + const last = list[list.length - 1]; + if (last && last.kind === kind && last.round === round) { + last.text += delta; + } else { + list.push({ kind, round, seq: list.length, text: delta }); + } + }, /** * Optimistically mark a detached background sub-agent as cancelled after the * user confirms a cancel via `openhuman.subagent_cancel`. The aborted run @@ -1640,6 +1807,9 @@ export const { setToolTimelineForThread, clearToolTimelineForThread, setTurnTimelinesForThread, + streamDeltaReceived, + toolCallReceived, + toolResultReceived, clearProcessingForThread, appendProcessingProse, recordProcessingTool,