feat(workflow): add durable observability contracts

src/db/migrations.ts

@@ -47,6 +47,11 @@ const migrations: Migration[] = [
     name: "workflow-agent-profiles",
     up: migrateWorkflowAgentProfiles,
   },
+  {
+    version: 9,
+    name: "workflow-observability",
+    up: migrateWorkflowObservability,
+  },
 ];
 
 export function migrateDatabase(sqlite: Database.Database): void {
@@ -324,6 +329,36 @@ function migrateWorkflowAgentProfiles(sqlite: Database.Database): void {
   addColumnIfMissing(sqlite, "workflow_agent_calls", "profile_fingerprint", "text");
 }
 
+function migrateWorkflowObservability(sqlite: Database.Database): void {
+  addColumnIfMissing(sqlite, "workflow_agent_calls", "usage_json", "text");
+  addColumnIfMissing(sqlite, "workflow_agent_calls", "final_usage_json", "text");
+  sqlite.exec(`
+    create table if not exists workflow_agent_observations (
+      run_id text not null,
+      call_index integer not null,
+      seq integer not null,
+      provider text not null,
+      kind text not null,
+      activity_id text,
+      message text,
+      tool_name text,
+      tool_status text,
+      detail text,
+      usage_json text,
+      data_json text,
+      created_at text not null,
+      primary key (run_id, call_index, seq),
+      foreign key (run_id) references workflow_runs(id) on delete cascade
+    );
+
+    create index if not exists workflow_agent_observations_call_seq_idx
+      on workflow_agent_observations(run_id, call_index, seq);
+
+    create index if not exists workflow_agent_observations_created_idx
+      on workflow_agent_observations(run_id, created_at);
+  `);
+}
+
 function addColumnIfMissing(
   sqlite: Database.Database,
   table: "workspace_sessions" | "local_agent_sessions" | "workflow_agent_calls",

src/db/schema.ts

@@ -181,6 +181,8 @@ export const workflowAgentCalls = sqliteTable(
     isolation: text("isolation").notNull().default("shared"),
     worktreePath: text("worktree_path"),
     dirty: text("dirty"),
+    usageJson: text("usage_json"),
+    finalUsageJson: text("final_usage_json"),
     createdAt: text("created_at").notNull(),
     startedAt: text("started_at"),
     completedAt: text("completed_at"),
@@ -196,6 +198,32 @@ export const workflowAgentCalls = sqliteTable(
   ],
 );
 
+export const workflowAgentObservations = sqliteTable(
+  "workflow_agent_observations",
+  {
+    runId: text("run_id")
+      .notNull()
+      .references(() => workflowRuns.id, { onDelete: "cascade" }),
+    callIndex: integer("call_index").notNull(),
+    seq: integer("seq").notNull(),
+    provider: text("provider").notNull(),
+    kind: text("kind").notNull(),
+    activityId: text("activity_id"),
+    message: text("message"),
+    toolName: text("tool_name"),
+    toolStatus: text("tool_status"),
+    detail: text("detail"),
+    usageJson: text("usage_json"),
+    dataJson: text("data_json"),
+    createdAt: text("created_at").notNull(),
+  },
+  (table) => [
+    primaryKey({ columns: [table.runId, table.callIndex, table.seq] }),
+    index("workflow_agent_observations_call_seq_idx").on(table.runId, table.callIndex, table.seq),
+    index("workflow_agent_observations_created_idx").on(table.runId, table.createdAt),
+  ],
+);
+
 export type WorkspaceSessionRow = typeof workspaceSessions.$inferSelect;
 export type NewWorkspaceSessionRow = typeof workspaceSessions.$inferInsert;
 export type LoadedAgentFileRow = typeof loadedAgentFiles.$inferSelect;
@@ -205,3 +233,4 @@ export type NewLocalAgentSessionRow = typeof localAgentSessions.$inferInsert;
 export type WorkflowRunRow = typeof workflowRuns.$inferSelect;
 export type WorkflowEventRow = typeof workflowEvents.$inferSelect;
 export type WorkflowAgentCallRow = typeof workflowAgentCalls.$inferSelect;
+export type WorkflowAgentObservationRow = typeof workflowAgentObservations.$inferSelect;

src/local-agent-observations.ts

@@ -0,0 +1,80 @@
+/** Provider-neutral observations emitted while a local agent is running. */
+
+export const LOCAL_AGENT_TOOL_STATUSES = [
+  "started",
+  "updated",
+  "completed",
+  "failed",
+] as const;
+
+export type LocalAgentToolStatus = (typeof LOCAL_AGENT_TOOL_STATUSES)[number];
+
+/** Token fields are optional because providers expose different subsets. */
+export interface LocalAgentTokenUsage {
+  inputTokens?: number;
+  outputTokens?: number;
+  totalTokens?: number;
+  cacheReadTokens?: number;
+  cacheWriteTokens?: number;
+}
+
+export interface LocalAgentActivityObservation {
+  kind: "activity";
+  /** Provider operation id, when the provider exposes one. */
+  activityId?: string;
+  message?: string;
+  toolName?: string;
+  toolStatus?: LocalAgentToolStatus;
+  detail?: string;
+}
+
+export interface LocalAgentUsageObservation {
+  kind: "usage";
+  usage: LocalAgentTokenUsage;
+}
+
+export type LocalAgentObservation =
+  | LocalAgentActivityObservation
+  | LocalAgentUsageObservation;
+
+export function normalizeLocalAgentTokenUsage(value: unknown): LocalAgentTokenUsage | undefined {
+  if (!value || typeof value !== "object" || Array.isArray(value)) return undefined;
+  const record = value as Record<string, unknown>;
+  const usage: LocalAgentTokenUsage = {
+    inputTokens: readNonNegativeNumber(record, ["inputTokens", "input_tokens", "promptTokens", "prompt_tokens"]),
+    outputTokens: readNonNegativeNumber(record, ["outputTokens", "output_tokens", "completionTokens", "completion_tokens"]),
+    totalTokens: readNonNegativeNumber(record, ["totalTokens", "total_tokens"]),
+    cacheReadTokens: readNonNegativeNumber(record, ["cacheReadTokens", "cache_read_tokens", "cachedInputTokens", "cached_input_tokens"]),
+    cacheWriteTokens: readNonNegativeNumber(record, ["cacheWriteTokens", "cache_write_tokens"]),
+  };
+  if (usage.totalTokens === undefined && usage.inputTokens !== undefined && usage.outputTokens !== undefined) {
+    usage.totalTokens = usage.inputTokens + usage.outputTokens;
+  }
+  return Object.values(usage).some((part) => part !== undefined) ? usage : undefined;
+}
+
+export function mergeLocalAgentTokenUsage(
+  previous: LocalAgentTokenUsage | undefined,
+  next: LocalAgentTokenUsage | undefined,
+): LocalAgentTokenUsage | undefined {
+  if (!previous && !next) return undefined;
+  return {
+    inputTokens: next?.inputTokens ?? previous?.inputTokens,
+    outputTokens: next?.outputTokens ?? previous?.outputTokens,
+    totalTokens: next?.totalTokens ?? previous?.totalTokens,
+    cacheReadTokens: next?.cacheReadTokens ?? previous?.cacheReadTokens,
+    cacheWriteTokens: next?.cacheWriteTokens ?? previous?.cacheWriteTokens,
+  };
+}
+
+function readNonNegativeNumber(
+  record: Record<string, unknown>,
+  keys: string[],
+): number | undefined {
+  for (const key of keys) {
+    const value = record[key];
+    if (typeof value !== "number" || !Number.isFinite(value) || value < 0) continue;
+    return Math.floor(value);
+  }
+  return undefined;
+}

src/oauth-store.test.ts

@@ -49,6 +49,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise<void> {
       { version: 6, name: "workflow-replay-provenance" },
       { version: 7, name: "workflow-exact-replay" },
       { version: 8, name: "workflow-agent-profiles" },
+      { version: 9, name: "workflow-observability" },
     ]);
   } finally {
     database.close();

src/workflow-store.test.ts

@@ -90,6 +90,7 @@ try {
     returnValueJson: JSON.stringify({ ok: true, exact: true }),
     providerSessionId: "sess_1",
     dirty: true,
+    usage: { inputTokens: 12, outputTokens: 8, totalTokens: 20 },
   });
   const call = store.getAgentCall(run.id, 0);
   assert.equal(call?.status, "completed");
