Persist bounded plan progress

Author: chubes4

packages/cli/src/bounded-recipe-plan.ts

@@ -1,14 +1,21 @@
-import { mkdir, readFile, writeFile } from "node:fs/promises"
+import { mkdir, readFile, rename, writeFile } from "node:fs/promises"
 import { join } from "node:path"
-import { commandArgValue, executeBoundedRuntimePlan, parseCommandJsonObject, type BoundedRuntimePlan, type BoundedRuntimePlanResult, type ExecutionResult, type Runtime } from "@automattic/wp-codebox-core"
+import { commandArgValue, executeBoundedRuntimePlan, parseCommandJsonObject, type BoundedRuntimePlan, type BoundedRuntimePlanEntryResult, type BoundedRuntimePlanResult, type ExecutionResult, type Runtime } from "@automattic/wp-codebox-core"
 import { recipeExecutionSpec } from "./agent-sandbox.js"
 
 export interface BoundedRecipePlanExecutionOptions {
   artifactRoot: string
   recipeDirectory: string
 }
 
+const BOUNDED_RUNTIME_PLAN_PROGRESS_SCHEMA = "wp-codebox/bounded-runtime-plan-progress/v1"
+
 export async function executeBoundedRecipePlan(runtime: Runtime, plan: BoundedRuntimePlan, options: BoundedRecipePlanExecutionOptions): Promise<BoundedRuntimePlanResult> {
+  const progressPath = join(options.artifactRoot, "bounded-plan", "progress.json")
+  const completed = new Map<string, BoundedRuntimePlanEntryResult>()
+  let progressWrite = Promise.resolve()
+  await mkdir(join(options.artifactRoot, "bounded-plan"), { recursive: true })
+  await writeBoundedPlanProgress(progressPath, plan, completed, false)
   const aggregate = await executeBoundedRuntimePlan(plan, {
     async materialize() { return { workspace: options.recipeDirectory, runtime } },
     async startServices() { return undefined },
@@ -50,10 +57,15 @@ export async function executeBoundedRecipePlan(runtime: Runtime, plan: BoundedRu
       }, null, 2)}\n`, "utf8")
       return { success: exitCode === 0, exitCode, message: stderr, stdoutRef, stderrRef, resultRef, artifactRefs }
     },
+    async onEntryResult(result) {
+      completed.set(result.id, result)
+      progressWrite = progressWrite.then(async () => writeBoundedPlanProgress(progressPath, plan, completed, false))
+      await progressWrite
+    },
     async stopServices() {},
     async dispose() {},
   })
-  await mkdir(join(options.artifactRoot, "bounded-plan"), { recursive: true })
+  await writeBoundedPlanProgress(progressPath, plan, completed, true)
   await writeFile(join(options.artifactRoot, "bounded-plan/result.json"), `${JSON.stringify(aggregate, null, 2)}\n`, "utf8")
   return aggregate
 }
@@ -76,3 +88,27 @@ function runtimeArtifactRefs(execution: ExecutionResult | undefined): string[] {
     return typeof value === "string" && value ? [value] : []
   })
 }
+
+async function writeBoundedPlanProgress(path: string, plan: BoundedRuntimePlan, completed: Map<string, BoundedRuntimePlanEntryResult>, complete: boolean): Promise<void> {
+  const entries = plan.entries.flatMap((entry) => {
+    const result = completed.get(entry.id)
+    return result ? [result] : []
+  })
+  const progress = {
+    schema: BOUNDED_RUNTIME_PLAN_PROGRESS_SCHEMA,
+    complete,
+    concurrency: Math.min(plan.concurrency, plan.entries.length),
+    counts: {
+      total: plan.entries.length,
+      succeeded: entries.filter((entry) => entry.status === "succeeded").length,
+      failed: entries.filter((entry) => entry.status === "failed").length,
+      timedOut: entries.filter((entry) => entry.status === "timed_out").length,
+      cancelled: entries.filter((entry) => entry.status === "cancelled").length,
+      unfinished: plan.entries.length - entries.length,
+    },
+    entries,
+  }
+  const temporaryPath = `${path}.tmp`
+  await writeFile(temporaryPath, `${JSON.stringify(progress, null, 2)}\n`, "utf8")
+  await rename(temporaryPath, path)
+}

packages/runtime-core/src/bounded-runtime-plan.ts

@@ -64,6 +64,7 @@ export interface BoundedRuntimePlanAdapter<TWorkspace = unknown, TRuntime = unkn
   materialize(): Promise<{ workspace: TWorkspace; runtime: TRuntime }>
   startServices(context: { workspace: TWorkspace; runtime: TRuntime }): Promise<TServices>
   execute(context: BoundedRuntimePlanExecution<TWorkspace, TRuntime, TServices>): Promise<{ success: boolean; exitCode?: number; message?: string; stdoutRef?: string; stderrRef?: string; resultRef?: string; artifactRefs?: string[] }>
+  onEntryResult?(result: BoundedRuntimePlanEntryResult): Promise<void>
   stopServices(context: { workspace: TWorkspace; runtime: TRuntime; services: TServices }): Promise<void>
   dispose(context: { workspace: TWorkspace; runtime: TRuntime }): Promise<void>
 }
@@ -134,10 +135,12 @@ async function executeEntries<TWorkspace, TRuntime, TServices>(plan: BoundedRunt
       const entry = plan.entries[index]!
       if (failed && plan.failFast) {
         results[index] = cancelledResult(entry)
+        await adapter.onEntryResult?.(results[index])
         continue
       }
       const result = await executeEntry(entry, materialized, services, adapter)
       results[index] = result
+      await adapter.onEntryResult?.(result)
       if (!result.success) failed = true
     }
   }

tests/bounded-recipe-plan.test.ts

@@ -57,6 +57,11 @@ try {
   assert.equal(await readFile(join(root, "entries/one/stdout.txt"), "utf8"), "password=[redacted]\n")
   assert.equal(await readFile(join(root, "entries/failed/stderr.txt"), "utf8"), "database [redacted] failed\n")
   assert.equal(JSON.parse(await readFile(join(root, "bounded-plan/result.json"), "utf8")).schema, "wp-codebox/bounded-runtime-plan-result/v1")
+  const progress = JSON.parse(await readFile(join(root, "bounded-plan/progress.json"), "utf8"))
+  assert.equal(progress.schema, "wp-codebox/bounded-runtime-plan-progress/v1")
+  assert.equal(progress.complete, true)
+  assert.deepEqual(progress.counts, { total: 2, succeeded: 1, failed: 1, timedOut: 0, cancelled: 0, unfinished: 0 })
+  assert.deepEqual(progress.entries.map((entry: { id: string }) => entry.id), ["one", "failed"])
 } finally {
   await rm(root, { recursive: true, force: true })
 }

tests/bounded-runtime-plan.test.ts

@@ -13,6 +13,7 @@ const plan: BoundedRuntimePlan = {
 }
 
 const lifecycle: string[] = []
+const completedEntries: string[] = []
 let active = 0
 let maximumActive = 0
 const executed: string[] = []
@@ -31,6 +32,7 @@ const adapter: BoundedRuntimePlanAdapter<{ root: string }, { id: string }, { id:
       active--
     }
   },
+  async onEntryResult(result) { completedEntries.push(result.id) },
   async stopServices() { lifecycle.push("stop-services") },
   async dispose() { lifecycle.push("dispose") },
 }
@@ -44,6 +46,7 @@ assert.deepEqual(result.counts, { total: 4, succeeded: 2, failed: 1, timedOut: 1
 assert.equal(result.entries.every((entry) => Number.isInteger(entry.durationMs) && entry.durationMs >= 0), true)
 assert.deepEqual(lifecycle, ["materialize", "start-services", "stop-services", "dispose"], "workspace, service, and runtime lifecycles are each owned once")
 assert.deepEqual(executed.sort(), ["failed", "first", "last", "slow"], "one failed entry does not isolate unrelated entries")
+assert.deepEqual(completedEntries.sort(), ["failed", "first", "last", "slow"], "each terminal entry is reported to the adapter")
 
 const retry = retryBoundedRuntimePlan(plan, result)
 assert.deepEqual(retry.entries.map((entry) => entry.id), ["failed", "slow"], "retry selects only unsuccessful prior entries")