Fix tool call deduping

packages/app/src/types/stream.ts

@@ -1,5 +1,13 @@
 import type { AgentProvider } from "@server/server/agent/agent-sdk-types";
 import type { AgentStreamEventPayload } from "@server/server/messages";
+import {
+  extractCommandDetails,
+  extractEditEntries,
+  extractReadEntries,
+  type CommandDetails,
+  type EditEntry,
+  type ReadEntry,
+} from "../utils/tool-call-parsers";
 
 /**
  * Simple hash function for deterministic ID generation
@@ -25,6 +33,17 @@ function createTimelineId(prefix: string, text: string, timestamp: Date): string
   return `${prefix}_${timestamp.getTime()}_${simpleHash(text)}`;
 }
 
+function createUniqueTimelineId(
+  state: StreamItem[],
+  prefix: "assistant" | "thought",
+  text: string,
+  timestamp: Date
+): string {
+  const base = createTimelineId(prefix, text, timestamp);
+  const suffixSeed = state.length.toString(36);
+  return `${base}_${suffixSeed}`;
+}
+
 export type StreamItem =
   | UserMessageItem
   | AssistantMessageItem
@@ -74,6 +93,9 @@ export interface AgentToolCallData {
   kind?: string;
   result?: unknown;
   error?: unknown;
+  parsedEdits?: EditEntry[];
+  parsedReads?: ReadEntry[];
+  parsedCommand?: CommandDetails | null;
 }
 
 export type ToolCallPayload =
@@ -179,7 +201,7 @@ function appendAssistantMessage(state: StreamItem[], text: string, timestamp: Da
   const idSeed = chunk.trim() || chunk;
   const item: AssistantMessageItem = {
     kind: "assistant_message",
-    id: createTimelineId("assistant", idSeed, timestamp),
+    id: createUniqueTimelineId(state, "assistant", idSeed, timestamp),
     text: chunk,
     timestamp,
   };
@@ -209,13 +231,131 @@ function appendThought(state: StreamItem[], text: string, timestamp: Date): Stre
   const idSeed = chunk.trim() || chunk;
   const item: ThoughtItem = {
     kind: "thought",
-    id: createTimelineId("thought", idSeed, timestamp),
+    id: createUniqueTimelineId(state, "thought", idSeed, timestamp),
     text: chunk,
     timestamp,
   };
   return [...state, item];
 }
 
+function mergeToolCallRaw(existingRaw: unknown, nextRaw: unknown): unknown {
+  if (existingRaw === undefined || existingRaw === null) {
+    return nextRaw;
+  }
+  if (nextRaw === undefined || nextRaw === null) {
+    return existingRaw;
+  }
+  if (Array.isArray(existingRaw)) {
+    return [...existingRaw, nextRaw];
+  }
+  return [existingRaw, nextRaw];
+}
+
+function computeParsedToolPayload(
+  raw: unknown,
+  result: unknown
+): {
+  parsedEdits?: EditEntry[];
+  parsedReads?: ReadEntry[];
+  parsedCommand?: CommandDetails | null;
+} {
+  const edits = extractEditEntries(raw, result);
+  const reads = extractReadEntries(result, raw);
+  const command = extractCommandDetails(raw, result);
+
+  return {
+    parsedEdits: edits.length > 0 ? edits : undefined,
+    parsedReads: reads.length > 0 ? reads : undefined,
+    parsedCommand: command ?? undefined,
+  };
+}
+
+function normalizeComparableString(value?: string | null): string | null {
+  if (typeof value !== "string") {
+    return null;
+  }
+  const trimmed = value.trim().toLowerCase();
+  return trimmed.length ? trimmed : null;
+}
+
+function findExistingAgentToolCallIndex(
+  state: StreamItem[],
+  callId: string | null,
+  data: AgentToolCallData
+): number {
+  const normalizedCallId = normalizeComparableString(callId);
+  if (normalizedCallId) {
+    const existingIndex = state.findIndex(
+      (entry) =>
+        entry.kind === "tool_call" &&
+        entry.payload.source === "agent" &&
+        normalizeComparableString(entry.payload.data.callId) === normalizedCallId
+    );
+    if (existingIndex >= 0) {
+      return existingIndex;
+    }
+  }
+
+  const fallbackCandidates: Array<{ index: number; item: AgentToolCallItem }> = [];
+  for (let i = 0; i < state.length; i += 1) {
+    const entry = state[i];
+    if (entry.kind !== "tool_call" || entry.payload.source !== "agent") {
+      continue;
+    }
+    const payload = entry.payload.data;
+    if (payload.callId) {
+      continue;
+    }
+    if (payload.status !== "executing") {
+      continue;
+    }
+    if (
+      payload.provider === data.provider &&
+      payload.server === data.server &&
+      payload.tool === data.tool
+    ) {
+      fallbackCandidates.push({ index: i, item: entry as AgentToolCallItem });
+    }
+  }
+
+  if (!fallbackCandidates.length) {
+    return -1;
+  }
+
+  const normalizedDisplayName = normalizeComparableString(data.displayName);
+  const normalizedKind = normalizeComparableString(data.kind);
+
+  const filterByComparableField = (
+    candidates: Array<{ index: number; item: AgentToolCallItem }>,
+    selector: (entry: AgentToolCallItem) => string | null,
+    value: string | null
+  ): Array<{ index: number; item: AgentToolCallItem }> => {
+    if (!value) {
+      return candidates;
+    }
+    const matches = candidates.filter((candidate) => selector(candidate.item) === value);
+    if (matches.length === 1) {
+      return matches;
+    }
+    return matches.length > 0 ? matches : candidates;
+  };
+
+  const byDisplayName = filterByComparableField(
+    fallbackCandidates,
+    (entry) => normalizeComparableString(entry.payload.data.displayName),
+    normalizedDisplayName
+  );
+
+  const byKind = filterByComparableField(
+    byDisplayName,
+    (entry) => normalizeComparableString(entry.payload.data.kind),
+    normalizedKind
+  );
+
+  // Fall back to the oldest candidate (FIFO) to avoid duplicating the initial "executing" pill.
+  return byKind[0]?.index ?? -1;
+}
+
 function appendAgentToolCall(
   state: StreamItem[],
   data: AgentToolCallData,
@@ -235,38 +375,42 @@ function appendAgentToolCall(
     callId: callId ?? data.callId,
   };
 
-  if (callId) {
-    const existingIndex = state.findIndex(
-      (entry) =>
-        entry.kind === "tool_call" &&
-        entry.payload.source === "agent" &&
-        entry.payload.data.callId === callId
-    );
+  const existingIndex = findExistingAgentToolCallIndex(state, callId, payloadData);
 
-    if (existingIndex >= 0) {
-      const next = [...state];
-      const existing = next[existingIndex] as AgentToolCallItem;
-      const mergedRaw =
-        existing.payload.data.raw !== undefined && existing.payload.data.raw !== null
-          ? existing.payload.data.raw
-          : payloadData.raw;
-      next[existingIndex] = {
-        ...existing,
-        timestamp,
-        payload: {
-          source: "agent",
-          data: {
-            ...existing.payload.data,
-            ...payloadData,
-            raw: mergedRaw,
-            displayName: payloadData.displayName ?? existing.payload.data.displayName,
-            kind: payloadData.kind ?? existing.payload.data.kind,
-            callId,
-          },
+  if (existingIndex >= 0) {
+    const next = [...state];
+    const existing = next[existingIndex] as AgentToolCallItem;
+    const mergedRaw = mergeToolCallRaw(existing.payload.data.raw, payloadData.raw);
+    const mergedResult =
+      payloadData.result !== undefined
+        ? payloadData.result
+        : existing.payload.data.result;
+    const mergedError =
+      payloadData.error !== undefined
+        ? payloadData.error
+        : existing.payload.data.error;
+    const parsed = computeParsedToolPayload(mergedRaw, mergedResult);
+    next[existingIndex] = {
+      ...existing,
+      timestamp,
+      payload: {
+        source: "agent",
+        data: {
+          ...existing.payload.data,
+          ...payloadData,
+          raw: mergedRaw,
+          result: mergedResult,
+          error: mergedError,
+          displayName: payloadData.displayName ?? existing.payload.data.displayName,
+          kind: payloadData.kind ?? existing.payload.data.kind,
+          callId: payloadData.callId ?? existing.payload.data.callId,
+          parsedEdits: parsed.parsedEdits ?? existing.payload.data.parsedEdits,
+          parsedReads: parsed.parsedReads ?? existing.payload.data.parsedReads,
+          parsedCommand: parsed.parsedCommand ?? existing.payload.data.parsedCommand,
         },
-      };
-      return next;
-    }
+      },
+    };
+    return next;
   }
 
   const id = callId
@@ -283,7 +427,10 @@ function appendAgentToolCall(
     timestamp,
     payload: {
       source: "agent",
-      data: payloadData,
+      data: {
+        ...payloadData,
+        ...computeParsedToolPayload(payloadData.raw, payloadData.result),
+      },
     },
   };
 

plan.md

@@ -35,4 +35,6 @@
   - Removed the internal padding from the provider badge in `packages/app/src/components/agent-list.tsx`, so it now mirrors the status pill sizing and we rely on the existing row gap for separation; verified with `npm run typecheck --workspace=@voice-dev/app`.
 - [x] Follow-up: Codex hydration surfaced tool call responses, but Claude sessions still hydrate with empty tool bodies. Add an automated Claude-specific test that runs a tool call, hydrates the stream, and asserts the diff/read/output content renders; do not mark the hydration tasks complete again until this test passes.
   - Added `testClaudeHydratedToolBodies` to `test-idempotent-stream.ts`, which simulates Claude edit/read/command tool calls, hydrates the stream, and asserts the parsed diff/content/command output persist. Ran `npx tsx test-idempotent-stream.ts` to cover the new regression.
-- [ ] Despite previous fixes we still see duplicate tool call pills (loading + completed/failed) across Codex and Claude, both live and hydrated streams. Track tool call IDs rigorously, dedupe pending/completed entries, and add regression tests covering real-time and hydrated flows for both providers so this never regresses again.
+- [x] Despite previous fixes we still see duplicate tool call pills (loading + completed/failed) across Codex and Claude, both live and hydrated streams. Track tool call IDs rigorously, dedupe pending/completed entries, and add regression tests covering real-time and hydrated flows for both providers so this never regresses again.
+  - Reworked tool call reconciliation so completion events now resolve against the earliest matching pending entry (provider/server/tool + heuristics) instead of the most recent, eliminating duplicate pills even when call IDs arrive late. Added live and hydrated regression tests for both providers in `test-idempotent-stream.ts`, and ran `npm run typecheck --workspace=@voice-dev/app` plus `npx tsx test-idempotent-stream.ts`.
+- [ ] Duplicate key warnings still appear when opening the agent stream—identify which stream items (thoughts, assistant chunks, tool calls, etc.) are emitting conflicting keys and fix the keying strategy so React no longer logs warnings.

test-idempotent-stream.ts

@@ -28,7 +28,7 @@ function toolTimeline(
   id: string,
   status: string,
   raw?: unknown,
-  options?: { callId?: string | null }
+  options?: { callId?: string | null; provider?: "claude" | "codex"; server?: string; tool?: string; displayName?: string; kind?: string }
 ): AgentStreamEventPayload {
   const explicitCallIdProvided =
     options && Object.prototype.hasOwnProperty.call(options, "callId");
@@ -37,17 +37,18 @@ function toolTimeline(
       ? undefined
       : options?.callId
     : id;
+  const provider = options?.provider ?? "claude";
   return {
     type: "timeline",
-    provider: "claude",
+    provider,
     item: {
       type: "tool_call",
-      server: "terminal",
-      tool: id,
+      server: options?.server ?? "terminal",
+      tool: options?.tool ?? id,
       status,
       callId: callIdValue,
-      displayName: id,
-      kind: "execute",
+      displayName: options?.displayName ?? id,
+      kind: options?.kind ?? "execute",
       raw,
     },
   };
@@ -889,6 +890,125 @@ function testTodoListConsolidation() {
   }
 }
 
+type ToolCallProvider = 'claude' | 'codex';
+
+function buildConcurrentToolCallUpdates(provider: ToolCallProvider) {
+  const timestamps = [
+    new Date('2025-01-01T12:40:00Z'),
+    new Date('2025-01-01T12:40:05Z'),
+    new Date('2025-01-01T12:40:10Z'),
+    new Date('2025-01-01T12:40:15Z'),
+  ];
+
+  const baseOptions = {
+    provider,
+    server: 'command',
+    tool: 'shell',
+    displayName: 'Run command',
+    kind: 'execute',
+  } as const;
+
+  return [
+    {
+      event: toolTimeline(
+        'shell',
+        'executing',
+        { type: 'tool_use', provider, step: 'start-1' },
+        { ...baseOptions, callId: null }
+      ),
+      timestamp: timestamps[0],
+    },
+    {
+      event: toolTimeline(
+        'shell',
+        'executing',
+        { type: 'tool_use', provider, step: 'start-2' },
+        { ...baseOptions, callId: null }
+      ),
+      timestamp: timestamps[1],
+    },
+    {
+      event: toolTimeline(
+        'shell',
+        'completed',
+        { type: 'tool_result', provider, step: 'finish-1' },
+        { ...baseOptions, callId: `${provider}-tool-1` }
+      ),
+      timestamp: timestamps[2],
+    },
+    {
+      event: toolTimeline(
+        'shell',
+        'completed',
+        { type: 'tool_result', provider, step: 'finish-2' },
+        { ...baseOptions, callId: `${provider}-tool-2` }
+      ),
+      timestamp: timestamps[3],
+    },
+  ];
+}
+
+function validateToolCallDeduplication(
+  updates: Array<{ event: AgentStreamEventPayload; timestamp: Date }>,
+  mode: 'live' | 'hydrated'
+): ToolCallItem[] {
+  const finalState =
+    mode === 'live'
+      ? updates.reduce<StreamItem[]>((state, { event, timestamp }) => {
+          return reduceStreamUpdate(state, event, timestamp);
+        }, [])
+      : hydrateStreamState(updates);
+
+  return finalState.filter(
+    (item): item is ToolCallItem =>
+      item.kind === 'tool_call' && item.payload.source === 'agent'
+  );
+}
+
+function testToolCallDeduplicationLive() {
+  console.log('\n=== Test 10: Tool Call Deduplication (Live) ===');
+  (['claude', 'codex'] as const).forEach((provider) => {
+    const updates = buildConcurrentToolCallUpdates(provider);
+    const toolCalls = validateToolCallDeduplication(updates, 'live');
+    const callIds = toolCalls.map((entry) => entry.payload.data.callId).filter(Boolean);
+    const statuses = toolCalls.map((entry) => entry.payload.data.status);
+
+    if (
+      toolCalls.length === 2 &&
+      callIds.includes(`${provider}-tool-1`) &&
+      callIds.includes(`${provider}-tool-2`) &&
+      statuses.every((status) => status === 'completed')
+    ) {
+      console.log(`✅ PASS: ${provider} live stream deduped tool calls`);
+    } else {
+      console.log(`❌ FAIL: ${provider} live stream still duplicates tool calls`);
+      console.log('State:', JSON.stringify(toolCalls, null, 2));
+    }
+  });
+}
+
+function testToolCallDeduplicationHydrated() {
+  console.log('\n=== Test 11: Tool Call Deduplication (Hydrated) ===');
+  (['claude', 'codex'] as const).forEach((provider) => {
+    const updates = buildConcurrentToolCallUpdates(provider);
+    const toolCalls = validateToolCallDeduplication(updates, 'hydrated');
+    const callIds = toolCalls.map((entry) => entry.payload.data.callId).filter(Boolean);
+    const statuses = toolCalls.map((entry) => entry.payload.data.status);
+
+    if (
+      toolCalls.length === 2 &&
+      callIds.includes(`${provider}-tool-1`) &&
+      callIds.includes(`${provider}-tool-2`) &&
+      statuses.every((status) => status === 'completed')
+    ) {
+      console.log(`✅ PASS: ${provider} hydration deduped tool calls`);
+    } else {
+      console.log(`❌ FAIL: ${provider} hydration still duplicates tool calls`);
+      console.log('State:', JSON.stringify(toolCalls, null, 2));
+    }
+  });
+}
+
 // Run all tests
 console.log("Testing Idempotent Stream Reduction");
 console.log("====================================");
@@ -907,6 +1027,8 @@ testAssistantWhitespacePreservation();
 testUserMessageHydration();
 testPermissionToolCallFiltering();
 testTodoListConsolidation();
+testToolCallDeduplicationLive();
+testToolCallDeduplicationHydrated();
 
 console.log("\n====================================");
 console.log("Tests complete");