@@ -102,11 +103,46 @@ try {
   assert.equal(call?.prompt, "review");
   assert.equal(call?.returnValueJson, JSON.stringify({ ok: true, exact: true }));
   assert.equal(call?.replayReason, "identity_changed:prompt");
+  assert.deepEqual(call?.usage, { inputTokens: 12, outputTokens: 8, totalTokens: 20 });
+  assert.deepEqual(call?.finalUsage, { inputTokens: 12, outputTokens: 8, totalTokens: 20 });
   assert.deepEqual(
     store.listEvents(run.id).slice(-2).map((event) => event.type),
     ["agent_call_started", "agent_call_completed"],
   );
 
+  store.startAgentCall({
+    runId: run.id,
+    callIndex: 2,
+    cacheKey: "key-observe",
+    prompt: "observe",
+    provider: "codex",
+  });
+  store.appendAgentObservation({
+    runId: run.id,
+    callIndex: 2,
+    observation: {
+      kind: "activity",
+      activityId: "tool-1",
+      toolName: "grep",
+      toolStatus: "started",
+      message: "Searching source",
+    },
+  });
+  store.appendAgentObservation({
+    runId: run.id,
+    callIndex: 2,
+    observation: {
+      kind: "usage",
+      usage: { inputTokens: 30, outputTokens: 10, totalTokens: 40 },
+    },
+  });
+  assert.equal(store.listAgentObservations(run.id, 2)[0]?.activityId, "tool-1");
+  assert.deepEqual(store.getAgentCall(run.id, 2)?.usage, {
+    inputTokens: 30,
+    outputTokens: 10,
+    totalTokens: 40,
+  });
+
   store.startAgentCall({
     runId: run.id,
     callIndex: 1,
@@ -122,7 +158,7 @@ try {
   });
   assert.equal(store.getAgentCall(run.id, 1)?.status, "failed");
   assert.equal(store.getAgentCall(run.id, 1)?.errorKind, "provider");
-  assert.equal(store.listAgentCalls(run.id).length, 2);
+  assert.equal(store.listAgentCalls(run.id).length, 3);
   assert.deepEqual(
     store.listEvents(run.id).slice(-2).map((event) => event.type),
     ["agent_call_started", "agent_call_failed"],