feat(workflow): observe Codex and Claude agent runs

src/local-agent-adapters.ts

@@ -19,11 +19,13 @@ import {
   ProviderSchemaUnsupportedError,
   type LocalAgentRunInput,
   type LocalAgentRunResult,
+  type LocalAgentObserver,
+  type LocalAgentUsageSnapshot,
 } from "./local-agent-runtime.js";
 
 export interface LocalAgentAdapter {
   readonly provider: LocalAgentProvider;
-  run(input: LocalAgentRunInput): Promise<LocalAgentRunResult>;
+  run(input: LocalAgentRunInput, observer?: LocalAgentObserver): Promise<LocalAgentRunResult>;
 }
 
 const ACP_COMMANDS: Record<"cursor" | "copilot", [string, ...string[]]> = {
@@ -35,18 +37,20 @@ const PI_AGENT_TIMEOUT_MS = 120_000;
 export async function runLocalAgentProvider(
   provider: LocalAgentProvider,
   input: LocalAgentRunInput,
+  observer?: LocalAgentObserver,
 ): Promise<LocalAgentRunResult> {
-  const result = await runLocalAgentProviderResult(provider, input);
+  const result = await runLocalAgentProviderResult(provider, input, observer);
   if (result.isErr()) throw result.error;
   return result.value;
 }
 
 export async function runLocalAgentProviderResult(
   provider: LocalAgentProvider,
   input: LocalAgentRunInput,
+  observer?: LocalAgentObserver,
 ): Promise<BetterResult<LocalAgentRunResult, AgentProviderError>> {
   return Result.tryPromise({
-    try: () => createLocalAgentAdapter(provider).run(input),
+    try: () => createLocalAgentAdapter(provider).run(input, observer),
     catch: (cause) => classifyAgentProviderError(provider, cause),
   });
 }
@@ -70,16 +74,16 @@ export function createLocalAgentAdapter(provider: LocalAgentProvider): LocalAgen
 class CodexLocalAgentAdapter implements LocalAgentAdapter {
   readonly provider = "codex" as const;
 
-  async run(input: LocalAgentRunInput): Promise<LocalAgentRunResult> {
+  async run(input: LocalAgentRunInput, observer?: LocalAgentObserver): Promise<LocalAgentRunResult> {
     const runtime = await createCodexSdkLocalAgentRuntime();
-    return runtime.run(input);
+    return runtime.run(input, observer);
   }
 }
 
 class ClaudeLocalAgentAdapter implements LocalAgentAdapter {
   readonly provider = "claude" as const;
 
-  async run(input: LocalAgentRunInput): Promise<LocalAgentRunResult> {
+  async run(input: LocalAgentRunInput, observer?: LocalAgentObserver): Promise<LocalAgentRunResult> {
     const { query } = await import("@anthropic-ai/claude-agent-sdk");
     const claudeExecutable = process.env.CLAUDE_COMMAND ?? resolveExecutable("claude");
     try {
@@ -107,7 +111,11 @@ class ClaudeLocalAgentAdapter implements LocalAgentAdapter {
       for await (const message of messages) {
         items.push(message);
         const record = message as Record<string, unknown>;
-        if (typeof record.session_id === "string") providerSessionId = record.session_id;
+        if (typeof record.session_id === "string") {
+          providerSessionId = record.session_id;
+          observer?.onSession?.(record.session_id);
+        }
+        notifyClaudeActivity(record, observer);
         if (record.type !== "result") continue;
         const resultError = claudeResultError(record);
         if (resultError) throw new Error(resultError);
@@ -116,14 +124,21 @@ class ClaudeLocalAgentAdapter implements LocalAgentAdapter {
           finalResponse = extracted.finalResponse;
           structured = extracted.structured;
         }
+        const usage = claudeUsage(record.usage, "final");
+        if (usage) observer?.onUsage?.(usage);
       }
 
       finalResponse = requireFinalResponse("Claude", finalResponse);
+      const usage = [...items]
+        .reverse()
+        .map((item) => claudeUsage((item as Record<string, unknown>).usage, "final"))
+        .find((snapshot) => snapshot !== undefined);
       return {
         provider: this.provider,
         providerSessionId,
         finalResponse,
         items,
+        ...(usage ? { usage } : {}),
         ...(structured !== undefined ? { structured } : {}),
       };
     } catch (error) {
@@ -135,6 +150,58 @@ class ClaudeLocalAgentAdapter implements LocalAgentAdapter {
   }
 }
 
+function notifyClaudeActivity(record: Record<string, unknown>, observer?: LocalAgentObserver): void {
+  if (record.type === "tool_progress" && typeof record.tool_name === "string") {
+    observer?.onActivity?.({ kind: "tool", status: "running", label: record.tool_name });
+    return;
+  }
+  if (record.type === "tool_use_summary" && typeof record.summary === "string") {
+    observer?.onActivity?.({ kind: "tool", status: "completed", label: record.summary });
+    return;
+  }
+  if (record.type !== "assistant") return;
+  const message = record.message as { content?: unknown[] } | undefined;
+  for (const block of message?.content ?? []) {
+    const content = block as Record<string, unknown>;
+    if (content.type !== "tool_use" || typeof content.name !== "string") continue;
+    observer?.onActivity?.({
+      kind: content.name === "Bash" ? "command" : content.name === "Write" || content.name === "Edit" ? "file" : "tool",
+      status: "running",
+      label: content.name,
+      detail: claudeToolDetail(content.input),
+    });
+  }
+}
+
+function claudeToolDetail(input: unknown): string | undefined {
+  if (!input || typeof input !== "object") return undefined;
+  const record = input as Record<string, unknown>;
+  for (const key of ["command", "file_path", "path", "query"]) {
+    if (typeof record[key] === "string") return record[key];
+  }
+  return undefined;
+}
+
+function claudeUsage(value: unknown, state: "partial" | "final"): LocalAgentUsageSnapshot | undefined {
+  if (!value || typeof value !== "object") return undefined;
+  const usage = value as Record<string, unknown>;
+  const inputTokens = nonNegativeInteger(usage.input_tokens);
+  const outputTokens = nonNegativeInteger(usage.output_tokens);
+  if (inputTokens === undefined && outputTokens === undefined) return undefined;
+  return {
+    inputTokens,
+    cachedInputTokens: nonNegativeInteger(usage.cache_read_input_tokens),
+    cacheCreationInputTokens: nonNegativeInteger(usage.cache_creation_input_tokens),
+    outputTokens,
+    totalTokens: (inputTokens ?? 0) + (outputTokens ?? 0),
+    state,
+  };
+}
+
+function nonNegativeInteger(value: unknown): number | undefined {
+  return typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : undefined;
+}
+
 /** Build Claude SDK outputFormat when a JSON Schema is requested. */
 export function claudeOutputFormatOptions(
   schema: JsonSchema | undefined,

src/local-agent-runtime.test.ts

@@ -1,5 +1,5 @@
 import assert from "node:assert/strict";
-import type { RunResult, ThreadOptions } from "@openai/codex-sdk";
+import type { RunResult, RunStreamedResult, ThreadEvent, ThreadOptions } from "@openai/codex-sdk";
 import {
   CodexSdkLocalAgentRuntime,
   createCodexSdkLocalAgentRuntime,
@@ -69,6 +69,63 @@ assert.deepEqual(codex.started[0], {
   modelReasoningEffort: undefined,
 });
 
+const streamedEvents: ThreadEvent[] = [
+  { type: "thread.started", thread_id: "stream-thread" },
+  {
+    type: "item.started",
+    item: {
+      id: "command-1",
+      type: "command_execution",
+      command: "npm test",
+      aggregated_output: "",
+      status: "in_progress",
+    },
+  },
+  {
+    type: "item.completed",
+    item: {
+      id: "command-1",
+      type: "command_execution",
+      command: "npm test",
+      aggregated_output: "ok",
+      exit_code: 0,
+      status: "completed",
+    },
+  },
+  { type: "item.completed", item: { id: "message-1", type: "agent_message", text: "done" } },
+  {
+    type: "turn.completed",
+    usage: { input_tokens: 100, cached_input_tokens: 20, output_tokens: 30, reasoning_output_tokens: 10 },
+  },
+];
+const streamingThread = {
+  id: "stream-thread",
+  async run(): Promise<RunResult> { throw new Error("unreachable"); },
+  async runStreamed(): Promise<RunStreamedResult> {
+    return { events: (async function* () { yield* streamedEvents; })() };
+  },
+};
+const observedSessions: string[] = [];
+const observedActivity: string[] = [];
+const observedUsage: number[] = [];
+const streamedRuntime = new CodexSdkLocalAgentRuntime({
+  startThread: () => streamingThread,
+  resumeThread: () => streamingThread,
+});
+const streamed = await streamedRuntime.run(
+  { prompt: "test", workspace: "/tmp/project" },
+  {
+    onSession: (id) => observedSessions.push(id),
+    onActivity: (activity) => observedActivity.push(`${activity.status}:${activity.label}`),
+    onUsage: (usage) => observedUsage.push(usage.totalTokens),
+  },
+);
+assert.equal(streamed.finalResponse, "done");
+assert.equal(streamed.usage?.totalTokens, 130);
+assert.deepEqual(observedSessions, ["stream-thread"]);
+assert.deepEqual(observedActivity, ["running:npm test", "completed:npm test"]);
+assert.deepEqual(observedUsage, [130]);
+
 await runtime.run({
   prompt: "make change",
   workspace: "/tmp/project",

src/local-agent-runtime.ts

@@ -3,7 +3,10 @@ import type {
   CodexOptions,
   ModelReasoningEffort,
   RunResult,
+  RunStreamedResult,
   SandboxMode,
+  ThreadEvent,
+  ThreadItem,
   ThreadOptions,
   TurnOptions,
 } from "@openai/codex-sdk";
@@ -41,16 +44,40 @@ export interface LocalAgentRunResult {
   items: unknown[];
   /** Provider-native structured object when schema was requested. */
   structured?: unknown;
+  usage?: LocalAgentUsageSnapshot;
+}
+
+export interface LocalAgentUsageSnapshot {
+  inputTokens?: number;
+  cachedInputTokens?: number;
+  cacheCreationInputTokens?: number;
+  outputTokens?: number;
+  totalTokens: number;
+  state: "partial" | "final";
+}
+
+export interface LocalAgentActivity {
+  kind: "tool" | "command" | "file" | "status";
+  status: "running" | "completed" | "failed";
+  label: string;
+  detail?: string;
+}
+
+export interface LocalAgentObserver {
+  onSession?(providerSessionId: string): void;
+  onUsage?(usage: LocalAgentUsageSnapshot): void;
+  onActivity?(activity: LocalAgentActivity): void;
 }
 
 export interface LocalAgentRuntime {
   readonly provider: LocalAgentProvider;
-  run(input: LocalAgentRunInput): Promise<LocalAgentRunResult>;
+  run(input: LocalAgentRunInput, observer?: LocalAgentObserver): Promise<LocalAgentRunResult>;
 }
 
 interface CodexThreadLike {
   readonly id: string | null;
   run(prompt: string, turnOptions?: TurnOptions): Promise<RunResult>;
+  runStreamed?(prompt: string, turnOptions?: TurnOptions): Promise<RunStreamedResult>;
 }
 
 interface CodexClientLike {
@@ -90,32 +117,109 @@ export class CodexSdkLocalAgentRuntime implements LocalAgentRuntime {
     this.codex = codex;
   }
 
-  async run(input: LocalAgentRunInput): Promise<LocalAgentRunResult> {
+  async run(input: LocalAgentRunInput, observer?: LocalAgentObserver): Promise<LocalAgentRunResult> {
     const options = threadOptionsFor(input);
     const thread = input.providerSessionId
       ? this.codex.resumeThread(input.providerSessionId, options)
       : this.codex.startThread(options);
     const turnOptions = input.schema ? { outputSchema: input.schema } : undefined;
     let turn: RunResult;
+    const streamed = thread.runStreamed !== undefined;
     try {
-      turn = await thread.run(input.prompt, turnOptions);
+      turn = thread.runStreamed
+        ? await collectCodexStream(await thread.runStreamed(input.prompt, turnOptions), observer)
+        : await thread.run(input.prompt, turnOptions);
     } catch (error) {
       if (input.schema && isNativeSchemaUnsupportedFailure(error)) {
         throw new ProviderSchemaUnsupportedError(this.provider, error);
       }
       throw error;
     }
 
+    if (!streamed && thread.id) observer?.onSession?.(thread.id);
+    const usage = turn.usage ? codexUsage(turn.usage) : undefined;
+    if (usage) observer?.onUsage?.(usage);
     return {
       provider: this.provider,
       providerSessionId: thread.id,
       finalResponse: turn.finalResponse,
       items: turn.items,
+      usage,
       ...(input.schema ? { structured: tryParseJson(turn.finalResponse) } : {}),
     };
   }
 }
 
+async function collectCodexStream(
+  streamed: RunStreamedResult,
+  observer?: LocalAgentObserver,
+): Promise<RunResult> {
+  const items: ThreadItem[] = [];
+  let finalResponse = "";
+  let usage: RunResult["usage"] = null;
+  for await (const event of streamed.events) {
+    if (event.type === "thread.started") observer?.onSession?.(event.thread_id);
+    if (event.type === "item.started") notifyCodexItem(event.item, "running", observer);
+    if (event.type === "item.completed") {
+      items.push(event.item);
+      notifyCodexItem(event.item, codexItemStatus(event.item), observer);
+      if (event.item.type === "agent_message") finalResponse = event.item.text;
+    }
+    if (event.type === "turn.completed") usage = event.usage;
+    if (event.type === "turn.failed") throw new Error(event.error.message);
+    if (event.type === "error") throw new Error(event.message);
+  }
+  return { items, finalResponse, usage };
+}
+
+function notifyCodexItem(
+  item: ThreadItem,
+  status: LocalAgentActivity["status"],
+  observer?: LocalAgentObserver,
+): void {
+  const activity = codexItemActivity(item, status);
+  if (activity) observer?.onActivity?.(activity);
+}
+
+function codexItemActivity(
+  item: ThreadItem,
+  status: LocalAgentActivity["status"],
+): LocalAgentActivity | undefined {
+  if (item.type === "command_execution") {
+    return { kind: "command", status, label: item.command };
+  }
+  if (item.type === "file_change") {
+    return {
+      kind: "file",
+      status,
+      label: "apply file changes",
+      detail: item.changes.map((change) => `${change.kind} ${change.path}`).join(", "),
+    };
+  }
+  if (item.type === "mcp_tool_call") {
+    return { kind: "tool", status, label: `${item.server}.${item.tool}` };
+  }
+  if (item.type === "web_search") return { kind: "tool", status, label: "web search", detail: item.query };
+  return undefined;
+}
+
+function codexItemStatus(item: ThreadItem): LocalAgentActivity["status"] {
+  if (item.type === "command_execution" || item.type === "mcp_tool_call" || item.type === "file_change") {
+    return item.status === "failed" ? "failed" : "completed";
+  }
+  return "completed";
+}
+
+function codexUsage(usage: NonNullable<RunResult["usage"]>): LocalAgentUsageSnapshot {
+  return {
+    inputTokens: usage.input_tokens,
+    cachedInputTokens: usage.cached_input_tokens,
+    outputTokens: usage.output_tokens,
+    totalTokens: usage.input_tokens + usage.output_tokens,
+    state: "final",
+  };
+}
+
 function tryParseJson(text: string): unknown | undefined {
   try {
     return JSON.parse(text) as unknown;