feat(agents): stream provider observations

package.json

@@ -28,7 +28,7 @@
     "dev": "node scripts/dev-server.mjs",
     "postinstall": "node scripts/fix-node-pty-permissions.mjs",
     "start": "node dist/cli.js serve",
-    "test": "tsx src/config.test.ts && tsx src/cli-workspace.test.ts && tsx src/cli-output.test.ts && tsx src/open-workspace-capabilities.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-capabilities.test.ts && tsx src/local-agent-catalog.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-resolution.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skill-install.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && tsx src/cli.test.ts && tsx src/workflow-contracts.test.ts && tsx src/workflow-errors.test.ts && tsx src/workflow-types.test.ts && tsx src/workflow-store.test.ts && tsx src/workflow-lifecycle.test.ts && tsx src/workflow-view.test.ts && tsx src/workflow-summary.test.ts && tsx src/workflow-tui.test.ts && tsx src/workflow-script.test.ts && tsx src/workflow-sandbox.test.ts && tsx src/workflow-engine.test.ts && tsx src/workflow-files.test.ts && tsx src/workflow-launch.test.ts && tsx src/workflow-replay.test.ts && tsx src/workflow-schema.test.ts",
+    "test": "tsx src/config.test.ts && tsx src/cli-workspace.test.ts && tsx src/cli-output.test.ts && tsx src/open-workspace-capabilities.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-provider-observations.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-capabilities.test.ts && tsx src/local-agent-catalog.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-resolution.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skill-install.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && tsx src/cli.test.ts && tsx src/workflow-contracts.test.ts && tsx src/workflow-errors.test.ts && tsx src/workflow-types.test.ts && tsx src/workflow-store.test.ts && tsx src/workflow-lifecycle.test.ts && tsx src/workflow-view.test.ts && tsx src/workflow-summary.test.ts && tsx src/workflow-tui.test.ts && tsx src/workflow-script.test.ts && tsx src/workflow-sandbox.test.ts && tsx src/workflow-engine.test.ts && tsx src/workflow-files.test.ts && tsx src/workflow-launch.test.ts && tsx src/workflow-replay.test.ts && tsx src/workflow-schema.test.ts",
     "typecheck": "tsc -p tsconfig.json --noEmit"
   },
   "keywords": [],

src/cli-output.test.ts

@@ -87,6 +87,7 @@ const call: WorkflowAgentCallRecord = {
   startedAt: now,
   completedAt: now,
   updatedAt: now,
+  finalUsage: { inputTokens: 8, outputTokens: 2, totalTokens: 10 },
 };
 const runJson = workflowRunOutput(run, [call]);
 assert.deepEqual(runJson.result, { ok: true });
@@ -98,11 +99,13 @@ assert.deepEqual(runJson.calls, {
   total: 1,
 });
 assert.equal("scriptHash" in runJson, false);
+assert.deepEqual(runJson.usage, { inputTokens: 8, outputTokens: 2, totalTokens: 10 });
 
 const callJson = workflowCallOutput(call, { detailed: true });
 assert.deepEqual(callJson.structured, { bugs: [] });
 assert.equal("cacheKey" in callJson, false);
 assert.equal("providerSessionId" in callJson, false);
 assert.equal("profileFingerprint" in callJson, false);
+assert.deepEqual(callJson.usage, { inputTokens: 8, outputTokens: 2, totalTokens: 10 });
 
 console.log("cli-output.test.ts: ok");

src/cli-output.ts

