From cd60a151be58bf5628be0ebc77f82619c183cd81 Mon Sep 17 00:00:00 2001 From: testikun <320479488+testikun@users.noreply.github.com> Date: Fri, 4 Sep 2026 18:17:13 +0800 Subject: [PATCH 1/4] fix(workflows): recover terminal artifact commits --- extensions/workflows/artifacts.ts | 304 +++++++++++++++++-- extensions/workflows/dashboard.ts | 2 + tests/extensions/workflows/artifacts.test.ts | 158 ++++++++++ tests/extensions/workflows/dashboard.test.ts | 65 ++++ 4 files changed, 512 insertions(+), 17 deletions(-) diff --git a/extensions/workflows/artifacts.ts b/extensions/workflows/artifacts.ts index 005e43db..af4aae86 100644 --- a/extensions/workflows/artifacts.ts +++ b/extensions/workflows/artifacts.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import * as fs from "node:fs"; import * as path from "node:path"; import { @@ -22,10 +23,14 @@ import { } from "./serialization.ts"; export const JOURNAL_FILE = "journal.json"; +export const WORKFLOW_COMMIT_FILE = ".workflow-commit.json"; const ARTIFACT_TRANSCRIPT_MAX_BYTES = 32 * 1024; const ARTIFACT_TRANSCRIPT_ENTRY_MAX_BYTES = 8 * 1024; const AGENT_RESULT_ARTIFACT_MAX_BYTES = 2 * 1024 * 1024; +const WORKFLOW_MANIFEST_MAX_BYTES = 1024 * 1024; +const WORKFLOW_TRANSCRIPTS_MAX_BYTES = 2 * 1024 * 1024; +const WORKFLOW_COMMIT_MAX_BYTES = 3 * 1024 * 1024; export const WORKFLOW_CHECKPOINT_INTERVAL_MS = 500; const ENTRY_TRUNCATION_MARKER = "\n[entry truncated]"; const TRANSCRIPT_TRUNCATION_MARKER = @@ -35,10 +40,254 @@ type WorkflowJournalSource = | readonly JournalEntry[] | WorkflowJournalAccumulator; +interface WorkflowArtifactWrite { + name: typeof JOURNAL_FILE | "result.json" | "transcripts.json"; + content: string; +} + +interface WorkflowCommitArtifact { + name: WorkflowArtifactWrite["name"]; + bytes: number; + sha256: string; +} + +interface WorkflowCommitMarker { + version: 1; + runId: string; + manifest: string; + artifacts: WorkflowCommitArtifact[]; +} + +export type WorkflowCommitRecovery = + | "none" + | "recovered" + | "already-committed" + | "incomplete" + | "invalid" + | "failed"; + +const artifactLimits = new Map([ + ["transcripts.json", WORKFLOW_TRANSCRIPTS_MAX_BYTES], + ["result.json", WORKFLOW_MANIFEST_MAX_BYTES], + [JOURNAL_FILE, JOURNAL_MAX_BYTES], +]); + function textBytes(text: string) { return Buffer.byteLength(text, "utf8"); } +function sha256(content: string | Buffer) { + return createHash("sha256").update(content).digest("hex"); +} + +function removeWorkflowCommit(runDir: string, strict = false) { + try { + fs.unlinkSync(path.join(runDir, WORKFLOW_COMMIT_FILE)); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return; + if (strict) throw error; + } +} + +function workflowCommitMarker( + details: WorkflowDetails, + manifest: string, + artifacts: WorkflowArtifactWrite[], +): WorkflowCommitMarker { + for (const { name, content } of artifacts) { + const bytes = textBytes(content); + const limit = artifactLimits.get(name); + if (limit === undefined || bytes > limit) { + throw new Error(`Workflow artifact ${name} exceeded its commit budget`); + } + } + return { + version: 1, + runId: details.runId, + manifest, + artifacts: artifacts.map(({ name, content }) => ({ + name, + bytes: textBytes(content), + sha256: sha256(content), + })), + }; +} + +function serializeWorkflowCommitMarker( + details: WorkflowDetails, + manifest: string, + artifacts: WorkflowArtifactWrite[], +) { + const content = JSON.stringify( + workflowCommitMarker(details, manifest, artifacts), + ); + if (textBytes(content) > WORKFLOW_COMMIT_MAX_BYTES) { + throw new Error( + "Workflow artifact commit receipt exceeded its byte budget", + ); + } + return content; +} + +function parseWorkflowCommitMarker( + runDir: string, +): WorkflowCommitMarker | "none" | "invalid" { + const markerPath = path.join(runDir, WORKFLOW_COMMIT_FILE); + let stat: fs.Stats; + try { + stat = fs.lstatSync(markerPath); + } catch (error) { + return (error as NodeJS.ErrnoException).code === "ENOENT" + ? "none" + : "invalid"; + } + if ( + !stat.isFile() || + stat.isSymbolicLink() || + stat.size <= 0 || + stat.size > WORKFLOW_COMMIT_MAX_BYTES + ) { + return "invalid"; + } + + let raw: unknown; + try { + raw = JSON.parse(fs.readFileSync(markerPath, "utf8")); + } catch { + return "invalid"; + } + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return "invalid"; + const record = raw as Record; + if ( + record.version !== 1 || + typeof record.runId !== "string" || + record.runId !== path.basename(runDir) || + typeof record.manifest !== "string" || + textBytes(record.manifest) > WORKFLOW_MANIFEST_MAX_BYTES || + !Array.isArray(record.artifacts) || + record.artifacts.length < 1 || + record.artifacts.length > artifactLimits.size + ) { + return "invalid"; + } + + let manifest: Record; + try { + const parsed: unknown = JSON.parse(record.manifest); + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { + return "invalid"; + } + manifest = parsed as Record; + } catch { + return "invalid"; + } + if ( + manifest.runId !== record.runId || + !["completed", "failed", "aborted", "uncertain"].includes( + String(manifest.status), + ) || + manifest.transcriptArtifact !== "transcripts.json" || + (manifest.resultArtifact !== undefined && + manifest.resultArtifact !== "result.json") + ) { + return "invalid"; + } + + const artifacts: WorkflowCommitArtifact[] = []; + const names = new Set(); + for (const value of record.artifacts) { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return "invalid"; + } + const artifact = value as Record; + if ( + typeof artifact.name !== "string" || + !artifactLimits.has(artifact.name as WorkflowArtifactWrite["name"]) || + names.has(artifact.name) || + typeof artifact.bytes !== "number" || + !Number.isInteger(artifact.bytes) || + artifact.bytes < 0 || + artifact.bytes > + (artifactLimits.get(artifact.name as WorkflowArtifactWrite["name"]) ?? + -1) || + typeof artifact.sha256 !== "string" || + !/^[0-9a-f]{64}$/u.test(artifact.sha256) + ) { + return "invalid"; + } + names.add(artifact.name); + artifacts.push({ + name: artifact.name as WorkflowArtifactWrite["name"], + bytes: artifact.bytes, + sha256: artifact.sha256, + }); + } + if ( + !names.has("transcripts.json") || + (manifest.resultArtifact === "result.json") !== names.has("result.json") + ) { + return "invalid"; + } + return { + version: 1, + runId: record.runId, + manifest: record.manifest, + artifacts, + }; +} + +/** + * Complete a terminal artifact commit only when every prepared file matches + * the exact bounded receipt written before the side-artifact sequence began. + */ +export function recoverPendingWorkflowCommit( + runDir: string, +): WorkflowCommitRecovery { + const marker = parseWorkflowCommitMarker(runDir); + if (marker === "none" || marker === "invalid") return marker; + + for (const artifact of marker.artifacts) { + const artifactPath = path.join(runDir, artifact.name); + let stat: fs.Stats; + let content: Buffer; + try { + stat = fs.lstatSync(artifactPath); + if ( + !stat.isFile() || + stat.isSymbolicLink() || + stat.size !== artifact.bytes + ) { + return "incomplete"; + } + content = fs.readFileSync(artifactPath); + } catch { + return "incomplete"; + } + if ( + content.byteLength !== artifact.bytes || + sha256(content) !== artifact.sha256 + ) { + return "incomplete"; + } + } + + const manifestPath = path.join(runDir, "workflow.json"); + try { + if ( + fs.existsSync(manifestPath) && + fs.readFileSync(manifestPath, "utf8") === marker.manifest + ) { + removeWorkflowCommit(runDir); + return "already-committed"; + } + writeFileAtomic(manifestPath, marker.manifest); + removeWorkflowCommit(runDir); + return "recovered"; + } catch { + return "failed"; + } +} + function boundEntry(entry: TranscriptEntry, maxBytes: number) { if (textBytes(entry.text) <= maxBytes) return { ...entry }; const markerBytes = textBytes(ENTRY_TRUNCATION_MARKER); @@ -177,15 +426,23 @@ export function persistWorkflowJson( // later artifact write fails, readers still see an explained terminal run // instead of the previous `running` manifest. The final manifest below adds // the artifact references once every dependent file has committed. - if (details.status !== "running") { + const terminal = details.status !== "running"; + if (terminal) { + // A retry supersedes an older unfinished receipt before it publishes a new + // terminal fact. Failing to remove it must stop the new commit rather than + // let a concurrent reader promote stale artifact identities. + removeWorkflowCommit(runDir, true); persistWorkflowTerminalState(runDir, details); } - writeRunFile( - runDir, - "transcripts.json", - safeStringify(transcripts, { maxBytes: 2 * 1024 * 1024 }), - ); + const artifactWrites: WorkflowArtifactWrite[] = [ + { + name: "transcripts.json", + content: safeStringify(transcripts, { + maxBytes: WORKFLOW_TRANSCRIPTS_MAX_BYTES, + }), + }, + ]; // Written alongside the rest so it inherits atomic write, 500ms coalescing, // and the final flush. Only present once a call has actually succeeded. // Accumulators already enforce the cap incrementally and can assemble the @@ -201,14 +458,15 @@ export function persistWorkflowJson( "toJson" in journal ? journal.toJson() : JSON.stringify(boundedJournal(journal).journal, null, 2); - writeRunFile(runDir, JOURNAL_FILE, content); + artifactWrites.push({ name: JOURNAL_FILE, content }); } if (details.result !== undefined) { - writeRunFile( - runDir, - "result.json", - safeStringify(details.result, { maxBytes: 1024 * 1024 }), - ); + artifactWrites.push({ + name: "result.json", + content: safeStringify(details.result, { + maxBytes: WORKFLOW_MANIFEST_MAX_BYTES, + }), + }); } const compact: WorkflowDetails = { ...details, @@ -218,11 +476,22 @@ export function persistWorkflowJson( transcriptArtifact: "transcripts.json", agents: details.agents.map((agent) => ({ ...agent, transcript: [] })), }; - writeRunFile( - runDir, - "workflow.json", - safeStringify(compact, { maxBytes: 1024 * 1024 }), - ); + const manifest = safeStringify(compact, { + maxBytes: WORKFLOW_MANIFEST_MAX_BYTES, + }); + + if (terminal) { + writeRunFile( + runDir, + WORKFLOW_COMMIT_FILE, + serializeWorkflowCommitMarker(details, manifest, artifactWrites), + ); + } + for (const artifact of artifactWrites) { + writeRunFile(runDir, artifact.name, artifact.content); + } + writeRunFile(runDir, "workflow.json", manifest); + if (terminal) removeWorkflowCommit(runDir); } /** @@ -235,6 +504,7 @@ export function persistWorkflowDeliveryState( runDir: string, delivery: WorkflowDelivery, ) { + recoverPendingWorkflowCommit(runDir); const file = path.join(runDir, "workflow.json"); const raw: unknown = JSON.parse(fs.readFileSync(file, "utf8")); if (!raw || typeof raw !== "object" || Array.isArray(raw)) { diff --git a/extensions/workflows/dashboard.ts b/extensions/workflows/dashboard.ts index 2a9b34ef..fe9db8bf 100644 --- a/extensions/workflows/dashboard.ts +++ b/extensions/workflows/dashboard.ts @@ -31,6 +31,7 @@ import { import { SPINNER_INTERVAL_MS, spinnerFrame } from "../shared/spinner.ts"; import { sanitizeTerminalText } from "../shared/terminal-text.ts"; import { isAcceptanceLedger } from "./acceptance.ts"; +import { recoverPendingWorkflowCommit } from "./artifacts.ts"; import { projectWorkflowGraph } from "./graph-projection.ts"; import { classifyInterruptedInvocation, @@ -143,6 +144,7 @@ export function readPersistedWorkflowDetails( runId: string, options: ReadPersistedRunOptions = {}, ): WorkflowDetails | undefined { + recoverPendingWorkflowCommit(path.join(runsDir(), runId)); let details: WorkflowDetails | undefined; try { const raw: unknown = JSON.parse( diff --git a/tests/extensions/workflows/artifacts.test.ts b/tests/extensions/workflows/artifacts.test.ts index 249b185c..72a59e6e 100644 --- a/tests/extensions/workflows/artifacts.test.ts +++ b/tests/extensions/workflows/artifacts.test.ts @@ -1,5 +1,7 @@ import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; import { + existsSync, mkdirSync, mkdtempSync, readFileSync, @@ -14,8 +16,11 @@ import { createWorkflowPersistence, loadJournal, persistWorkflowAgentResult, + persistWorkflowDeliveryState, persistWorkflowJson, persistWorkflowTerminalState, + recoverPendingWorkflowCommit, + WORKFLOW_COMMIT_FILE, } from "../../../extensions/workflows/artifacts.ts"; import { createJournalAccumulator, @@ -39,6 +44,66 @@ function workflowDetails(): WorkflowDetails { }; } +function artifactDigest(content: string) { + return createHash("sha256").update(content).digest("hex"); +} + +function stageTerminalCommit( + root: string, + options: { omitResult?: boolean; committedManifest?: boolean } = {}, +) { + const runId = "wf_crash"; + const prepared = join(root, "prepared"); + const runDir = join(root, runId); + mkdirSync(prepared); + mkdirSync(runDir); + const details: WorkflowDetails = { + ...workflowDetails(), + runId, + status: "completed", + finishedAt: 2, + result: { verdict: "complete" }, + }; + persistWorkflowJson(prepared, details); + const manifest = readFileSync(join(prepared, "workflow.json"), "utf8"); + const transcripts = readFileSync(join(prepared, "transcripts.json"), "utf8"); + const result = readFileSync(join(prepared, "result.json"), "utf8"); + writeFileSync( + join(runDir, "workflow.json"), + options.committedManifest + ? manifest + : JSON.stringify({ + ...details, + result: undefined, + resultArtifact: undefined, + transcriptArtifact: undefined, + }), + ); + writeFileSync(join(runDir, "transcripts.json"), transcripts); + if (!options.omitResult) writeFileSync(join(runDir, "result.json"), result); + writeFileSync( + join(runDir, WORKFLOW_COMMIT_FILE), + JSON.stringify({ + version: 1, + runId, + manifest, + artifacts: [ + { + name: "transcripts.json", + bytes: Buffer.byteLength(transcripts), + sha256: artifactDigest(transcripts), + }, + { + name: "result.json", + bytes: Buffer.byteLength(result), + sha256: artifactDigest(result), + }, + ], + }), + ); + return { runDir, manifest }; +} + test("artifact transcript keeps the initial prompt, marker, and newest entries", () => { const prompt = `initial:${"p".repeat(70)}`; const transcript = [ @@ -283,11 +348,103 @@ test("terminal persistence publishes status before dependent artifacts", () => { JSON.parse(readFileSync(join(directory, "result.json"), "utf8")), { partial: true }, ); + assert.equal(existsSync(join(directory, WORKFLOW_COMMIT_FILE)), false); } finally { rmSync(directory, { recursive: true, force: true }); } }); +test("a complete pending artifact receipt recovers the exact terminal manifest", () => { + const root = mkdtempSync(join(tmpdir(), "pi-workflow-commit-recovery-")); + try { + const { runDir, manifest } = stageTerminalCommit(root); + + assert.equal(recoverPendingWorkflowCommit(runDir), "recovered"); + assert.equal(readFileSync(join(runDir, "workflow.json"), "utf8"), manifest); + assert.equal(existsSync(join(runDir, WORKFLOW_COMMIT_FILE)), false); + assert.equal(recoverPendingWorkflowCommit(runDir), "none"); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("an incomplete pending artifact receipt cannot publish terminal references", () => { + const root = mkdtempSync(join(tmpdir(), "pi-workflow-commit-incomplete-")); + try { + const { runDir } = stageTerminalCommit(root, { omitResult: true }); + + assert.equal(recoverPendingWorkflowCommit(runDir), "incomplete"); + const stored = JSON.parse( + readFileSync(join(runDir, "workflow.json"), "utf8"), + ) as WorkflowDetails; + assert.equal(stored.resultArtifact, undefined); + assert.equal(stored.transcriptArtifact, undefined); + assert.equal(existsSync(join(runDir, WORKFLOW_COMMIT_FILE)), true); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("a same-size artifact substitution cannot satisfy the commit receipt", () => { + const root = mkdtempSync(join(tmpdir(), "pi-workflow-commit-digest-")); + try { + const { runDir } = stageTerminalCommit(root); + const resultPath = join(runDir, "result.json"); + const result = readFileSync(resultPath, "utf8"); + writeFileSync(resultPath, result.replace("complete", "tampered")); + + assert.equal( + readFileSync(resultPath).byteLength, + Buffer.byteLength(result), + ); + assert.equal(recoverPendingWorkflowCommit(runDir), "incomplete"); + const stored = JSON.parse( + readFileSync(join(runDir, "workflow.json"), "utf8"), + ) as WorkflowDetails; + assert.equal(stored.resultArtifact, undefined); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("delivery persistence first recovers a complete pending artifact commit", () => { + const root = mkdtempSync(join(tmpdir(), "pi-workflow-delivery-recovery-")); + try { + const { runDir } = stageTerminalCommit(root); + + persistWorkflowDeliveryState(runDir, { + id: "workflow:wf_crash:terminal", + state: "delivered", + attempts: 1, + updatedAt: 3, + }); + + const stored = JSON.parse( + readFileSync(join(runDir, "workflow.json"), "utf8"), + ) as WorkflowDetails; + assert.equal(stored.resultArtifact, "result.json"); + assert.equal(stored.transcriptArtifact, "transcripts.json"); + assert.equal(stored.delivery?.state, "delivered"); + assert.equal(existsSync(join(runDir, WORKFLOW_COMMIT_FILE)), false); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("recovery removes a receipt whose manifest was already committed", () => { + const root = mkdtempSync(join(tmpdir(), "pi-workflow-commit-idempotent-")); + try { + const { runDir } = stageTerminalCommit(root, { + committedManifest: true, + }); + + assert.equal(recoverPendingWorkflowCommit(runDir), "already-committed"); + assert.equal(existsSync(join(runDir, WORKFLOW_COMMIT_FILE)), false); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + test("a dependent artifact write failure cannot leave the prior running manifest", () => { const directory = mkdtempSync( join(tmpdir(), "pi-workflow-terminal-failure-"), @@ -311,6 +468,7 @@ test("a dependent artifact write failure cannot leave the prior running manifest assert.equal(stored.status, "completed"); assert.equal(stored.resultArtifact, undefined); assert.equal(stored.transcriptArtifact, undefined); + assert.equal(existsSync(join(directory, WORKFLOW_COMMIT_FILE)), true); } finally { rmSync(directory, { recursive: true, force: true }); } diff --git a/tests/extensions/workflows/dashboard.test.ts b/tests/extensions/workflows/dashboard.test.ts index 77ebbfc1..ab829388 100644 --- a/tests/extensions/workflows/dashboard.test.ts +++ b/tests/extensions/workflows/dashboard.test.ts @@ -1,6 +1,8 @@ import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; import { chmodSync, + existsSync, mkdirSync, mkdtempSync, readFileSync, @@ -17,6 +19,7 @@ import { } from "@earendil-works/pi-coding-agent"; import type { TUI } from "@earendil-works/pi-tui"; import { SPINNER_INTERVAL_MS } from "../../../extensions/shared/spinner.ts"; +import { WORKFLOW_COMMIT_FILE } from "../../../extensions/workflows/artifacts.ts"; import type { Theme, WorkflowDetails, @@ -32,6 +35,7 @@ const { buildWorkflowReport, loadRunEntries, normalizePersistedWorkflowDetails, + readPersistedWorkflowDetails, recoverStaleWorkflowDetails, workflowGraphSummary, WorkflowDashboard, @@ -39,6 +43,67 @@ const { const SESSION = "session-1"; +test("persisted reads recover a fully prepared terminal artifact commit", () => { + const runId = "wf_commit_read"; + const dir = join(agentDir, "workflows", runId); + mkdirSync(dir, { recursive: true }); + const manifest = JSON.stringify({ + runId, + sessionId: SESSION, + background: true, + status: "completed", + startedAt: 1, + finishedAt: 2, + phases: [], + agents: [], + result: "[stored in result.json]", + resultArtifact: "result.json", + transcriptArtifact: "transcripts.json", + }); + const result = JSON.stringify({ verdict: "complete" }); + const transcripts = JSON.stringify({}); + const artifact = (name: string, content: string) => ({ + name, + bytes: Buffer.byteLength(content), + sha256: createHash("sha256").update(content).digest("hex"), + }); + writeFileSync( + join(dir, "workflow.json"), + JSON.stringify({ + runId, + sessionId: SESSION, + background: true, + status: "completed", + startedAt: 1, + finishedAt: 2, + phases: [], + agents: [], + }), + ); + writeFileSync(join(dir, "result.json"), result); + writeFileSync(join(dir, "transcripts.json"), transcripts); + writeFileSync( + join(dir, WORKFLOW_COMMIT_FILE), + JSON.stringify({ + version: 1, + runId, + manifest, + artifacts: [ + artifact("transcripts.json", transcripts), + artifact("result.json", result), + ], + }), + ); + + const restored = readPersistedWorkflowDetails(runId, { + hydrateArtifacts: true, + }); + assert.equal(restored?.status, "completed"); + assert.equal(restored?.resultArtifact, "result.json"); + assert.deepEqual(restored?.result, { verdict: "complete" }); + assert.equal(existsSync(join(dir, WORKFLOW_COMMIT_FILE)), false); +}); + function writeRun( runId: string, startedAt: number, From 0bf2089c009fea84a90414b85cf1b52c9e85cb76 Mon Sep 17 00:00:00 2001 From: testikun <320479488+testikun@users.noreply.github.com> Date: Fri, 4 Sep 2026 18:17:58 +0800 Subject: [PATCH 2/4] docs: record workflow artifact commit protocol --- docs/architecture/WORKFLOW_ARTIFACT_COMMIT.md | 51 +++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 docs/architecture/WORKFLOW_ARTIFACT_COMMIT.md diff --git a/docs/architecture/WORKFLOW_ARTIFACT_COMMIT.md b/docs/architecture/WORKFLOW_ARTIFACT_COMMIT.md new file mode 100644 index 00000000..7ab85683 --- /dev/null +++ b/docs/architecture/WORKFLOW_ARTIFACT_COMMIT.md @@ -0,0 +1,51 @@ +# Workflow terminal artifact commit protocol + +- Status: `validated` +- Created: 2026-09-04 +- Verified: 2026-09-04 +- Source boundary: implementation commit `cd60a15`, based on `72fbba5` +- Affected Pi primitive: the OpenPI Workflow extension's run-directory persistence; Pi Sessions, messages, providers, and child lifecycle remain unchanged +- Related Issue: [#110](https://github.com/openpi-dev/openpi/issues/110) +- Related Decision: [0001 — documentation and evidence governance](../decisions/0001-documentation-and-evidence-governance.md) +- Supersedes: none + +## Ownership + +`workflow.json` remains the canonical persisted projection of a Workflow run. `result.json`, `transcripts.json`, and `journal.json` are dependent side artifacts. The hidden `.workflow-commit.json` file is a bounded recovery receipt, not another run manifest and not model-visible state. + +The runtime still owns terminalization, delivery, cancellation, and cleanup. This protocol only makes the existing filesystem projection recoverable across a process crash between individually atomic file replacements. + +## Commit sequence + +For a terminal run, persistence follows this order: + +1. Remove any receipt from an older attempt, failing closed if that cannot be done. +2. Atomically publish a terminal `workflow.json` without side-artifact references. A later write failure therefore cannot leave a known terminal run recorded as `running`. +3. Build the final compact manifest and every dependent artifact in memory. +4. Atomically write `.workflow-commit.json`. It contains version `1`, the exact run id, the exact final manifest bytes, and a filename, byte count, and SHA-256 digest for each artifact. +5. Atomically replace each side artifact. +6. Atomically replace `workflow.json` with the exact manifest recorded by the receipt. +7. Best-effort remove the receipt. A crash or unlink failure after step 6 is harmless because recovery recognizes the already-committed manifest. + +Running checkpoints retain the existing lightweight path and do not create commit receipts. Successful terminal persistence leaves no receipt behind. + +## Recovery invariants + +Persisted Workflow reads and delivery-receipt updates check for a pending commit before consuming `workflow.json`. Recovery promotes the recorded manifest only when all of these facts hold: + +- the receipt is a regular, non-symlink file within its byte budget; +- the receipt version and run id match the containing generated run directory; +- the recorded manifest is bounded JSON for a known terminal state; +- artifact names are unique members of the fixed `result.json`, `transcripts.json`, and `journal.json` set; +- manifest references agree exactly with the receipt's artifact set; +- every artifact is a regular, non-symlink file whose byte count and SHA-256 digest match the receipt. + +If the final manifest is already byte-identical, recovery only removes the stale receipt. If every artifact validates and the manifest is still the earlier terminal projection, recovery atomically completes the manifest commit. Missing, truncated, substituted, oversized, malformed, or path-traversing evidence never gains an artifact reference. + +An incomplete or invalid receipt stays available for inspection and for a concurrently finishing writer; the next terminal persistence attempt replaces the single fixed receipt. Legacy runs without a receipt keep their existing compatibility behavior. In particular, recovery does not infer completion merely from an orphan `result.json`, because that file alone does not carry a trustworthy terminal identity. + +## Evidence and limits + +At `cd60a15`, focused tests cover full preparation followed by recovery, incomplete preparation, same-size content substitution, an already-committed manifest, delivery mutation after recovery, normal receipt cleanup, and the dashboard/startup read path. `bun run check` passed; the full suite passed with 1247 Node tests, 0 failures, 1 skip, and 30 Vitest tests. + +The guarantee is process-crash recovery at the repository's existing per-file atomic-replace boundary. It does not claim a filesystem-wide transaction or power-loss durability beyond `writeFileAtomic`, which does not currently fsync file and directory metadata. From c0ff5f3367cacf6aebc523374568aca91977d59e Mon Sep 17 00:00:00 2001 From: testikun <320479488+testikun@users.noreply.github.com> Date: Fri, 4 Sep 2026 18:18:55 +0800 Subject: [PATCH 3/4] docs: link workflow artifact recovery PR --- docs/architecture/WORKFLOW_ARTIFACT_COMMIT.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/architecture/WORKFLOW_ARTIFACT_COMMIT.md b/docs/architecture/WORKFLOW_ARTIFACT_COMMIT.md index 7ab85683..3e97183b 100644 --- a/docs/architecture/WORKFLOW_ARTIFACT_COMMIT.md +++ b/docs/architecture/WORKFLOW_ARTIFACT_COMMIT.md @@ -6,6 +6,7 @@ - Source boundary: implementation commit `cd60a15`, based on `72fbba5` - Affected Pi primitive: the OpenPI Workflow extension's run-directory persistence; Pi Sessions, messages, providers, and child lifecycle remain unchanged - Related Issue: [#110](https://github.com/openpi-dev/openpi/issues/110) +- Related PR: [#386](https://github.com/openpi-dev/openpi/pull/386) - Related Decision: [0001 — documentation and evidence governance](../decisions/0001-documentation-and-evidence-governance.md) - Supersedes: none From fd2842f8a5ee1e67d37db2ba10f48a957e884b9b Mon Sep 17 00:00:00 2001 From: testikun <320479488+testikun@users.noreply.github.com> Date: Sat, 5 Sep 2026 10:33:18 +0800 Subject: [PATCH 4/4] fix(workflows): preserve newer manifest during recovery --- extensions/workflows/artifacts.ts | 27 +++++++++++++++++- tests/extensions/workflows/artifacts.test.ts | 29 ++++++++++++++++++++ 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/extensions/workflows/artifacts.ts b/extensions/workflows/artifacts.ts index af4aae86..675fdd76 100644 --- a/extensions/workflows/artifacts.ts +++ b/extensions/workflows/artifacts.ts @@ -236,6 +236,31 @@ function parseWorkflowCommitMarker( }; } +function hasCommittedManifest( + manifestPath: string, + marker: WorkflowCommitMarker, +) { + try { + const parsed: unknown = JSON.parse(fs.readFileSync(manifestPath, "utf8")); + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { + return false; + } + const manifest = parsed as Record; + const markerManifest = JSON.parse(marker.manifest) as Record< + string, + unknown + >; + return ( + manifest.runId === marker.runId && + manifest.status === markerManifest.status && + manifest.transcriptArtifact === markerManifest.transcriptArtifact && + manifest.resultArtifact === markerManifest.resultArtifact + ); + } catch { + return false; + } +} + /** * Complete a terminal artifact commit only when every prepared file matches * the exact bounded receipt written before the side-artifact sequence began. @@ -275,7 +300,7 @@ export function recoverPendingWorkflowCommit( try { if ( fs.existsSync(manifestPath) && - fs.readFileSync(manifestPath, "utf8") === marker.manifest + hasCommittedManifest(manifestPath, marker) ) { removeWorkflowCommit(runDir); return "already-committed"; diff --git a/tests/extensions/workflows/artifacts.test.ts b/tests/extensions/workflows/artifacts.test.ts index 72a59e6e..0bbfd393 100644 --- a/tests/extensions/workflows/artifacts.test.ts +++ b/tests/extensions/workflows/artifacts.test.ts @@ -445,6 +445,35 @@ test("recovery removes a receipt whose manifest was already committed", () => { } }); +test("recovery preserves newer delivery fields in an already committed manifest", () => { + const root = mkdtempSync( + join(tmpdir(), "pi-workflow-commit-newer-manifest-"), + ); + try { + const { runDir } = stageTerminalCommit(root, { committedManifest: true }); + const manifestPath = join(runDir, "workflow.json"); + const manifest = JSON.parse(readFileSync(manifestPath, "utf8")) as Record< + string, + unknown + >; + manifest.delivery = { state: "delivered", attempts: 2, updatedAt: 9 }; + writeFileSync(manifestPath, JSON.stringify(manifest)); + + assert.equal(recoverPendingWorkflowCommit(runDir), "already-committed"); + assert.deepEqual( + ( + JSON.parse(readFileSync(manifestPath, "utf8")) as Record< + string, + unknown + > + ).delivery, + { state: "delivered", attempts: 2, updatedAt: 9 }, + ); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + test("a dependent artifact write failure cannot leave the prior running manifest", () => { const directory = mkdtempSync( join(tmpdir(), "pi-workflow-terminal-failure-"),