Surface codex provider tool events live

packages/app/src/types/stream.test.ts

@@ -58,6 +58,24 @@ function toolTimeline(
   };
 }
 
+function codexProviderEvent(
+  eventType: string,
+  itemType: string,
+  overrides: Record<string, unknown>
+): AgentStreamEventPayload {
+  return {
+    type: "provider_event",
+    provider: "codex",
+    raw: {
+      type: eventType,
+      item: {
+        type: itemType,
+        ...overrides,
+      },
+    },
+  } as AgentStreamEventPayload;
+}
+
 function permissionTimeline(id: string, status: string): AgentStreamEventPayload {
   return {
     type: "timeline",
@@ -1236,6 +1254,91 @@ function testFallbackToolCallIdsStayUnique() {
   );
 }
 
+function testCodexProviderEventsProduceToolCalls() {
+  const timestamp = new Date('2025-01-02T08:00:00Z');
+
+  const commandUpdates = [
+    {
+      event: codexProviderEvent('item.started', 'command_execution', {
+        id: 'cmd-provider-1',
+        command: 'ls',
+        aggregated_output: '',
+        status: 'in_progress',
+      }),
+      timestamp,
+    },
+    {
+      event: codexProviderEvent('item.completed', 'command_execution', {
+        id: 'cmd-provider-1',
+        command: 'ls',
+        aggregated_output: 'README.md',
+        exit_code: 0,
+        status: 'completed',
+      }),
+      timestamp,
+    },
+  ];
+
+  const commandState = commandUpdates.reduce<StreamItem[]>((state, { event, timestamp: ts }) => {
+    return reduceStreamUpdate(state, event, ts);
+  }, []);
+
+  const commandTool = commandState.find(
+    (item): item is ToolCallItem =>
+      item.kind === 'tool_call' &&
+      item.payload.source === 'agent' &&
+      item.payload.data.callId === 'cmd-provider-1'
+  );
+
+  assert.ok(commandTool, 'Codex provider command events should create tool call entries');
+  assert.strictEqual(commandTool.payload.data.status, 'completed');
+  assert.strictEqual(commandTool.payload.data.displayName, 'ls');
+  assert.strictEqual(commandTool.payload.data.parsedCommand?.command, 'ls');
+  assert.ok(
+    commandTool.payload.data.parsedCommand?.output?.includes('README.md'),
+    'Command output should hydrate via parsed payloads'
+  );
+
+  const mcpUpdates = [
+    {
+      event: codexProviderEvent('item.started', 'mcp_tool_call', {
+        id: 'mcp-provider-1',
+        server: 'filesystem',
+        tool: 'read_file',
+        arguments: { path: 'package.json' },
+        status: 'in_progress',
+      }),
+      timestamp,
+    },
+    {
+      event: codexProviderEvent('item.completed', 'mcp_tool_call', {
+        id: 'mcp-provider-1',
+        server: 'filesystem',
+        tool: 'read_file',
+        result: { content: [{ type: 'text', text: 'Hello' }] },
+        status: 'completed',
+      }),
+      timestamp,
+    },
+  ];
+
+  const mcpState = mcpUpdates.reduce<StreamItem[]>((state, { event, timestamp: ts }) => {
+    return reduceStreamUpdate(state, event, ts);
+  }, []);
+
+  const mcpTool = mcpState.find(
+    (item): item is ToolCallItem =>
+      item.kind === 'tool_call' &&
+      item.payload.source === 'agent' &&
+      item.payload.data.callId === 'mcp-provider-1'
+  );
+
+  assert.ok(mcpTool, 'Codex provider MCP events should create tool call entries');
+  assert.strictEqual(mcpTool.payload.data.status, 'completed');
+  assert.strictEqual(mcpTool.payload.data.tool, 'read_file');
+  assert.strictEqual(mcpTool.payload.data.displayName, 'filesystem.read_file');
+}
+
 describe('stream timeline reducers', () => {
   it('produces deterministic hydration results', testIdempotentReduction);
   it('deduplicates pending/completed tool entries in place', testUserMessageDeduplication);
@@ -1260,4 +1363,5 @@ describe('stream timeline reducers', () => {
   it('replays metadata-only tool calls without duplicating entries (live)', testMetadataReplayDeduplicationLive);
   it('replays metadata-only tool calls without duplicating entries (hydrated)', testMetadataReplayDeduplicationHydrated);
   it('assigns unique ids for fallback tool calls without call ids', testFallbackToolCallIdsStayUnique);
+  it('surfaces Codex provider events as tool calls', testCodexProviderEventsProduceToolCalls);
 });

packages/app/src/types/stream.ts

@@ -163,6 +163,179 @@ function normalizeChunk(text: string): { chunk: string; hasContent: boolean } {
   return { chunk, hasContent: /\S/.test(chunk) };
 }
 
+function coerceString(value: unknown): string | null {
+  if (typeof value !== "string") {
+    return null;
+  }
+  const trimmed = value.trim();
+  return trimmed.length ? trimmed : null;
+}
+
+function buildCodexCommandLabel(command?: unknown): string {
+  const normalized = coerceString(command);
+  return normalized ?? "Command";
+}
+
+function buildCodexFileChangeSummary(changes: unknown): string {
+  if (!Array.isArray(changes) || changes.length === 0) {
+    return "File change";
+  }
+
+  if (changes.length === 1) {
+    const change = changes[0] as { path?: unknown; kind?: unknown };
+    const kind = coerceString(change?.kind) ?? "edit";
+    const path = coerceString(change?.path) ?? "file";
+    return `${kind}: ${path}`;
+  }
+
+  return `${changes.length} file changes`;
+}
+
+function normalizeCodexStatus(
+  status: unknown,
+  fallback?: "executing" | "completed"
+): "executing" | "completed" | "failed" {
+  if (typeof status === "string") {
+    const normalized = status.trim().toLowerCase();
+    if (normalized === "failed") {
+      return "failed";
+    }
+    if (normalized === "completed") {
+      return "completed";
+    }
+    if (normalized === "in_progress") {
+      return "executing";
+    }
+  }
+  return fallback ?? "executing";
+}
+
+function coerceCodexCallId(item: Record<string, unknown>): string | undefined {
+  const idCandidates = [item.call_id, item.callId, item.tool_use_id, item.id];
+  for (const candidate of idCandidates) {
+    const value = coerceString(candidate);
+    if (value) {
+      return value;
+    }
+  }
+  return undefined;
+}
+
+function convertCodexProviderEvent(
+  provider: AgentProvider,
+  raw: unknown
+): AgentToolCallData | null {
+  if (provider !== "codex" || !raw || typeof raw !== "object") {
+    return null;
+  }
+
+  const rawRecord = raw as { type?: unknown; item?: unknown };
+  const eventType = coerceString(rawRecord.type);
+  if (!eventType || !eventType.startsWith("item.")) {
+    return null;
+  }
+
+  const item = rawRecord.item;
+  if (!item || typeof item !== "object") {
+    return null;
+  }
+
+  const typedItem = item as Record<string, unknown> & { type?: unknown };
+  const itemType = coerceString(typedItem.type);
+  if (!itemType) {
+    return null;
+  }
+
+  const baseStatus = normalizeCodexStatus(
+    typedItem.status,
+    eventType === "item.completed" ? "completed" : undefined
+  );
+  const callId = coerceCodexCallId(typedItem);
+
+  if (itemType === "command_execution") {
+    const command = typedItem.command;
+    const aggregatedOutput = coerceString(
+      (typedItem as { aggregated_output?: unknown }).aggregated_output
+    );
+    const exitCode = typeof typedItem.exit_code === "number" ? typedItem.exit_code : undefined;
+    const resultPayload =
+      aggregatedOutput !== null || exitCode !== undefined
+        ? {
+            output: aggregatedOutput ?? undefined,
+            exitCode,
+            command,
+          }
+        : undefined;
+
+    return {
+      provider,
+      server: "command",
+      tool: "shell",
+      status: baseStatus,
+      raw: typedItem,
+      callId,
+      displayName: buildCodexCommandLabel(command),
+      kind: "execute",
+      result: resultPayload,
+    };
+  }
+
+  if (itemType === "file_change") {
+    const changes = Array.isArray(typedItem.changes) ? typedItem.changes : [];
+    return {
+      provider,
+      server: "file_change",
+      tool: "apply_patch",
+      status: baseStatus,
+      raw: typedItem,
+      callId,
+      displayName: buildCodexFileChangeSummary(changes),
+      kind: "edit",
+      result: { files: changes },
+    };
+  }
+
+  if (itemType === "mcp_tool_call") {
+    const serverName = coerceString(typedItem.server) ?? "mcp";
+    const toolName = coerceString(typedItem.tool) ?? "tool";
+    const argumentsPayload = (typedItem as { arguments?: unknown }).arguments;
+    const resultPayload = (typedItem as { result?: unknown }).result;
+    const errorPayload = (typedItem as { error?: unknown }).error;
+    return {
+      provider,
+      server: serverName,
+      tool: toolName,
+      status: baseStatus,
+      raw: {
+        ...typedItem,
+        arguments: argumentsPayload,
+      },
+      callId,
+      displayName: serverName && toolName ? `${serverName}.${toolName}` : toolName,
+      kind: "tool",
+      result: resultPayload,
+      error: errorPayload,
+    };
+  }
+
+  if (itemType === "web_search") {
+    const query = coerceString(typedItem.query);
+    return {
+      provider,
+      server: "web_search",
+      tool: "web_search",
+      status: baseStatus,
+      raw: typedItem,
+      callId,
+      displayName: query ? `Web search: ${query}` : "Web search",
+      kind: "search",
+      result: { query },
+    };
+  }
+
+  return null;
+}
+
 function appendUserMessage(
   state: StreamItem[],
   text: string,
@@ -795,6 +968,13 @@ export function reduceStreamUpdate(
           return state;
       }
     }
+    case "provider_event": {
+      const converted = convertCodexProviderEvent(event.provider, event.raw);
+      if (!converted) {
+        return state;
+      }
+      return appendAgentToolCall(state, converted, timestamp);
+    }
     default:
       return state;
   }

plan.md

@@ -74,7 +74,8 @@
   - Split OpenAI TTS responses into chunked `audio_output` payloads via a streaming `TTSManager` pipeline (`packages/server/src/server/agent/tts-manager.ts`, `packages/server/src/server/agent/tts-openai.ts`) that tracks per-chunk acknowledgements, adds utterance IDs, and resolves once every chunk finishes. The app’s session context now keeps audio groups active until the last chunk (or an error) lands so playback begins as soon as each chunk arrives without waiting for the full clip (`packages/app/src/contexts/session-context.tsx`). `npm run typecheck --workspace=@voice-dev/server` and `npm run typecheck --workspace=@voice-dev/app` still fail in the pre-existing cross-workspace stream harness files (see earlier tasks); no new regressions were introduced.
 - [x] Add telemetry for barge-in latency (speech start → playback stop, chunk arrival → LLM abort) to measure improvements per milestone.
   - `RealtimeProvider` now records when speech detection interrupts active playback and logs `[Telemetry] barge_in.playback_stop_latency` once the app reports audio stopped, while the server's `Session.handleRealtimeSpeechStart` logs `[Telemetry] barge_in.llm_abort_latency` after aborting live LLM runs—giving us measurable client + server latency metrics per turn.
-- [ ] Codex tool calls aren’t visible in the live stream (they only show up after hydrating/refreshing); fix the reducer/rendering path so Codex tool events appear in real time the same way Claude’s do.
+- [x] Codex tool calls aren’t visible in the live stream (they only show up after hydrating/refreshing); fix the reducer/rendering path so Codex tool events appear in real time the same way Claude’s do.
+  - Stream reducer now inspects Codex `provider_event` payloads and converts command/file/MCP/web search items into agent tool entries immediately, so the UI shows the pills while the turn is running instead of waiting for hydration; `stream.test.ts` gained coverage proving provider events alone still surface command + MCP calls, and `npm run test --workspace=@voice-dev/app -- src/types/stream.test.ts` passes.
 - [ ] Update the AgentInput “Cancel” control to use a square stop icon button (consistent with media players) so interrupting an agent is visually distinct from other actions.
 - [ ] When the agent input has text and we’re not in the cancellable state, replace both the mic (Dictate) and Realtime buttons with a single Send button—only surface Dictate/Realtime when the input is empty or the agent is running.
 - [ ] Remove the agent status pill next to the permission selector; instead show a “working” indicator inside the chat scroll view itself (three bouncing dots animation) whenever the agent is busy.