Skip to content

Commit d12fc43

Browse files
WaishnavWaishnav
committed
fix(workflow): replay only deterministic call prefixes
1 parent 57229cb commit d12fc43

4 files changed

Lines changed: 170 additions & 120 deletions

File tree

‎src/workflow-api.ts‎

Lines changed: 34 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ export interface WorkflowReplayHit {
6464
value: JsonValue;
6565
responseText?: string;
6666
structuredJson?: string;
67+
returnValueJson: string;
6768
providerSessionId?: string;
6869
replayMatch: "same_index" | "compatible_key";
6970
replayedFromRunId: string;
@@ -75,7 +76,11 @@ export interface WorkflowReplayMiss {
7576
| "no_compatible_call"
7677
| "prior_call_not_replayable"
7778
| "compatible_result_consumed"
78-
| "identity_changed";
79+
| "identity_changed"
80+
| "prefix_diverged"
81+
| "worktree_not_restored"
82+
| "result_not_persisted"
83+
| "stored_result_invalid";
7984
changedFields?: Array<keyof AgentCacheKeyInput>;
8085
}
8186

@@ -118,6 +123,7 @@ export interface WorkflowJournal {
118123
callIndex: number;
119124
responseText?: string;
120125
structuredJson?: string;
126+
returnValueJson?: string;
121127
providerSessionId?: string;
122128
dirty?: boolean;
123129
worktreePath?: string;
@@ -285,6 +291,7 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi {
285291
callIndex: index,
286292
responseText: hit.responseText,
287293
structuredJson: hit.structuredJson,
294+
returnValueJson: hit.returnValueJson,
288295
providerSessionId: hit.providerSessionId,
289296
fromCache: true,
290297
});
@@ -444,9 +451,8 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi {
444451
runId: deps.runId,
445452
callIndex: index,
446453
responseText: truncate(result.finalResponse, WORKFLOW_LIMITS.responseTextBytes),
447-
structuredJson: structuredJson
448-
? truncate(structuredJson, WORKFLOW_LIMITS.structuredJsonBytes)
449-
: undefined,
454+
structuredJson: boundedStructuredJson(structuredJson),
455+
returnValueJson: serializeReplayValue(returnValue),
450456
providerSessionId: result.providerSessionId,
451457
dirty,
452458
worktreePath,
@@ -721,10 +727,30 @@ function cancelledError(): WorkflowEngineError {
721727

722728
function truncate(text: string, maxBytes: number): string {
723729
if (Buffer.byteLength(text, "utf8") <= maxBytes) return text;
724-
// rough char truncate for journal safety
725-
let end = Math.min(text.length, maxBytes);
726-
while (end > 0 && Buffer.byteLength(text.slice(0, end), "utf8") > maxBytes) end -= 1;
727-
return `${text.slice(0, end)}…`;
730+
const marker = "…";
731+
const budget = Math.max(0, maxBytes - Buffer.byteLength(marker, "utf8"));
732+
let end = Math.min(text.length, budget);
733+
while (end > 0 && Buffer.byteLength(text.slice(0, end), "utf8") > budget) end -= 1;
734+
return `${text.slice(0, end)}${marker}`;
735+
}
736+
737+
function boundedStructuredJson(value: string | undefined): string | undefined {
738+
if (value === undefined) return undefined;
739+
return Buffer.byteLength(value, "utf8") <= WORKFLOW_LIMITS.structuredJsonBytes
740+
? value
741+
: undefined;
742+
}
743+
744+
function serializeReplayValue(value: unknown): string | undefined {
745+
try {
746+
const json = JSON.stringify(value);
747+
if (json === undefined) return undefined;
748+
return Buffer.byteLength(json, "utf8") <= WORKFLOW_LIMITS.replayValueJsonBytes
749+
? json
750+
: undefined;
751+
} catch {
752+
return undefined;
753+
}
728754
}
729755

730756
/** Minimal JSON extract for schema path until Ajv module lands. */

‎src/workflow-engine.test.ts‎

Lines changed: 36 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ import {
1212
type WorkflowProviderRunInput,
1313
type CreateAgentWorktree,
1414
} from "./workflow-api.js";
15-
import { createStubBudget } from "./workflow-types.js";
15+
import { createStubBudget, WORKFLOW_LIMITS } from "./workflow-types.js";
1616

1717
// ---------------------------------------------------------------------------
1818
// Semaphore
@@ -73,6 +73,8 @@ import { createStubBudget } from "./workflow-types.js";
7373
]);
7474
assert.deepEqual(results, ["ok:a", null, "ok:b"]);
7575
assert.equal(api.getCallCount(), 3);
76+
assert.equal(store.getAgentCall(run.id, 0)?.returnValueJson, JSON.stringify("ok:a"));
77+
assert.equal(store.getAgentCall(run.id, 2)?.returnValueJson, JSON.stringify("ok:b"));
7678
store.close();
7779
await rm(dir, { recursive: true, force: true });
7880
}
@@ -373,11 +375,44 @@ import { createStubBudget } from "./workflow-types.js";
373375
assert.equal(calls[0]?.providerSessionId, undefined);
374376
assert.equal(calls[1]?.schema, undefined);
375377
assert.equal(calls[1]?.providerSessionId, "sess-1");
378+
assert.equal(store.getAgentCall(run.id, 0)?.returnValueJson, JSON.stringify({ n: 2 }));
376379

