diff --git a/package.json b/package.json index a6f01226..3cffe4d1 100644 --- a/package.json +++ b/package.json @@ -26,6 +26,7 @@ "build": "npm run clean && npm run build:app && tsc -p tsconfig.build.json", "build:app": "vite build", "dev": "node scripts/dev-server.mjs", + "dev:tui-fixture": "tsx scripts/workflow-tui-fixture.ts", "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", diff --git a/scripts/workflow-tui-fixture.ts b/scripts/workflow-tui-fixture.ts new file mode 100644 index 00000000..3e5d536e --- /dev/null +++ b/scripts/workflow-tui-fixture.ts @@ -0,0 +1,328 @@ +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { parseArgs } from "node:util"; +import { databasePath } from "../src/db/client.js"; +import { WorkflowStore } from "../src/workflow-store.js"; +import type { WorkflowRunRecord } from "../src/workflow-types.js"; + +const FIXTURE_VERSION = "large-v1"; +const WORKFLOW_NAME = "Ship multi-service authentication"; + +const fixtureNames = [ + "empty", + "starting", + "running", + "phased-running", + "replayed", + "call-failed", + "completed", + "failed", + "cancelled", +] as const; + +type FixtureName = (typeof fixtureNames)[number]; + +interface FixtureResult { + name: FixtureName; + stateDir: string; + run?: WorkflowRunRecord; +} + +const { values } = parseArgs({ + options: { + state: { type: "string", default: "all" }, + "state-dir": { type: "string" }, + "workspace-root": { type: "string" }, + }, + strict: true, +}); + +const requestedState = values.state; +const selectedFixtures = requestedState === "all" + ? [...fixtureNames] + : fixtureNames.includes(requestedState as FixtureName) + ? [requestedState as FixtureName] + : fail(`Unknown fixture state: ${requestedState}. Use all or one of: ${fixtureNames.join(", ")}`); +const fixtureRoot = resolve( + values["state-dir"] ?? join(tmpdir(), "devspace-workflow-tui-fixtures"), +); +const workspaceRoot = resolve(values["workspace-root"] ?? process.cwd()); + +const results = selectedFixtures.map((name) => seedFixture(name, fixtureRoot, workspaceRoot)); + +console.log(`Workflow TUI fixtures for ${workspaceRoot}`); +console.log(""); +for (const result of results) { + console.log(`${result.name}:`); + console.log(` database: ${databasePath(result.stateDir)}`); + if (result.run) console.log(` run: ${result.run.id}`); + const runArgument = result.run && !["starting", "running"].includes(result.run.status) + ? ` ${result.run.id}` + : ""; + console.log( + ` DEVSPACE_STATE_DIR=${JSON.stringify(result.stateDir)} DEVSPACE_WORKFLOWS=1 devspace workflow tui${runArgument}`, + ); + console.log(""); +} + +function seedFixture( + name: FixtureName, + root: string, + workspace: string, +): FixtureResult { + const stateDir = join(root, name); + const store = new WorkflowStore(stateDir); + try { + if (name === "empty") return { name, stateDir }; + + const scriptHash = `workflow-tui-fixture:${name}:${FIXTURE_VERSION}`; + const existing = store + .listRunsForWorkspace(workspace) + .find((run) => run.scriptHash === scriptHash); + if (existing) return { name, stateDir, run: existing }; + + const run = store.createRun({ + name: WORKFLOW_NAME, + source: name === "replayed" ? "resume" : "inline", + scriptPath: join(stateDir, "fixtures", `${name}.js`), + scriptHash, + workspaceRoot: workspace, + resumedFromRunId: name === "replayed" ? "wfr_previous_fixture" : undefined, + }); + + if (name === "starting") return { name, stateDir, run }; + + store.claimRun(run.id, process.pid); + store.appendEvent({ + runId: run.id, + type: "run_started", + data: { name: run.name, scriptHash, concurrency: 2 }, + }); + + if (name === "running") { + startPhase(store, run.id, "Discovery"); + addCompletedCall(store, run.id, 0, "Discovery", "Map authentication services", "codex"); + addCompletedCall(store, run.id, 1, "Discovery", "Audit token storage", "claude"); + startCall(store, run.id, 2, "Trace client login flows", "codex", "Discovery"); + startCall(store, run.id, 3, "Inventory migration risks", "claude", "Discovery"); + } else if (name === "phased-running") { + addCompletedPhase(store, run.id, "Discovery", 0, [ + ["Map authentication services", "codex"], + ["Audit token storage", "claude"], + ["Trace client login flows", "codex"], + ]); + addCompletedPhase(store, run.id, "Architecture", 3, [ + ["Design session boundaries", "claude"], + ["Plan database migration", "codex"], + ]); + startPhase(store, run.id, "Backend implementation"); + startCall(store, run.id, 5, "Implement OAuth store", "codex", "Backend implementation", true); + startCall(store, run.id, 6, "Add session rotation", "claude", "Backend implementation"); + startCall(store, run.id, 7, "Migrate authentication API", "codex", "Backend implementation", true); + store.appendEvent({ + runId: run.id, + type: "log", + phase: "Backend implementation", + data: { message: "Running service-level authentication tests" }, + }); + } else if (name === "replayed") { + startPhase(store, run.id, "Discovery"); + addCachedCall(store, run.id, 0, "Discovery", "Map authentication services", "codex"); + addCachedCall(store, run.id, 1, "Discovery", "Audit token storage", "claude"); + addCachedCall(store, run.id, 2, "Discovery", "Trace client login flows", "codex"); + startPhase(store, run.id, "Architecture"); + addCachedCall(store, run.id, 3, "Architecture", "Design session boundaries", "claude"); + addCachedCall(store, run.id, 4, "Architecture", "Plan database migration", "codex"); + startPhase(store, run.id, "Backend implementation"); + startCall(store, run.id, 5, "Implement OAuth store", "codex", "Backend implementation", true); + startCall(store, run.id, 6, "Add session rotation", "claude", "Backend implementation"); + } else if (name === "call-failed") { + addCompletedPhase(store, run.id, "Discovery", 0, [ + ["Map authentication services", "codex"], + ["Audit token storage", "claude"], + ["Trace client login flows", "codex"], + ]); + addCompletedPhase(store, run.id, "Architecture", 3, [ + ["Design session boundaries", "claude"], + ["Plan database migration", "codex"], + ]); + startPhase(store, run.id, "Backend implementation"); + startCall(store, run.id, 5, "Migrate authentication API", "codex", "Backend implementation", true); + store.failAgentCall({ + runId: run.id, + callIndex: 5, + error: "Provider process exited while updating the API", + errorKind: "provider", + }); + startCall(store, run.id, 6, "Implement OAuth store", "claude", "Backend implementation"); + startCall(store, run.id, 7, "Inspect client impact", "codex", "Backend implementation"); + } else if (name === "completed") { + seedCompletedWorkflow(store, run.id); + store.completeRun(run.id, { resultJson: JSON.stringify({ ok: true }), callCount: 12 }); + } else if (name === "failed") { + addCompletedPhase(store, run.id, "Discovery", 0, [ + ["Map authentication services", "codex"], + ["Audit token storage", "claude"], + ]); + addCompletedPhase(store, run.id, "Architecture", 2, [ + ["Design session boundaries", "claude"], + ["Plan database migration", "codex"], + ]); + addCompletedPhase(store, run.id, "Backend implementation", 4, [ + ["Implement OAuth store", "codex"], + ["Add session rotation", "claude"], + ["Migrate authentication API", "codex"], + ]); + addCompletedPhase(store, run.id, "Frontend integration", 7, [ + ["Update login experience", "claude"], + ["Handle session expiry", "codex"], + ]); + startPhase(store, run.id, "Verification"); + startCall(store, run.id, 9, "Run cross-service integration tests", "claude", "Verification"); + store.failAgentCall({ + runId: run.id, + callIndex: 9, + error: "Cross-service integration tests failed", + errorKind: "internal", + }); + store.failRun(run.id, { + error: "Workflow stopped because cross-service integration tests failed", + errorKind: "internal", + }); + } else if (name === "cancelled") { + addCompletedPhase(store, run.id, "Discovery", 0, [ + ["Map authentication services", "codex"], + ["Audit token storage", "claude"], + ["Trace client login flows", "codex"], + ]); + addCompletedPhase(store, run.id, "Architecture", 3, [ + ["Design session boundaries", "claude"], + ["Plan database migration", "codex"], + ]); + startPhase(store, run.id, "Backend implementation"); + store.cancelRun(run.id, "Cancelled by user"); + } + + return { name, stateDir, run: store.getRun(run.id) ?? run }; + } finally { + store.close(); + } +} + +function startCall( + store: WorkflowStore, + runId: string, + callIndex: number, + label: string, + provider: "codex" | "claude", + phase?: string, + worktree = false, +): void { + store.startAgentCall({ + runId, + callIndex, + cacheKey: `fixture-${callIndex}`, + prompt: label, + provider, + model: provider === "codex" ? "gpt-5.4" : "sonnet", + label, + phase, + isolation: worktree ? "worktree" : "shared", + worktreePath: worktree ? `/tmp/devspace-fixture-worktree-${callIndex}` : undefined, + }); +} + +function startPhase(store: WorkflowStore, runId: string, phase: string): void { + store.appendEvent({ + runId, + type: "phase_started", + phase, + data: { title: phase }, + }); +} + +function addCompletedCall( + store: WorkflowStore, + runId: string, + callIndex: number, + phase: string, + label: string, + provider: "codex" | "claude", + worktree = false, +): void { + startCall(store, runId, callIndex, label, provider, phase, worktree); + store.completeAgentCall({ runId, callIndex, responseText: `${label} completed` }); +} + +function addCompletedPhase( + store: WorkflowStore, + runId: string, + phase: string, + firstCallIndex: number, + calls: ReadonlyArray, +): void { + startPhase(store, runId, phase); + calls.forEach(([label, provider], offset) => { + addCompletedCall(store, runId, firstCallIndex + offset, phase, label, provider, offset % 3 === 2); + }); +} + +function addCachedCall( + store: WorkflowStore, + runId: string, + callIndex: number, + phase: string, + label: string, + provider: "codex" | "claude", +): void { + store.cacheAgentCall({ + runId, + callIndex, + cacheKey: `fixture-replayed-${callIndex}`, + prompt: label, + provider, + model: provider === "codex" ? "gpt-5.4" : "sonnet", + label, + phase, + replayMatch: "same_index", + replayedFromRunId: "wfr_previous_fixture", + replayedFromCallIndex: callIndex, + responseText: `${label} reused from the previous run`, + }); +} + +function seedCompletedWorkflow(store: WorkflowStore, runId: string): void { + addCompletedPhase(store, runId, "Discovery", 0, [ + ["Map authentication services", "codex"], + ["Audit token storage", "claude"], + ["Trace client login flows", "codex"], + ]); + addCompletedPhase(store, runId, "Architecture", 3, [ + ["Design session boundaries", "claude"], + ["Plan database migration", "codex"], + ]); + addCompletedPhase(store, runId, "Backend implementation", 5, [ + ["Implement OAuth store", "codex"], + ["Add session rotation", "claude"], + ["Migrate authentication API", "codex"], + ]); + addCompletedPhase(store, runId, "Frontend integration", 8, [ + ["Update login experience", "claude"], + ["Handle session expiry", "codex"], + ]); + addCompletedPhase(store, runId, "Verification", 10, [ + ["Run cross-service integration tests", "claude"], + ["Review security boundaries", "codex"], + ]); + startPhase(store, runId, "Release"); + store.appendEvent({ + runId, + type: "log", + phase: "Release", + data: { message: "Authentication rollout is ready" }, + }); +} + +function fail(message: string): never { + throw new Error(message); +} diff --git a/src/db/migrations.ts b/src/db/migrations.ts index 43389ee5..bd00e71c 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,9 +329,40 @@ function migrateWorkflowAgentProfiles(sqlite: Database.Database): void { addColumnIfMissing(sqlite, "workflow_agent_calls", "profile_fingerprint", "text"); } +function migrateWorkflowObservability(sqlite: Database.Database): void { + addColumnIfMissing(sqlite, "workflow_runs", "phases_json", "text not null default '[]'"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "usage_input_tokens", "integer"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "usage_cached_input_tokens", "integer"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "usage_cache_creation_input_tokens", "integer"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "usage_output_tokens", "integer"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "usage_total_tokens", "integer"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "usage_state", "text"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "usage_updated_at", "text"); + + sqlite.exec(` + create table if not exists workflow_agent_activity ( + run_id text not null, + call_index integer not null, + seq integer not null, + kind text not null, + status text not null, + label text not null, + detail text, + started_at text, + completed_at 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_activity_call_seq_idx + on workflow_agent_activity(run_id, call_index, seq); + `); +} + function addColumnIfMissing( sqlite: Database.Database, - table: "workspace_sessions" | "local_agent_sessions" | "workflow_agent_calls", + table: "workspace_sessions" | "local_agent_sessions" | "workflow_runs" | "workflow_agent_calls", column: string, definition: string, ): void { diff --git a/src/db/schema.ts b/src/db/schema.ts index a087bae9..62e755d1 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -108,6 +108,7 @@ export const workflowRuns = sqliteTable( workspaceRoot: text("workspace_root").notNull(), workspaceId: text("workspace_id"), argsJson: text("args_json").notNull().default("null"), + phasesJson: text("phases_json").notNull().default("[]"), status: text("status").notNull(), error: text("error"), errorKind: text("error_kind"), @@ -169,6 +170,13 @@ export const workflowAgentCalls = sqliteTable( status: text("status").notNull(), fromCache: text("from_cache").notNull().default("false"), providerSessionId: text("provider_session_id"), + usageInputTokens: integer("usage_input_tokens"), + usageCachedInputTokens: integer("usage_cached_input_tokens"), + usageCacheCreationInputTokens: integer("usage_cache_creation_input_tokens"), + usageOutputTokens: integer("usage_output_tokens"), + usageTotalTokens: integer("usage_total_tokens"), + usageState: text("usage_state"), + usageUpdatedAt: text("usage_updated_at"), responseText: text("response_text"), structuredJson: text("structured_json"), returnValueJson: text("return_value_json"), @@ -196,6 +204,26 @@ export const workflowAgentCalls = sqliteTable( ], ); +export const workflowAgentActivity = sqliteTable( + "workflow_agent_activity", + { + runId: text("run_id").notNull().references(() => workflowRuns.id, { onDelete: "cascade" }), + callIndex: integer("call_index").notNull(), + seq: integer("seq").notNull(), + kind: text("kind").notNull(), + status: text("status").notNull(), + label: text("label").notNull(), + detail: text("detail"), + startedAt: text("started_at"), + completedAt: text("completed_at"), + createdAt: text("created_at").notNull(), + }, + (table) => [ + primaryKey({ columns: [table.runId, table.callIndex, table.seq] }), + index("workflow_agent_activity_call_seq_idx").on(table.runId, table.callIndex, table.seq), + ], +); + 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 WorkflowAgentActivityRow = typeof workflowAgentActivity.$inferSelect; 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-contracts.ts b/src/workflow-contracts.ts index 8c15c75a..197185c2 100644 --- a/src/workflow-contracts.ts +++ b/src/workflow-contracts.ts @@ -6,20 +6,18 @@ import { jsonSchemaSchema, type JsonSchema, type JsonValue } from "./json-types. export const localAgentProviderSchema = z.enum(LOCAL_AGENT_PROVIDERS); +export const workflowPhaseMetaSchema = z + .object({ + title: z.string().trim().min(1), + detail: z.string().trim().min(1).optional(), + }) + .strict(); + export const workflowMetaSchema = z .object({ name: z.string().trim().min(1).regex(/^[a-z0-9-]+$/), description: z.string().trim().min(1), - phases: z - .array( - z - .object({ - title: z.string().trim().min(1), - detail: z.string().trim().min(1).optional(), - }) - .strict(), - ) - .optional(), + phases: z.array(workflowPhaseMetaSchema).optional(), whenToUse: z.string().trim().min(1).optional(), defaultProvider: localAgentProviderSchema.optional(), concurrency: z.number().finite().int().positive().optional(), diff --git a/src/workflow-launch.test.ts b/src/workflow-launch.test.ts index 3f40f919..1fe9b700 100644 --- a/src/workflow-launch.test.ts +++ b/src/workflow-launch.test.ts @@ -14,7 +14,7 @@ import { launchWorkflowRun } from "./workflow-launch.js"; workspaceRoot: dir, source: { kind: "inline", - script: `export const meta = { name: 'launch-demo', description: 'd' }\nreturn 1\n`, + script: `export const meta = { name: 'launch-demo', description: 'd', phases: [{ title: 'Plan' }, { title: 'Build', detail: 'Implement it' }] }\nreturn 1\n`, }, args: { n: 1 }, cliEntry: "/tmp/devspace-cli-not-used", @@ -26,6 +26,10 @@ import { launchWorkflowRun } from "./workflow-launch.js"; assert.equal(launched.value.run.status, "starting"); assert.match(launched.value.run.scriptPath.replaceAll("\\", "/"), /workflow-scripts\//); assert.equal(launched.value.run.argsJson, JSON.stringify({ n: 1 })); + assert.deepEqual(launched.value.run.phases, [ + { title: "Plan" }, + { title: "Build", detail: "Implement it" }, + ]); await mkdir(join(dir, ".devspace", "workflows"), { recursive: true }); await writeFile( diff --git a/src/workflow-launch.ts b/src/workflow-launch.ts index ef324fce..f32c8f79 100644 --- a/src/workflow-launch.ts +++ b/src/workflow-launch.ts @@ -94,6 +94,7 @@ export async function launchWorkflowRun( workspaceRoot: input.workspaceRoot, workspaceId: input.workspaceId, argsJson: JSON.stringify(args === undefined ? null : args), + phases: parsed.meta.phases, resumedFromRunId: priorRunId, baseSha, }); diff --git a/src/workflow-store.test.ts b/src/workflow-store.test.ts index 1b03d61e..935662c6 100644 --- a/src/workflow-store.test.ts +++ b/src/workflow-store.test.ts @@ -21,12 +21,20 @@ try { workspaceRoot: join(root, "project"), workspaceId: "ws_1", argsJson: JSON.stringify({ files: ["a.ts"] }), + phases: [ + { title: "Planning", detail: "Understand the change" }, + { title: "Review" }, + ], }); assert.match(run.id, /^wfr_[a-f0-9]{12}$/); assert.equal(run.status, "starting"); assert.equal(run.cancelRequested, false); assert.equal(store.getRun(run.id)?.name, "fanout"); + assert.deepEqual(store.getRun(run.id)?.phases, [ + { title: "Planning", detail: "Understand the change" }, + { title: "Review" }, + ]); const claimed = store.claimRun(run.id, process.pid); assert.equal(claimed?.status, "running"); @@ -82,6 +90,34 @@ try { worktreePath: "/tmp/wt", replayReason: "identity_changed:prompt", }); + store.attachAgentSession(run.id, 0, "sess_live"); + const partialUsage = store.updateAgentUsage(run.id, 0, { + inputTokens: 1_000, + cachedInputTokens: 700, + outputTokens: 200, + totalTokens: 1_200, + state: "partial", + }); + assert.equal(partialUsage.state, "partial"); + store.appendAgentActivity({ + runId: run.id, + callIndex: 0, + kind: "command", + status: "running", + label: "npm test", + }); + store.appendAgentActivity({ + runId: run.id, + callIndex: 0, + kind: "command", + status: "completed", + label: "npm test", + detail: "passed", + }); + assert.deepEqual( + store.listAgentActivity(run.id, 0).map((activity) => activity.status), + ["running", "completed"], + ); store.completeAgentCall({ runId: run.id, callIndex: 0, @@ -91,11 +127,20 @@ try { providerSessionId: "sess_1", dirty: true, }); + store.updateAgentUsage(run.id, 0, { + inputTokens: 1_100, + cachedInputTokens: 700, + outputTokens: 250, + totalTokens: 1_350, + state: "final", + }); const call = store.getAgentCall(run.id, 0); assert.equal(call?.status, "completed"); assert.equal(call?.isolation, "worktree"); assert.equal(call?.dirty, true); assert.equal(call?.providerSessionId, "sess_1"); + assert.equal(call?.usage?.totalTokens, 1_350); + assert.equal(call?.usage?.state, "final"); assert.equal(call?.effort, "high"); assert.equal(call?.profileName, "reviewer"); assert.equal(call?.profileFingerprint, "profile-hash"); diff --git a/src/workflow-store.ts b/src/workflow-store.ts index ea40d3dc..254ce3d6 100644 --- a/src/workflow-store.ts +++ b/src/workflow-store.ts @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; import { resolve } from "node:path"; import { Result, type Result as BetterResult } from "better-result"; +import * as z from "zod/v4"; import { openDatabase, type DatabaseHandle } from "./db/client.js"; import type { ServerConfig } from "./config.js"; import { @@ -9,11 +10,16 @@ import { type AppendWorkflowEventInput, type WorkflowAgentCallRecord, type WorkflowAgentCallStatus, + type WorkflowAgentActivityKind, + type WorkflowAgentActivityRecord, + type WorkflowAgentActivityStatus, type WorkflowErrorKind, type WorkflowEventRecord, type WorkflowRunRecord, type WorkflowRunSource, type WorkflowRunStatus, + type WorkflowPhaseMeta, + type WorkflowTokenUsage, } from "./workflow-types.js"; import { localAgentProviderSchema, @@ -22,6 +28,7 @@ import { workflowEventTypeSchema, workflowRunSourceSchema, workflowRunStatusSchema, + workflowPhaseMetaSchema, } from "./workflow-contracts.js"; import { InvalidRunTransitionError, @@ -42,10 +49,22 @@ export interface CreateWorkflowRunInput { workspaceRoot: string; workspaceId?: string; argsJson?: string; + phases?: WorkflowPhaseMeta[]; resumedFromRunId?: string; baseSha?: string; } +export interface AppendWorkflowAgentActivityInput { + runId: string; + callIndex: number; + kind: WorkflowAgentActivityKind; + status: WorkflowAgentActivityStatus; + label: string; + detail?: string; + startedAt?: string; + completedAt?: string; +} + export interface BeginAgentCallInput { runId: string; callIndex: number; @@ -131,6 +150,7 @@ interface WorkflowRunRow { workspace_root: string; workspace_id: string | null; args_json: string; + phases_json: string; status: string; error: string | null; error_kind: string | null; @@ -172,6 +192,13 @@ interface WorkflowAgentCallRow { status: string; from_cache: string; provider_session_id: string | null; + usage_input_tokens: number | null; + usage_cached_input_tokens: number | null; + usage_cache_creation_input_tokens: number | null; + usage_output_tokens: number | null; + usage_total_tokens: number | null; + usage_state: string | null; + usage_updated_at: string | null; response_text: string | null; structured_json: string | null; return_value_json: string | null; @@ -190,6 +217,19 @@ interface WorkflowAgentCallRow { updated_at: string; } +interface WorkflowAgentActivityRow { + run_id: string; + call_index: number; + seq: number; + kind: string; + status: string; + label: string; + detail: string | null; + started_at: string | null; + completed_at: string | null; + created_at: string; +} + const TERMINAL_STATUSES = new Set(["completed", "failed", "cancelled"]); export class WorkflowStore { @@ -202,6 +242,7 @@ export class WorkflowStore { createRun(input: CreateWorkflowRunInput): WorkflowRunRecord { const now = isoNow(); const argsJson = input.argsJson ?? "null"; + const phasesJson = JSON.stringify(input.phases ?? []); assertArgsSize(argsJson); const record: WorkflowRunRecord = { @@ -213,6 +254,7 @@ export class WorkflowStore { workspaceRoot: resolve(input.workspaceRoot), workspaceId: input.workspaceId, argsJson, + phases: input.phases ?? [], status: "starting", cancelRequested: false, resumedFromRunId: input.resumedFromRunId, @@ -225,9 +267,9 @@ export class WorkflowStore { .prepare( `insert into workflow_runs ( id, name, source, script_path, script_hash, workspace_root, workspace_id, - args_json, status, cancel_requested, resumed_from_run_id, base_sha, + args_json, phases_json, status, cancel_requested, resumed_from_run_id, base_sha, created_at, updated_at - ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ) .run( record.id, @@ -238,6 +280,7 @@ export class WorkflowStore { record.workspaceRoot, record.workspaceId ?? null, record.argsJson, + phasesJson, record.status, "false", record.resumedFromRunId ?? null, @@ -949,6 +992,128 @@ export class WorkflowStore { return rows.map(rowToAgentCall); } + attachAgentSession(runId: string, callIndex: number, providerSessionId: string): void { + const sessionId = providerSessionId.trim(); + if (!sessionId) throw new Error("providerSessionId cannot be empty"); + const now = isoNow(); + const update = this.database.sqlite + .prepare( + `update workflow_agent_calls + set provider_session_id = ?, updated_at = ? + where run_id = ? and call_index = ?`, + ) + .run(sessionId, now, runId, callIndex); + if (update.changes === 0) this.requireAgentCall(runId, callIndex); + } + + updateAgentUsage( + runId: string, + callIndex: number, + usage: Omit, + ): WorkflowTokenUsage { + for (const value of [ + usage.inputTokens, + usage.cachedInputTokens, + usage.cacheCreationInputTokens, + usage.outputTokens, + usage.totalTokens, + ]) { + if (value !== undefined && (!Number.isSafeInteger(value) || value < 0)) { + throw new Error("Workflow token usage must contain non-negative integers"); + } + } + const now = isoNow(); + const update = this.database.sqlite + .prepare( + `update workflow_agent_calls set + usage_input_tokens = ?, + usage_cached_input_tokens = ?, + usage_cache_creation_input_tokens = ?, + usage_output_tokens = ?, + usage_total_tokens = ?, + usage_state = ?, + usage_updated_at = ?, + updated_at = ? + where run_id = ? and call_index = ?`, + ) + .run( + usage.inputTokens ?? null, + usage.cachedInputTokens ?? null, + usage.cacheCreationInputTokens ?? null, + usage.outputTokens ?? null, + usage.totalTokens, + usage.state, + now, + now, + runId, + callIndex, + ); + if (update.changes === 0) this.requireAgentCall(runId, callIndex); + return { ...usage, updatedAt: now }; + } + + appendAgentActivity(input: AppendWorkflowAgentActivityInput): WorkflowAgentActivityRecord { + const now = isoNow(); + const transaction = this.database.sqlite.transaction(() => { + this.requireAgentCall(input.runId, input.callIndex); + const next = this.database.sqlite + .prepare( + `select coalesce(max(seq), 0) + 1 as next_seq + from workflow_agent_activity where run_id = ? and call_index = ?`, + ) + .get(input.runId, input.callIndex) as { next_seq: number }; + this.database.sqlite + .prepare( + `insert into workflow_agent_activity ( + run_id, call_index, seq, kind, status, label, detail, + started_at, completed_at, created_at + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + input.runId, + input.callIndex, + next.next_seq, + input.kind, + input.status, + input.label, + input.detail ?? null, + input.startedAt ?? null, + input.completedAt ?? null, + now, + ); + this.database.sqlite + .prepare( + `delete from workflow_agent_activity + where run_id = ? and call_index = ? and seq <= ?`, + ) + .run(input.runId, input.callIndex, next.next_seq - WORKFLOW_LIMITS.activityPerCall); + return { + ...input, + seq: next.next_seq, + createdAt: now, + }; + }); + return transaction.immediate(); + } + + listAgentActivity( + runId: string, + callIndex: number, + limit = WORKFLOW_LIMITS.activityPerCall, + ): WorkflowAgentActivityRecord[] { + const capped = Math.max(1, Math.min(limit, WORKFLOW_LIMITS.activityPerCall)); + const rows = this.database.sqlite + .prepare( + `select * from ( + select * from workflow_agent_activity + where run_id = ? and call_index = ? + order by seq desc limit ? + ) order by seq asc`, + ) + .all(runId, callIndex, capped) as WorkflowAgentActivityRow[]; + return rows.map(rowToAgentActivity); + } + /** * Mark abandoned starting runs and running runs with a dead worker as failed. * staleBeforeMs: start/update or heartbeat older than this and no live pid. @@ -1050,6 +1215,7 @@ function rowToRun(row: WorkflowRunRow): WorkflowRunRecord { workspaceRoot: row.workspace_root, workspaceId: row.workspace_id ?? undefined, argsJson: row.args_json, + phases: z.array(workflowPhaseMetaSchema).parse(JSON.parse(row.phases_json)), status: workflowRunStatusSchema.parse(row.status), error: row.error ?? undefined, errorKind: (row.error_kind as WorkflowErrorKind | null) ?? undefined, @@ -1095,6 +1261,17 @@ function rowToAgentCall(row: WorkflowAgentCallRow): WorkflowAgentCallRecord { status: workflowAgentCallStatusSchema.parse(row.status), fromCache: row.from_cache === "true", providerSessionId: row.provider_session_id ?? undefined, + usage: row.usage_total_tokens === null || row.usage_updated_at === null + ? undefined + : { + inputTokens: row.usage_input_tokens ?? undefined, + cachedInputTokens: row.usage_cached_input_tokens ?? undefined, + cacheCreationInputTokens: row.usage_cache_creation_input_tokens ?? undefined, + outputTokens: row.usage_output_tokens ?? undefined, + totalTokens: row.usage_total_tokens, + state: row.usage_state === "final" ? "final" : "partial", + updatedAt: row.usage_updated_at, + }, responseText: row.response_text ?? undefined, structuredJson: row.structured_json ?? undefined, returnValueJson: row.return_value_json ?? undefined, @@ -1117,6 +1294,29 @@ function rowToAgentCall(row: WorkflowAgentCallRow): WorkflowAgentCallRecord { }; } +function rowToAgentActivity(row: WorkflowAgentActivityRow): WorkflowAgentActivityRecord { + const kinds: WorkflowAgentActivityKind[] = ["tool", "command", "file", "status"]; + const statuses: WorkflowAgentActivityStatus[] = ["running", "completed", "failed"]; + if (!kinds.includes(row.kind as WorkflowAgentActivityKind)) { + throw new Error(`Unknown workflow agent activity kind: ${row.kind}`); + } + if (!statuses.includes(row.status as WorkflowAgentActivityStatus)) { + throw new Error(`Unknown workflow agent activity status: ${row.status}`); + } + return { + runId: row.run_id, + callIndex: row.call_index, + seq: row.seq, + kind: row.kind as WorkflowAgentActivityKind, + status: row.status as WorkflowAgentActivityStatus, + label: row.label, + detail: row.detail ?? undefined, + startedAt: row.started_at ?? undefined, + completedAt: row.completed_at ?? undefined, + createdAt: row.created_at, + }; +} + function isoNow(): string { return new Date().toISOString(); } diff --git a/src/workflow-types.ts b/src/workflow-types.ts index 75286338..8535092b 100644 --- a/src/workflow-types.ts +++ b/src/workflow-types.ts @@ -67,6 +67,7 @@ export const WORKFLOW_LIMITS = { scriptSourceBytes: 512 * 1024, eventDrainDefault: 200, eventDrainMax: 500, + activityPerCall: 500, } as const; export type AgentProviderId = LocalAgentProvider; @@ -88,6 +89,7 @@ export interface WorkflowRunRecord { workspaceRoot: string; workspaceId?: string; argsJson: string; + phases?: WorkflowPhaseMeta[]; status: WorkflowRunStatus; error?: string; errorKind?: WorkflowErrorKind; @@ -104,6 +106,32 @@ export interface WorkflowRunRecord { updatedAt: string; } +export interface WorkflowTokenUsage { + inputTokens?: number; + cachedInputTokens?: number; + cacheCreationInputTokens?: number; + outputTokens?: number; + totalTokens: number; + state: "partial" | "final"; + updatedAt: string; +} + +export type WorkflowAgentActivityKind = "tool" | "command" | "file" | "status"; +export type WorkflowAgentActivityStatus = "running" | "completed" | "failed"; + +export interface WorkflowAgentActivityRecord { + runId: string; + callIndex: number; + seq: number; + kind: WorkflowAgentActivityKind; + status: WorkflowAgentActivityStatus; + label: string; + detail?: string; + startedAt?: string; + completedAt?: string; + createdAt: string; +} + export interface WorkflowEventRecord { runId: string; seq: number; @@ -130,6 +158,7 @@ export interface WorkflowAgentCallRecord { status: WorkflowAgentCallStatus; fromCache: boolean; providerSessionId?: string; + usage?: WorkflowTokenUsage; responseText?: string; structuredJson?: string; returnValueJson?: string;