diff --git a/src/db/migrations.ts b/src/db/migrations.ts index 43389ee5..f6d9d2df 100644 --- a/src/db/migrations.ts +++ b/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", diff --git a/src/db/schema.ts b/src/db/schema.ts index a087bae9..eaffc61a 100644 --- a/src/db/schema.ts +++ b/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; diff --git a/src/local-agent-observations.ts b/src/local-agent-observations.ts new file mode 100644 index 00000000..e0ce854a --- /dev/null +++ b/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; + 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, + 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; +} diff --git a/src/oauth-store.test.ts b/src/oauth-store.test.ts index 78a66214..63762ecf 100644 --- a/src/oauth-store.test.ts +++ b/src/oauth-store.test.ts @@ -49,6 +49,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise { { 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(); diff --git a/src/workflow-store.test.ts b/src/workflow-store.test.ts index 1b03d61e..137f8304 100644 --- a/src/workflow-store.test.ts +++ b/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"], diff --git a/src/workflow-store.ts b/src/workflow-store.ts index ea40d3dc..728bb6fc 100644 --- a/src/workflow-store.ts +++ b/src/workflow-store.ts @@ -5,10 +5,13 @@ import { openDatabase, type DatabaseHandle } from "./db/client.js"; import type { ServerConfig } from "./config.js"; import { WORKFLOW_LIMITS, + type LocalAgentObservation, + type LocalAgentTokenUsage, type AgentIsolationMode, type AppendWorkflowEventInput, type WorkflowAgentCallRecord, type WorkflowAgentCallStatus, + type WorkflowAgentObservationRecord, type WorkflowErrorKind, type WorkflowEventRecord, type WorkflowRunRecord, @@ -77,6 +80,7 @@ export interface CompleteAgentCallInput { dirty?: boolean; worktreePath?: string; fromCache?: boolean; + usage?: LocalAgentTokenUsage; } export interface CacheAgentCallInput extends BeginAgentCallInput { @@ -87,6 +91,15 @@ export interface CacheAgentCallInput extends BeginAgentCallInput { structuredJson?: string; returnValueJson?: string; providerSessionId?: string; + usage?: LocalAgentTokenUsage; +} + +export interface AppendAgentObservationInput { + runId: string; + callIndex: number; + observation: LocalAgentObservation; + /** Optional provider payload retained for diagnostics, subject to the cap. */ + dataJson?: string; } export interface FailAgentCallInput { @@ -184,12 +197,30 @@ interface WorkflowAgentCallRow { isolation: string; worktree_path: string | null; dirty: string | null; + usage_json: string | null; + final_usage_json: string | null; created_at: string; started_at: string | null; completed_at: string | null; updated_at: string; } +interface WorkflowAgentObservationRow { + run_id: string; + call_index: number; + seq: number; + provider: string; + kind: string; + activity_id: string | null; + message: string | null; + tool_name: string | null; + tool_status: string | null; + detail: string | null; + usage_json: string | null; + data_json: string | null; + created_at: string; +} + const TERMINAL_STATUSES = new Set(["completed", "failed", "cancelled"]); export class WorkflowStore { @@ -728,6 +759,7 @@ export class WorkflowStore { structuredJson: input.structuredJson, returnValueJson: input.returnValueJson, providerSessionId: input.providerSessionId, + usage: input.usage, fromCache: true, }, now, @@ -836,6 +868,8 @@ export class WorkflowStore { structured_json = ?, return_value_json = ?, provider_session_id = coalesce(?, provider_session_id), + usage_json = coalesce(?, usage_json), + final_usage_json = coalesce(?, final_usage_json), worktree_path = coalesce(?, worktree_path), dirty = ?, completed_at = ?, @@ -849,6 +883,8 @@ export class WorkflowStore { input.structuredJson ?? null, input.returnValueJson ?? null, input.providerSessionId ?? null, + input.usage ? JSON.stringify(input.usage) : null, + input.usage ? JSON.stringify(input.usage) : null, input.worktreePath ?? null, input.dirty === undefined ? null : input.dirty ? "true" : "false", now, @@ -859,6 +895,92 @@ export class WorkflowStore { return this.requireAgentCall(input.runId, input.callIndex); } + appendAgentObservation(input: AppendAgentObservationInput): WorkflowAgentObservationRecord { + const call = this.requireAgentCall(input.runId, input.callIndex); + const observation = input.observation; + const usageJson = observation.kind === "usage" + ? JSON.stringify(observation.usage) + : null; + const dataJson = input.dataJson + ? truncateText(input.dataJson, WORKFLOW_LIMITS.observationDataJsonBytes, "dataJson") + : null; + const createdAt = isoNow(); + const transaction = this.database.sqlite.transaction(() => { + const next = this.database.sqlite + .prepare( + `select coalesce(max(seq), 0) + 1 as next_seq + from workflow_agent_observations where run_id = ? and call_index = ?`, + ) + .get(input.runId, input.callIndex) as { next_seq: number }; + this.database.sqlite + .prepare( + `insert into workflow_agent_observations ( + run_id, call_index, seq, provider, kind, activity_id, message, + tool_name, tool_status, detail, usage_json, data_json, created_at + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + input.runId, + input.callIndex, + next.next_seq, + call.provider, + observation.kind, + observation.kind === "activity" ? observation.activityId ?? null : null, + observation.kind === "activity" ? observation.message ?? null : null, + observation.kind === "activity" ? observation.toolName ?? null : null, + observation.kind === "activity" ? observation.toolStatus ?? null : null, + observation.kind === "activity" ? observation.detail ?? null : null, + usageJson, + dataJson, + createdAt, + ); + if (usageJson) { + this.database.sqlite + .prepare( + `update workflow_agent_calls + set usage_json = ?, updated_at = ? + where run_id = ? and call_index = ?`, + ) + .run(usageJson, createdAt, input.runId, input.callIndex); + } + return next.next_seq; + }); + const seq = transaction.immediate(); + return { + runId: input.runId, + callIndex: input.callIndex, + seq, + provider: call.provider, + kind: observation.kind, + activityId: observation.kind === "activity" ? observation.activityId : undefined, + message: observation.kind === "activity" ? observation.message : undefined, + toolName: observation.kind === "activity" ? observation.toolName : undefined, + toolStatus: observation.kind === "activity" ? observation.toolStatus : undefined, + detail: observation.kind === "activity" ? observation.detail : undefined, + usage: observation.kind === "usage" ? observation.usage : undefined, + dataJson: dataJson ?? undefined, + createdAt, + }; + } + + listAgentObservations( + runId: string, + callIndex: number, + limit = 100, + ): WorkflowAgentObservationRecord[] { + const capped = Math.max(1, Math.min(limit, WORKFLOW_LIMITS.eventDrainMax)); + const rows = this.database.sqlite + .prepare( + `select * from ( + select * from workflow_agent_observations + where run_id = ? and call_index = ? + order by seq desc limit ? + ) order by seq asc`, + ) + .all(runId, callIndex, capped) as WorkflowAgentObservationRow[]; + return rows.map(rowToAgentObservation); + } + failAgentCall(input: FailAgentCallInput): WorkflowAgentCallRecord { const now = isoNow(); const transaction = this.database.sqlite.transaction(() => { @@ -1110,6 +1232,8 @@ function rowToAgentCall(row: WorkflowAgentCallRow): WorkflowAgentCallRecord { isolation: row.isolation === "worktree" ? "worktree" : "shared", worktreePath: row.worktree_path ?? undefined, dirty: row.dirty === null ? undefined : row.dirty === "true", + usage: parseUsageJson(row.usage_json), + finalUsage: parseUsageJson(row.final_usage_json), createdAt: row.created_at, startedAt: row.started_at ?? undefined, completedAt: row.completed_at ?? undefined, @@ -1117,6 +1241,41 @@ function rowToAgentCall(row: WorkflowAgentCallRow): WorkflowAgentCallRecord { }; } +function rowToAgentObservation(row: WorkflowAgentObservationRow): WorkflowAgentObservationRecord { + return { + runId: row.run_id, + callIndex: row.call_index, + seq: row.seq, + provider: localAgentProviderSchema.parse(row.provider), + kind: row.kind === "usage" ? "usage" : "activity", + activityId: row.activity_id ?? undefined, + message: row.message ?? undefined, + toolName: row.tool_name ?? undefined, + toolStatus: isToolStatus(row.tool_status) ? row.tool_status : undefined, + detail: row.detail ?? undefined, + usage: parseUsageJson(row.usage_json), + dataJson: row.data_json ?? undefined, + createdAt: row.created_at, + }; +} + +function parseUsageJson(value: string | null | undefined): LocalAgentTokenUsage | undefined { + if (!value) return undefined; + try { + const parsed = JSON.parse(value) as unknown; + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return undefined; + return parsed as LocalAgentTokenUsage; + } catch { + return undefined; + } +} + +function isToolStatus( + value: string | null, +): value is "started" | "updated" | "completed" | "failed" { + return value === "started" || value === "updated" || value === "completed" || value === "failed"; +} + function isoNow(): string { return new Date().toISOString(); } @@ -1146,6 +1305,13 @@ function assertTextSize(value: string, maxBytes: number, label: string): void { } } +function truncateText(value: string, maxBytes: number, label: string): string { + if (Buffer.byteLength(value, "utf8") <= maxBytes) return value; + const truncated = Buffer.from(value, "utf8").subarray(0, maxBytes).toString("utf8"); + if (!truncated) throw new Error(`${label} exceeds limit (${maxBytes} bytes)`); + return truncated; +} + function truncateJson(value: unknown, maxBytes: number): string { let text: string; try { diff --git a/src/workflow-types.ts b/src/workflow-types.ts index 75286338..65821f4c 100644 --- a/src/workflow-types.ts +++ b/src/workflow-types.ts @@ -10,6 +10,10 @@ */ import type { LocalAgentProvider } from "./local-agent-profiles.js"; +import type { + LocalAgentActivityObservation, + LocalAgentTokenUsage, +} from "./local-agent-observations.js"; import type { JsonSchema } from "./json-types.js"; import type { AgentIsolationMode, @@ -24,6 +28,13 @@ import type { } from "./workflow-contracts.js"; export type { JsonObject, JsonPrimitive, JsonSchema, JsonValue } from "./json-types.js"; +export type { + LocalAgentActivityObservation, + LocalAgentObservation, + LocalAgentToolStatus, + LocalAgentTokenUsage, + LocalAgentUsageObservation, +} from "./local-agent-observations.js"; export type { AgentIsolationMode, AgentOpts, @@ -59,6 +70,7 @@ export const WORKFLOW_MCP_YIELD_MS = 110_000; /** Soft/hard transport + storage caps (not semantic coverage truncation). */ export const WORKFLOW_LIMITS = { eventDataJsonBytes: 8 * 1024, + observationDataJsonBytes: 16 * 1024, responseTextBytes: 1 * 1024 * 1024, structuredJsonBytes: 256 * 1024, replayValueJsonBytes: 1 * 1024 * 1024, @@ -142,12 +154,30 @@ export interface WorkflowAgentCallRecord { isolation: AgentIsolationMode; worktreePath?: string; dirty?: boolean; + usage?: LocalAgentTokenUsage; + finalUsage?: LocalAgentTokenUsage; createdAt: string; startedAt?: string; completedAt?: string; updatedAt: string; } +export interface WorkflowAgentObservationRecord { + runId: string; + callIndex: number; + seq: number; + provider: AgentProviderId; + kind: "activity" | "usage"; + activityId?: string; + message?: string; + toolName?: string; + toolStatus?: "started" | "updated" | "completed" | "failed"; + detail?: string; + usage?: LocalAgentTokenUsage; + dataJson?: string; + createdAt: string; +} + // --------------------------------------------------------------------------- // Cache key // ---------------------------------------------------------------------------