377380
store.close();
378381
await rm(dir, { recursive: true, force: true });
379382
}
380383

384+
// ---------------------------------------------------------------------------
385+
// oversized exact replay values do not fail the live call
386+
// ---------------------------------------------------------------------------
387+
{
388+
const dir = await mkdtemp(join(tmpdir(), "wf-replay-size-"));
389+
const store = new WorkflowStore(dir);
390+
const run = store.createRun({
391+
name: "replay-size",
392+
source: "inline",
393+
scriptPath: "inline",
394+
scriptHash: "h",
395+
workspaceRoot: dir,
396+
});
397+
const response = "x".repeat(WORKFLOW_LIMITS.replayValueJsonBytes + 1);
398+
const api = createWorkflowApi({
399+
runId: run.id,
400+
journal: store,
401+
meta: { name: "replay-size", description: "d" },
402+
args: undefined,
403+
concurrency: 1,
404+
signal: new AbortController().signal,
405+
workspaceRoot: dir,
406+
enabledProviders: ["codex"],
407+
runProvider: async () => ({ finalResponse: response }),
408+
});
409+
410+
assert.equal(await api.agent("large"), response);
411+
assert.equal(store.getAgentCall(run.id, 0)?.returnValueJson, undefined);
412+
store.close();
413+
await rm(dir, { recursive: true, force: true });
414+
}
415+
381416
// ---------------------------------------------------------------------------
382417
// executeWorkflow end-to-end with sandbox + nest depth
383418
// ---------------------------------------------------------------------------

‎src/workflow-replay.test.ts‎

Lines changed: 45 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import type { WorkflowAgentCallRecord } from "./workflow-types.js";
44

