From c4e45af565c9ad84d530a472ecf223074837bfda Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 17 May 2026 20:53:03 +0700 Subject: [PATCH] Coalesce prepended tool call history --- .../timeline/session-stream-reducers.test.ts | 93 ++++++++++++++++++- .../src/timeline/session-stream-reducers.ts | 13 ++- packages/app/src/types/stream.ts | 91 ++++++++++++------ 3 files changed, 163 insertions(+), 34 deletions(-) diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 2879ddadb..b5a5ba9c0 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/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, +) { + 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", diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 630a11e24..71e7e5c52 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/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: { diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index e0197e64e..f7ff4bc6c 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -438,6 +438,36 @@ function mergeAgentToolCallStatus( return "running"; } +function mergeAgentToolCallItem( + existing: AgentToolCallItem, + data: AgentToolCallData, + timestamp: Date, +): AgentToolCallItem { + const mergedStatus = mergeAgentToolCallStatus(existing.payload.data.status, data.status); + const mergedError = + mergedStatus === "failed" + ? (data.error ?? existing.payload.data.error ?? { message: "Tool call failed" }) + : null; + const mergedMetadata = mergeToolCallMetadata(existing.payload.data.metadata, data.metadata); + const mergedDetail = mergeToolCallDetail(existing.payload.data.detail, data.detail); + + return { + ...existing, + timestamp, + payload: { + source: "agent", + data: { + ...existing.payload.data, + ...data, + status: mergedStatus, + error: mergedError, + detail: mergedDetail, + metadata: mergedMetadata, + }, + }, + }; +} + function appendAgentToolCall( state: StreamItem[], data: AgentToolCallData, @@ -450,42 +480,23 @@ function appendAgentToolCall( if (!existing || !isAgentToolCallItem(existing)) { return state; } - const mergedStatus = mergeAgentToolCallStatus(existing.payload.data.status, data.status); - const mergedError = - mergedStatus === "failed" - ? (data.error ?? existing.payload.data.error ?? { message: "Tool call failed" }) - : null; - const mergedMetadata = mergeToolCallMetadata(existing.payload.data.metadata, data.metadata); - const mergedDetail = mergeToolCallDetail(existing.payload.data.detail, data.detail); + const merged = mergeAgentToolCallItem(existing, data, timestamp); 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 + 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] = { - ...existing, - timestamp, - payload: { - source: "agent", - data: { - ...existing.payload.data, - ...data, - status: mergedStatus, - error: mergedError, - detail: mergedDetail, - metadata: mergedMetadata, - }, - }, - }; + 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(); + 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) {