From 656e5311d81e576d4451f3a66312855361cc938b Mon Sep 17 00:00:00 2001 From: Evanfeenstra Date: Mon, 31 Aug 2026 08:08:03 -0700 Subject: [PATCH] =?UTF-8?q?evolve:=20phase-1=20measurement=20infra=20?= =?UTF-8?q?=E2=80=94=20foreach=20concurrency,=20partial=20run=20summaries,?= =?UTF-8?q?=20eval/matrix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of plans/evolve-scoreboard-and-task-matrix.md (#1623), scoped to avoid #1621's files. Three independent pieces: vein foreach `concurrency` (default 1): bounded worker pool over iterations — results stay in input order, `#i` event paths and journal replay are unchanged, the cooperative checkpoint moves to just before each iteration starts (pause parks new starts, in-flight drain), a failure stops new starts and surfaces the lowest-index error, and cancellation always wins over an ordinary error. The prod evolve run spent 17h strictly serial on 225 produce runs; repeats are unaffordable without this. vein partial run summaries: `GET /workflows/:name/runs/:runId` now serves a summary reconstructed from the event log (`partial: true`, status from the live controller or "stale") instead of a 404 when `run.json` is missing — in-flight and crash-orphaned runs alike. The event log is already durable per-step; refusing to read it is what turned a dead 17-hour run into event-log archaeology. Reconstruction is a pure `summarizeFromEvents` (exported) over run.start / top-level step.end / step.error. No write-path changes: run.json stays terminal-only, so "summary exists = finalized" consumers are untouched. eval/matrix (seeded lab step): the task×version matrix across measurements — per-task bands (floor / movable / ceiling), an EMPIRICAL noise floor from same-version re-measurements (fitness deltas + task flips on identical YAML; UNKNOWN rather than 0 when no version has n≥2), and bias-vs-variance tags on never-correct tasks (byte-identical wrong answer ×≥3 = bias, distinct answers = variance). Verdict channel only — gold never enters (EVOLVE_SPEC §6). Offline smoke: `npx tsx src/lab/eval/matrix-smoke.ts`, fixtures modeled on prod run 1788061734710 ("36"×9 bias, the v11 0.76/0.80 resample pair). vein: 544/544 tests pass (6 new foreach-concurrency, 4 summarize, 1 endpoint). gaia evolve-smoke + matrix-smoke pass; both packages typecheck. Co-Authored-By: Claude Fable 5 --- mcp/src/lab/AGENTS.md | 9 + mcp/src/lab/eval/matrix-smoke.ts | 133 ++++++++++++++ mcp/src/lab/eval/seed.ts | 5 + mcp/src/lab/eval/steps/matrix.ts | 297 +++++++++++++++++++++++++++++++ vein/src/control-flow.test.ts | 177 ++++++++++++++++++ vein/src/createVein.test.ts | 35 ++++ vein/src/createVein.ts | 13 +- vein/src/index.ts | 2 + vein/src/runner.ts | 64 ++++++- vein/src/steps/core/foreach.ts | 3 +- vein/src/store.test.ts | 59 +++++- vein/src/store.ts | 81 +++++++++ 12 files changed, 867 insertions(+), 11 deletions(-) create mode 100644 mcp/src/lab/eval/matrix-smoke.ts create mode 100644 mcp/src/lab/eval/steps/matrix.ts diff --git a/mcp/src/lab/AGENTS.md b/mcp/src/lab/AGENTS.md index 8e6f2794e..ec4054394 100644 --- a/mcp/src/lab/AGENTS.md +++ b/mcp/src/lab/AGENTS.md @@ -544,6 +544,15 @@ Domain-agnostic eval substrate, shared by every experiment. See (never a mid-generation kill), both null by default. Generation count is a poor budget on its own: authors reliably evolve toward more expensive architectures, so per-generation cost and wall-clock GROW over a run. +- `eval/steps/matrix.ts` (`eval/matrix`) — the task×version MATRIX across + measurements (plans/evolve-scoreboard-and-task-matrix.md, Phase 1): folds + every `{ version, results }` measurement into per-task bands + (floor/movable/ceiling), an EMPIRICAL noise floor from same-version + re-measurements (identical-YAML fitness deltas + task flips; UNKNOWN, not + 0, when no version has n≥2), and bias-vs-variance tags on never-correct + tasks (byte-identical wrong answer ×≥3 = bias — immune to redundancy and + prompt nudges; distinct wrong answers = variance). Verdict channel only — + gold never enters. Smoke: `npx tsx src/lab/eval/matrix-smoke.ts`. **Naming rule:** `eval/*` = generic. The eval *workflows* that wire these with a rubric/task/dataset belong to the experiment and are named `-…`. diff --git a/mcp/src/lab/eval/matrix-smoke.ts b/mcp/src/lab/eval/matrix-smoke.ts new file mode 100644 index 000000000..c89970ad6 --- /dev/null +++ b/mcp/src/lab/eval/matrix-smoke.ts @@ -0,0 +1,133 @@ +/** + * Offline validation for eval/matrix (the task×version matrix — + * plans/evolve-scoreboard-and-task-matrix.md, Phase 1). The fixtures are + * modeled on prod run gaia-evolve/1788061734710, whose pathologies the step + * exists to surface: + * 1. bands: tasks correct in every / no / some measurement(s) + * 2. empirical noise floor from same-version re-measurements (v11 was + * accidentally measured twice, 0.76 vs 0.80 — identical YAML) + * 3. bias tag: a never-correct task answering byte-identically ≥3 times + * ("36" ×9 in prod) vs a variance-tagged one (distinct wrong answers) + * 4. no same-version pairs → the floor reads UNKNOWN, never zero + * 5. the three correctness shapes gaia-run / gaia-candidate-run emit + * No LLM, no network, no dataset. + * Run: npx tsx src/lab/eval/matrix-smoke.ts + */ +import assert from "node:assert/strict"; +import matrix from "./steps/matrix.js"; + +type AnyRec = Record; + +/** gaia-candidate-run shape: boolean `correct`. */ +function res(taskId: string, correct: boolean, answer: string, level = 2, extra: AnyRec = {}): AnyRec { + return { taskId, level, correct, answer, ...extra }; +} + +const ctx = { runId: "smoke", path: "matrix", scope: {}, input: {}, emit: async () => {}, services: {}, registry: {} } as never; + +async function main() { + // ── the main fixture: 4 tasks × 4 measurements of 3 versions ─────────── + // floor-1: correct everywhere; bias-1: never correct, identical "36"; + // var-1: never correct, scattered answers; flip-1: movable, flips on the + // v11 re-measurement (identical YAML — pure sampling noise). + const meas = (version: string, flip1: boolean, varAns: string) => ({ + version, + results: [ + res("floor-1", true, "right"), + res("bias-1", false, "36", 3), + res("var-1", false, varAns, 3), + res("flip-1", true && flip1, flip1 ? "ok" : "nope"), + ], + }); + const out = (await matrix.run( + { + measurements: [ + meas("gaia-produce", false, "0.00073"), + meas("v10", true, "0.00022"), + meas("v11", true, "0.0031"), + meas("v11", false, "7"), // the accidental re-measurement + ], + maxAnswerChars: 120, + }, + ctx, + )) as AnyRec; + + // 1. bands + const tasks = out.tasks as AnyRec[]; + const byId = Object.fromEntries(tasks.map((t) => [t.taskId as string, t])); + assert.equal(byId["floor-1"]!.band, "floor"); + assert.equal(byId["bias-1"]!.band, "ceiling"); + assert.equal(byId["var-1"]!.band, "ceiling"); + assert.equal(byId["flip-1"]!.band, "movable"); + assert.deepEqual(out.bands, { floor: 1, movable: 1, ceiling: 2 }); + console.log("✔ bands: floor / movable / ceiling split as observed"); + + // 2. noise floor from the v11 pair: fitness 3/4 vs 2/4 → |Δ| = 0.25, 1 flip + const noise = out.noise as AnyRec; + assert.equal(noise.sameVersionPairs, 1); + assert.equal(noise.floorKnown, true); + assert.equal(noise.maxAbsFitnessDelta, 0.25); + assert.equal(noise.maxTaskFlips, 1); + assert.equal(noise.suggestedMargin, 0.25); + console.log("✔ noise floor measured from the same-version pair (Δ0.25, 1 flip)"); + + // 3. bias vs variance on ceiling tasks + assert.equal(byId["bias-1"]!.bias, true); + assert.equal(byId["bias-1"]!.repeatedAnswer, "36"); + assert.ok(!byId["var-1"]!.bias, "distinct wrong answers must not tag as bias"); + assert.ok(((byId["var-1"]!.wrongAnswers as string[]) ?? []).length >= 3); + console.log("✔ bias (identical ×4) vs variance (distinct answers) tagged"); + + // versions aggregate + text rendering carries the actionable lines + const versions = out.versions as AnyRec[]; + assert.deepEqual(versions.map((v) => v.version), ["gaia-produce", "v10", "v11"]); + assert.deepEqual((versions[2] as AnyRec).fitness, [0.5, 0.25]); + const text = out.text as string; + assert.ok(text.includes("BIAS"), "text names the bias failure"); + assert.ok(text.includes("±0.25"), "text states the measured margin"); + console.log("✔ version rows + text rendering"); + + // 4. single measurement per version → floor UNKNOWN, never zero + const single = (await matrix.run( + { measurements: [meas("gaia-produce", true, "x"), meas("v10", false, "y")], maxAnswerChars: 120 }, + ctx, + )) as AnyRec; + const n2 = single.noise as AnyRec; + assert.equal(n2.sameVersionPairs, 0); + assert.equal(n2.floorKnown, false); + assert.equal(n2.suggestedMargin, undefined); + assert.ok((single.text as string).includes("UNKNOWN"), "text says the floor is unmeasured"); + console.log("✔ no same-version pair → noise floor UNKNOWN (not 0)"); + + // 5. correctness shape normalization: gaia-run's results-array and count + const shapes = (await matrix.run( + { + measurements: [ + { + version: "base", + results: [ + { taskId: "a", level: 1, results: [{ correct: true }], answer: "1" }, + { taskId: "b", level: 1, correct: 1, total: 1, answer: "2" }, + { taskId: "c", level: 1, correct: 0, total: 1, answer: "3" }, + { taskId: "d", level: 1, answer: "unreadable shape" }, + ], + }, + ], + maxAnswerChars: 120, + }, + ctx, + )) as AnyRec; + const sById = Object.fromEntries((shapes.tasks as AnyRec[]).map((t) => [t.taskId as string, t])); + assert.equal(sById["a"]!.band, "floor"); + assert.equal(sById["b"]!.band, "floor"); + assert.equal(sById["c"]!.band, "ceiling"); + assert.equal(sById["d"]!.band, "ceiling"); // unreadable → false, never true + console.log("✔ all three graded-result shapes normalize"); + + console.log("\nmatrix smoke: all checks passed"); +} + +main().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/mcp/src/lab/eval/seed.ts b/mcp/src/lab/eval/seed.ts index aeed346a9..dbbb9d190 100644 --- a/mcp/src/lab/eval/seed.ts +++ b/mcp/src/lab/eval/seed.ts @@ -19,6 +19,10 @@ import type { WorkspaceManager } from "vein"; * generations (EVOLVE_SPEC §5.3.3 generalized): a domain supplies its * one-generation workflow + a digest with a `fitness`; harvey-evolve and * gaia-evolve are the two instances. + * - `eval/matrix` — the task×version matrix across MEASUREMENTS: + * bands (floor/movable/ceiling), the empirical noise floor from + * same-version re-runs, and bias-vs-variance tags for never-correct + * tasks (plans/evolve-scoreboard-and-task-matrix.md, Phase 1). */ const SEED_STEPS: Array<{ file: string; type: string }> = [ @@ -26,6 +30,7 @@ const SEED_STEPS: Array<{ file: string; type: string }> = [ { file: "reflect.ts", type: "eval/reflect" }, { file: "optimize.ts", type: "eval/optimize" }, { file: "evolve-loop.ts", type: "eval/evolve-loop" }, + { file: "matrix.ts", type: "eval/matrix" }, ]; const HERE = dirname(fileURLToPath(import.meta.url)); diff --git a/mcp/src/lab/eval/steps/matrix.ts b/mcp/src/lab/eval/steps/matrix.ts new file mode 100644 index 000000000..68f268dd7 --- /dev/null +++ b/mcp/src/lab/eval/steps/matrix.ts @@ -0,0 +1,297 @@ +import { z, defineStep } from "vein"; + +/** + * The task×version MATRIX — the evolve harness's cross-measurement memory + * (plans/evolve-scoreboard-and-task-matrix.md, Phase 1). Where a digest + * summarizes ONE measurement of one version, this step folds EVERY + * measurement of every version into the per-task view that single digests + * structurally cannot show: + * + * - BANDS: floor (correct in every measurement — regression ballast, not + * signal), movable (flips between measurements — where ALL the fitness + * dynamic range lives), ceiling (never correct — unreachable by the + * approaches measured so far). + * - EMPIRICAL NOISE FLOOR: same-version re-measurements are re-runs of + * identical YAML, so their fitness deltas and per-task flips measure + * produce-sampling noise directly. `noise.maxAbsFitnessDelta` is the + * margin a challenger must clear before a comparison means anything — + * an observed number, not a hand-set param. + * - BIAS vs VARIANCE: a never-correct task whose non-empty wrong answers + * are byte-identical across ≥3 measurements is a BIAS failure — the + * approach is systematically wrong (broken data path, wrong method) and + * provably immune to redundancy/reconciliation and to prompt nudging. + * Distinct wrong answers are VARIANCE — sampling scatter, where + * redundancy helps. The two demand opposite fixes; one measurement can + * never tell them apart. + * + * GOLD DISCIPLINE (EVOLVE_SPEC §6): verdicts, the candidates' own answers, + * and produce/tool errors only — the gold never enters or leaves. + * + * Input entries are gaia-run / gaia-candidate-run style graded results (the + * same shapes gaia/digest-results normalizes; the correctness/answer/error + * normalizers are duplicated here because seeded steps are self-contained + * files). Measurements are ordered oldest → newest; a version may appear in + * any number of measurements (baseline k-samples, incumbent re-measurements, + * a challenger's first sample). + */ + +interface AnyRec { + [k: string]: unknown; +} + +function str(v: unknown): string | undefined { + return typeof v === "string" && v.trim() ? v : undefined; +} + +function num(v: unknown): number | undefined { + return typeof v === "number" && Number.isFinite(v) ? v : undefined; +} + +/** An error field may be a string or an { message } object. */ +function errStr(v: unknown): string | undefined { + if (typeof v === "string") return str(v); + if (v && typeof v === "object") return str((v as AnyRec)["message"]); + return undefined; +} + +/** Same three correctness shapes gaia/digest-results accepts: boolean + * (candidate-run), single-entry score-results array (gaia-run), 0/1 count + * with total 1 (gaia-run). Unreadable → false, never true. */ +function correctOf(r: AnyRec): boolean { + if (typeof r["correct"] === "boolean") return r["correct"] as boolean; + const arr = Array.isArray(r["results"]) ? (r["results"] as AnyRec[]) : []; + if (arr.length === 1 && typeof arr[0]?.["correct"] === "boolean") return arr[0]["correct"] as boolean; + if (typeof r["correct"] === "number" && num(r["total"]) === 1) return r["correct"] === 1; + return false; +} + +function truncate(s: string, max: number): string { + const t = s.replace(/\s+/g, " ").trim(); + return t.length > max ? t.slice(0, max) + " […]" : t; +} + +function round3(n: number): number { + return Math.round(n * 1000) / 1000; +} + +/** One task's observation in one measurement. */ +interface Obs { + version: string; + measurement: number; + correct: boolean; + answer: string; + error?: string; +} + +export default defineStep({ + type: "eval/matrix", + description: + "Fold MULTIPLE graded measurements (each: one version run over the task set) into the task×version matrix: per-task bands (floor / movable / ceiling), per-version fitness samples, an EMPIRICAL noise floor from same-version re-measurements (fitness deltas + task flips on identical YAML), and bias-vs-variance tags for never-correct tasks (byte-identical wrong answer across ≥3 measurements = bias; distinct wrong answers = variance). Gold never enters or leaves. Config: measurements (array of { version, results }, oldest → newest; results are gaia-run / gaia-candidate-run style graded outputs), maxAnswerChars? (default 120). Output: { tasks, versions, bands, noise, text }.", + input: z.object({ + measurements: z + .array( + z.object({ + version: z.string().describe("the workflow version this measurement graded (baseline runs use the base workflow name)"), + results: z.array(z.any()).describe("graded results, one per task"), + }), + ) + .min(1) + .describe("all measurements so far, oldest → newest"), + maxAnswerChars: z.number().int().positive().default(120), + }), + output: z.any(), + async run(cfg) { + // ── fold every measurement into per-task observations ──────────────── + const byTask = new Map(); + const versionOrder: string[] = []; + const byVersion = new Map[] }>(); + + cfg.measurements.forEach((m, mi) => { + if (!byVersion.has(m.version)) { + byVersion.set(m.version, { fitness: [], vectors: [] }); + versionOrder.push(m.version); + } + const vec = new Map(); + let correctCount = 0; + const entries = m.results as AnyRec[]; + for (const raw of entries) { + const r = (raw ?? {}) as AnyRec; + const taskId = str(r["taskId"]); + if (!taskId) continue; + const correct = correctOf(r); + if (correct) correctCount++; + vec.set(taskId, correct); + const rec = byTask.get(taskId) ?? { level: num(r["level"]) ?? null, obs: [] }; + rec.level ??= num(r["level"]) ?? null; + rec.obs.push({ + version: m.version, + measurement: mi, + correct, + answer: typeof r["answer"] === "string" ? (r["answer"] as string) : "", + error: errStr(r["error"]) ?? errStr(r["produceError"]) ?? errStr(r["gradeError"]), + }); + byTask.set(taskId, rec); + } + const v = byVersion.get(m.version)!; + v.fitness.push(entries.length ? round3(correctCount / entries.length) : 0); + v.vectors.push(vec); + }); + + // ── per-task rows: band, flips, bias-vs-variance ───────────────────── + const tasks = [...byTask.entries()].map(([taskId, { level, obs }]) => { + const n = obs.length; + const solved = obs.filter((o) => o.correct).length; + const band = solved === n ? "floor" : solved === 0 ? "ceiling" : "movable"; + // Flips in measurement order — how unstable this task is overall. + let flips = 0; + for (let i = 1; i < obs.length; i++) if (obs[i]!.correct !== obs[i - 1]!.correct) flips++; + const wrong = obs.filter((o) => !o.correct); + const wrongAnswers = [...new Set(wrong.map((o) => o.answer.trim()).filter(Boolean))]; + const emptyCount = wrong.filter((o) => !o.answer.trim()).length; + const errors = [...new Set(wrong.map((o) => o.error).filter((e): e is string => Boolean(e)))]; + // BIAS: never correct, one distinct non-empty wrong answer, seen ≥3×. + const bias = + band === "ceiling" && wrongAnswers.length === 1 && wrong.length - emptyCount >= 3; + return { + taskId, + level, + band, + n, + solved, + flips, + ...(wrongAnswers.length + ? { wrongAnswers: wrongAnswers.slice(0, 5).map((a) => truncate(a, cfg.maxAnswerChars)) } + : {}), + ...(emptyCount ? { emptyCount } : {}), + ...(bias ? { bias: true, repeatedAnswer: truncate(wrongAnswers[0]!, cfg.maxAnswerChars) } : {}), + ...(errors.length ? { errors: errors.slice(0, 3).map((e) => truncate(e, 200)) } : {}), + }; + }); + const bandOrder = { floor: 0, movable: 1, ceiling: 2 } as const; + tasks.sort( + (a, b) => + bandOrder[a.band as keyof typeof bandOrder] - bandOrder[b.band as keyof typeof bandOrder] || + b.solved / b.n - a.solved / a.n || + a.taskId.localeCompare(b.taskId), + ); + + // ── per-version rows ───────────────────────────────────────────────── + const versions = versionOrder.map((version) => { + const v = byVersion.get(version)!; + const mean = v.fitness.reduce((s, f) => s + f, 0) / v.fitness.length; + return { + version, + n: v.fitness.length, + fitness: v.fitness, + meanFitness: round3(mean), + minFitness: Math.min(...v.fitness), + maxFitness: Math.max(...v.fitness), + }; + }); + + // ── empirical noise floor: same-version re-measurement pairs ───────── + // Identical YAML re-run — every fitness delta and task flip between such + // a pair is pure produce-sampling noise, measured for free. + let pairs = 0; + let maxAbsFitnessDelta = 0; + let sumAbsFitnessDelta = 0; + let maxTaskFlips = 0; + for (const v of byVersion.values()) { + for (let i = 0; i < v.fitness.length; i++) { + for (let j = i + 1; j < v.fitness.length; j++) { + pairs++; + const d = Math.abs(v.fitness[i]! - v.fitness[j]!); + maxAbsFitnessDelta = Math.max(maxAbsFitnessDelta, d); + sumAbsFitnessDelta += d; + let flips = 0; + for (const [taskId, ci] of v.vectors[i]!) { + const cj = v.vectors[j]!.get(taskId); + if (cj !== undefined && cj !== ci) flips++; + } + maxTaskFlips = Math.max(maxTaskFlips, flips); + } + } + } + const noise = { + sameVersionPairs: pairs, + ...(pairs + ? { + maxAbsFitnessDelta: round3(maxAbsFitnessDelta), + meanAbsFitnessDelta: round3(sumAbsFitnessDelta / pairs), + maxTaskFlips, + } + : {}), + // The margin a fitness comparison must clear to be signal. With no + // same-version pairs there is NO measured floor — "unknown" must read + // as "re-measure something", never as zero. + floorKnown: pairs > 0, + ...(pairs ? { suggestedMargin: round3(maxAbsFitnessDelta) } : {}), + }; + + const bands = { + floor: tasks.filter((t) => t.band === "floor").length, + movable: tasks.filter((t) => t.band === "movable").length, + ceiling: tasks.filter((t) => t.band === "ceiling").length, + }; + + // ── text rendering for briefings ───────────────────────────────────── + const lines: string[] = []; + lines.push( + `TASK×VERSION MATRIX — ${tasks.length} task(s) × ${cfg.measurements.length} measurement(s) of ${versions.length} version(s).`, + ); + lines.push( + `Bands: ${bands.floor} floor (correct in every measurement — not signal), ` + + `${bands.movable} movable (the fitness dynamic range), ` + + `${bands.ceiling} ceiling (never correct by any measured approach).`, + ); + if (noise.floorKnown) { + lines.push( + `Measured noise floor (from ${pairs} same-version re-measurement pair(s)): identical YAML ` + + `re-runs differed by up to ${noise.maxAbsFitnessDelta} fitness (${maxTaskFlips} task flip(s)). ` + + `Fitness deltas within ±${noise.suggestedMargin} are NOISE — treat them as ties.`, + ); + } else { + lines.push( + `Noise floor UNKNOWN — no version has been measured twice yet. Until one is, no fitness ` + + `comparison here is trustworthy; re-measure the incumbent before believing any delta.`, + ); + } + lines.push(""); + lines.push("VERSIONS (oldest → newest):"); + for (const v of versions) { + lines.push( + `- ${v.version}: ${v.n === 1 ? `fitness ${v.fitness[0]}` : `fitness ${v.fitness.join(" / ")} (mean ${v.meanFitness})`} over ${v.n} measurement(s)`, + ); + } + const movable = tasks.filter((t) => t.band === "movable"); + if (movable.length) { + lines.push(""); + lines.push("MOVABLE tasks (all fitness movement lives here; high flip counts are sampling noise, not approach signal):"); + for (const t of movable) { + lines.push( + `- ${t.taskId}${t.level != null ? ` (L${t.level})` : ""}: correct ${t.solved}/${t.n}, ${t.flips} flip(s)` + + (t.wrongAnswers?.length ? ` — wrong answers seen: ${t.wrongAnswers.map((a) => `"${a}"`).join(", ")}` : ""), + ); + } + } + const ceiling = tasks.filter((t) => t.band === "ceiling"); + if (ceiling.length) { + lines.push(""); + lines.push("CEILING tasks (0 correct across every measurement):"); + for (const t of ceiling) { + lines.push( + `- ${t.taskId}${t.level != null ? ` (L${t.level})` : ""}: ` + + (t.bias + ? `BIAS — answered "${t.repeatedAnswer}" identically in every non-empty measurement. The approach is ` + + `systematically wrong (bad data path or method); redundancy, reconciliation, and prompt nudges cannot fix it — root-cause it.` + : t.wrongAnswers?.length + ? `VARIANCE — distinct wrong answers across measurements: ${t.wrongAnswers.map((a) => `"${a}"`).join(", ")}` + : `every measurement returned an empty answer`) + + (t.errors?.length ? ` [errors: ${t.errors.join(" | ")}]` : ""), + ); + } + } + + return { tasks, versions, bands, noise, text: lines.join("\n") }; + }, +}); diff --git a/vein/src/control-flow.test.ts b/vein/src/control-flow.test.ts index e6eb75e98..40231adc4 100644 --- a/vein/src/control-flow.test.ts +++ b/vein/src/control-flow.test.ts @@ -631,6 +631,183 @@ describe("foreach step", () => { assert.equal(result.status, "error"); assert.ok(result.error?.message.includes("body")); }); + + describe("concurrency", () => { + /** A step that records how many bodies are in flight at once. */ + function gauge() { + let active = 0; + let peak = 0; + const step_ = defineStep({ + type: "gauge", + input: z.object({ n: z.number() }), + output: z.any(), + async run(cfg) { + active++; + peak = Math.max(peak, active); + await new Promise((r) => setTimeout(r, 15)); + active--; + return cfg.n * 10; + }, + }); + return { step_, peakActive: () => peak }; + } + + it("runs up to N iterations at once, results in input order", async () => { + const { step_, peakActive } = gauge(); + const wf = flow("foreach-conc", { + input: z.object({ items: z.array(z.number()) }), + steps: [ + step("each", "foreach", { + items: "{{ input.items }}", + concurrency: 3, + body: step("g", "gauge", { n: "{{ $current }}" }), + }), + ], + }); + + const result = await runWorkflow( + wf, + { items: [1, 2, 3, 4, 5, 6, 7] }, + makeRegistry({ gauge: step_ }), + ); + assert.equal(result.status, "success"); + assert.deepEqual(result.output, [10, 20, 30, 40, 50, 60, 70]); + assert.ok(peakActive() >= 2, `expected overlap, peak was ${peakActive()}`); + assert.ok(peakActive() <= 3, `pool exceeded its bound: ${peakActive()}`); + }); + + it("caps in-flight iterations at the configured bound", async () => { + const { step_, peakActive } = gauge(); + const wf = flow("foreach-conc-cap", { + input: z.object({}), + steps: [ + step("each", "foreach", { + items: [1, 2, 3, 4, 5, 6], + concurrency: 2, + body: step("g", "gauge", { n: "{{ $current }}" }), + }), + ], + }); + + await runWorkflow(wf, {}, makeRegistry({ gauge: step_ })); + assert.ok(peakActive() <= 2, `pool exceeded its bound: ${peakActive()}`); + }); + + it("a failing iteration stops new starts and surfaces the lowest-index error", async () => { + const started: number[] = []; + const flaky = defineStep({ + type: "flaky", + input: z.object({ i: z.number() }), + output: z.any(), + async run(cfg) { + started.push(cfg.i); + await new Promise((r) => setTimeout(r, 5)); + if (cfg.i === 1 || cfg.i === 2) throw new Error(`boom at ${cfg.i}`); + return cfg.i; + }, + }); + + const wf = flow("foreach-conc-fail", { + input: z.object({}), + steps: [ + step("each", "foreach", { + items: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9], + concurrency: 3, + body: step("f", "flaky", { i: "{{ $current }}" }), + }), + ], + }); + + const result = await runWorkflow(wf, {}, makeRegistry({ flaky })); + assert.equal(result.status, "error"); + // Both 1 and 2 fail near-simultaneously; the lowest index must win. + assert.ok( + result.error?.message.includes("boom at 1"), + `expected the lowest-index error, got: ${result.error?.message}`, + ); + // The pool must not have raced through the whole list after failing. + assert.ok( + started.length < 10, + `expected no new starts after failure, but ${started.length}/10 started`, + ); + }); + + it("replays journaled iterations and runs the rest live under concurrency", async () => { + const ran: number[] = []; + const tracker = defineStep({ + type: "tracker", + input: z.object({ i: z.number() }), + output: z.any(), + async run(cfg) { + ran.push(cfg.i); + return cfg.i * 10; + }, + }); + + const wf = flow("foreach-conc-journal", { + input: z.object({}), + steps: [ + step("each", "foreach", { + items: [0, 1, 2, 3], + concurrency: 2, + body: step("t", "tracker", { i: "{{ $current }}" }), + }), + ], + }); + + const store = new MemoryRunStore(); + const result = await runWorkflow(wf, {}, makeRegistry({ tracker }), { + runId: "fc-j", + store, + journal: { + "foreach-conc-journal/each#0": 0, + "foreach-conc-journal/each#2": 20, + }, + }); + + assert.equal(result.status, "success"); + assert.deepEqual(result.output, [0, 10, 20, 30]); + assert.deepEqual(ran.sort(), [1, 3]); + const replayed = eventsOfType(store, "foreach-conc-journal", "fc-j", "step.replayed"); + assert.deepEqual(replayed.map((e) => e.iteration).sort(), [0, 2]); + }); + + it("concurrency defaults to sequential when omitted", async () => { + const { step_, peakActive } = gauge(); + const wf = flow("foreach-seq-default", { + input: z.object({}), + steps: [ + step("each", "foreach", { + items: [1, 2, 3], + body: step("g", "gauge", { n: "{{ $current }}" }), + }), + ], + }); + + const result = await runWorkflow(wf, {}, makeRegistry({ gauge: step_ })); + assert.deepEqual(result.output, [10, 20, 30]); + assert.equal(peakActive(), 1); + }); + + it("resolves concurrency from params", async () => { + const { step_, peakActive } = gauge(); + const wf = flow("foreach-conc-param", { + input: z.object({}), + params: { width: 2 }, + steps: [ + step("each", "foreach", { + items: [1, 2, 3, 4], + concurrency: "{{ params.width }}", + body: step("g", "gauge", { n: "{{ $current }}" }), + }), + ], + }); + + const result = await runWorkflow(wf, {}, makeRegistry({ gauge: step_ })); + assert.deepEqual(result.output, [10, 20, 30, 40]); + assert.ok(peakActive() >= 2 && peakActive() <= 2, `peak ${peakActive()}`); + }); + }); }); // ── DAG depends (parallel via shared dependency) ─────────────────────────── diff --git a/vein/src/createVein.test.ts b/vein/src/createVein.test.ts index cedff659f..a441decd6 100644 --- a/vein/src/createVein.test.ts +++ b/vein/src/createVein.test.ts @@ -230,6 +230,41 @@ describe("createVein", () => { assert.equal(result.status, "success"); }); + it("serves a partial summary for a run with events but no run.json", async () => { + const ws = new WorkspaceManager(tempDir); + const vein = await createVein({ + workspace: ws, + serveUi: false, + enableChat: false, + }); + // Simulate a run orphaned before finalize: events on disk, no run.json. + const ev = (over: Record) => ({ + ts: "2026-01-01T00:00:00.000Z", + runId: "9999", + path: "dead-wf", + type: "run.start", + ...over, + }); + await vein.store.append("dead-wf", "9999", ev({ input: { taskId: "t1" } }) as never); + await vein.store.append( + "dead-wf", + "9999", + ev({ type: "step.end", path: "dead-wf/first", output: { n: 1 }, ts: "2026-01-01T00:01:00.000Z" }) as never, + ); + + const res = await vein.app.request("/workflows/dead-wf/runs/9999"); + assert.equal(res.status, 200); + const body = (await res.json()) as Record; + assert.equal(body.partial, true); + assert.equal(body.status, "stale"); + assert.deepEqual(body.input, { taskId: "t1" }); + assert.deepEqual(body.steps, { first: { n: 1 } }); + + // A run with no events at all is still a 404. + const missing = await vein.app.request("/workflows/dead-wf/runs/1234"); + assert.equal(missing.status, 404); + }); + it("exposes a working /health endpoint", async () => { const vein = await createVein({ workspace: new WorkspaceManager(tempDir), diff --git a/vein/src/createVein.ts b/vein/src/createVein.ts index c1e67c4d9..4c06cf34d 100644 --- a/vein/src/createVein.ts +++ b/vein/src/createVein.ts @@ -10,7 +10,7 @@ import { z } from "zod"; import type { Flow, StepRegistry, RunEvent, RunResult } from "./core.js"; import type { RunStore } from "./store.js"; -import { FileRunStore, generateRunId } from "./store.js"; +import { FileRunStore, generateRunId, summarizeFromEvents } from "./store.js"; import type { ChatStore, ChatEvent } from "./chat-store.js"; import { FileChatStore, @@ -519,10 +519,17 @@ export async function createVein( return c.json({ error: "Run lookup requires a FileRunStore" }, 501); } const summary = await store.getRunSummary(name, runId); - if (!summary) { + if (summary) return c.json(summary); + // No run.json — in-flight, or orphaned before finalize (crash/restart). + // The event log is durable per-step, so serve a summary reconstructed + // from it (`partial: true` is the discriminator) instead of a 404: a + // 17-hour run that dies mid-generation must not cost its whole report. + const events = await store.getRunEvents(name, runId); + const partial = summarizeFromEvents(name, runId, events, liveStatus(name, runId)); + if (!partial) { return c.json({ error: `Run "${runId}" not found for workflow "${name}"` }, 404); } - return c.json(summary); + return c.json(partial); }); app.get("/workflows/:name/runs/:runId/events", async (c) => { diff --git a/vein/src/index.ts b/vein/src/index.ts index f17db004a..47c8bc8d2 100644 --- a/vein/src/index.ts +++ b/vein/src/index.ts @@ -57,10 +57,12 @@ export { // Persistence export { type RunStore, + type PartialRunSummary, FileRunStore, MemoryRunStore, generateRunId, tailJsonl, + summarizeFromEvents, } from "./store.js"; // Chat persistence (detached AI-builder background jobs) diff --git a/vein/src/runner.ts b/vein/src/runner.ts index f783815c5..f66950c5a 100644 --- a/vein/src/runner.ts +++ b/vein/src/runner.ts @@ -704,6 +704,15 @@ async function executeLoop( * * Body config is re-resolved each iteration so templates referencing * `$current` / `$index` see the right values. + * + * `concurrency: N` (default 1) runs up to N iterations at once through a + * bounded worker pool. Results stay in input order, each iteration keeps its + * own `#i` event path (so journal replay and per-iteration inspection are + * unchanged), and the cooperative checkpoint moves to just before each + * iteration STARTS — pause parks new starts while in-flight iterations + * drain, cancel stops the pool at the same boundary. The practical ceiling + * is usually the targets the body talks to (rate limits), not CPU — set it + * per call site, low. */ async function executeForeach( step: Step, @@ -742,10 +751,16 @@ async function executeForeach( throw new Error(`foreach step "${step.id}" requires a "body" step`); } - const results: unknown[] = []; + const concurrencyRaw = step.config["concurrency"] != null + ? (resolveConfig(step.config["concurrency"], scope) as number) + : 1; + const concurrency = Math.max(1, Math.floor(Number(concurrencyRaw) || 1)); + + const results: unknown[] = new Array(items.length); - for (let i = 0; i < items.length; i++) { - // Cooperative boundary: between foreach iterations (§2.1). + const runIteration = async (i: number): Promise => { + // Cooperative boundary: before each iteration STARTS (§2.1) — under a + // pool, pause parks new starts while in-flight iterations drain. await exec.controller?.checkpoint(); const iterPath = `${path}#${i}`; @@ -761,8 +776,8 @@ async function executeForeach( output, iteration: i, }); - results.push(output); - continue; + results[i] = output; + return; } const iterScope = { ...scope, $current: items[i], $index: i }; @@ -788,7 +803,44 @@ async function executeForeach( iteration: i, }); - results.push(output); + results[i] = output; + }; + + if (concurrency === 1) { + for (let i = 0; i < items.length; i++) { + await runIteration(i); + } + return results; + } + + // Bounded pool: workers pull the next index off a shared cursor. On the + // first failure no NEW iterations start; in-flight ones finish (they are + // mid-spend and aborting a body mid-step isn't supported), then the error + // that would have surfaced first sequentially — the lowest-index one — + // propagates. A cancellation always wins over an ordinary error so the run + // finalizes as cancelled, never as error (RUN_CONTROL_SPEC §3). + let cursor = 0; + const failures: { i: number; err: unknown }[] = []; + const worker = async (): Promise => { + while (failures.length === 0) { + const i = cursor++; + if (i >= items.length) return; + try { + await runIteration(i); + } catch (err) { + failures.push({ i, err }); + return; + } + } + }; + await Promise.all( + Array.from({ length: Math.min(concurrency, items.length) }, () => worker()), + ); + if (failures.length > 0) { + const cancelled = failures.find((f) => isCancelledError(f.err)); + if (cancelled) throw cancelled.err; + failures.sort((a, b) => a.i - b.i); + throw failures[0]!.err; } return results; diff --git a/vein/src/steps/core/foreach.ts b/vein/src/steps/core/foreach.ts index dd7df1847..feeb09665 100644 --- a/vein/src/steps/core/foreach.ts +++ b/vein/src/steps/core/foreach.ts @@ -16,11 +16,12 @@ const EXAMPLE = `- id: process_changes export default defineStep({ type: "foreach", - description: `Iterate over a list, running "body" once per item. Config: "items" (template expression evaluating to an array), "body" (single Step). Inside body: "$current" is the current item, "$index" is the zero-based position. Output is the array of body results, one per item, in order.\n\n${EXAMPLE}`, + description: `Iterate over a list, running "body" once per item. Config: "items" (template expression evaluating to an array), "body" (single Step), "concurrency" (optional, default 1: run up to N iterations at once through a bounded pool — results stay in input order and each iteration keeps its own #i event path; size it to what the body's targets tolerate, e.g. rate-limited sites want 1-4). Inside body: "$current" is the current item, "$index" is the zero-based position. Output is the array of body results, one per item, in order.\n\n${EXAMPLE}`, input: z.object({ items: z.any(), // template expression → array (resolved per-iteration scope) body: z.any(), // Step object maxIterations: z.number().int().positive().optional(), // optional safety cap + concurrency: z.number().int().positive().optional(), // bounded-pool width (default 1 = sequential) }), output: z.any(), async run() { diff --git a/vein/src/store.test.ts b/vein/src/store.test.ts index fd53e8866..fa0dc5068 100644 --- a/vein/src/store.test.ts +++ b/vein/src/store.test.ts @@ -5,7 +5,7 @@ import { join } from "node:path"; import { tmpdir } from "node:os"; import { randomUUID } from "node:crypto"; import type { RunEvent, RunSummary } from "./core.js"; -import { FileRunStore, MemoryRunStore } from "./store.js"; +import { FileRunStore, MemoryRunStore, summarizeFromEvents } from "./store.js"; const WF = "test-workflow"; @@ -343,3 +343,60 @@ describe("FileRunStore", () => { assert.equal(seen[0]?.type, "run.start"); }); }); + +// ── summarizeFromEvents ──────────────────────────────────────────────────── + +describe("summarizeFromEvents", () => { + const ev = (over: Partial & { type: RunEvent["type"] }): RunEvent => ({ + ts: "2026-01-01T00:00:00.000Z", + runId: "r1", + path: WF, + ...over, + }); + + it("returns null for an empty log", () => { + assert.equal(summarizeFromEvents(WF, "r1", []), null); + }); + + it("reconstructs input, top-level step outputs, and the last event", () => { + const events: RunEvent[] = [ + ev({ type: "run.start", input: { q: 1 }, ts: "2026-01-01T00:00:00.000Z" }), + ev({ type: "step.start", path: `${WF}/a` }), + ev({ type: "step.end", path: `${WF}/a`, output: { got: "a" } }), + ev({ type: "step.start", path: `${WF}/b` }), + // nested + iteration paths must NOT appear as top-level steps + ev({ type: "step.end", path: `${WF}/b/inner`, output: "nested" }), + ev({ type: "step.end", path: `${WF}/b#0`, output: "iter" }), + ev({ type: "step.end", path: `${WF}/b`, output: { got: "b" }, ts: "2026-01-01T00:05:00.000Z" }), + ]; + const s = summarizeFromEvents(WF, "r1", events); + assert.ok(s); + assert.equal(s.partial, true); + assert.equal(s.status, "stale"); + assert.equal(s.startedAt, "2026-01-01T00:00:00.000Z"); + assert.deepEqual(s.input, { q: 1 }); + assert.deepEqual(s.steps, { a: { got: "a" }, b: { got: "b" } }); + assert.equal(s.lastEventAt, "2026-01-01T00:05:00.000Z"); + assert.equal(s.lastEvent?.path, `${WF}/b`); + assert.equal(s.eventCount, events.length); + }); + + it("keeps the LATEST output for a re-run step and records the last error", () => { + const events: RunEvent[] = [ + ev({ type: "run.start", input: {} }), + ev({ type: "step.end", path: `${WF}/a`, output: "first" }), + ev({ type: "step.error", path: `${WF}/a/tool`, error: { message: "rate limited" }, ts: "t-err" }), + ev({ type: "step.replayed", path: `${WF}/a`, output: "second" }), + ]; + const s = summarizeFromEvents(WF, "r1", events); + assert.ok(s); + assert.deepEqual(s.steps, { a: "second" }); + assert.equal(s.lastError?.message, "rate limited"); + assert.equal(s.lastError?.path, `${WF}/a/tool`); + }); + + it("threads a caller-supplied live status through", () => { + const s = summarizeFromEvents(WF, "r1", [ev({ type: "run.start" })], "running"); + assert.equal(s?.status, "running"); + }); +}); diff --git a/vein/src/store.ts b/vein/src/store.ts index 400d14b7c..dd079e97a 100644 --- a/vein/src/store.ts +++ b/vein/src/store.ts @@ -122,6 +122,87 @@ export interface RunStore { finalize(workflow: string, runId: string, summary: RunSummary): Promise; } +// ── Partial summary (reconstructed from events) ──────────────────────────── + +/** + * A best-effort summary for a run with no `run.json` — in-flight, or + * orphaned by a crash/restart before `finalize` ran. Everything here is + * derived from the append-only event log, which IS durable per-step: the + * run's input from `run.start`, the latest output of every top-level step, + * and the last error seen anywhere in the tree. `partial: true` is the + * discriminator — a consumer that needs a terminal result must not treat + * this as one. + */ +export interface PartialRunSummary { + runId: string; + workflow: string; + partial: true; + /** Live state when the caller knows it ("running" / "paused"), else + * "stale" (no controller — the process that ran it is gone; resumable). */ + status: string; + startedAt?: string; + lastEventAt?: string; + eventCount: number; + input?: unknown; + /** Latest completed output per TOP-LEVEL step (path `/` with + * no deeper segment and no `#iteration`), in completion order. */ + steps: Record; + /** The last `step.error` seen at any depth — where a dead run stopped. */ + lastError?: { path: string; message: string; ts: string }; + /** The last event of any kind — how far the log got. */ + lastEvent?: { type: string; path: string; ts: string }; +} + +/** + * Reconstruct a `PartialRunSummary` from a run's event log. Pure over the + * events array so it is equally usable on a live tail, a stale run's log, + * or in tests. Returns null for an empty log (no such run). + */ +export function summarizeFromEvents( + workflow: string, + runId: string, + events: RunEvent[], + status = "stale", +): PartialRunSummary | null { + if (events.length === 0) return null; + + const prefix = `${workflow}/`; + const isTopLevelStep = (path: string): boolean => { + if (!path.startsWith(prefix)) return false; + const rest = path.slice(prefix.length); + return rest.length > 0 && !rest.includes("/") && !rest.includes("#"); + }; + + const summary: PartialRunSummary = { + runId, + workflow, + partial: true, + status, + eventCount: events.length, + steps: {}, + }; + + for (const e of events) { + if (e.type === "run.start") { + summary.startedAt ??= e.ts; + if (e.input !== undefined) summary.input = e.input; + } + if ((e.type === "step.end" || e.type === "step.replayed") && isTopLevelStep(e.path)) { + const stepId = e.path.slice(prefix.length); + delete summary.steps[stepId]; // re-insert so key order tracks completion order + summary.steps[stepId] = e.output; + } + if (e.type === "step.error") { + summary.lastError = { path: e.path, message: e.error?.message ?? "unknown", ts: e.ts }; + } + } + + const last = events[events.length - 1]!; + summary.lastEventAt = last.ts; + summary.lastEvent = { type: last.type, path: last.path, ts: last.ts }; + return summary; +} + // ── Filesystem implementation ────────────────────────────────────────────── /**