Coalesce prepended tool call history

packages/app/src/timeline/session-stream-reducers.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect, it } from "vitest";
 import type { AgentStreamEventPayload } from "@server/shared/messages";
-import type { StreamItem } from "@/types/stream";
+import { hydrateStreamState, type AgentToolCallItem, type StreamItem } from "@/types/stream";
 import {
   createAgentStreamReducerQueue,
   processTimelineResponse,
@@ -31,6 +31,28 @@ function makeTimelineEntry(
   };
 }
 
+function makeToolCallTimelineEntry(
+  seq: number,
+  callId: string,
+  status: "running" | "completed",
+  detail: Record<string, unknown>,
+) {
+  return {
+    seqStart: seq,
+    seqEnd: seq,
+    provider: "claude",
+    item: {
+      type: "tool_call",
+      callId,
+      name: "Read",
+      status,
+      detail,
+      error: null,
+    },
+    timestamp: new Date(1000 + seq).toISOString(),
+  };
+}
+
 function makeTimelineEvent(
   text: string,
   type: string = "assistant_message",
@@ -97,6 +119,13 @@ function getUserTexts(items: StreamItem[]): string[] {
     .map((item) => item.text);
 }
 
+function getAgentToolCalls(items: StreamItem[]) {
+  return items.filter(
+    (item): item is AgentToolCallItem =>
+      item.kind === "tool_call" && item.payload.source === "agent",
+  );
+}
+
 const baseTimelineInput: ProcessTimelineResponseInput = {
   payload: {
     agentId: "agent-1",
@@ -510,6 +539,68 @@ describe("processTimelineResponse", () => {
     });
   });
 
+  it("coalesces tool call lifecycle rows across the older-page prepend boundary", () => {
+    const callId = "toolu_boundary";
+    const currentTail = hydrateStreamState(
+      [
+        {
+          event: {
+            type: "timeline",
+            provider: "claude",
+            item: makeToolCallTimelineEntry(3, callId, "completed", {
+              type: "read",
+              filePath: "/tmp/example.ts",
+            }).item,
+          } as AgentStreamEventPayload,
+          timestamp: new Date(3000),
+        },
+      ],
+      { source: "canonical" },
+    );
+    const existingCursor: TimelineCursor = {
+      epoch: "epoch-1",
+      startSeq: 3,
+      endSeq: 5,
+    };
+
+    const result = processTimelineResponse({
+      ...baseTimelineInput,
+      currentTail,
+      currentCursor: existingCursor,
+      payload: {
+        ...baseTimelineInput.payload,
+        direction: "before",
+        epoch: "epoch-1",
+        startCursor: { seq: 1 },
+        endCursor: { seq: 2 },
+        entries: [
+          makeToolCallTimelineEntry(1, callId, "running", {
+            type: "unknown",
+            input: { file_path: "/tmp/example.ts" },
+            output: null,
+          }),
+        ],
+      },
+    });
+
+    const toolCalls = getAgentToolCalls(result.tail);
+    expect(
+      toolCalls.map((item) => ({
+        id: item.id,
+        callId: item.payload.data.callId,
+        status: item.payload.data.status,
+        detailType: item.payload.data.detail.type,
+      })),
+    ).toEqual([
+      {
+        id: `agent_tool_${callId}`,
+        callId,
+        status: "completed",
+        detailType: "read",
+      },
+    ]);
+  });
+
   it("requests canonical catch-up when a projected entry overlaps unseen seqs", () => {
     const existingCursor: TimelineCursor = {
       epoch: "epoch-1",

packages/app/src/timeline/session-stream-reducers.ts

@@ -3,7 +3,12 @@ import type { AgentLifecycleStatus } from "@server/shared/agent-lifecycle";
 import type { Agent } from "@/stores/session-store";
 import { useSessionStore } from "@/stores/session-store";
 import type { StreamItem } from "@/types/stream";
-import { applyStreamEvent, hydrateStreamState, reduceStreamUpdate } from "@/types/stream";
+import {
+  applyStreamEvent,
+  coalesceAgentToolCallItems,
+  hydrateStreamState,
+  reduceStreamUpdate,
+} from "@/types/stream";
 
 const AGENT_STREAM_REDUCER_FLUSH_DELAY_MS = 16 * 3;
 
@@ -347,10 +352,10 @@ function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: Strea
   const olderLast = olderTail.at(-1);
   const currentFirst = currentTail[0];
   if (olderLast?.kind !== "assistant_message" || currentFirst?.kind !== "assistant_message") {
-    return [...olderTail, ...currentTail];
+    return coalesceAgentToolCallItems([...olderTail, ...currentTail]);
   }
 
-  return [
+  return coalesceAgentToolCallItems([
     ...olderTail.slice(0, -1),
     {
       ...olderLast,
@@ -358,7 +363,7 @@ function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: Strea
       timestamp: currentFirst.timestamp,
     },
     ...currentTail.slice(1),
-  ];
+  ]);
 }
 
 function applyTimelineIncrementalPath(args: {

packages/app/src/types/stream.ts

@@ -438,18 +438,11 @@ function mergeAgentToolCallStatus(
   return "running";
 }
 
-function appendAgentToolCall(
-  state: StreamItem[],
+function mergeAgentToolCallItem(
+  existing: AgentToolCallItem,
   data: AgentToolCallData,
   timestamp: Date,
-): StreamItem[] {
-  const existingIndex = findExistingAgentToolCallIndex(state, data.callId);
-
-  if (existingIndex >= 0) {
-    const existing = state[existingIndex];
-    if (!existing || !isAgentToolCallItem(existing)) {
-      return state;
-    }
+): AgentToolCallItem {
   const mergedStatus = mergeAgentToolCallStatus(existing.payload.data.status, data.status);
   const mergedError =
     mergedStatus === "failed"
@@ -458,20 +451,7 @@ function appendAgentToolCall(
   const mergedMetadata = mergeToolCallMetadata(existing.payload.data.metadata, data.metadata);
   const mergedDetail = mergeToolCallDetail(existing.payload.data.detail, data.detail);
 
-    if (
-      data.provider === existing.payload.data.provider &&
-      data.callId === existing.payload.data.callId &&
-      data.name === existing.payload.data.name &&
-      mergedStatus === existing.payload.data.status &&
-      mergedError === existing.payload.data.error &&
-      mergedDetail === existing.payload.data.detail &&
-      mergedMetadata === existing.payload.data.metadata
-    ) {
-      return state;
-    }
-
-    const next = [...state];
-    next[existingIndex] = {
+  return {
     ...existing,
     timestamp,
     payload: {
@@ -486,6 +466,37 @@ function appendAgentToolCall(
       },
     },
   };
+}
+
+function appendAgentToolCall(
+  state: StreamItem[],
+  data: AgentToolCallData,
+  timestamp: Date,
+): StreamItem[] {
+  const existingIndex = findExistingAgentToolCallIndex(state, data.callId);
+
+  if (existingIndex >= 0) {
+    const existing = state[existingIndex];
+    if (!existing || !isAgentToolCallItem(existing)) {
+      return state;
+    }
+    const merged = mergeAgentToolCallItem(existing, data, timestamp);
+
+    if (
+      merged.timestamp === existing.timestamp &&
+      merged.payload.data.provider === existing.payload.data.provider &&
+      merged.payload.data.callId === existing.payload.data.callId &&
+      merged.payload.data.name === existing.payload.data.name &&
+      merged.payload.data.status === existing.payload.data.status &&
+      merged.payload.data.error === existing.payload.data.error &&
+      merged.payload.data.detail === existing.payload.data.detail &&
+      merged.payload.data.metadata === existing.payload.data.metadata
+    ) {
+      return state;
+    }
+
+    const next = [...state];
+    next[existingIndex] = merged;
     return next;
   }
 
@@ -505,6 +516,28 @@ function appendAgentToolCall(
   return [...state, item];
 }
 
+export function coalesceAgentToolCallItems(items: StreamItem[]): StreamItem[] {
+  const state: StreamItem[] = [];
+  const toolCallIndexByCallId = new Map<string, number>();
+  for (const item of items) {
+    if (!isAgentToolCallItem(item)) {
+      state.push(item);
+      continue;
+    }
+    const existingIndex = toolCallIndexByCallId.get(item.payload.data.callId);
+    if (existingIndex === undefined) {
+      toolCallIndexByCallId.set(item.payload.data.callId, state.length);
+      state.push(item);
+      continue;
+    }
+    const existing = state[existingIndex];
+    if (existing && isAgentToolCallItem(existing)) {
+      state[existingIndex] = mergeAgentToolCallItem(existing, item.payload.data, item.timestamp);
+    }
+  }
+  return state;
+}
+
 function appendActivityLog(state: StreamItem[], entry: ActivityLogItem): StreamItem[] {
   const index = state.findIndex((existing) => existing.id === entry.id);
   if (index >= 0) {