diff --git a/skills/dynamic-workflows/SKILL.md b/skills/dynamic-workflows/SKILL.md index ccb34570..87322cb8 100644 --- a/skills/dynamic-workflows/SKILL.md +++ b/skills/dynamic-workflows/SKILL.md @@ -17,11 +17,14 @@ review, migrate-and-verify, research panels — **not** a single subagent turn. ```bash devspace workflow run --file path/to/script.js [--arg k=v]... [--follow] +devspace workflow run --script-path path/to/script.js [--resume ] [--follow] devspace workflow run --name review-auth [--follow] devspace workflow run --resume devspace workflow status [--follow] devspace workflow cancel devspace workflow ls +devspace workflow calls +devspace workflow call ``` Named scripts: `.devspace/workflows/.js` or `workflows/.js`. @@ -57,7 +60,6 @@ return { summary, findings } | `pipeline(items, ...stages)` | Per-item chains; no cross-item barrier | | `phase(title)` / `log(msg)` | Progress; journaled | | `args` | Run input (object preferred) | -| `budget` | Stub: `total: null`, `remaining(): Infinity` — do not loop on budget alone | | `workflow(name\|{scriptPath}, args?)` | Nested, depth 1, shared call index | **No `writeMode`.** Teach read-only vs write in the prompt. Use `isolation: 'worktree'` when parallel mutators would conflict (git required). @@ -86,7 +88,25 @@ Default: first **enabled ∩ available** provider (`agentProviders.enabled` in c ### Resume -`devspace workflow run --resume ` creates a **new** run that replays completed agent calls by cache key (callIndex+key, then consume-once by key). +Failed and cancelled runs are terminal. Recovery creates a **new** run: + +1. Inspect the prior run with `workflow status`, `workflow calls`, and + `workflow call`. +2. Edit the persisted `scriptPath` reported by the run, or pass a different + `--script-path`. +3. Keep prompts and agent options stable for completed calls whose return values + should be reused. +4. Run `devspace workflow run --resume ` (optionally with + `--script-path `). + +Replay first matches the same call index and cache key, then consumes one +compatible prior cache key after reordering. The new run records whether each +call was reused by same-index or compatible-key matching, and where it came +from. Failed, interrupted, changed, or unmatched calls execute live. + +Replay restores an agent's **return value**. It does not recreate shared-checkout +edits or reapply a prior worktree diff. Verify required filesystem state before +depending on a replayed mutating call. ### Cancel diff --git a/src/db/migrations.ts b/src/db/migrations.ts index b3e481ad..85f2fa68 100644 --- a/src/db/migrations.ts +++ b/src/db/migrations.ts @@ -32,6 +32,11 @@ const migrations: Migration[] = [ name: "workflow-journal", up: migrateWorkflowJournal, }, + { + version: 6, + name: "workflow-replay-provenance", + up: migrateWorkflowReplayProvenance, + }, ]; export function migrateDatabase(sqlite: Database.Database): void { @@ -282,9 +287,24 @@ function migrateWorkflowJournal(sqlite: Database.Database): void { `); } +function migrateWorkflowReplayProvenance(sqlite: Database.Database): void { + addColumnIfMissing(sqlite, "workflow_agent_calls", "prompt", "text not null default ''"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "schema_json", "text"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "error_kind", "text"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "replay_match", "text"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "replayed_from_run_id", "text"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "replayed_from_call_index", "integer"); + addColumnIfMissing(sqlite, "workflow_agent_calls", "replay_reason", "text"); + + sqlite.exec(` + create index if not exists workflow_agent_calls_replay_source_idx + on workflow_agent_calls(replayed_from_run_id, replayed_from_call_index); + `); +} + function addColumnIfMissing( sqlite: Database.Database, - table: "workspace_sessions" | "local_agent_sessions", + table: "workspace_sessions" | "local_agent_sessions" | "workflow_agent_calls", column: string, definition: string, ): void { diff --git a/src/db/schema.ts b/src/db/schema.ts index 36104a97..8f7e1518 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -157,6 +157,8 @@ export const workflowAgentCalls = sqliteTable( .references(() => workflowRuns.id, { onDelete: "cascade" }), callIndex: integer("call_index").notNull(), cacheKey: text("cache_key").notNull(), + prompt: text("prompt").notNull().default(""), + schemaJson: text("schema_json"), provider: text("provider").notNull(), model: text("model"), effort: text("effort"), @@ -168,6 +170,11 @@ export const workflowAgentCalls = sqliteTable( responseText: text("response_text"), structuredJson: text("structured_json"), error: text("error"), + errorKind: text("error_kind"), + replayMatch: text("replay_match"), + replayedFromRunId: text("replayed_from_run_id"), + replayedFromCallIndex: integer("replayed_from_call_index"), + replayReason: text("replay_reason"), isolation: text("isolation").notNull().default("shared"), worktreePath: text("worktree_path"), dirty: text("dirty"), @@ -179,6 +186,10 @@ export const workflowAgentCalls = sqliteTable( (table) => [ primaryKey({ columns: [table.runId, table.callIndex] }), index("workflow_agent_calls_cache_key_idx").on(table.runId, table.cacheKey), + index("workflow_agent_calls_replay_source_idx").on( + table.replayedFromRunId, + table.replayedFromCallIndex, + ), ], ); diff --git a/src/oauth-store.test.ts b/src/oauth-store.test.ts index a5cfaec4..bae52669 100644 --- a/src/oauth-store.test.ts +++ b/src/oauth-store.test.ts @@ -46,6 +46,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise { { version: 3, name: "local-agent-sessions" }, { version: 4, name: "local-agent-effort-rename" }, { version: 5, name: "workflow-journal" }, + { version: 6, name: "workflow-replay-provenance" }, ]); } finally { database.close(); diff --git a/src/workflow-api.ts b/src/workflow-api.ts index 78953f54..8ae13c92 100644 --- a/src/workflow-api.ts +++ b/src/workflow-api.ts @@ -11,6 +11,7 @@ import { buildAgentCacheKeyInput, createStubBudget, type AgentIsolationMode, + type AgentCacheKeyInput, type AgentOpts, type AppendWorkflowEventInput, type WorkflowMeta, @@ -64,10 +65,30 @@ export interface WorkflowReplayHit { responseText?: string; structuredJson?: string; providerSessionId?: string; + replayMatch: "same_index" | "compatible_key"; + replayedFromRunId: string; + replayedFromCallIndex: number; } +export interface WorkflowReplayMiss { + reason: + | "no_compatible_call" + | "prior_call_not_replayable" + | "compatible_result_consumed" + | "identity_changed"; + changedFields?: Array; +} + +export type WorkflowReplayDecision = + | { hit: WorkflowReplayHit; miss?: never } + | { hit?: never; miss: WorkflowReplayMiss }; + export interface WorkflowReplay { - match(callIndex: number, cacheKey: string): WorkflowReplayHit | null; + decide( + callIndex: number, + cacheKey: string, + input: AgentCacheKeyInput, + ): WorkflowReplayDecision; } export interface WorkflowJournal { @@ -78,6 +99,8 @@ export interface WorkflowJournal { runId: string; callIndex: number; cacheKey: string; + prompt: string; + schemaJson?: string; provider: LocalAgentProvider; model?: string; effort?: string; @@ -85,6 +108,10 @@ export interface WorkflowJournal { phase?: string; isolation?: AgentIsolationMode; worktreePath?: string; + replayMatch?: "same_index" | "compatible_key"; + replayedFromRunId?: string; + replayedFromCallIndex?: number; + replayReason?: string; }): unknown; completeAgentCall(input: { runId: string; @@ -100,6 +127,7 @@ export interface WorkflowJournal { runId: string; callIndex: number; error: string; + errorKind?: import("./workflow-types.js").WorkflowErrorKind; worktreePath?: string; dirty?: boolean; }): unknown; @@ -233,19 +261,24 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { }); const cacheKey = hashCacheKey(cacheKeyInput); - if (deps.replay) { - const hit = deps.replay.match(index, cacheKey); - if (hit) { + const replayDecision = deps.replay?.decide(index, cacheKey, cacheKeyInput); + if (replayDecision?.hit) { + const hit = replayDecision.hit; deps.journal.beginAgentCall({ runId: deps.runId, callIndex: index, cacheKey, + prompt, + schemaJson: agentOpts.schema ? JSON.stringify(agentOpts.schema) : undefined, provider, model: agentOpts.model, effort: agentOpts.effort, label: agentOpts.label, phase, isolation, + replayMatch: hit.replayMatch, + replayedFromRunId: hit.replayedFromRunId, + replayedFromCallIndex: hit.replayedFromCallIndex, }); deps.journal.completeAgentCall({ runId: deps.runId, @@ -260,10 +293,16 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { type: "agent_call_cached", phase, label: agentOpts.label, - data: { callIndex: index, cacheKey, provider }, + data: { + callIndex: index, + cacheKey, + provider, + replayMatch: hit.replayMatch, + replayedFromRunId: hit.replayedFromRunId, + replayedFromCallIndex: hit.replayedFromCallIndex, + }, }); return hit.value; - } } await semaphore.acquire(deps.signal); @@ -300,6 +339,8 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { runId: deps.runId, callIndex: index, cacheKey, + prompt, + schemaJson: agentOpts.schema ? JSON.stringify(agentOpts.schema) : undefined, provider, model: agentOpts.model, effort: agentOpts.effort, @@ -307,6 +348,9 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { phase, isolation, worktreePath, + replayReason: replayDecision?.miss + ? formatReplayMiss(replayDecision.miss) + : undefined, }); agentCallBegun = true; deps.journal.appendEvent({ @@ -453,6 +497,7 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { runId: deps.runId, callIndex: index, error: message, + errorKind: error instanceof WorkflowEngineError ? error.kind : "internal", worktreePath, }); } @@ -594,6 +639,12 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { }; } +function formatReplayMiss(miss: WorkflowReplayMiss): string { + return miss.reason === "identity_changed" && miss.changedFields?.length + ? `${miss.reason}:${miss.changedFields.join(",")}` + : miss.reason; +} + /** Test helper: read current ALS phase (undefined outside phase). */ export function getCurrentWorkflowPhase(): string | undefined { return phaseAls.getStore(); diff --git a/src/workflow-cli.ts b/src/workflow-cli.ts index 5ed5bc2c..10e3cc0d 100644 --- a/src/workflow-cli.ts +++ b/src/workflow-cli.ts @@ -29,6 +29,7 @@ import { WORKFLOW_LIMITS, resolveWorkflowConcurrency, type WorkflowEventRecord, + type WorkflowAgentCallRecord, type WorkflowRunRecord, type WorkflowRunSource, } from "./workflow-types.js"; @@ -62,6 +63,12 @@ export async function runWorkflowCommand( case "list": await runWorkflowList(config); return; + case "calls": + await runWorkflowCalls(rest, config); + return; + case "call": + await runWorkflowCall(rest, config); + return; case "__worker": await runWorkflowWorker(rest, config); return; @@ -82,11 +89,13 @@ export function printWorkflowHelp(): void { "DevSpace workflows", "", "Usage:", - " devspace workflow run (--file | --name | --resume )", + " devspace workflow run [--file|--script-path | --name ] [--resume ]", " [--arg key=value]... [--follow]", " devspace workflow status [--follow]", " devspace workflow cancel ", " devspace workflow ls", + " devspace workflow calls ", + " devspace workflow call ", ].join("\n"), ); } @@ -94,18 +103,24 @@ export function printWorkflowHelp(): void { async function runWorkflowRun(args: string[], config: ServerConfig): Promise { const { flags } = splitFlags(args); const follow = flags.has("follow"); - const file = flagValue(flags, "file"); + const file = flagValue(flags, "script-path") ?? flagValue(flags, "file"); const name = flagValue(flags, "name"); const resumeFrom = flagValue(flags, "resume"); const parsedArgs = parseWorkflowArgFlagsResult(collectArgTokens(args)); if (parsedArgs.isErr()) throw parsedArgs.error; const workflowArgs = parsedArgs.value.args; + if (file && name) { + throw new InvalidWorkflowInputError({ + code: "ambiguous_source", + message: "Provide only one of --file/--script-path or --name", + }); + } if (!file && !name && !resumeFrom) { throw new InvalidWorkflowInputError({ code: "missing_source", message: - "Usage: devspace workflow run (--file | --name | --resume )", + "Usage: devspace workflow run [--file|--script-path | --name ] [--resume ]", }); } @@ -125,13 +140,20 @@ async function runWorkflowRun(args: string[], config: ServerConfig): Promise { } } +async function runWorkflowCalls(args: string[], config: ServerConfig): Promise { + const runId = args[0]; + if (!runId) throw new Error("Usage: devspace workflow calls "); + const store = createWorkflowStore(config); + try { + if (!store.getRun(runId)) throw new WorkflowNotFoundError(runId); + const calls = store.listAgentCalls(runId); + if (calls.length === 0) { + console.log("No workflow agent calls."); + return; + } + for (const call of calls) console.log(formatCallLine(call)); + } finally { + store.close(); + } +} + +async function runWorkflowCall(args: string[], config: ServerConfig): Promise { + const runId = args[0]; + const callIndex = Number(args[1]); + if (!runId || !Number.isInteger(callIndex) || callIndex < 0) { + throw new Error("Usage: devspace workflow call "); + } + const store = createWorkflowStore(config); + try { + if (!store.getRun(runId)) throw new WorkflowNotFoundError(runId); + const call = store.getAgentCall(runId, callIndex); + if (!call) throw new Error(`Unknown workflow agent call: ${runId}#${callIndex}`); + console.log(JSON.stringify(formatCallDetail(call), null, 2)); + } finally { + store.close(); + } +} + /** Detached worker entry: claim run, heartbeat, execute, complete/fail. */ export async function runWorkflowWorker( args: string[], @@ -501,10 +555,62 @@ function printEvent(event: WorkflowEventRecord): void { } function formatRunLine( - run: Pick, + run: Pick< + WorkflowRunRecord, + "id" | "status" | "name" | "error" | "scriptPath" | "scriptHash" | "resumedFromRunId" + >, ): string { const err = run.error ? ` error=${JSON.stringify(run.error)}` : ""; - return `${run.id} ${run.status} ${run.name}${err}`; + const resumed = run.resumedFromRunId ? ` resumedFrom=${run.resumedFromRunId}` : ""; + return `${run.id} ${run.status} ${run.name} scriptPath=${JSON.stringify(run.scriptPath)} scriptHash=${run.scriptHash}${resumed}${err}`; +} + +function formatCallLine(call: WorkflowAgentCallRecord): string { + const label = call.label ? ` label=${JSON.stringify(call.label)}` : ""; + const phase = call.phase ? ` phase=${JSON.stringify(call.phase)}` : ""; + const model = call.model ? ` model=${call.model}` : ""; + const duration = callDurationMs(call); + const replay = call.fromCache + ? ` replay=${call.replayMatch ?? "cached"}:${call.replayedFromRunId ?? "?"}#${call.replayedFromCallIndex ?? "?"}` + : call.replayReason + ? ` replayMiss=${call.replayReason}` + : ""; + const worktree = call.worktreePath + ? ` worktree=${JSON.stringify(call.worktreePath)} dirty=${String(call.dirty)}` + : ""; + return `#${call.callIndex} ${call.status} ${call.provider}${model}${label}${phase} durationMs=${duration}${replay}${worktree}`; +} + +function formatCallSummary(calls: WorkflowAgentCallRecord[]): string { + const reused = calls.filter((call) => call.fromCache).length; + const failed = calls.filter((call) => call.status === "failed").length; + const live = calls.filter( + (call) => !call.fromCache && call.status === "completed", + ).length; + const running = calls.filter((call) => call.status === "running").length; + return `calls reused=${reused} live=${live} failed=${failed} running=${running} total=${calls.length}`; +} + +function formatCallDetail(call: WorkflowAgentCallRecord): Record { + return { + ...call, + durationMs: callDurationMs(call), + schema: call.schemaJson ? safeParseJson(call.schemaJson) : undefined, + structured: call.structuredJson ? safeParseJson(call.structuredJson) : undefined, + }; +} + +function callDurationMs(call: WorkflowAgentCallRecord): number | undefined { + if (!call.startedAt || !call.completedAt) return undefined; + return Math.max(0, Date.parse(call.completedAt) - Date.parse(call.startedAt)); +} + +function safeParseJson(text: string): unknown { + try { + return JSON.parse(text) as unknown; + } catch { + return text; + } } function resolveEnabledProviders( diff --git a/src/workflow-contracts.ts b/src/workflow-contracts.ts index a6947c63..1944b1c5 100644 --- a/src/workflow-contracts.ts +++ b/src/workflow-contracts.ts @@ -196,6 +196,9 @@ export const workflowEventPayloadSchemas = { callIndex: z.number().int().nonnegative(), cacheKey: z.string(), provider: localAgentProviderSchema, + replayMatch: z.enum(["same_index", "compatible_key"]), + replayedFromRunId: z.string(), + replayedFromCallIndex: z.number().int().nonnegative(), }) .strict(), schema_retry: z diff --git a/src/workflow-replay.test.ts b/src/workflow-replay.test.ts index 5258809e..7b0ac993 100644 --- a/src/workflow-replay.test.ts +++ b/src/workflow-replay.test.ts @@ -8,6 +8,7 @@ function call( ): WorkflowAgentCallRecord { return { runId: "wfr_prior", + prompt: "prompt", provider: "codex", status: "completed", fromCache: false, @@ -18,14 +19,25 @@ function call( }; } +function identity(prompt = "prompt") { + return { + prompt, + provider: "codex" as const, + model: null, + effort: null, + schema: null, + isolation: "shared" as const, + }; +} + { const replay = createWorkflowReplay([ call({ callIndex: 0, cacheKey: "k0", responseText: "a" }), call({ callIndex: 1, cacheKey: "k1", responseText: "b" }), ]); - assert.equal(replay.match(0, "k0")?.value, "a"); - assert.equal(replay.match(1, "k1")?.value, "b"); - assert.equal(replay.match(2, "k0"), null); + assert.equal(replay.decide(0, "k0", identity()).hit?.value, "a"); + assert.equal(replay.decide(1, "k1", identity()).hit?.value, "b"); + assert.equal(replay.decide(2, "k0", identity()).miss?.reason, "compatible_result_consumed"); } { @@ -35,9 +47,14 @@ function call( call({ callIndex: 1, cacheKey: "kb", responseText: "B" }), ]); // new run asks index0 for kb first - assert.equal(replay.match(0, "kb")?.value, "B"); - assert.equal(replay.match(1, "ka")?.value, "A"); - assert.equal(replay.match(2, "ka"), null); + const reorderedB = replay.decide(0, "kb", identity()).hit; + assert.equal(reorderedB?.value, "B"); + assert.equal(reorderedB?.replayMatch, "compatible_key"); + assert.equal(replay.decide(1, "ka", identity()).hit?.value, "A"); + assert.equal( + replay.decide(2, "ka", identity()).miss?.reason, + "compatible_result_consumed", + ); } { @@ -49,7 +66,16 @@ function call( structuredJson: '{"ok":true}', }), ]); - assert.deepEqual(replay.match(0, "ks")?.value, { ok: true }); + assert.deepEqual(replay.decide(0, "ks", identity()).hit?.value, { ok: true }); +} + +{ + const replay = createWorkflowReplay([ + call({ callIndex: 0, cacheKey: "old", prompt: "old prompt", responseText: "a" }), + ]); + const miss = replay.decide(0, "new", identity("new prompt")).miss; + assert.equal(miss?.reason, "identity_changed"); + assert.deepEqual(miss?.changedFields, ["prompt"]); } console.log("workflow-replay.test.ts: ok"); diff --git a/src/workflow-replay.ts b/src/workflow-replay.ts index 24f35418..25e2273a 100644 --- a/src/workflow-replay.ts +++ b/src/workflow-replay.ts @@ -1,5 +1,10 @@ import type { WorkflowAgentCallRecord } from "./workflow-types.js"; -import type { WorkflowReplay, WorkflowReplayHit } from "./workflow-api.js"; +import type { + WorkflowReplay, + WorkflowReplayDecision, + WorkflowReplayHit, +} from "./workflow-api.js"; +import type { AgentCacheKeyInput } from "./workflow-types.js"; import { parseJsonText } from "./json-types.js"; import { WorkflowStoredDataError } from "./workflow-errors.js"; @@ -26,20 +31,46 @@ export function createWorkflowReplay( const consumed = new Set(); // `${callIndex}` of prior rows consumed return { - match(callIndex: number, cacheKey: string): WorkflowReplayHit | null { + decide( + callIndex: number, + cacheKey: string, + input: AgentCacheKeyInput, + ): WorkflowReplayDecision { const exact = byIndex.get(callIndex); if (exact && exact.cacheKey === cacheKey && !consumed.has(indexKey(exact))) { consumed.add(indexKey(exact)); removeFromKeyQueue(byKeyQueue, exact); - return toHit(exact); + return { hit: toHit(exact, "same_index") }; } const queue = byKeyQueue.get(cacheKey); - if (!queue || queue.length === 0) return null; - const next = queue.shift()!; - consumed.add(indexKey(next)); - if (queue.length === 0) byKeyQueue.delete(cacheKey); - return toHit(next); + if (queue && queue.length > 0) { + const next = queue.shift()!; + consumed.add(indexKey(next)); + if (queue.length === 0) byKeyQueue.delete(cacheKey); + return { hit: toHit(next, "compatible_key") }; + } + + const priorAtIndex = priorCalls.find((call) => call.callIndex === callIndex); + if (priorAtIndex) { + if (priorAtIndex.status !== "completed" && priorAtIndex.status !== "from_cache") { + return { miss: { reason: "prior_call_not_replayable" } }; + } + if (priorAtIndex.cacheKey !== cacheKey) { + return { + miss: { + reason: "identity_changed", + changedFields: changedIdentityFields(priorAtIndex, input), + }, + }; + } + return { miss: { reason: "compatible_result_consumed" } }; + } + + if (priorCalls.some((call) => call.cacheKey === cacheKey)) { + return { miss: { reason: "compatible_result_consumed" } }; + } + return { miss: { reason: "no_compatible_call" } }; }, }; } @@ -61,7 +92,15 @@ function removeFromKeyQueue( if (queue.length === 0) map.delete(call.cacheKey); } -function toHit(call: WorkflowAgentCallRecord): WorkflowReplayHit { +function toHit( + call: WorkflowAgentCallRecord, + replayMatch: WorkflowReplayHit["replayMatch"], +): WorkflowReplayHit { + const provenance = { + replayMatch, + replayedFromRunId: call.runId, + replayedFromCallIndex: call.callIndex, + } as const; if (call.structuredJson) { try { return { @@ -69,6 +108,7 @@ function toHit(call: WorkflowAgentCallRecord): WorkflowReplayHit { responseText: call.responseText, structuredJson: call.structuredJson, providerSessionId: call.providerSessionId, + ...provenance, }; } catch (cause) { throw new WorkflowStoredDataError( @@ -82,5 +122,22 @@ function toHit(call: WorkflowAgentCallRecord): WorkflowReplayHit { responseText: call.responseText, structuredJson: call.structuredJson, providerSessionId: call.providerSessionId, + ...provenance, }; } + +function changedIdentityFields( + prior: WorkflowAgentCallRecord, + current: AgentCacheKeyInput, +): Array { + const changed: Array = []; + if (prior.prompt !== current.prompt) changed.push("prompt"); + if (prior.provider !== current.provider) changed.push("provider"); + if ((prior.model ?? null) !== current.model) changed.push("model"); + if ((prior.effort ?? null) !== current.effort) changed.push("effort"); + const priorSchema = prior.schemaJson ? JSON.stringify(parseJsonText(prior.schemaJson)) : null; + const currentSchema = current.schema === null ? null : JSON.stringify(current.schema); + if (priorSchema !== currentSchema) changed.push("schema"); + if (prior.isolation !== current.isolation) changed.push("isolation"); + return changed.length > 0 ? changed : ["prompt"]; +} diff --git a/src/workflow-store.test.ts b/src/workflow-store.test.ts index 314ab289..142e6a2c 100644 --- a/src/workflow-store.test.ts +++ b/src/workflow-store.test.ts @@ -67,12 +67,15 @@ try { runId: run.id, callIndex: 0, cacheKey: "key-a", + prompt: "review", + schemaJson: JSON.stringify({ type: "object" }), provider: "codex", model: "gpt-5.4", effort: "high", phase: "Review", isolation: "worktree", worktreePath: "/tmp/wt", + replayReason: "identity_changed:prompt", }); store.completeAgentCall({ runId: run.id, @@ -88,15 +91,24 @@ try { assert.equal(call?.dirty, true); assert.equal(call?.providerSessionId, "sess_1"); assert.equal(call?.effort, "high"); + assert.equal(call?.prompt, "review"); + assert.equal(call?.replayReason, "identity_changed:prompt"); store.beginAgentCall({ runId: run.id, callIndex: 1, cacheKey: "key-b", + prompt: "review two", provider: "claude", }); - store.failAgentCall({ runId: run.id, callIndex: 1, error: "boom" }); + store.failAgentCall({ + runId: run.id, + callIndex: 1, + error: "boom", + errorKind: "provider", + }); 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); const cancelled = store.requestCancel(run.id); diff --git a/src/workflow-store.ts b/src/workflow-store.ts index 085757d2..0cfdaae3 100644 --- a/src/workflow-store.ts +++ b/src/workflow-store.ts @@ -50,6 +50,8 @@ export interface BeginAgentCallInput { runId: string; callIndex: number; cacheKey: string; + prompt: string; + schemaJson?: string; provider: string; model?: string; effort?: string; @@ -57,6 +59,10 @@ export interface BeginAgentCallInput { phase?: string; isolation?: AgentIsolationMode; worktreePath?: string; + replayMatch?: "same_index" | "compatible_key"; + replayedFromRunId?: string; + replayedFromCallIndex?: number; + replayReason?: string; } export interface CompleteAgentCallInput { @@ -74,6 +80,7 @@ export interface FailAgentCallInput { runId: string; callIndex: number; error: string; + errorKind?: WorkflowErrorKind; worktreePath?: string; dirty?: boolean; } @@ -132,6 +139,8 @@ interface WorkflowAgentCallRow { run_id: string; call_index: number; cache_key: string; + prompt: string; + schema_json: string | null; provider: string; model: string | null; effort: string | null; @@ -143,6 +152,11 @@ interface WorkflowAgentCallRow { response_text: string | null; structured_json: string | null; error: string | null; + error_kind: string | null; + replay_match: string | null; + replayed_from_run_id: string | null; + replayed_from_call_index: number | null; + replay_reason: string | null; isolation: string; worktree_path: string | null; dirty: string | null; @@ -562,14 +576,18 @@ export class WorkflowStore { this.database.sqlite .prepare( `insert into workflow_agent_calls ( - run_id, call_index, cache_key, provider, model, effort, label, phase, - status, from_cache, isolation, worktree_path, created_at, started_at, updated_at - ) values (?, ?, ?, ?, ?, ?, ?, ?, 'running', 'false', ?, ?, ?, ?, ?)`, + run_id, call_index, cache_key, prompt, schema_json, provider, model, effort, label, phase, + status, from_cache, isolation, worktree_path, replay_match, + replayed_from_run_id, replayed_from_call_index, replay_reason, + created_at, started_at, updated_at + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'running', 'false', ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ) .run( input.runId, input.callIndex, input.cacheKey, + input.prompt, + input.schemaJson ?? null, input.provider, input.model ?? null, input.effort ?? null, @@ -577,6 +595,10 @@ export class WorkflowStore { input.phase ?? null, isolation, input.worktreePath ?? null, + input.replayMatch ?? null, + input.replayedFromRunId ?? null, + input.replayedFromCallIndex ?? null, + input.replayReason ?? null, now, now, now, @@ -630,6 +652,7 @@ export class WorkflowStore { `update workflow_agent_calls set status = 'failed', error = ?, + error_kind = ?, worktree_path = coalesce(?, worktree_path), dirty = ?, completed_at = ?, @@ -638,6 +661,7 @@ export class WorkflowStore { ) .run( input.error, + input.errorKind ?? "internal", input.worktreePath ?? null, input.dirty === undefined ? null : input.dirty ? "true" : "false", now, @@ -756,6 +780,8 @@ function rowToAgentCall(row: WorkflowAgentCallRow): WorkflowAgentCallRecord { runId: row.run_id, callIndex: row.call_index, cacheKey: row.cache_key, + prompt: row.prompt, + schemaJson: row.schema_json ?? undefined, provider: localAgentProviderSchema.parse(row.provider), model: row.model ?? undefined, effort: row.effort ?? undefined, @@ -767,6 +793,14 @@ function rowToAgentCall(row: WorkflowAgentCallRow): WorkflowAgentCallRecord { responseText: row.response_text ?? undefined, structuredJson: row.structured_json ?? undefined, error: row.error ?? undefined, + errorKind: (row.error_kind as WorkflowErrorKind | null) ?? undefined, + replayMatch: + row.replay_match === "same_index" || row.replay_match === "compatible_key" + ? row.replay_match + : undefined, + replayedFromRunId: row.replayed_from_run_id ?? undefined, + replayedFromCallIndex: row.replayed_from_call_index ?? undefined, + replayReason: row.replay_reason ?? undefined, isolation: row.isolation === "worktree" ? "worktree" : "shared", worktreePath: row.worktree_path ?? undefined, dirty: row.dirty === null ? undefined : row.dirty === "true", diff --git a/src/workflow-tools.ts b/src/workflow-tools.ts index 7de46bf6..5f8c2c6c 100644 --- a/src/workflow-tools.ts +++ b/src/workflow-tools.ts @@ -41,7 +41,7 @@ Workflow scripts (JS only): agent(prompt, { label?, phase?, schema?, model?, effort?, provider?, isolation?: 'worktree' }) parallel(thunks) → Array // barrier; throw → null pipeline(items, ...stages) // no cross-item barrier - phase(title); log(msg); args; budget (stub) + phase(title); log(msg); args workflow(name | { scriptPath }, args?) // nest depth 1 Bans: Date.now(), Math.random(), new Date() without args. No writeMode — teach RO vs write in prompts; isolation contains writes. @@ -69,6 +69,10 @@ export function registerWorkflowTools( .optional() .describe("Inline workflow script source (export const meta = …)."), name: z.string().optional().describe("Named workflow under .devspace/workflows/.js"), + scriptPath: z + .string() + .optional() + .describe("Existing workflow script path. May be combined with resumeFromRunId."), resumeFromRunId: z.string().optional().describe("Prior run id to resume (new run + cache)."), args: jsonValueSchema.optional().describe("JSON args passed to script as `args`."), yieldTimeMs: z @@ -82,15 +86,16 @@ export function registerWorkflowTools( annotations: { readOnlyHint: false }, _meta: {}, }, - async ({ workspaceId, script, name, resumeFromRunId, args, yieldTimeMs }) => { + async ({ workspaceId, script, name, scriptPath, resumeFromRunId, args, yieldTimeMs }) => { const workspace = workspaces.getWorkspace(workspaceId); const store = createWorkflowStore(config); try { - const provided = [script, name, resumeFromRunId].filter((v) => v !== undefined); - if (provided.length !== 1) { + const providedSources = [script, name, scriptPath].filter((v) => v !== undefined); + if (providedSources.length > 1 || (providedSources.length === 0 && !resumeFromRunId)) { throw new InvalidWorkflowInputError({ - code: provided.length === 0 ? "missing_source" : "ambiguous_source", - message: "Provide exactly one of script, name, or resumeFromRunId", + code: providedSources.length === 0 ? "missing_source" : "ambiguous_source", + message: + "Provide one of script, name, or scriptPath; resumeFromRunId may accompany that source or reuse the prior script", }); } @@ -105,13 +110,30 @@ export function registerWorkflowTools( const prior = store.getRun(resumeFromRunId); if (!prior) throw new WorkflowNotFoundError(resumeFromRunId); priorRunId = prior.id; - priorScriptPath = prior.scriptPath; - const resolvedResult = await readWorkflowScriptFileResult(prior.scriptPath); - if (resolvedResult.isErr()) throw resolvedResult.error; - const resolved = resolvedResult.value; - source = resolved.source; - scriptHash = prior.scriptHash; - nameHint = prior.name; + const overridePath = scriptPath; + if (script !== undefined) { + source = script; + const overrideParsed = parseWorkflowScript(source); + scriptHash = overrideParsed.scriptHash; + nameHint = overrideParsed.meta.name; + } else if (name) { + const resolvedResult = await resolveNamedWorkflowScriptResult({ + name, + workspaceRoot: workspace.root, + stateDir: config.stateDir, + }); + if (resolvedResult.isErr()) throw resolvedResult.error; + source = resolvedResult.value.source; + scriptHash = resolvedResult.value.scriptHash; + nameHint = resolvedResult.value.nameHint; + } else { + priorScriptPath = overridePath ?? prior.scriptPath; + const resolvedResult = await readWorkflowScriptFileResult(priorScriptPath); + if (resolvedResult.isErr()) throw resolvedResult.error; + source = resolvedResult.value.source; + scriptHash = resolvedResult.value.scriptHash; + nameHint = prior.name; + } runSource = "resume"; if (args === undefined && prior.argsJson && prior.argsJson !== "null") { try { @@ -132,6 +154,12 @@ export function registerWorkflowTools( scriptHash = resolved.scriptHash; nameHint = resolved.nameHint; runSource = "named"; + } else if (scriptPath) { + const resolvedResult = await readWorkflowScriptFileResult(scriptPath); + if (resolvedResult.isErr()) throw resolvedResult.error; + source = resolvedResult.value.source; + scriptHash = resolvedResult.value.scriptHash; + nameHint = resolvedResult.value.nameHint; } else { source = script!; const parsed = parseWorkflowScript(source); @@ -145,7 +173,7 @@ export function registerWorkflowTools( const run = store.createRun({ name: parsed.meta.name || nameHint, source: runSource, - scriptPath: priorScriptPath ?? "pending", + scriptPath: "pending", scriptHash, workspaceRoot: workspace.root, workspaceId, @@ -154,19 +182,16 @@ export function registerWorkflowTools( baseSha, }); - let persisted = priorScriptPath; - if (!persisted) { - const persistedResult = await persistWorkflowScriptResult({ - stateDir: config.stateDir, - runId: run.id, - source, - preferredName: parsed.meta.name || nameHint, - }); - if (persistedResult.isErr()) throw persistedResult.error; - persisted = persistedResult.value; - const updated = store.setScriptPathResult(run.id, persisted); - if (updated.isErr()) throw updated.error; - } + const persistedResult = await persistWorkflowScriptResult({ + stateDir: config.stateDir, + runId: run.id, + source, + preferredName: parsed.meta.name || nameHint, + }); + if (persistedResult.isErr()) throw persistedResult.error; + const persisted = persistedResult.value; + const updated = store.setScriptPathResult(run.id, persisted); + if (updated.isErr()) throw updated.error; const cliEntry = fileURLToPath( import.meta.url.replace(/workflow-tools\.(ts|js)$/, "cli.$1"), @@ -271,6 +296,7 @@ async function yieldEvents( events: WorkflowEventRecord[]; nextSeq: number; terminal: boolean; + callSummary: ReturnType; }> { const deadline = Date.now() + Math.min(yieldMs, WORKFLOW_MCP_YIELD_MS); let cursor = sinceSeq; @@ -288,7 +314,13 @@ async function yieldEvents( await sleep(250); } - return { run, events, nextSeq: cursor, terminal }; + return { + run, + events, + nextSeq: cursor, + terminal, + callSummary: summarizeCalls(store.listAgentCalls(runId)), + }; } function toolResult(page: { @@ -296,10 +328,17 @@ function toolResult(page: { events: WorkflowEventRecord[]; nextSeq: number; terminal: boolean; + callSummary: ReturnType; }) { const payload = { runId: page.run.id, status: page.run.status, + name: page.run.name, + source: page.run.source, + scriptPath: page.run.scriptPath, + scriptHash: page.run.scriptHash, + resumedFromRunId: page.run.resumedFromRunId, + callSummary: page.callSummary, events: page.events.map((e) => ({ seq: e.seq, type: e.type, @@ -318,6 +357,16 @@ function toolResult(page: { }; } +function summarizeCalls(calls: ReturnType["listAgentCalls"]>) { + return { + reused: calls.filter((call) => call.fromCache).length, + live: calls.filter((call) => !call.fromCache && call.status === "completed").length, + failed: calls.filter((call) => call.status === "failed").length, + running: calls.filter((call) => call.status === "running").length, + total: calls.length, + }; +} + function workflowToolError(error: Parameters[0]) { const payload = { error: serializeWorkflowError(error) }; return { diff --git a/src/workflow-types.ts b/src/workflow-types.ts index ad9b9d8a..ad050e1c 100644 --- a/src/workflow-types.ts +++ b/src/workflow-types.ts @@ -137,6 +137,8 @@ export interface WorkflowAgentCallRecord { runId: string; callIndex: number; cacheKey: string; + prompt: string; + schemaJson?: string; provider: AgentProviderId; model?: string; effort?: string; @@ -148,6 +150,11 @@ export interface WorkflowAgentCallRecord { responseText?: string; structuredJson?: string; error?: string; + errorKind?: WorkflowErrorKind; + replayMatch?: "same_index" | "compatible_key"; + replayedFromRunId?: string; + replayedFromCallIndex?: number; + replayReason?: string; isolation: AgentIsolationMode; worktreePath?: string; dirty?: boolean;