@@ -56,6 +56,7 @@ export function workflowRunOutput(
     resumedFromRunId: run.resumedFromRunId,
     cancelRequested: run.cancelRequested,
     calls: calls ? workflowCallCounts(calls) : undefined,
+    usage: calls ? sumUsage(calls.map((call) => call.finalUsage ?? call.usage)) : undefined,
     result: parseStoredJson(run.resultJson),
     error: run.error
       ? { kind: run.errorKind, message: parseStoredJson(run.error) }
@@ -82,6 +83,7 @@ export function workflowCallOutput(
     effort: call.effort,
     cached: call.fromCache,
     durationMs: workflowCallDurationMs(call),
+    usage: call.finalUsage ?? call.usage,
     isolation: call.isolation,
     worktree: call.worktreePath
       ? { path: call.worktreePath, dirty: call.dirty }
@@ -130,6 +132,17 @@ function workflowCallDurationMs(call: WorkflowAgentCallRecord): number | undefin
   return Math.max(0, Date.parse(call.completedAt) - Date.parse(call.startedAt));
 }
 
+function sumUsage(calls: Array<WorkflowAgentCallRecord["usage"]>): WorkflowAgentCallRecord["usage"] | undefined {
+  const result: NonNullable<WorkflowAgentCallRecord["usage"]> = {};
+  for (const key of ["inputTokens", "outputTokens", "totalTokens", "cacheReadTokens", "cacheWriteTokens"] as const) {
+    const values = calls
+      .map((usage) => usage?.[key])
+      .filter((value): value is number => value !== undefined);
+    if (values.length > 0) result[key] = values.reduce((total, value) => total + value, 0);
+  }
+  return Object.keys(result).length > 0 ? result : undefined;
+}
+
 function parseStoredJson(value: string | undefined): unknown {
   if (value === undefined) return undefined;
   try {

src/local-agent-adapters.ts

@@ -15,11 +15,24 @@ import type { LocalAgentProvider } from "./local-agent-profiles.js";
 import { removeDevspaceNodeModulesBinFromPath } from "./local-agent-path.js";
 import {
   createCodexSdkLocalAgentRuntime,
+  createLocalAgentObservationEmitter,
   isNativeSchemaUnsupportedFailure,
   ProviderSchemaUnsupportedError,
   type LocalAgentRunInput,
   type LocalAgentRunResult,
 } from "./local-agent-runtime.js";
+import {
+  extractAcpObservations,
+  extractAcpUsage,
+  extractClaudeObservations,
+  extractClaudeUsage,
+  extractCodexObservations,
+  extractCodexUsage,
+  extractOpenCodeObservations,
+  extractOpenCodeUsage,
+  extractPiObservations,
+  extractPiUsage,
+} from "./local-agent-provider-observations.js";
 
 export interface LocalAgentAdapter {
   readonly provider: LocalAgentProvider;
@@ -103,9 +116,13 @@ class ClaudeLocalAgentAdapter implements LocalAgentAdapter {
       let providerSessionId = input.providerSessionId ?? null;
       let finalResponse = "";
       let structured: unknown | undefined;
+      let usage: LocalAgentRunResult["usage"];
+      const emitObservation = createLocalAgentObservationEmitter(input);
       const items: unknown[] = [];
       for await (const message of messages) {
         items.push(message);
+        for (const observation of extractClaudeObservations(message)) emitObservation(observation);
+        usage = extractClaudeUsage(message) ?? usage;
         const record = message as Record<string, unknown>;
         if (typeof record.session_id === "string") providerSessionId = record.session_id;
         if (record.type !== "result") continue;
@@ -124,6 +141,7 @@ class ClaudeLocalAgentAdapter implements LocalAgentAdapter {
         providerSessionId,
         finalResponse,
         items,
+        usage,
         ...(structured !== undefined ? { structured } : {}),
       };
     } catch (error) {
@@ -226,8 +244,13 @@ class OpencodeLocalAgentAdapter implements LocalAgentAdapter {
     try {
       const sessionId = input.providerSessionId ?? await createOpencodeSession(client, input);
       const promptResult = await promptOpencodeSession(client, sessionId, input);
+      const emitObservation = createLocalAgentObservationEmitter(input);
+      for (const observation of extractOpenCodeObservations(promptResult)) emitObservation(observation);
       await waitForOpencodeSession(client, sessionId);
       const messages = await readOpencodeMessages(client, sessionId);
+      for (const observation of extractOpenCodeObservations(messages)) emitObservation(observation);
+      const usage = extractOpenCodeUsage(messages) ?? extractOpenCodeUsage(promptResult);
+      if (usage) emitObservation({ kind: "usage", usage });
       const finalResponse = requireFinalResponse(
         "OpenCode",
         extractOpenCodeFinalResponse(messages) || extractOpenCodeFinalResponse(promptResult),
@@ -237,6 +260,7 @@ class OpencodeLocalAgentAdapter implements LocalAgentAdapter {
         providerSessionId: sessionId,
         finalResponse,
         items: [promptResult, messages],
+        usage,
       };
     } finally {
       server.close();
@@ -263,6 +287,8 @@ class AcpLocalAgentAdapter implements LocalAgentAdapter {
     });
     assertPipedChild(child);
     let stderr = "";
+    const emitObservation = createLocalAgentObservationEmitter(input);
+    let usage: LocalAgentRunResult["usage"];
     child.stderr.on("data", (chunk: Buffer) => {
       stderr += chunk.toString("utf8");
     });
@@ -302,6 +328,8 @@ class AcpLocalAgentAdapter implements LocalAgentAdapter {
               }
 
               const update = message.update;
+              for (const observation of extractAcpObservations(update)) emitObservation(observation);
+              usage = extractAcpUsage(update) ?? usage;
               if (update.sessionUpdate !== "agent_message_chunk") continue;
               const content = update.content;
               if (content.type === "text") textParts.push(content.text);
@@ -315,6 +343,7 @@ class AcpLocalAgentAdapter implements LocalAgentAdapter {
         providerSessionId,
         finalResponse: finalResponse.trim(),
         items: [],
+        usage,
       };
     } catch (error) {
       throw new Error(`${this.provider} ACP run failed: ${errorMessage(error)}${stderr ? `\n${stderr.trim()}` : ""}`);
@@ -429,14 +458,24 @@ class PiRpcLocalAgentAdapter implements LocalAgentAdapter {
     assertPipedChild(child);
     const rpc = new JsonLineRpc(child);
     const events: unknown[] = [];
-    rpc.onEvent((event) => events.push(event));
+    const emitObservation = createLocalAgentObservationEmitter(input);
+    let usage: LocalAgentRunResult["usage"];
+    rpc.onEvent((event) => {
+      events.push(event);
+      for (const observation of extractPiObservations(event)) emitObservation(observation);
+      usage = extractPiUsage(event) ?? usage;
+    });
     try {
       const state = await rpc.request({ type: "get_state" });
       const providerSessionId = readNestedString(state, ["sessionId"]) ?? input.providerSessionId ?? null;
       const done = rpc.waitForEvent((event) => asRecord(event)?.type === "agent_end", PI_AGENT_TIMEOUT_MS);
       await rpc.request({ type: "prompt", message: input.prompt });
       const agentEnd = await done;
       const sessionMessages = await rpc.request({ type: "get_messages" });
+      for (const observation of extractPiObservations(agentEnd)) emitObservation(observation);
+      for (const observation of extractPiObservations(sessionMessages)) emitObservation(observation);
+      usage = extractPiUsage(agentEnd) ?? extractPiUsage(sessionMessages) ?? usage;
+      if (usage) emitObservation({ kind: "usage", usage });
       const finalResponse =
         extractPiFinalResponse(agentEnd) ||
         extractPiFinalResponse(sessionMessages) ||
@@ -454,6 +493,7 @@ class PiRpcLocalAgentAdapter implements LocalAgentAdapter {
         providerSessionId,
         finalResponse,
         items: [...events, sessionMessages],
+        usage,
       };
     } finally {
       child.kill();

src/local-agent-provider-observations.test.ts

@@ -0,0 +1,80 @@
+import assert from "node:assert/strict";
+import {
+  extractAcpObservations,
+  extractAcpUsage,
+  extractClaudeObservations,
+  extractClaudeUsage,
+  extractCodexObservations,
+  extractCodexUsage,
+  extractOpenCodeObservations,
+  extractOpenCodeUsage,
+  extractPiObservations,
+  extractPiUsage,
+} from "./local-agent-provider-observations.js";
+
+{
+  const item = {
+    type: "command_execution",
+    id: "cmd-1",
+    command: "npm test",
+    status: "completed",
+    usage: { input_tokens: 12, output_tokens: 8 },
+  };
+  const codexActivity = extractCodexObservations([item]).find((entry) => entry.kind === "activity");
+  assert.equal(codexActivity?.activityId, "cmd-1");
+  assert.equal(codexActivity?.toolStatus, "completed");
+  assert.deepEqual(extractCodexUsage({ usage: { input_tokens: 12, output_tokens: 8 } }), {
+    inputTokens: 12,
+    outputTokens: 8,
+    totalTokens: 20,
+  });
+}
+
+{
+  const messages = [
+    { type: "assistant", content: [{ type: "tool_use", id: "tool-1", name: "bash", input: { command: "pwd" } }] },
+    { type: "user", content: [{ type: "tool_result", tool_use_id: "tool-1", content: "ok" }] },
+    { type: "result", usage: { input_tokens: 20, output_tokens: 4 } },
+  ];
+  assert.deepEqual(extractClaudeObservations(messages).filter((entry) => entry.kind === "activity").map((entry) => entry.toolStatus), ["started", "completed"]);
+  assert.deepEqual(extractClaudeUsage(messages), { inputTokens: 20, outputTokens: 4, totalTokens: 24 });
+}
+
+{
+  const messages = [{
+    info: { tokens: { input: 30, output: 5 } },
+    parts: [{ type: "tool", callID: "call-1", tool: "grep", state: { status: "completed", output: "match" } }],
+  }];
+  const openCodeActivity = extractOpenCodeObservations(messages).find((entry) => entry.kind === "activity");
+  assert.equal(openCodeActivity?.toolName, "grep");
+  assert.equal(openCodeActivity?.toolStatus, "completed");
+  assert.deepEqual(extractOpenCodeUsage(messages), { inputTokens: 30, outputTokens: 5, totalTokens: 35 });
+}
+
+{
+  const event = { type: "tool_result", toolCallId: "pi-1", toolName: "read", status: "success", usage: { input: 4, output: 3 } };
+  const piActivity = extractPiObservations([event]).find((entry) => entry.kind === "activity");
+  assert.equal(piActivity?.activityId, "pi-1");
+  assert.equal(piActivity?.toolStatus, "completed");
+  assert.deepEqual(extractPiUsage(event), { inputTokens: 4, outputTokens: 3, totalTokens: 7 });
+}
+
+for (const provider of ["cursor", "copilot"] as const) {
+  const update = {
+    sessionUpdate: "tool_call_update",
+    toolCallId: `${provider}-1`,
+    toolName: "edit",
+    status: "completed",
+    usage_update: { input_tokens: 9, output_tokens: 2 },
+  };
+  const acpActivity = extractAcpObservations(update).find((entry) => entry.kind === "activity");
+  assert.equal(acpActivity?.toolName, "edit");
+  assert.equal(acpActivity?.toolStatus, "completed");
+  assert.deepEqual(extractAcpUsage(update), {
+    inputTokens: 9,
+    outputTokens: 2,
+    totalTokens: 11,
+  });
+}
+
+console.log("local-agent-provider-observations.test.ts: ok");