feat(workflow): persist replay provenance

Author: Waishnav

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 {

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,
+    ),
   ],
 );
 

src/oauth-store.test.ts

@@ -46,6 +46,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise<void> {
       { 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();

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<keyof AgentCacheKeyInput>;
+}
+
+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,13 +99,19 @@ export interface WorkflowJournal {
     runId: string;
     callIndex: number;
     cacheKey: string;
+    prompt: string;
+    schemaJson?: string;
     provider: LocalAgentProvider;
     model?: string;
     effort?: string;
     label?: string;
     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,13 +339,18 @@ 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,
         label: agentOpts.label,
         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();

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

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");