refactor(chat): move socket-event merges into chatRuntimeSlice reducers (Phase 3, slices 1–2) (#4622)

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',
           })
         );
       },

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();
+  });
+});

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,