55
function call(
66
partial: Partial<WorkflowAgentCallRecord> &
7-
Pick<WorkflowAgentCallRecord, "callIndex" | "cacheKey" | "responseText">,
7+
Pick<WorkflowAgentCallRecord, "callIndex" | "cacheKey">,
88
): WorkflowAgentCallRecord {
99
return {
1010
runId: "wfr_prior",
@@ -15,6 +15,7 @@ function call(
1515
isolation: "shared",
1616
createdAt: "t",
1717
updatedAt: "t",
18+
returnValueJson: JSON.stringify(`result-${partial.callIndex}`),
1819
...partial,
1920
};
2021
}
@@ -32,50 +33,75 @@ function identity(prompt = "prompt") {
3233

3334
{
3435
const replay = createWorkflowReplay([
35-
call({ callIndex: 0, cacheKey: "k0", responseText: "a" }),
36-
call({ callIndex: 1, cacheKey: "k1", responseText: "b" }),
36+
call({ callIndex: 0, cacheKey: "k0", returnValueJson: JSON.stringify("a") }),
37+
call({ callIndex: 1, cacheKey: "k1", returnValueJson: JSON.stringify("b") }),
3738
]);
3839
assert.equal(replay.decide(0, "k0", identity()).hit?.value, "a");
3940
assert.equal(replay.decide(1, "k1", identity()).hit?.value, "b");
40-
assert.equal(replay.decide(2, "k0", identity()).miss?.reason, "compatible_result_consumed");
41+
assert.equal(replay.decide(2, "k2", identity()).miss?.reason, "no_compatible_call");
4142
}
4243

4344
{
44-
// fan-out reorder: callIndex mismatch, consume-once by key
4545
const replay = createWorkflowReplay([
46-
call({ callIndex: 0, cacheKey: "ka", responseText: "A" }),
47-
call({ callIndex: 1, cacheKey: "kb", responseText: "B" }),
46+
call({ callIndex: 0, cacheKey: "ka", returnValueJson: JSON.stringify("A") }),
47+
call({ callIndex: 1, cacheKey: "kb", returnValueJson: JSON.stringify("B") }),
4848
]);
49-
// new run asks index0 for kb first
50-
const reorderedB = replay.decide(0, "kb", identity()).hit;
51-
assert.equal(reorderedB?.value, "B");
52-
assert.equal(reorderedB?.replayMatch, "compatible_key");
53-
assert.equal(replay.decide(1, "ka", identity()).hit?.value, "A");
54-
assert.equal(
55-
replay.decide(2, "ka", identity()).miss?.reason,
56-
"compatible_result_consumed",
57-
);
49+
const changed = replay.decide(0, "kb", identity()).miss;
50+
assert.equal(changed?.reason, "identity_changed");
51+
assert.equal(replay.decide(1, "kb", identity()).miss?.reason, "prefix_diverged");
5852
}
5953

6054
{
6155
const replay = createWorkflowReplay([
6256
call({
6357
callIndex: 0,
6458
cacheKey: "ks",
65-
responseText: '{"ok":true}',
59+
responseText: "bounded preview",
6660
structuredJson: '{"ok":true}',
61+
returnValueJson: '{"ok":true,"text":"exact"}',
6762
}),
6863
]);
69-
assert.deepEqual(replay.decide(0, "ks", identity()).hit?.value, { ok: true });
64+
assert.deepEqual(replay.decide(0, "ks", identity()).hit?.value, {
65+
ok: true,
66+
text: "exact",
67+
});
7068
}
7169

7270
{
7371
const replay = createWorkflowReplay([
74-
call({ callIndex: 0, cacheKey: "old", prompt: "old prompt", responseText: "a" }),
72+
call({ callIndex: 0, cacheKey: "old", prompt: "old prompt" }),
73+
call({ callIndex: 1, cacheKey: "later" }),
7574
]);
7675
const miss = replay.decide(0, "new", identity("new prompt")).miss;
7776
assert.equal(miss?.reason, "identity_changed");
7877
assert.deepEqual(miss?.changedFields, ["prompt"]);
78+
assert.equal(replay.decide(1, "later", identity()).miss?.reason, "prefix_diverged");
79+
}
80+
81+
{
82+
const replay = createWorkflowReplay([
83+
call({ callIndex: 0, cacheKey: "worktree", isolation: "worktree" }),
84+
call({ callIndex: 1, cacheKey: "later" }),
85+
]);
86+
assert.equal(
87+
replay.decide(0, "worktree", { ...identity(), isolation: "worktree" }).miss?.reason,
88+
"worktree_not_restored",
89+
);
90+
assert.equal(replay.decide(1, "later", identity()).miss?.reason, "prefix_diverged");
91+
}
92+
93+
{
94+
const replay = createWorkflowReplay([
95+
call({ callIndex: 0, cacheKey: "legacy", returnValueJson: undefined }),
96+
]);
97+
assert.equal(replay.decide(0, "legacy", identity()).miss?.reason, "result_not_persisted");
98+
}
99+
100+
{
101+
const replay = createWorkflowReplay([
102+
call({ callIndex: 0, cacheKey: "corrupt", returnValueJson: "{" }),
103+
]);
104+
assert.equal(replay.decide(0, "corrupt", identity()).miss?.reason, "stored_result_invalid");
79105
}
80106

81107
console.log("workflow-replay.test.ts: ok");

0 commit comments

Comments
 (0)