From 4e7a06a410f6d172fb37305171eb6254f05aac86 Mon Sep 17 00:00:00 2001 From: Evanfeenstra Date: Fri, 28 Aug 2026 13:41:26 -0700 Subject: [PATCH 1/5] =?UTF-8?q?vein:=20run=20control=20=E2=80=94=20cancel,?= =?UTF-8?q?=20pause/resume,=20durable=20resume=20(RUN=5FCONTROL=5FSPEC=20r?= =?UTF-8?q?ungs=201-3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Engine + server implementation of RUN_CONTROL_SPEC.md: - RunController (src/run-control.ts): cooperative checkpoint at every boundary (DAG steps, loop/foreach iterations, retry attempts, agent tool loop), tree linkage for nested runs, quiesce accounting via unit-scoped control views. - Cancel: CancelledError is a distinct outcome — run.cancelled event, status "cancelled", never the error path (no retry/onError), teardown fires, partial outputs stay journaled. executeFlow now settles every in-flight branch before finalizing (no events after the terminal). - Pause/resume (in-memory): parks every active branch at its next boundary; run.paused/run.resumed markers record the gap; runs listing reports paused; agent step checkpoints between tool calls (prepareStep). - Durable resume (journal replay): step.end outputs keyed by path replay as step.replayed (zero cost); first missing path executes live. run.start records workflowHash + params for the validity guard and faithful re-execution; failed/cancelled/stale runs resume, successful runs need from= (forced invalidation incl. transitive dependents and sequential-loop tails); torn JSONL tails skipped; run.resumed reopens SSE tails past an old terminal event. - Endpoints: POST /workflows/:name/runs/:runId/{cancel,pause,resume}. - Tree linkage threaded through every launch site: launchDetached, vein.run (parentRunId), authoring runWorkflow (meta/run-workflow passes ctx.runId), chat run_workflow (AiDeps.trackRun). Co-Authored-By: Claude Fable 5 --- vein/RUN_CONTROL_SPEC.md | 53 +- vein/package.json | 2 +- vein/src/ai/notifier.ts | 2 +- vein/src/ai/prompts.ts | 9 + vein/src/ai/tools.ts | 8 +- vein/src/authoring.ts | 30 +- vein/src/core.ts | 35 +- vein/src/createVein.ts | 279 +++++- vein/src/index.ts | 18 + vein/src/journal.ts | 233 +++++ vein/src/run-control.test.ts | 1185 +++++++++++++++++++++++ vein/src/run-control.ts | 180 ++++ vein/src/run-step.ts | 2 +- vein/src/runner.ts | 481 +++++---- vein/src/steps/core/agent.ts | 9 + vein/src/steps/lib/meta/run-workflow.ts | 7 +- vein/src/store.ts | 79 +- vein/src/workspace.ts | 18 + 18 files changed, 2372 insertions(+), 258 deletions(-) create mode 100644 vein/src/journal.ts create mode 100644 vein/src/run-control.test.ts create mode 100644 vein/src/run-control.ts diff --git a/vein/RUN_CONTROL_SPEC.md b/vein/RUN_CONTROL_SPEC.md index bcf092aa4..928e97baa 100644 --- a/vein/RUN_CONTROL_SPEC.md +++ b/vein/RUN_CONTROL_SPEC.md @@ -252,7 +252,42 @@ Optionally, the server can offer auto-resume of summary-less runs on boot (off by default — a human choosing Resume on a "stale" run is the right v1 ergonomics). -**Same runId, same artifacts.** Resume CONTINUES the original run: it +### 5.2 Resume after failure — retry from the failed step + +A run that finalized `status: "error"` (an infra hiccup: a 529 from the +provider that outlived retries, a grader subprocess OOM) is the same +journal with a terminal summary on top. The replay mechanics need +NOTHING new: the failed step has no journaled `step.end`, so plain +resume replays the completed prefix and re-executes exactly the failed +step (fresh retry budget, same onError config) and everything +downstream. A failed foreach iteration re-runs alone — completed +iterations replay by their `#i` paths. What failure-resume actually +adds is lifecycle bookkeeping: + +- **The terminal-summary guard relaxes.** Resume refuses only + SUCCESSFUL runs (nothing to resume — unless `from` below). `error` + and `cancelled` runs are resumable; on the resumed run's completion, + `store.finalize` supersedes the old summary (the log keeps the + original `run.error` + `run.resumed` marker, so history stays + honest). +- **Tail terminality.** `run.error`/`run.cancelled` are no longer + unconditionally terminal: a later `run.resumed` in the log reopens + the stream (historical tails scan ahead; live tails consult the + controllers map). Without this, the UI would freeze a resumed run's + event panel at the old failure. +- **`from`: forced invalidation (the "re-run from this step" gesture).** + Resume accepts an optional step path: that path, its transitive + dependents, and its iteration children are DROPPED from the journal + before replay, forcing re-execution even though they completed. This + covers the step that returned garbage without erroring (a judge that + produced empty criteria, a fetch that 200'd with an error page). + With `from`, even a successful run is resumable — "re-grade from + candeval onward" costs the grades, not the memo. + +UI: a failed run's view offers **Resume** (retry the failed step); a +step node's flyout offers **Re-run from here** (resume with +`from: `). Both show what will replay vs re-execute before +confirming — the journal makes that computable upfront. Resume CONTINUES the original run: it appends to the same JSONL (after a `run.resumed` marker event) and keeps the runId — critical because artifact directories are keyed by runId (`artifacts//...`): a drafted memo written before the crash is @@ -286,14 +321,17 @@ the runId (it's not dead, pause it instead); the workflow's current content hash differs from the one recorded at `run.start` (the runner must record it there — replaying outputs into a DIFFERENT DAG is undefined; power users may override with an explicit flag); or the run -already has a terminal summary. Registry drift (a custom step edited +finished successfully with no `from` invalidation (§5.2 — `error` and +`cancelled` runs ARE resumable). Registry drift (a custom step edited between crash and resume) is allowed but WARNED — steps re-executing post-resume use current code, same as any new run. -**API/UI.** `POST /workflows/:name/runs/:runId/resume` (only valid on -summary-less, controller-less runs — exactly today's "stale"). UI: the -"stale" badge becomes a Resume affordance. This retroactively gives -"stale" a purpose: it is the set of resumable runs. +**API/UI.** `POST /workflows/:name/runs/:runId/resume` with optional +`{ from: }` (§5.2). Valid on controller-less runs that are +summary-less ("stale" — crashed), `error`, or `cancelled`; a `success` +run needs `from`. UI: the "stale" badge becomes a Resume affordance. +This retroactively gives "stale" a purpose: it is (part of) the set of +resumable runs. --- @@ -341,7 +379,8 @@ summary-less, controller-less runs — exactly today's "stale"). UI: the 2. Pause/resume + agent tool-loop checkpoint + paused/quiesced surfacing. 3. Content-hash recording at `run.start` (ship early — resume needs history to exist); journal replay; `ctx.journal`; evolve-loop - iteration resume; UI resume-from-stale. + iteration resume; failure-resume + `from` invalidation (§5.2); UI + resume-from-stale, resume-failed, re-run-from-here. Each rung lands with runner tests (cancel/pause mid-DAG, mid-foreach, mid-retry; resume replay correctness incl. skip/gate reconstruction) and diff --git a/vein/package.json b/vein/package.json index 64565f2d9..85c8ed76f 100644 --- a/vein/package.json +++ b/vein/package.json @@ -22,7 +22,7 @@ "build:web": "npm --prefix web run build", "dev": "npm run build:web && tsx --env-file=.env src/server.ts", "start": "node build/server.js", - "test": "tsx --test src/expr.test.ts src/core.test.ts src/runner.test.ts src/control-flow.test.ts src/store.test.ts src/workspace.test.ts src/integration.test.ts src/services.test.ts src/cassette.test.ts src/run-step.test.ts src/createVein.test.ts src/authoring.test.ts src/ai-integration.test.ts src/chat-store.test.ts src/chat-endpoints.test.ts src/steps/registry.test.ts src/steps/core/agent.test.ts src/auth.test.ts src/secret-store.test.ts src/artifacts.test.ts src/slack.test.ts src/gdrive.test.ts src/html-extract.test.ts src/shell.test.ts" + "test": "tsx --test src/expr.test.ts src/core.test.ts src/runner.test.ts src/run-control.test.ts src/control-flow.test.ts src/store.test.ts src/workspace.test.ts src/integration.test.ts src/services.test.ts src/cassette.test.ts src/run-step.test.ts src/createVein.test.ts src/authoring.test.ts src/ai-integration.test.ts src/chat-store.test.ts src/chat-endpoints.test.ts src/steps/registry.test.ts src/steps/core/agent.test.ts src/auth.test.ts src/secret-store.test.ts src/artifacts.test.ts src/slack.test.ts src/gdrive.test.ts src/html-extract.test.ts src/shell.test.ts" }, "dependencies": { "@ai-sdk/anthropic": "3.0.92", diff --git a/vein/src/ai/notifier.ts b/vein/src/ai/notifier.ts index b75ad0f93..052025b8f 100644 --- a/vein/src/ai/notifier.ts +++ b/vein/src/ai/notifier.ts @@ -28,7 +28,7 @@ export const NOTIFICATION_PREFIX = "[run-notification]"; export interface RunNotificationInfo { workflow: string; runId: string; - status: "success" | "error"; + status: "success" | "error" | "cancelled"; durationMs?: number; output?: unknown; error?: { message: string }; diff --git a/vein/src/ai/prompts.ts b/vein/src/ai/prompts.ts index 590d3407f..b006b65b8 100644 --- a/vein/src/ai/prompts.ts +++ b/vein/src/ai/prompts.ts @@ -50,6 +50,15 @@ export interface AiDeps { promise: Promise; }) => void; }; + /** Register a chat-launched run with the host's controller registry + * (RUN_CONTROL_SPEC §2.2) so it is cancellable/pausable and listed as + * live. Optional: without it, runs are simply uncontrolled (tests, + * embedders). The returned untrack belongs in the launch's finally. */ + trackRun?: ( + workflow: string, + runId: string, + parentRunId?: string, + ) => { controller?: import("../run-control.js").RunController; untrack: () => void }; } // ── System prompt ────────────────────────────────────────────────────────── diff --git a/vein/src/ai/tools.ts b/vein/src/ai/tools.ts index 2378cf550..e58881211 100644 --- a/vein/src/ai/tools.ts +++ b/vein/src/ai/tools.ts @@ -334,13 +334,19 @@ export function buildTools(deps: AiDeps) { // can report it before the run finishes. const runId = generateRunId(); const startedAt = Date.now(); + // Register with the host's controller registry (when wired) so the + // run is cancellable/pausable and lists as live from launch. + const tracked = deps.trackRun?.(name, runId); const promise = runWorkflow(flow, coerceJsonArg(input) ?? {}, deps.registry, { runId, store: deps.store, workspace: deps.workspace, services: deps.services, params: coerceJsonArg(params) as Record | undefined, - }); + controller: tracked?.controller, + workflowHash: + (await deps.workspace.getWorkflowHash(name, version)) ?? undefined, + }).finally(() => tracked?.untrack()); // No detach seam (tests / non-chat embedders) → await as before. const detach = deps.detach; diff --git a/vein/src/authoring.ts b/vein/src/authoring.ts index c1a45478d..5a78fe983 100644 --- a/vein/src/authoring.ts +++ b/vein/src/authoring.ts @@ -388,6 +388,9 @@ export interface AuthoringCapability { input?: unknown, params?: Record, version?: string, + /** `parentRunId` = the calling step's `ctx.runId`, linking the nested + * run's controller under the launching run's (subtree control). */ + opts?: { parentRunId?: string }, ): Promise; listRuns(name: string, limit?: number): Promise; getRun(name: string, runId: string, fullEvents?: boolean): Promise; @@ -403,11 +406,16 @@ export interface AuthoringDeps extends StepPublishDeps { /** Read-only view of the deployment's secret store (NAMES only — never * values). Optional: `listSecrets` degrades gracefully when absent. */ secrets?: { list(): Promise }; - /** Register a nested run as in-flight (returns the untrack fn) so the - * server's runs listing reports it "running" rather than "stale" while - * its run.json doesn't exist yet. Optional: embedders without a live - * server need not care. */ - trackRun?: (workflow: string, runId: string) => () => void; + /** Register a nested run as in-flight, creating its RunController — + * attached to the launching run's controller when `parentRunId` is given, + * so cancelling/pausing the parent reaches this run (RUN_CONTROL_SPEC + * §2.2 tree linkage). Also drives the runs listing ("running" vs + * "stale"). Optional: embedders without a live server need not care. */ + trackRun?: ( + workflow: string, + runId: string, + parentRunId?: string, + ) => { controller?: import("./run-control.js").RunController; untrack: () => void }; } export function buildAuthoringCapability(deps: AuthoringDeps): AuthoringCapability { @@ -562,7 +570,7 @@ export function buildAuthoringCapability(deps: AuthoringDeps): AuthoringCapabili } }, - async runWorkflow(name, input, params, version) { + async runWorkflow(name, input, params, version, opts) { const gate = await notOwned(name, "runs"); if (gate) return { error: gate }; let flow; @@ -577,7 +585,10 @@ export function buildAuthoringCapability(deps: AuthoringDeps): AuthoringCapabili // invisible to the enclosing run's registry snapshot. const registry = await deps.getRegistry(); const runId = generateRunId(); - const untrack = deps.trackRun?.(flow.name, runId); + // Tree linkage: attach this nested run's controller to the launching + // run's (opts.parentRunId = the calling step's ctx.runId), so controls + // on the parent reach it (RUN_CONTROL_SPEC §2.2). + const tracked = deps.trackRun?.(flow.name, runId, opts?.parentRunId); try { return await runWorkflow(flow, coerceJsonArg(input) ?? {}, registry, { runId, @@ -585,9 +596,12 @@ export function buildAuthoringCapability(deps: AuthoringDeps): AuthoringCapabili workspace, services: deps.services, params: coerceJsonArg(params) as Record | undefined, + controller: tracked?.controller, + workflowHash: + (await workspace.getWorkflowHash(flow.name, version)) ?? undefined, }); } finally { - untrack?.(); + tracked?.untrack(); } }, diff --git a/vein/src/core.ts b/vein/src/core.ts index c6821df89..7c0d504fa 100644 --- a/vein/src/core.ts +++ b/vein/src/core.ts @@ -54,6 +54,18 @@ export interface StepContext { * runner-handled container step. Optional: absent when a step is invoked * outside the runner (e.g. unit tests). Read-only by convention. */ registry?: StepRegistry; + /** Cooperative run control (RUN_CONTROL_SPEC §6). A step with a long + * internal loop should `await ctx.control?.checkpoint()` per iteration so + * pause/cancel take effect between iterations rather than only at the + * step's boundary. Optional (absent outside the runner, like `registry`); + * ignoring it just leaves the step coarse-grained. */ + control?: import("./run-control.js").RunControl; + /** On resume: this step's own prior synthetic `step.end` outputs (keys are + * full event paths under this step's path, e.g. `wf/evolve#3`). A step + * that emits per-iteration synthetic `step.end` events can consume this to + * skip completed iterations (RUN_CONTROL_SPEC §5/§6). Absent on a fresh + * run or when there is nothing journaled under this step. */ + journal?: Record; } /** Error handling options for a step. */ @@ -124,9 +136,18 @@ export type RunEventType = | "step.error" | "step.retry" | "step.skipped" + /** A completed step's journaled output was replayed on resume — zero cost, + * no side effects re-executed. Never a fake `step.end` (honest timings). */ + | "step.replayed" | "run.start" | "run.end" - | "run.error"; + | "run.error" + /** Terminal: the run tree was cooperatively cancelled (RUN_CONTROL_SPEC §3). */ + | "run.cancelled" + /** Non-terminal markers so parked time is visible in the log (§4), and so a + * `run.resumed` after a terminal event reopens tails (§5.2). */ + | "run.paused" + | "run.resumed"; /** A single event in the run log. */ export interface RunEvent { @@ -140,12 +161,20 @@ export interface RunEvent { error?: { message: string; stack?: string }; durationMs?: number; iteration?: number; + /** Content hash of the workflow version this run executes, recorded on + * `run.start` — resume refuses to replay a journal into a DIFFERENT DAG + * (RUN_CONTROL_SPEC §5, validity guards). */ + workflowHash?: string; + /** Per-run param overrides, recorded on `run.start` so a durable resume + * re-executes steps with the SAME knob values the original run used. */ + params?: Record; + paramOverrides?: Record>; } /** Result of running a workflow. */ export interface RunResult { runId: string; - status: "success" | "error"; + status: "success" | "error" | "cancelled"; output?: unknown; error?: { message: string; stack?: string }; } @@ -157,7 +186,7 @@ export interface RunSummary { startedAt: string; finishedAt: string; durationMs: number; - status: "success" | "error"; + status: "success" | "error" | "cancelled"; input: unknown; output?: unknown; error?: { message: string; stack?: string }; diff --git a/vein/src/createVein.ts b/vein/src/createVein.ts index 1b810ed25..d89478d14 100644 --- a/vein/src/createVein.ts +++ b/vein/src/createVein.ts @@ -23,6 +23,8 @@ import { buildRegistry, readStepSourceFromDisk } from "./steps/registry.js"; import { maxOutputTokensFor } from "./pricing.js"; import type { StepSources } from "./steps/registry.js"; import { runWorkflow } from "./runner.js"; +import { RunController } from "./run-control.js"; +import { buildJournal, invalidateFrom, readRunStart } from "./journal.js"; import { requireApiKey, warnIfUnconfigured } from "./auth.js"; import { standardServices, fileArtifactsCapability } from "./capabilities.js"; import type { ArtifactsCapability } from "./capabilities.js"; @@ -165,6 +167,10 @@ export interface VeinRunOptions { runId?: string; /** Workflow version (only meaningful when `workflow` is a string). */ version?: string; + /** The launching run's id (the calling step's `ctx.runId`) — attaches this + * run's controller under the parent's, so cancel/pause on the parent + * reach it (RUN_CONTROL_SPEC §2.2 tree linkage). */ + parentRunId?: string; /** Per-event hook — useful for SSE streaming. */ onEvent?: (event: RunEvent) => void | Promise; /** Override the instance-level services for a single run. */ @@ -284,22 +290,54 @@ export async function createVein( ): Promise> { const workspace = opts.workspace ?? new WorkspaceManager(); const store = opts.store ?? new FileRunStore(workspace.path); - // Run IDs currently executing **in this process** (keyed `${workflow}/${runId}`). - // Detached execution is in-memory, so a run with no `run.json` summary is only - // genuinely "running" if it's in this set; otherwise it never finalized (server - // restart / crash / cancellation) and is reported as "stale" rather than a - // perpetual "running". - const activeRuns = new Set(); - /** Register a run as in-flight for the runs-listing status fallback (a run - * dir with events but no run.json yet is "running" if registered here, - * "stale" — i.e. orphaned by a crash/restart — otherwise). Every launch - * path must register: HTTP (launchDetached), programmatic (vein.run), and - * in-process nested runs (authoring's meta/run-workflow). Returns the - * untrack fn for the caller's finally. */ - const trackRun = (workflow: string, runId: string) => { + // Controllers for runs currently executing **in this process** (keyed + // `${workflow}/${runId}`) — RUN_CONTROL_SPEC §2.2. A controller's presence + // IS "in-flight" (superseding the old `activeRuns` set): detached execution + // is in-memory, so a run with no `run.json` summary is only genuinely live + // if it's in this map; otherwise it never finalized (server restart / + // crash) and is reported as "stale" — i.e. resumable (§5). + const controllers = new Map(); + // Secondary index for tree linkage: a nested launch names only its + // `parentRunId` (the calling step's ctx.runId), not the parent workflow. + const controllersByRunId = new Map(); + /** Register a run as in-flight, creating its controller (attached to the + * launching run's controller when `parentRunId` resolves — controls apply + * to whole subtrees). Every launch path must register: HTTP + * (launchDetached), programmatic (vein.run), in-process nested runs + * (authoring's meta/run-workflow), and chat-detached runs. The returned + * untrack fn belongs in the launcher's finally. */ + const trackRun = (workflow: string, runId: string, parentRunId?: string) => { const key = `${workflow}/${runId}`; - activeRuns.add(key); - return () => activeRuns.delete(key); + const parent = parentRunId ? controllersByRunId.get(parentRunId) : undefined; + const controller = new RunController(runId, workflow, parent); + controllers.set(key, controller); + controllersByRunId.set(runId, controller); + const untrack = () => { + controllers.delete(key); + if (controllersByRunId.get(runId) === controller) controllersByRunId.delete(runId); + controller.detach(); + }; + return { controller, untrack }; + }; + /** Listing/stream status for a summary-less run: the controller's live + * state, or "stale" (orphaned by a crash/restart — resumable). */ + const liveStatus = (workflow: string, runId: string): string => { + const controller = controllers.get(`${workflow}/${runId}`); + return controller ? controller.state : "stale"; + }; + /** Append a control marker (run.paused / run.resumed) to a run's log so + * the parked gap is visible in the record (§4 observability). */ + const appendControlEvent = async ( + workflow: string, + runId: string, + type: "run.paused" | "run.resumed", + ) => { + await store.append(workflow, runId, { + ts: new Date().toISOString(), + runId, + path: workflow, + type, + }); }; // Deployment-scoped secret store backing the `secrets` capability + the // `/secrets` admin endpoints. Mirrors the run/chat store defaults: encrypted @@ -469,8 +507,7 @@ export async function createVein( if (summary) { runs.push(summary); } else { - const status = activeRuns.has(`${name}/${runId}`) ? "running" : "stale"; - runs.push({ runId, workflow: name, status }); + runs.push({ runId, workflow: name, status: liveStatus(name, runId) }); } } return c.json(runs); @@ -509,18 +546,164 @@ export async function createVein( return streamSSE(c, async (stream) => { const ac = new AbortController(); stream.onAbort(() => ac.abort()); - for await (const event of store.tailEvents(name, runId, { signal: ac.signal })) { + for await (const event of store.tailEvents(name, runId, { + signal: ac.signal, + // A resumed run appends past its old terminal event — keep following + // while a live controller exists (§5.2 tail terminality). + stillLive: () => controllers.has(`${name}/${runId}`), + })) { await stream.writeSSE({ data: JSON.stringify(event) }); } if (ac.signal.aborted) return; const summary = await store.getRunSummary(name, runId); const result = summary ? { runId, status: summary.status, output: summary.output, error: summary.error } - : { runId, status: activeRuns.has(`${name}/${runId}`) ? "running" : "stale" }; + : { runId, status: liveStatus(name, runId) }; await stream.writeSSE({ event: "done", data: JSON.stringify(result) }); }); }); + // ── Run control (RUN_CONTROL_SPEC §3–§5) ──────────────────────────────── + // + // Cancel / pause / resume act on the LIVE controller (whole subtree via the + // effective-state walk); resume additionally covers DURABLE resume (§5): + // with no live controller, it replays the journal of a "stale" (crashed), + // `error`, or `cancelled` run — or, with `from`, re-runs a completed run + // from a chosen step. + + /** Resolve a control request target: its live controller (if any) and + * whether the run exists at all. */ + const findRun = async (name: string, runId: string) => { + const controller = controllers.get(`${name}/${runId}`) ?? null; + const summary = + store instanceof FileRunStore ? await store.getRunSummary(name, runId) : null; + const events = + store instanceof FileRunStore ? await store.getRunEvents(name, runId) : []; + return { controller, summary, exists: controller != null || events.length > 0 }; + }; + + app.post("/workflows/:name/runs/:runId/cancel", async (c) => { + const { name, runId } = c.req.param(); + const { controller, summary, exists } = await findRun(name, runId); + if (!exists) return c.json({ error: `Run "${runId}" not found` }, 404); + if (!controller) { + return c.json( + { error: summary ? `Run already terminal (${summary.status})` : "Run is not live (stale) — nothing to cancel" }, + 409, + ); + } + controller.cancel(); + return c.json({ ok: true, runId, state: controller.state }, 202); + }); + + app.post("/workflows/:name/runs/:runId/pause", async (c) => { + const { name, runId } = c.req.param(); + const { controller, summary, exists } = await findRun(name, runId); + if (!exists) return c.json({ error: `Run "${runId}" not found` }, 404); + if (!controller) { + return c.json( + { error: summary ? `Run already terminal (${summary.status})` : "Run is not live (stale) — nothing to pause" }, + 409, + ); + } + controller.pause(); + await appendControlEvent(name, runId, "run.paused"); + return c.json( + { ok: true, runId, state: controller.state, quiesced: controller.quiesced() }, + 202, + ); + }); + + app.post("/workflows/:name/runs/:runId/resume", async (c) => { + const { name, runId } = c.req.param(); + const body = await c.req + .json<{ from?: string; force?: boolean }>() + .catch(() => ({}) as { from?: string; force?: boolean }); + + const { controller, summary, exists } = await findRun(name, runId); + + // Live controller → in-memory resume of a paused run (§4). + if (controller) { + if (body.from) { + return c.json( + { error: "Run is live — `from` applies to durable resume of a dead run. Pause/cancel it first." }, + 409, + ); + } + controller.resume(); + await appendControlEvent(name, runId, "run.resumed"); + return c.json({ ok: true, runId, state: controller.state, resumed: "in-memory" }, 202); + } + + // No controller → durable resume: journal replay (§5). + if (!(store instanceof FileRunStore)) { + return c.json({ error: "Durable resume requires a FileRunStore" }, 501); + } + if (!exists) return c.json({ error: `Run "${runId}" not found` }, 404); + if (summary?.status === "success" && !body.from) { + return c.json( + { error: "Run completed successfully — nothing to resume (pass `from` to force re-execution from a step)" }, + 400, + ); + } + + const events = await store.getRunEvents(name, runId); + const runStart = readRunStart(events); + if (!runStart) { + return c.json({ error: "Run log has no run.start event — cannot resume" }, 409); + } + + let flow: Flow; + try { + flow = await workspace.getWorkflow(name); + } catch (err) { + return c.json({ error: err instanceof Error ? err.message : String(err) }, 404); + } + + // Validity guard (§5): replaying outputs into a DIFFERENT DAG is + // undefined — refuse on a content-hash mismatch unless forced. + const currentHash = await workspace.getWorkflowHash(name); + if (runStart.workflowHash && currentHash && runStart.workflowHash !== currentHash && !body.force) { + return c.json( + { + error: + `Workflow content changed since this run started (recorded ${runStart.workflowHash}, ` + + `active ${currentHash}) — replaying its journal into a different DAG is refused. ` + + `Pass { force: true } to override.`, + }, + 409, + ); + } + if (!runStart.workflowHash) { + console.warn( + `[run ${runId}] no workflow hash recorded at run.start (pre-run-control log) — resuming without the DAG guard`, + ); + } + + let journal = buildJournal(events); + if (body.from) { + try { + const inv = await invalidateFrom(journal, body.from, flow, workspace); + journal = inv.journal; + for (const w of inv.warnings) console.warn(`[run ${runId}] resume from=${body.from}: ${w}`); + } catch (err) { + return c.json({ error: err instanceof Error ? err.message : String(err) }, 400); + } + } + + launchDetached( + flow, + { + input: runStart.input, + runId, + params: runStart.params, + paramOverrides: runStart.paramOverrides, + }, + { journal, resume: true }, + ); + return c.json({ ok: true, runId, resumed: "journal", replaying: Object.keys(journal).length }, 202); + }); + // Resolve this workflow's declared `promotes` against a run's OUTPUT — the // review surface for "promote a winner". For each spec: the resolved value // from the run output (`value`) and the target param's CURRENT default @@ -968,17 +1151,29 @@ export async function createVein( * throws (e.g. a store write failure) so they don't become unhandled * rejections. */ - function launchDetached(flow: Flow, body: RunBody): string { + function launchDetached( + flow: Flow, + body: RunBody, + extra?: { journal?: Record; resume?: boolean; version?: string }, + ): string { const runId = body.runId ?? generateRunId(); - const untrack = trackRun(flow.name, runId); - void runWorkflow(flow, body.input ?? {}, registry, { - runId, - store, - workspace, - services, - params: body.params, - paramOverrides: body.paramOverrides, - }) + const { controller, untrack } = trackRun(flow.name, runId); + void (async () => { + const workflowHash = + (await workspace.getWorkflowHash(flow.name, extra?.version)) ?? undefined; + return runWorkflow(flow, body.input ?? {}, registry, { + runId, + store, + workspace, + services, + params: body.params, + paramOverrides: body.paramOverrides, + controller, + ...(workflowHash ? { workflowHash } : {}), + ...(extra?.journal ? { journal: extra.journal } : {}), + ...(extra?.resume ? { resume: true } : {}), + }); + })() .catch((err) => { console.error(`[run ${runId}] launch failed:`, err); }) @@ -1008,7 +1203,7 @@ export async function createVein( } catch (err) { return c.json({ error: err instanceof Error ? err.message : String(err) }, 404); } - const runId = launchDetached(flow, body); + const runId = launchDetached(flow, body, { version }); return c.json({ runId }, 202); }); @@ -1077,9 +1272,13 @@ export async function createVein( stepSources = bundle.sources; return bundle.registry; }, + // Controller registration for chat-launched runs — cancellable/ + // pausable like any other launch path (tracked from launch inside + // the run_workflow tool, not only on detach). + trackRun, // Dispatch-mode run_workflow: a run that outlives the wait window - // converts to detached — track it like an HTTP-launched run and - // wake this chat with a [run-notification] when it settles. + // converts to detached — wake this chat with a [run-notification] + // when it settles. detach: { waitMs: chatRunWaitMs, onDetach: ({ @@ -1093,8 +1292,6 @@ export async function createVein( startedAt: number; promise: Promise; }) => { - const key = `${workflow}/${runId}`; - activeRuns.add(key); promise .then( (res) => @@ -1128,8 +1325,7 @@ export async function createVein( ) .catch((err) => console.error(`[chat ${chatId}] run-notification delivery failed:`, err), - ) - .finally(() => activeRuns.delete(key)); + ); }, }, }; @@ -1351,8 +1547,15 @@ export async function createVein( // the run can be registered as in-flight — otherwise nested runs (e.g. the // optimizer capability's generation runs) list as "stale" while running. const runId = runOpts?.runId ?? generateRunId(); - const untrack = trackRun(flow.name, runId); + // Tree linkage (RUN_CONTROL_SPEC §2.2): `parentRunId` (the calling + // step's ctx.runId — set by e.g. the lab's optimizer capability) attaches + // this run's controller under the launching run's. + const { controller, untrack } = trackRun(flow.name, runId, runOpts?.parentRunId); try { + const workflowHash = + typeof workflow === "string" + ? ((await workspace.getWorkflowHash(workflow, runOpts?.version)) ?? undefined) + : undefined; return await runWorkflow(flow, input, registry, { runId, store, @@ -1361,6 +1564,8 @@ export async function createVein( params: runOpts?.params, paramOverrides: runOpts?.paramOverrides, onEvent: runOpts?.onEvent, + controller, + ...(workflowHash ? { workflowHash } : {}), }); } finally { untrack(); diff --git a/vein/src/index.ts b/vein/src/index.ts index 942910d83..f17db004a 100644 --- a/vein/src/index.ts +++ b/vein/src/index.ts @@ -27,6 +27,24 @@ export { // Runner export { runWorkflow, type RunOptions } from "./runner.js"; +// Run control — cancel / pause / resume for run trees (RUN_CONTROL_SPEC.md) +export { + RunController, + CancelledError, + isCancelledError, + type RunControl, + type ControlState, +} from "./run-control.js"; + +// Resume journal — replay completed step outputs from the event log +export { + buildJournal, + invalidateFrom, + readRunStart, + transitiveDependents, + type InvalidateResult, +} from "./journal.js"; + // Expression engine export { evaluateExpr, diff --git a/vein/src/journal.ts b/vein/src/journal.ts new file mode 100644 index 000000000..a11203c9e --- /dev/null +++ b/vein/src/journal.ts @@ -0,0 +1,233 @@ +/** + * Resume journal (RUN_CONTROL_SPEC §5) — the event log already contains + * everything a resume needs: every completed step's `step.end` carries its + * `output`, keyed by a deterministic `path`. This module turns a run's + * events into the `{ path → output }` journal `runWorkflow` replays from, + * and implements the `from` invalidation ("re-run from this step", §5.2). + */ + +import type { Flow, RunEvent, Step } from "./core.js"; +import type { SubflowResolver } from "./runner.js"; + +/** Completed units: last `step.end` per path wins — a resumed log may carry + * a pre-failure entry AND a post-resume re-execution for the same path. */ +export function buildJournal(events: RunEvent[]): Record { + const journal: Record = {}; + for (const event of events) { + if (event.type === "step.end") journal[event.path] = event.output; + } + return journal; +} + +/** The run's original input, params, and recorded workflow hash, from + * `run.start` — a durable resume re-invokes with exactly these. */ +export function readRunStart( + events: RunEvent[], +): { + input: unknown; + workflowHash?: string; + params?: Record; + paramOverrides?: Record>; +} | null { + const start = events.find((e) => e.type === "run.start"); + if (!start) return null; + return { + input: start.input, + workflowHash: start.workflowHash, + params: start.params, + paramOverrides: start.paramOverrides, + }; +} + +export interface InvalidateResult { + journal: Record; + /** Journal keys dropped (forced to re-execute), sorted. */ + dropped: string[]; + /** Levels where dependent-computation was impossible (dynamic subflow + * name, missing resolver) — the target subtree is still dropped, but + * same-level dependents of that container may replay stale outputs. */ + warnings: string[]; +} + +/** + * §5.2 `from`: forced invalidation. Drop `from`'s own subtree, its ancestor + * container entries (so containers re-execute and re-reach it), the later + * iterations of any enclosing `loop` (sequential — they consumed its + * output), and the transitive dependents of every step on the ancestor + * chain at its own flow level. Everything left in the journal replays. + */ +export async function invalidateFrom( + journal: Record, + from: string, + entryFlow: Flow, + resolver?: SubflowResolver, +): Promise { + const kept: Record = { ...journal }; + const dropped = new Set(); + const warnings: string[] = []; + + const dropExact = (key: string) => { + if (key in kept) { + delete kept[key]; + dropped.add(key); + } + }; + const dropSubtree = (prefix: string) => { + for (const key of Object.keys(kept)) { + if (key === prefix || key.startsWith(`${prefix}/`) || key.startsWith(`${prefix}#`)) { + delete kept[key]; + dropped.add(key); + } + } + }; + + const prefix = `${entryFlow.name}/`; + if (!from.startsWith(prefix)) { + throw new Error( + `from path "${from}" is not under workflow "${entryFlow.name}"`, + ); + } + const segments = from.slice(prefix.length).split("/"); + + // Walk the ancestor chain level by level. At each level we know the flow's + // steps, so we can compute same-level transitive dependents; descending + // requires resolving the container's child flow (subflow → its workflow; + // foreach/loop → their body, which may itself be a subflow). + let flow: Flow | null = entryFlow; + let pathPrefix = entryFlow.name; // path up to (excluding) the current segment + + for (let level = 0; level < segments.length; level++) { + const segment = segments[level]!; + const hash = segment.indexOf("#"); + const stepId = hash >= 0 ? segment.slice(0, hash) : segment; + const iteration = hash >= 0 ? Number(segment.slice(hash + 1)) : undefined; + const segmentPath = `${pathPrefix}/${segment}`; + const stepPath = `${pathPrefix}/${stepId}`; + const isLast = level === segments.length - 1; + + if (!flow) { + warnings.push( + `could not resolve the flow at "${pathPrefix}" — dependents of "${segment}" at that level may replay stale outputs`, + ); + // Still drop the target subtree + remaining ancestor entries coarsely. + dropExact(stepPath); + if (!isLast) { + dropExact(segmentPath); + pathPrefix = segmentPath; + continue; + } + dropSubtree(segmentPath); + break; + } + + const step = flow.steps.find((s) => s.id === stepId); + if (!step) { + throw new Error( + `from path "${from}": step "${stepId}" not found in workflow "${flow.name}"`, + ); + } + + // Same-level transitive dependents of this ancestor: they consumed its + // output, so they re-execute. + for (const dep of transitiveDependents(flow.steps, stepId)) { + dropSubtree(`${pathPrefix}/${dep}`); + } + + // Enclosing `loop` iterations are SEQUENTIAL: later iterations consumed + // this one's `$current`, so invalidating #i invalidates every #j > i. + // (`foreach` iterations are independent — siblings replay.) + if (iteration !== undefined && step.type === "loop") { + for (const key of Object.keys(kept)) { + const m = keyIteration(key, stepPath); + if (m !== null && m > iteration) dropSubtree(`${stepPath}#${m}`); + } + } + + if (isLast) { + // The target itself: subtree (exact + iterations + descendants). + dropSubtree(segmentPath); + if (iteration !== undefined) dropExact(stepPath); // container entry too + break; + } + + // An ancestor container: drop its own completed entries (whole-step and, + // when the path descends through an iteration, that iteration's entry) so + // it re-executes down to the target — sibling iterations stay journaled. + dropExact(stepPath); + if (iteration !== undefined) dropExact(segmentPath); + + flow = await childFlowOf(step, resolver); + pathPrefix = segmentPath; + } + + return { journal: kept, dropped: [...dropped].sort(), warnings }; +} + +/** `#i` of `key` when it is exactly `#` or a descendant of it. */ +function keyIteration(key: string, stepPath: string): number | null { + if (!key.startsWith(`${stepPath}#`)) return null; + const rest = key.slice(stepPath.length + 1); + const end = rest.search(/[/#]/); + const n = Number(end === -1 ? rest : rest.slice(0, end)); + return Number.isInteger(n) ? n : null; +} + +/** Transitive dependents of `stepId` under the runner's dependency + * semantics: explicit `depends`, or implicit previous-step. */ +export function transitiveDependents(steps: Step[], stepId: string): Set { + const depsOf = new Map(); + for (let i = 0; i < steps.length; i++) { + const s = steps[i]!; + depsOf.set( + s.id, + s.depends != null + ? Array.isArray(s.depends) + ? s.depends + : [s.depends] + : i > 0 + ? [steps[i - 1]!.id] + : [], + ); + } + const dependents = new Set(); + let grew = true; + while (grew) { + grew = false; + for (const s of steps) { + if (dependents.has(s.id) || s.id === stepId) continue; + const deps = depsOf.get(s.id)!; + if (deps.some((d) => d === stepId || dependents.has(d))) { + dependents.add(s.id); + grew = true; + } + } + } + return dependents; +} + +/** Resolve the flow a path descends INTO through `step`: a subflow's target + * workflow, or a foreach/loop whose body is a subflow. Only statically + * named targets resolve (no `{{ }}`); anything else → null (the caller + * records a warning and degrades to coarse dropping). */ +async function childFlowOf( + step: Step, + resolver?: SubflowResolver, +): Promise { + let target = step; + if (step.type === "foreach" || step.type === "loop") { + const body = step.config["body"] as Step | undefined; + if (!body || body.type !== "subflow") return null; + target = body; + } + if (target.type !== "subflow" || !resolver) return null; + const name = target.config["workflow"]; + const version = target.config["version"]; + if (typeof name !== "string" || name.includes("{{")) return null; + try { + return typeof version === "string" && !version.includes("{{") + ? await resolver.getWorkflowVersion(name, version) + : await resolver.getWorkflow(name); + } catch { + return null; + } +} diff --git a/vein/src/run-control.test.ts b/vein/src/run-control.test.ts new file mode 100644 index 000000000..a957894e2 --- /dev/null +++ b/vein/src/run-control.test.ts @@ -0,0 +1,1185 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import { mkdir, rm, writeFile, appendFile, readFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { randomUUID } from "node:crypto"; +import { z } from "zod"; + +import { flow, step, defineStep, type StepRegistry, type RunEvent } from "./core.js"; +import { runWorkflow } from "./runner.js"; +import { MemoryRunStore, FileRunStore } from "./store.js"; +import { createVein } from "./createVein.js"; +import { WorkspaceManager } from "./workspace.js"; +import { RunController, CancelledError, isCancelledError } from "./run-control.js"; +import { buildJournal, invalidateFrom, readRunStart, transitiveDependents } from "./journal.js"; + +// ── Test helpers ─────────────────────────────────────────────────────────── + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +async function waitFor(cond: () => boolean, ms = 2000): Promise { + const deadline = Date.now() + ms; + while (!cond()) { + if (Date.now() > deadline) throw new Error("waitFor timed out"); + await sleep(5); + } +} + +function deferred() { + let resolve!: (v: T) => void; + const promise = new Promise((r) => (resolve = r)); + return { promise, resolve }; +} + +/** A step that parks until released from the test, recording starts. */ +function createGateStep() { + const gates = new Map; resolve: () => void }>(); + const started: string[] = []; + const entry = (name: string) => { + let e = gates.get(name); + if (!e) { + const d = deferred(); + e = { promise: d.promise, resolve: () => d.resolve() }; + gates.set(name, e); + } + return e; + }; + const stepDef = defineStep({ + type: "gate", + input: z.object({ name: z.string() }), + output: z.any(), + async run(cfg) { + started.push(cfg.name); + await entry(cfg.name).promise; + return cfg.name; + }, + }); + return { + stepDef, + started, + release: (name: string) => entry(name).resolve(), + waitForStart: (name: string) => waitFor(() => started.includes(name)), + }; +} + +const valueStep = defineStep({ + type: "value", + input: z.object({ result: z.any() }), + output: z.any(), + async run(cfg) { + return cfg.result; + }, +}); + +function createCounterStep() { + let count = 0; + const stepDef = defineStep({ + type: "counter", + input: z.any(), + output: z.number(), + async run() { + count++; + return count; + }, + }); + return { stepDef, calls: () => count }; +} + +/** Fails the first `failCount` invocations, then succeeds. */ +function createFlakeyStep(failCount: number) { + let attempts = 0; + const stepDef = defineStep({ + type: "flakey", + input: z.any(), + output: z.any(), + async run() { + attempts++; + if (attempts <= failCount) throw new Error(`Attempt ${attempts} failed`); + return { attempts }; + }, + }); + return { stepDef, attempts: () => attempts }; +} + +function reg(extra: Record): StepRegistry { + return { value: valueStep, ...extra } as StepRegistry; +} + +function types(store: MemoryRunStore, wf: string, runId: string): string[] { + return store.getEvents(wf, runId).map((e) => e.type); +} + +// ── RunController unit tests ─────────────────────────────────────────────── + +describe("RunController", () => { + it("checkpoint resolves immediately while running", async () => { + const c = new RunController("r1", "wf"); + await c.checkpoint(); // must not hang + assert.equal(c.state, "running"); + }); + + it("cancel makes checkpoint throw CancelledError, idempotently", async () => { + const c = new RunController("r1", "wf"); + c.cancel(); + c.cancel(); + await assert.rejects(() => c.checkpoint(), (e: unknown) => isCancelledError(e)); + assert.equal(c.state, "cancelling"); + }); + + it("pause parks checkpoint; resume releases it", async () => { + const c = new RunController("r1", "wf"); + c.pause(); + let passed = false; + const p = c.checkpoint().then(() => (passed = true)); + await sleep(20); + assert.equal(passed, false); + c.resume(); + await p; + assert.equal(passed, true); + }); + + it("cancel releases a PARKED checkpoint with CancelledError", async () => { + const c = new RunController("r1", "wf"); + c.pause(); + const p = c.checkpoint(); + await sleep(10); + c.cancel(); + await assert.rejects(() => p, (e: unknown) => isCancelledError(e)); + }); + + it("effective state inherits the strictest ancestor (subtree control)", async () => { + const parent = new RunController("p", "wf"); + const child = new RunController("c", "gen", parent); + const grandchild = new RunController("g", "cand", child); + + parent.pause(); + assert.equal(grandchild.state, "paused"); // nothing busy → quiesced + let passed = false; + const p = grandchild.checkpoint().then(() => (passed = true)); + await sleep(10); + assert.equal(passed, false); + + parent.cancel(); // strictest wins; also wakes the parked waiter + await assert.rejects(() => p, (e: unknown) => isCancelledError(e)); + assert.equal(child.state, "cancelling"); + }); + + it("pausing a child does not pause the parent or a sibling", async () => { + const parent = new RunController("p", "wf"); + const a = new RunController("a", "genA", parent); + const b = new RunController("b", "genB", parent); + a.pause(); + assert.equal(a.state, "paused"); + assert.equal(parent.state, "running"); + await b.checkpoint(); // sibling unaffected + }); + + it("quiesced reflects busy units, including released parked units (forUnit)", async () => { + const c = new RunController("r1", "wf"); + assert.equal(c.quiesced(), true); + c.beginUnit(); + c.pause(); + assert.equal(c.quiesced(), false); // a unit is mid-flight + assert.equal(c.state, "pausing"); // not yet parked + + // A unit-scoped checkpoint releases the unit while parked → quiesced. + const unit = c.forUnit(); + const p = unit.checkpoint(); + await sleep(10); + assert.equal(c.quiesced(), true); + assert.equal(c.state, "paused"); + + c.resume(); + await p; + assert.equal(c.quiesced(), false); // unit re-acquired + c.endUnit(); + assert.equal(c.quiesced(), true); + }); + + it("quiesced requires every descendant to be parked", () => { + const parent = new RunController("p", "wf"); + const child = new RunController("c", "gen", parent); + child.beginUnit(); + parent.pause(); + assert.equal(parent.quiesced(), false); + child.endUnit(); + assert.equal(parent.quiesced(), true); + }); + + it("detach unlinks a completed child from the parent's quiescence", () => { + const parent = new RunController("p", "wf"); + const child = new RunController("c", "gen", parent); + child.beginUnit(); + parent.pause(); + assert.equal(parent.quiesced(), false); + child.detach(); + assert.equal(parent.quiesced(), true); + }); + + it("resume does not clear cancelling", async () => { + const c = new RunController("r1", "wf"); + c.cancel(); + c.resume(); + await assert.rejects(() => c.checkpoint(), (e: unknown) => isCancelledError(e)); + }); +}); + +// ── Rung 1: cancel ───────────────────────────────────────────────────────── + +describe("cancel", () => { + it("cancels a run at the next DAG boundary; in-flight unit completes and is journaled", async () => { + const gate = createGateStep(); + const store = new MemoryRunStore(); + const controller = new RunController("r1", "wf"); + const wf = flow("wf", { + input: z.any(), + steps: [ + step("a", "gate", { name: "a" }), + step("b", "value", { result: "never" }), + ], + }); + + const run = runWorkflow(wf, {}, reg({ gate: gate.stepDef }), { + runId: "r1", + store, + controller, + }); + + await gate.waitForStart("a"); + controller.cancel(); + gate.release("a"); // the in-flight unit completes... + + const result = await run; + assert.equal(result.status, "cancelled"); + + const evts = types(store, "wf", "r1"); + assert.ok(evts.includes("step.end")); // ...and its output was journaled + assert.ok(evts.includes("run.cancelled")); + assert.ok(!evts.includes("run.error")); + // b never started + const bEvents = store.getEvents("wf", "r1").filter((e) => e.path === "wf/b"); + assert.equal(bEvents.length, 0); + + const summary = store.getSummary("wf", "r1"); + assert.equal(summary?.status, "cancelled"); + assert.equal(summary?.error, undefined); + }); + + it("fires the onRunEnd teardown hook on cancellation", async () => { + const gate = createGateStep(); + const controller = new RunController("r1", "wf"); + const disposed: string[] = []; + const wf = flow("wf", { + input: z.any(), + steps: [step("a", "gate", { name: "a" })], + }); + const run = runWorkflow(wf, {}, reg({ gate: gate.stepDef }), { + runId: "r1", + controller, + services: { onRunEnd: async (id: string) => void disposed.push(id) }, + }); + await gate.waitForStart("a"); + controller.cancel(); + gate.release("a"); + await run; + assert.deepEqual(disposed, ["r1"]); + }); + + it("stops a foreach between iterations", async () => { + const gate = createGateStep(); + const store = new MemoryRunStore(); + const controller = new RunController("r1", "wf"); + const wf = flow("wf", { + input: z.any(), + steps: [ + step("each", "foreach", { + items: ["x", "y", "z"], + body: step("body", "gate", { name: "{{ $current }}" }), + }), + ], + }); + + const run = runWorkflow(wf, {}, reg({ gate: gate.stepDef }), { + runId: "r1", + store, + controller, + }); + + await gate.waitForStart("x"); + controller.cancel(); + gate.release("x"); + + const result = await run; + assert.equal(result.status, "cancelled"); + assert.deepEqual(gate.started, ["x"]); // y, z never started + // iteration 0 completed and journaled + const iter0 = store.getEvents("wf", "r1").find( + (e) => e.type === "step.end" && e.path === "wf/each#0", + ); + assert.ok(iter0); + }); + + it("stops retrying at the retry boundary and skips onError (cancel is not the error path)", async () => { + const flakey = createFlakeyStep(99); + const fallbackRan = { value: false }; + const fallbackStep = defineStep({ + type: "fallback", + input: z.any(), + output: z.any(), + async run() { + fallbackRan.value = true; + return "fallback"; + }, + }); + const store = new MemoryRunStore(); + const controller = new RunController("r1", "wf"); + const wf = flow("wf", { + input: z.any(), + steps: [ + step( + "shaky", + "flakey", + {}, + { + retry: { max: 5, delayMs: 30 }, + onError: step("rescue", "fallback", {}), + }, + ), + ], + }); + + const run = runWorkflow(wf, {}, reg({ flakey: flakey.stepDef, fallback: fallbackStep }), { + runId: "r1", + store, + controller, + }); + + await waitFor(() => flakey.attempts() >= 1); + controller.cancel(); + + const result = await run; + assert.equal(result.status, "cancelled"); + assert.equal(flakey.attempts(), 1); // no retry after cancel + assert.equal(fallbackRan.value, false); // onError never diverted + }); + + it("cancelling a parent controller cancels a nested run attached to it", async () => { + const gate = createGateStep(); + const parent = new RunController("parent", "outer"); + const child = new RunController("child", "inner", parent); + const wf = flow("inner", { + input: z.any(), + steps: [ + step("a", "gate", { name: "a" }), + step("b", "value", { result: "never" }), + ], + }); + const run = runWorkflow(wf, {}, reg({ gate: gate.stepDef }), { + runId: "child", + controller: child, + }); + await gate.waitForStart("a"); + parent.cancel(); + gate.release("a"); + const result = await run; + assert.equal(result.status, "cancelled"); + }); +}); + +// ── Rung 2: pause / resume (in-memory) ───────────────────────────────────── + +describe("pause/resume", () => { + it("parks between DAG steps, quiesces, and resumes to completion", async () => { + const gate = createGateStep(); + const store = new MemoryRunStore(); + const controller = new RunController("r1", "wf"); + const wf = flow("wf", { + input: z.any(), + steps: [ + step("a", "gate", { name: "a" }), + step("b", "value", { result: "done" }), + ], + }); + + const run = runWorkflow(wf, {}, reg({ gate: gate.stepDef }), { + runId: "r1", + store, + controller, + }); + + await gate.waitForStart("a"); + controller.pause(); + assert.equal(controller.quiesced(), false); // a is mid-unit + gate.release("a"); + + await waitFor(() => controller.quiesced()); + assert.equal(controller.state, "paused"); + // a completed, b has not started + const evts = store.getEvents("wf", "r1"); + assert.ok(evts.some((e) => e.type === "step.end" && e.path === "wf/a")); + assert.ok(!evts.some((e) => e.path === "wf/b")); + + controller.resume(); + const result = await run; + assert.equal(result.status, "success"); + assert.equal(result.output, "done"); + }); + + it("a step's ctx.control checkpoint parks mid-step and counts as quiesced", async () => { + const iterations: number[] = []; + const loopy = defineStep({ + type: "loopy", + input: z.any(), + output: z.any(), + async run(_cfg, ctx) { + for (let i = 0; i < 4; i++) { + await ctx.control?.checkpoint(); + iterations.push(i); + } + return iterations.length; + }, + }); + const controller = new RunController("r1", "wf"); + const wf = flow("wf", { + input: z.any(), + steps: [step("l", "loopy", {})], + }); + + const run = runWorkflow(wf, {}, reg({ loopy }), { runId: "r1", controller }); + await waitFor(() => iterations.length >= 1); + controller.pause(); + await waitFor(() => controller.quiesced()); // parked INSIDE the step + const parkedAt = iterations.length; + await sleep(30); + assert.equal(iterations.length, parkedAt); // truly parked + + controller.resume(); + const result = await run; + assert.equal(result.status, "success"); + assert.equal(iterations.length, 4); + }); + + it("pause between loop iterations", async () => { + const gate = createGateStep(); + const store = new MemoryRunStore(); + const controller = new RunController("r1", "wf"); + const wf = flow("wf", { + input: z.any(), + steps: [ + step("l", "loop", { + maxIterations: 5, + until: "{{ $current === 'g1' }}", + body: step("body", "gate", { name: "g{{ 1 + 0 }}" }), + }), + ], + }); + // Only one iteration needed: until matches after the first. + const run = runWorkflow(wf, {}, reg({ gate: gate.stepDef }), { + runId: "r1", + store, + controller, + }); + await gate.waitForStart("g1"); + controller.pause(); + gate.release("g1"); + await waitFor(() => controller.quiesced()); + controller.resume(); + const result = await run; + assert.equal(result.status, "success"); + }); +}); + +// ── Rung 3: durable resume (journal replay) ──────────────────────────────── + +describe("journal", () => { + it("buildJournal maps step.end paths to outputs, last write wins", () => { + const events = [ + { type: "step.end", path: "wf/a", output: 1 }, + { type: "step.end", path: "wf/b", output: "old" }, + { type: "step.end", path: "wf/b", output: "new" }, + { type: "step.start", path: "wf/c", input: 0 }, + ] as RunEvent[]; + const j = buildJournal(events); + assert.deepEqual(j, { "wf/a": 1, "wf/b": "new" }); + }); + + it("readRunStart recovers input, hash, and params", () => { + const events = [ + { + type: "run.start", + path: "wf", + input: { x: 1 }, + workflowHash: "abc", + params: { knob: 2 }, + }, + ] as RunEvent[]; + const rs = readRunStart(events); + assert.deepEqual(rs?.input, { x: 1 }); + assert.equal(rs?.workflowHash, "abc"); + assert.deepEqual(rs?.params, { knob: 2 }); + }); + + it("transitiveDependents follows explicit and implicit deps", () => { + const steps = [ + step("a", "value", { result: 1 }), + step("b", "value", { result: 2 }), // implicit: depends a + step("c", "value", { result: 3 }, { depends: ["a"] }), + step("d", "value", { result: 4 }, { depends: ["c"] }), + ]; + assert.deepEqual([...transitiveDependents(steps, "a")].sort(), ["b", "c", "d"]); + assert.deepEqual([...transitiveDependents(steps, "c")].sort(), ["d"]); + }); + + it("invalidateFrom drops the target, its dependents, and container entries — keeping foreach siblings", async () => { + const wf = flow("wf", { + input: z.any(), + steps: [ + step("prep", "value", { result: 1 }), + step("each", "foreach", { + items: [1, 2, 3], + body: step("body", "value", { result: "{{ $current }}" }), + }), + step("after", "value", { result: "{{ each }}" }), + ], + }); + const journal = { + "wf/prep": 1, + "wf/each": [1, 2, 3], + "wf/each#0": 1, + "wf/each#1": 2, + "wf/each#2": 3, + "wf/after": "x", + }; + const inv = await invalidateFrom(journal, "wf/each#1", wf); + assert.deepEqual(Object.keys(inv.journal).sort(), ["wf/each#0", "wf/each#2", "wf/prep"]); + assert.deepEqual(inv.dropped.sort(), ["wf/after", "wf/each", "wf/each#1"]); + }); + + it("invalidateFrom drops LATER iterations of a loop (sequential), keeping earlier ones", async () => { + const wf = flow("wf", { + input: z.any(), + steps: [ + step("l", "loop", { + maxIterations: 5, + until: "{{ $current === 3 }}", + body: step("body", "value", { result: 1 }), + }), + ], + }); + const journal = { + "wf/l": 3, + "wf/l#0": 1, + "wf/l#1": 2, + "wf/l#2": 3, + }; + const inv = await invalidateFrom(journal, "wf/l#1", wf); + assert.deepEqual(Object.keys(inv.journal), ["wf/l#0"]); + }); + + it("invalidateFrom on a plain top-level step drops it plus downstream only", async () => { + const wf = flow("wf", { + input: z.any(), + steps: [ + step("a", "value", { result: 1 }), + step("b", "value", { result: 2 }), + step("c", "value", { result: 3 }), + ], + }); + const journal = { "wf/a": 1, "wf/b": 2, "wf/c": 3 }; + const inv = await invalidateFrom(journal, "wf/b", wf); + assert.deepEqual(Object.keys(inv.journal), ["wf/a"]); + assert.deepEqual(inv.dropped.sort(), ["wf/b", "wf/c"]); + }); +}); + +describe("durable resume", () => { + it("replays completed steps and re-executes from the first incomplete one", async () => { + const counter = createCounterStep(); + const flakey = createFlakeyStep(1); + const store = new MemoryRunStore(); + const wf = flow("wf", { + input: z.object({ x: z.number() }), + steps: [ + step("one", "counter", {}), + step("two", "flakey", {}), + step("three", "value", { result: "{{ two.attempts }}" }), + ], + }); + const registry = reg({ counter: counter.stepDef, flakey: flakey.stepDef }); + + // First run: `two` fails → status error. `one` completed and journaled. + const first = await runWorkflow(wf, { x: 1 }, registry, { runId: "r1", store }); + assert.equal(first.status, "error"); + assert.equal(counter.calls(), 1); + + // Resume: replay the journal, re-execute the failed step + downstream. + const journal = buildJournal(store.getEvents("wf", "r1")); + assert.ok("wf/one" in journal); + const second = await runWorkflow(wf, { x: 1 }, registry, { + runId: "r1", + store, + journal, + resume: true, + }); + assert.equal(second.status, "success"); + assert.equal(second.output, 2); // flakey succeeded on its 2nd attempt + assert.equal(counter.calls(), 1); // `one` was REPLAYED, not re-executed + + const evts = store.getEvents("wf", "r1"); + assert.ok(evts.some((e) => e.type === "run.resumed")); + assert.ok(evts.some((e) => e.type === "step.replayed" && e.path === "wf/one")); + // history stays honest: the original failure is still in the log + assert.ok(evts.some((e) => e.type === "run.error")); + assert.ok(evts.some((e) => e.type === "run.end")); + assert.equal(store.getSummary("wf", "r1")?.status, "success"); // superseded + }); + + it("re-runs only the failed foreach iteration; completed ones replay by #i path", async () => { + let failOnce = true; + const executed: number[] = []; + const failAt2 = defineStep({ + type: "failAt2", + input: z.object({ idx: z.number() }), + output: z.any(), + async run(cfg) { + if (cfg.idx === 2 && failOnce) { + failOnce = false; + throw new Error("boom at 2"); + } + executed.push(cfg.idx); + return cfg.idx * 10; + }, + }); + const store = new MemoryRunStore(); + const wf = flow("wf", { + input: z.any(), + steps: [ + step("each", "foreach", { + items: [0, 1, 2, 3], + body: step("body", "failAt2", { idx: "{{ $index }}" }), + }), + ], + }); + const registry = reg({ failAt2 }); + + const first = await runWorkflow(wf, {}, registry, { runId: "r1", store }); + assert.equal(first.status, "error"); + assert.deepEqual(executed, [0, 1]); // 2 failed, 3 never ran + + const journal = buildJournal(store.getEvents("wf", "r1")); + const second = await runWorkflow(wf, {}, registry, { + runId: "r1", + store, + journal, + resume: true, + }); + assert.equal(second.status, "success"); + assert.deepEqual(second.output, [0, 10, 20, 30]); + assert.deepEqual(executed, [0, 1, 2, 3]); // 0,1 NOT re-executed + + const replayed = store + .getEvents("wf", "r1") + .filter((e) => e.type === "step.replayed") + .map((e) => e.path); + assert.deepEqual(replayed.sort(), ["wf/each#0", "wf/each#1"]); + }); + + it("replays completed loop iterations and re-evaluates `until` against replayed $current", async () => { + let calls = 0; + let failOnce = true; + const inc = defineStep({ + type: "inc", + input: z.any(), + output: z.number(), + async run() { + calls++; + if (calls === 3 && failOnce) { + failOnce = false; + throw new Error("boom at iteration 2"); + } + return calls; + }, + }); + const store = new MemoryRunStore(); + const wf = flow("wf", { + input: z.any(), + steps: [ + step("l", "loop", { + maxIterations: 10, + until: "{{ $current >= 4 }}", + body: step("body", "inc", {}), + }), + ], + }); + const registry = reg({ inc }); + + const first = await runWorkflow(wf, {}, registry, { runId: "r1", store }); + assert.equal(first.status, "error"); // died on iteration index 2 (call 3) + + const journal = buildJournal(store.getEvents("wf", "r1")); + const second = await runWorkflow(wf, {}, registry, { + runId: "r1", + store, + journal, + resume: true, + }); + assert.equal(second.status, "success"); + assert.equal(second.output, 4); + // iterations 0 and 1 replayed; 2 and 3 executed live (calls 3 → failed, then 4, 5... ) + const replayed = store + .getEvents("wf", "r1") + .filter((e) => e.type === "step.replayed") + .map((e) => e.path); + assert.deepEqual(replayed.sort(), ["wf/l#0", "wf/l#1"]); + }); + + it("reconstructs skip/gate logic from replayed outputs", async () => { + const flakey = createFlakeyStep(1); + const ran: string[] = []; + const record = defineStep({ + type: "record", + input: z.object({ tag: z.string() }), + output: z.string(), + async run(cfg) { + ran.push(cfg.tag); + return cfg.tag; + }, + }); + const store = new MemoryRunStore(); + const wf = flow("wf", { + input: z.any(), + steps: [ + step("check", "value", { result: false }), // the boolean gate + step("yes", "record", { tag: "yes" }, { depends: ["check"], when: true }), + step("no", "record", { tag: "no" }, { depends: ["check"], when: false }), + step("after", "flakey", {}, { depends: ["yes", "no"] }), + ], + }); + const registry = reg({ record, flakey: flakey.stepDef }); + + const first = await runWorkflow(wf, {}, registry, { runId: "r1", store }); + assert.equal(first.status, "error"); + assert.deepEqual(ran, ["no"]); + + const journal = buildJournal(store.getEvents("wf", "r1")); + const second = await runWorkflow(wf, {}, registry, { + runId: "r1", + store, + journal, + resume: true, + }); + assert.equal(second.status, "success"); + // gate + `no` replayed; `yes` skipped AGAIN (reconstructed, not run) + assert.deepEqual(ran, ["no"]); + const resumedEvents = store.getEvents("wf", "r1"); + const skips = resumedEvents.filter((e) => e.type === "step.skipped" && e.path === "wf/yes"); + assert.equal(skips.length, 2); // one per invocation + }); + + it("hands a step its own synthetic-iteration journal slice as ctx.journal", async () => { + const seenJournals: Array | undefined> = []; + let failOnce = true; + const gens = defineStep({ + type: "gens", + input: z.any(), + output: z.any(), + async run(_cfg, ctx) { + seenJournals.push(ctx.journal); + const done: number[] = []; + for (let g = 0; g < 3; g++) { + const key = `${ctx.path}#${g}`; + if (ctx.journal && key in ctx.journal) { + done.push(ctx.journal[key] as number); + continue; // completed generation — skip + } + if (g === 2 && failOnce) { + failOnce = false; + throw new Error("boom at gen 2"); + } + await ctx.emit({ + ts: new Date().toISOString(), + runId: ctx.runId, + path: key, + type: "step.end", + output: g * 100, + iteration: g, + }); + done.push(g * 100); + } + return done; + }, + }); + const store = new MemoryRunStore(); + const wf = flow("wf", { + input: z.any(), + steps: [step("evolve", "gens", {})], + }); + const registry = reg({ gens }); + + const first = await runWorkflow(wf, {}, registry, { runId: "r1", store }); + assert.equal(first.status, "error"); + assert.equal(seenJournals[0], undefined); + + const journal = buildJournal(store.getEvents("wf", "r1")); + const second = await runWorkflow(wf, {}, registry, { + runId: "r1", + store, + journal, + resume: true, + }); + assert.equal(second.status, "success"); + assert.deepEqual(second.output, [0, 100, 200]); + // resume handed the step exactly its two completed generations + assert.deepEqual(Object.keys(seenJournals[1] ?? {}).sort(), ["wf/evolve#0", "wf/evolve#1"]); + }); +}); + +// ── Crash hardening: torn tail + tail reopening ──────────────────────────── + +describe("crash hardening", () => { + it("getRunEvents skips an unparseable trailing line (torn tail)", async () => { + const dir = join(tmpdir(), `vein-test-${randomUUID()}`); + const store = new FileRunStore(dir); + await store.append("wf", "r1", { + ts: "t", + runId: "r1", + path: "wf", + type: "run.start", + input: {}, + }); + await store.append("wf", "r1", { + ts: "t", + runId: "r1", + path: "wf/a", + type: "step.end", + output: 1, + }); + // Simulate a SIGKILL mid-append: truncated JSON, no newline. + await appendFile( + join(dir, "workflows", "wf", "runs", "r1", "events.jsonl"), + '{"ts":"t","runId":"r1","path":"wf/b","type":"step.e', + "utf-8", + ); + const events = await store.getRunEvents("wf", "r1"); + assert.equal(events.length, 2); + assert.equal(events[1]!.path, "wf/a"); + await rm(dir, { recursive: true, force: true }); + }); + + it("tailEvents scans past run.error when a run.resumed follows (historical reopen)", async () => { + const dir = join(tmpdir(), `vein-test-${randomUUID()}`); + const store = new FileRunStore(dir); + const base = { ts: "t", runId: "r1" }; + const log: RunEvent[] = [ + { ...base, path: "wf", type: "run.start", input: {} }, + { ...base, path: "wf/a", type: "step.end", output: 1 }, + { ...base, path: "wf", type: "run.error", error: { message: "boom" } }, + { ...base, path: "wf", type: "run.resumed" }, + { ...base, path: "wf/a", type: "step.replayed", output: 1 }, + { ...base, path: "wf", type: "run.end", output: "ok" }, + ]; + for (const e of log) await store.append("wf", "r1", e); + + const seen: string[] = []; + for await (const e of store.tailEvents("wf", "r1", { intervalMs: 10 })) { + seen.push(e.type); + } + assert.deepEqual(seen, [ + "run.start", + "step.end", + "run.error", + "run.resumed", + "step.replayed", + "run.end", + ]); + await rm(dir, { recursive: true, force: true }); + }); + + it("tailEvents still terminates at a plain terminal event with nothing after it", async () => { + const dir = join(tmpdir(), `vein-test-${randomUUID()}`); + const store = new FileRunStore(dir); + await store.append("wf", "r1", { + ts: "t", + runId: "r1", + path: "wf", + type: "run.cancelled", + }); + const seen: string[] = []; + for await (const e of store.tailEvents("wf", "r1", { intervalMs: 10 })) { + seen.push(e.type); + } + assert.deepEqual(seen, ["run.cancelled"]); + await rm(dir, { recursive: true, force: true }); + }); +}); + +// ── HTTP endpoints: cancel / pause / resume ──────────────────────────────── + +describe("run control endpoints", () => { + async function makeServer(steps: Record) { + const dir = join(tmpdir(), `vein-test-${randomUUID()}`); + await mkdir(dir, { recursive: true }); + const workspace = new WorkspaceManager(dir); + const registry = { + value: valueStep, + ...steps, + } as StepRegistry; + const vein = await createVein({ + workspace, + registry, + serveUi: false, + enableChat: false, + }); + const store = vein.store as FileRunStore; + const cleanup = () => rm(dir, { recursive: true, force: true }); + return { vein, workspace, store, cleanup }; + } + + async function waitForSummary(store: FileRunStore, wf: string, runId: string) { + const deadline = Date.now() + 3000; + for (;;) { + const s = await store.getRunSummary(wf, runId); + if (s) return s; + if (Date.now() > deadline) throw new Error("summary never appeared"); + await sleep(10); + } + } + + it("POST cancel stops a live run tree; 404 unknown; 409 terminal", async () => { + const gate = createGateStep(); + const { vein, workspace, store, cleanup } = await makeServer({ gate: gate.stepDef }); + try { + await workspace.publishWorkflowByContent( + "cancellable", + [ + "name: cancellable", + "steps:", + " - id: a", + " type: gate", + " config: { name: a }", + " - id: b", + " type: gate", + " config: { name: b }", + ].join("\n"), + ); + + // 404 for an unknown run + const notFound = await vein.app.request("/workflows/cancellable/runs/9999/cancel", { + method: "POST", + }); + assert.equal(notFound.status, 404); + + const launch = await vein.app.request("/workflows/cancellable/run", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ input: {} }), + }); + assert.equal(launch.status, 202); + const { runId } = (await launch.json()) as { runId: string }; + + await gate.waitForStart("a"); + const cancel = await vein.app.request(`/workflows/cancellable/runs/${runId}/cancel`, { + method: "POST", + }); + assert.equal(cancel.status, 202); + gate.release("a"); + + const summary = await waitForSummary(store, "cancellable", runId); + assert.equal(summary.status, "cancelled"); + assert.ok(!gate.started.includes("b")); // subtree stopped at the boundary + + // 409 once terminal + const again = await vein.app.request(`/workflows/cancellable/runs/${runId}/cancel`, { + method: "POST", + }); + assert.equal(again.status, 409); + } finally { + await cleanup(); + } + }); + + it("POST pause parks a run (listing shows paused, log records the gap); resume releases it", async () => { + const gate = createGateStep(); + const { vein, workspace, store, cleanup } = await makeServer({ gate: gate.stepDef }); + try { + await workspace.publishWorkflowByContent( + "pausable", + [ + "name: pausable", + "steps:", + " - id: a", + " type: gate", + " config: { name: a }", + " - id: b", + " type: value", + " config: { result: done }", + ].join("\n"), + ); + + const launch = await vein.app.request("/workflows/pausable/run", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ input: {} }), + }); + const { runId } = (await launch.json()) as { runId: string }; + + await gate.waitForStart("a"); + const pause = await vein.app.request(`/workflows/pausable/runs/${runId}/pause`, { + method: "POST", + }); + assert.equal(pause.status, 202); + gate.release("a"); + + // Parked between a and b: no summary, listing reports paused. + await sleep(50); + assert.equal(await store.getRunSummary("pausable", runId), null); + const listing = await vein.app.request("/workflows/pausable/runs"); + const runs = (await listing.json()) as Array<{ runId: string; status: string }>; + assert.equal(runs.find((r) => r.runId === runId)?.status, "paused"); + + const resume = await vein.app.request(`/workflows/pausable/runs/${runId}/resume`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({}), + }); + assert.equal(resume.status, 202); + + const summary = await waitForSummary(store, "pausable", runId); + assert.equal(summary.status, "success"); + const evts = await store.getRunEvents("pausable", runId); + assert.ok(evts.some((e) => e.type === "run.paused")); + assert.ok(evts.some((e) => e.type === "run.resumed")); + } finally { + await cleanup(); + } + }); + + it("POST resume durably resumes a failed run (replay + retry) and enforces the hash guard", async () => { + const counter = createCounterStep(); + const flakey = createFlakeyStep(1); + const { vein, workspace, store, cleanup } = await makeServer({ + counter: counter.stepDef, + flakey: flakey.stepDef, + }); + try { + await workspace.publishWorkflowByContent( + "resumable", + [ + "name: resumable", + "steps:", + " - id: one", + " type: counter", + " - id: two", + " type: flakey", + ].join("\n"), + ); + + const launch = await vein.app.request("/workflows/resumable/run", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ input: {} }), + }); + const { runId } = (await launch.json()) as { runId: string }; + const failed = await waitForSummary(store, "resumable", runId); + assert.equal(failed.status, "error"); + assert.equal(counter.calls(), 1); + + // A successful run refuses resume without `from` — but first, resume the failed one. + const resume = await vein.app.request(`/workflows/resumable/runs/${runId}/resume`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({}), + }); + assert.equal(resume.status, 202); + + const deadline = Date.now() + 3000; + let summary = failed; + while (summary.status !== "success") { + if (Date.now() > deadline) throw new Error("resume never succeeded"); + await sleep(10); + summary = (await store.getRunSummary("resumable", runId))!; + } + assert.equal(counter.calls(), 1); // `one` replayed, not re-run + assert.equal(flakey.attempts(), 2); // `two` re-executed and succeeded + + // Now refuse resuming the (successful) run without `from`. + const refuse = await vein.app.request(`/workflows/resumable/runs/${runId}/resume`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({}), + }); + assert.equal(refuse.status, 400); + + // Hash guard: publish a CHANGED active version → resume with `from` is refused… + await workspace.publishWorkflowByContent( + "resumable", + [ + "name: resumable", + "steps:", + " - id: one", + " type: counter", + " - id: two", + " type: flakey", + " - id: three", + " type: value", + " config: { result: changed }", + ].join("\n"), + ); + const mismatch = await vein.app.request(`/workflows/resumable/runs/${runId}/resume`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ from: "resumable/two" }), + }); + assert.equal(mismatch.status, 409); + } finally { + await cleanup(); + } + }); + + it("POST resume with `from` re-runs a completed run from a chosen step", async () => { + const counter = createCounterStep(); + const tally = createCounterStep(); + const { vein, workspace, store, cleanup } = await makeServer({ + counter: counter.stepDef, + tally: tally.stepDef, + }); + try { + await workspace.publishWorkflowByContent( + "regrade", + [ + "name: regrade", + "steps:", + " - id: memo", + " type: counter", + " - id: grade", + " type: tally", + ].join("\n"), + ); + + const launch = await vein.app.request("/workflows/regrade/run", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ input: {} }), + }); + const { runId } = (await launch.json()) as { runId: string }; + const first = await waitForSummary(store, "regrade", runId); + assert.equal(first.status, "success"); + + // "re-grade from grade onward" — costs the grade, not the memo. + const resume = await vein.app.request(`/workflows/regrade/runs/${runId}/resume`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ from: "regrade/grade" }), + }); + assert.equal(resume.status, 202); + + await waitFor(() => tally.calls() === 2); + await waitFor(() => counter.calls() === 1); // memo replayed + const evts = await store.getRunEvents("regrade", runId); + assert.ok(evts.some((e) => e.type === "step.replayed" && e.path === "regrade/memo")); + } finally { + await cleanup(); + } + }); +}); diff --git a/vein/src/run-control.ts b/vein/src/run-control.ts new file mode 100644 index 000000000..baf1d4c38 --- /dev/null +++ b/vein/src/run-control.ts @@ -0,0 +1,180 @@ +/** + * Run control — cancel, pause/resume for run TREES (RUN_CONTROL_SPEC.md §2). + * + * One mechanism: every in-flight run has a `RunController` registered at its + * launch site. Nested launches (meta/run-workflow, the lab's optimizer + * capability) attach their controller to the launching run's, so controls + * apply to whole subtrees: cancelling an evolve run cancels its generations + * and candidates; pausing a single candidate pauses only that candidate. + * + * All control is COOPERATIVE, at boundaries: the current unit of work (an LLM + * call, a grader subprocess) completes — and is paid for, its output landing + * in the journal — and the run consults `checkpoint()` before starting the + * next unit. The runner awaits `checkpoint()` between DAG steps, loop/foreach + * iterations and retry attempts; leaf steps with long internal loops (the + * agent step's tool loop, evolve-loop's generations) opt in via + * `ctx.control`. + */ + +export type ControlState = "running" | "pausing" | "paused" | "cancelling"; + +/** Thrown out of `checkpoint()` when the run (or an ancestor) is cancelling. + * The runner treats it as a distinct outcome — `status: "cancelled"`, never + * the generic error path. Detected structurally (`isCancelledError`) rather + * than by `instanceof` because SDK stream plumbing may re-wrap errors. */ +export class CancelledError extends Error { + readonly isVeinCancelled = true; + constructor(runId: string) { + super(`Run ${runId} was cancelled`); + this.name = "CancelledError"; + } +} + +/** True when `err` is (or wraps, via `cause`) a CancelledError. */ +export function isCancelledError(err: unknown): boolean { + let e: unknown = err; + for (let depth = 0; depth < 10 && e != null && typeof e === "object"; depth++) { + const o = e as { isVeinCancelled?: unknown; name?: unknown; cause?: unknown }; + if (o.isVeinCancelled === true || o.name === "CancelledError") return true; + e = o.cause; + } + return false; +} + +/** The cooperative surface a step sees as `ctx.control` (a unit-scoped view + * of its run's `RunController` — see `RunController.forUnit`). */ +export interface RunControl { + readonly state: ControlState; + /** Resolves immediately when running; blocks while (effectively) paused; + * throws `CancelledError` when (effectively) cancelling. */ + checkpoint(): Promise; +} + +export class RunController implements RunControl { + readonly runId: string; + readonly workflow: string; + readonly parent?: RunController; + readonly children = new Set(); + + /** OWN state — the effective state additionally inherits the strictest + * ancestor (any ancestor cancelling → cancelling; else any ancestor + * pausing → pausing). */ + private own: ControlState = "running"; + /** Parked `checkpoint()` calls, woken by `poke()` to re-read state. */ + private waiters: Array<() => void> = []; + /** Units of work currently executing between boundaries (leaf step bodies). + * A unit-scoped checkpoint releases its unit while parked, so a subtree + * with every branch parked at a boundary reads `busy === 0`. */ + private busy = 0; + + constructor(runId: string, workflow: string, parent?: RunController) { + this.runId = runId; + this.workflow = workflow; + if (parent) { + this.parent = parent; + parent.children.add(this); + } + } + + /** Effective state: strictest of self + ancestors; `pausing` reads as + * `paused` once the whole subtree is parked. */ + get state(): ControlState { + const eff = this.effective(); + if (eff === "pausing" && this.quiesced()) return "paused"; + return eff; + } + + private effective(): "running" | "pausing" | "cancelling" { + let pausing = false; + for (let c: RunController | undefined = this; c; c = c.parent) { + if (c.own === "cancelling") return "cancelling"; + if (c.own === "pausing" || c.own === "paused") pausing = true; + } + return pausing ? "pausing" : "running"; + } + + /** The cooperative checkpoint (RUN_CONTROL_SPEC §2.2). */ + async checkpoint(): Promise { + for (;;) { + const eff = this.effective(); + if (eff === "cancelling") throw new CancelledError(this.runId); + if (eff === "running") return; + // pausing/paused → park until a control call pokes us to re-check. + await new Promise((resolve) => { + this.waiters.push(resolve); + }); + } + } + + /** True when this run AND all descendants are parked at a boundary — + * nothing is mid-unit, so a restart loses no in-flight work. */ + quiesced(): boolean { + if (this.busy > 0) return false; + for (const child of this.children) if (!child.quiesced()) return false; + return true; + } + + /** Idempotent; applies to the whole subtree via the effective-state walk. */ + cancel(): void { + this.own = "cancelling"; + this.poke(); + } + + pause(): void { + if (this.own === "running") this.own = "pausing"; + this.poke(); + } + + resume(): void { + if (this.own !== "cancelling") this.own = "running"; + this.poke(); + } + + /** Mark a unit of work (a leaf step body) as executing. Callers MUST pair + * with `endUnit` in a finally. Kept adjacent to a passed `checkpoint()` + * (no interleaving await) so pause can never observe a false quiesce + * between the two. */ + beginUnit(): void { + this.busy++; + } + + endUnit(): void { + this.busy--; + } + + /** A unit-scoped view for `ctx.control`: its `checkpoint()` releases the + * enclosing unit while parked (so an agent step paused between tool calls + * counts as quiesced) and re-acquires it before continuing. */ + forUnit(): RunControl { + const controller = this; + return { + get state() { + return controller.state; + }, + async checkpoint() { + controller.endUnit(); + try { + await controller.checkpoint(); + } finally { + controller.beginUnit(); + } + }, + }; + } + + /** Unlink from the parent on unregister so a completed nested run stops + * counting toward the parent's quiescence. */ + detach(): void { + this.parent?.children.delete(this); + } + + /** Wake every parked checkpoint in the subtree to re-evaluate the + * effective state (resume releases them; cancel makes them throw; a + * waiter still effectively paused re-parks). */ + private poke(): void { + const woken = this.waiters; + this.waiters = []; + for (const wake of woken) wake(); + for (const child of this.children) child.poke(); + } +} diff --git a/vein/src/run-step.ts b/vein/src/run-step.ts index 8d85d2a09..2ab52734c 100644 --- a/vein/src/run-step.ts +++ b/vein/src/run-step.ts @@ -39,7 +39,7 @@ export interface RunStepOptions { } export interface RunStepResult { - status: "success" | "error"; + status: "success" | "error" | "cancelled"; output?: unknown; error?: { message: string; stack?: string }; /** Every event the step emitted (start/end/error, plus nested for containers). */ diff --git a/vein/src/runner.ts b/vein/src/runner.ts index ebf05416b..f783815c5 100644 --- a/vein/src/runner.ts +++ b/vein/src/runner.ts @@ -9,6 +9,7 @@ import type { import { resolveConfig } from "./expr.js"; import type { RunStore } from "./store.js"; import { MemoryRunStore, generateRunId } from "./store.js"; +import { RunController, isCancelledError } from "./run-control.js"; // ── Runner ───────────────────────────────────────────────────────────────── @@ -47,6 +48,24 @@ export interface RunOptions { * `params` defaults. Precedence: step `.default()` < flow `params` default * < `paramOverrides[name]` < (entry only) `params`. */ paramOverrides?: Record>; + /** Cooperative run control (RUN_CONTROL_SPEC §2.2). Registered at the + * launch site; the runner awaits `checkpoint()` at every boundary (between + * DAG steps, loop/foreach iterations, retry attempts) and exposes a + * unit-scoped view to steps as `ctx.control`. Absent → uncontrolled run + * (unit tests, bare embedders), zero overhead. */ + controller?: RunController; + /** Resume journal (RUN_CONTROL_SPEC §5): completed step outputs keyed by + * event path. A step whose path is journaled REPLAYS its output (emitting + * `step.replayed`) instead of executing; the first path not in the journal + * executes live and everything downstream follows. */ + journal?: Record; + /** True when this invocation CONTINUES an interrupted run (§5.2): emits a + * `run.resumed` marker instead of a fresh `run.start`, appending to the + * same log under the same runId. */ + resume?: boolean; + /** Content hash of the workflow version being run, recorded on `run.start` + * so resume can refuse to replay a journal into a different DAG (§5). */ + workflowHash?: string; } /** Sentinel returned by steps that were skipped because their `when` didn't match. */ @@ -56,6 +75,19 @@ function isSkipped(v: unknown): boolean { return v === SKIP; } +/** Everything the execution tree threads through unchanged — bundled so the + * recursive executors don't each grow another positional parameter. */ +interface Exec { + registry: StepRegistry; + runId: string; + emit: EmitFn; + workspace: SubflowResolver | undefined; + services: unknown; + paramOverrides?: Record>; + controller?: RunController; + journal?: Record; +} + export async function runWorkflow( workflow: Flow, input: unknown, @@ -105,21 +137,35 @@ export async function runWorkflow( return { runId, status: "error", error }; } - await emit({ type: "run.start", path: wfName, input: parsedInput }); + if (opts?.resume) { + // Continuing an interrupted run: same runId, same log — the marker both + // records the gap and reopens tails past an earlier terminal event. + await emit({ type: "run.resumed", path: wfName }); + } else { + await emit({ + type: "run.start", + path: wfName, + input: parsedInput, + ...(opts?.workflowHash ? { workflowHash: opts.workflowHash } : {}), + // Recorded so a durable resume re-executes with the same knob values. + ...(opts?.params ? { params: opts.params } : {}), + ...(opts?.paramOverrides ? { paramOverrides: opts.paramOverrides } : {}), + }); + } + + const exec: Exec = { + registry, + runId, + emit, + workspace: opts?.workspace, + services, + paramOverrides: opts?.paramOverrides, + controller: opts?.controller, + journal: opts?.journal, + }; try { - const output = await executeFlow( - workflow, - parsedInput, - registry, - runId, - wfName, - emit, - opts?.workspace, - services, - opts?.params, - opts?.paramOverrides, - ); + const output = await executeFlow(workflow, parsedInput, exec, wfName, opts?.params); const finishedAt = new Date().toISOString(); await emit({ type: "run.end", path: wfName, output }); @@ -135,23 +181,31 @@ export async function runWorkflow( }); return { runId, status: "success", output }; } catch (err) { + const cancelled = isCancelledError(err); const error = { message: err instanceof Error ? err.message : String(err), stack: err instanceof Error ? err.stack : undefined, }; const finishedAt = new Date().toISOString(); - await emit({ type: "run.error", path: wfName, error }); + // Cancellation is a DISTINCT outcome, never conflated with error + // (RUN_CONTROL_SPEC §3): the run finalizes honestly as `cancelled`, its + // partial outputs stay inspectable in the log, and it is never "stale". + await emit( + cancelled + ? { type: "run.cancelled", path: wfName } + : { type: "run.error", path: wfName, error }, + ); await store.finalize(wfName, runId, { runId, workflow: wfName, startedAt, finishedAt, durationMs: Date.parse(finishedAt) - Date.parse(startedAt), - status: "error", + status: cancelled ? "cancelled" : "error", input: parsedInput, - error, + ...(cancelled ? {} : { error }), }); - return { runId, status: "error", error }; + return cancelled ? { runId, status: "cancelled" } : { runId, status: "error", error }; } finally { // Generic per-run teardown hook. A consumer's services bag may implement // `onRunEnd(runId)` to dispose any per-run resources it allocated during the @@ -192,14 +246,9 @@ function getDeps(step: Step, index: number, steps: Step[]): string[] { async function executeFlow( workflow: Flow, input: unknown, - registry: StepRegistry, - runId: string, + exec: Exec, basePath: string, - emit: EmitFn, - workspace: SubflowResolver | undefined, - services: unknown, paramsOverride?: Record, - paramOverrides?: Record>, ): Promise { // `params` = workflow defaults, shallow-merged with per-run overrides. // Exposed to step configs via `{{ params.* }}`. Distinct from `input`. @@ -208,7 +257,7 @@ async function executeFlow( // Precedence: flow defaults < keyed override < entry flat override. const params = { ...(workflow.params ?? {}), - ...(paramOverrides?.[workflow.name] ?? {}), + ...(exec.paramOverrides?.[workflow.name] ?? {}), ...(paramsOverride ?? {}), }; const scope: Record = { input, params }; @@ -227,84 +276,119 @@ async function executeFlow( // Track completion const completed = new Set(); - // Each pending promise resolves with the step's output (or SKIP if skipped) - const pending = new Map void; promise: Promise }>(); + // Each pending promise resolves with the step's output (or SKIP if skipped), + // or REJECTS when the step failed — so dependents settle (fail fast) instead + // of hanging, letting the flow wait for every in-flight branch before it + // reports its outcome (required for honest cancel/error finalization: no + // step events land after the terminal event). + const pending = new Map< + string, + { resolve: (v: unknown) => void; reject: (e: unknown) => void; promise: Promise } + >(); for (const s of steps) { let resolve!: (v: unknown) => void; - const promise = new Promise((r) => { resolve = r; }); - pending.set(s.id, { resolve, promise }); + let reject!: (e: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + // A rejected step promise with no dependents must not surface as an + // unhandled rejection — the error still propagates via runStep's throw. + promise.catch(() => {}); + pending.set(s.id, { resolve, reject, promise }); } // Execute a single step once its deps are met async function runStep(s: Step): Promise { - // Wait for all dependencies, collecting their outputs - const deps = depMap.get(s.id) ?? []; - const depOutputs = await Promise.all( - deps.map((d) => pending.get(d)?.promise ?? Promise.resolve(undefined)), - ); + try { + // Wait for all dependencies, collecting their outputs + const deps = depMap.get(s.id) ?? []; + const depOutputs = await Promise.all( + deps.map((d) => pending.get(d)?.promise ?? Promise.resolve(undefined)), + ); - // Skip propagation: if this step has deps and ALL of them were skipped, - // skip this step too (no useful inputs available). - // A step with at least one real dep still runs — fan-in pattern. - // Steps with `when` use gate logic below regardless. - const hasGate = s.when != null; - if (deps.length > 0 && !hasGate) { - const allSkipped = depOutputs.every(isSkipped); - if (allSkipped) { - scope[s.id] = undefined; - await emit({ - type: "step.skipped", - path: `${basePath}/${s.id}`, - stepType: s.type, - }); - pending.get(s.id)!.resolve(SKIP); - return; + // Skip propagation: if this step has deps and ALL of them were skipped, + // skip this step too (no useful inputs available). + // A step with at least one real dep still runs — fan-in pattern. + // Steps with `when` use gate logic below regardless. + const hasGate = s.when != null; + if (deps.length > 0 && !hasGate) { + const allSkipped = depOutputs.every(isSkipped); + if (allSkipped) { + scope[s.id] = undefined; + await exec.emit({ + type: "step.skipped", + path: `${basePath}/${s.id}`, + stepType: s.type, + }); + pending.get(s.id)!.resolve(SKIP); + return; + } } - } - // Gate check: if `when` is set, find the gate dependency (the one whose - // boolean output matters). The gate is the dep whose output is boolean — - // but more precisely, we check ALL non-skipped deps for a boolean that - // matches `when`. If any boolean dep doesn't match, skip. - if (hasGate) { - // Find boolean deps (these are gates) and verify at least one matches. - // A skipped dep can never satisfy a gate (the gate didn't run). - let matched = false; - let sawGate = false; - for (let i = 0; i < deps.length; i++) { - const out = depOutputs[i]; - if (isSkipped(out)) continue; - if (typeof out === "boolean") { - sawGate = true; - if (out === s.when) { - matched = true; - break; + // Gate check: if `when` is set, find the gate dependency (the one whose + // boolean output matters). The gate is the dep whose output is boolean — + // but more precisely, we check ALL non-skipped deps for a boolean that + // matches `when`. If any boolean dep doesn't match, skip. + if (hasGate) { + // Find boolean deps (these are gates) and verify at least one matches. + // A skipped dep can never satisfy a gate (the gate didn't run). + let matched = false; + let sawGate = false; + for (let i = 0; i < deps.length; i++) { + const out = depOutputs[i]; + if (isSkipped(out)) continue; + if (typeof out === "boolean") { + sawGate = true; + if (out === s.when) { + matched = true; + break; + } } } + if (!sawGate || !matched) { + // Either no gate ran, or its value doesn't match `when` → skip + scope[s.id] = undefined; + await exec.emit({ + type: "step.skipped", + path: `${basePath}/${s.id}`, + stepType: s.type, + }); + pending.get(s.id)!.resolve(SKIP); + return; + } } - if (!sawGate || !matched) { - // Either no gate ran, or its value doesn't match `when` → skip - scope[s.id] = undefined; - await emit({ - type: "step.skipped", - path: `${basePath}/${s.id}`, - stepType: s.type, - }); - pending.get(s.id)!.resolve(SKIP); - return; - } - } - const stepPath = `${basePath}/${s.id}`; - const output = await executeStep(s, scope, registry, runId, stepPath, emit, workspace, services, paramOverrides); - scope[s.id] = output; - completed.add(s.id); - pending.get(s.id)!.resolve(output); + // Cooperative boundary: between DAG steps (RUN_CONTROL_SPEC §2.1). + // Blocks while paused; throws CancelledError while cancelling. + await exec.controller?.checkpoint(); + + const stepPath = `${basePath}/${s.id}`; + const output = await executeStep(s, scope, exec, stepPath); + scope[s.id] = output; + completed.add(s.id); + pending.get(s.id)!.resolve(output); + } catch (err) { + pending.get(s.id)!.reject(err); + throw err; + } } - // Launch all steps — each waits for its own deps internally - await Promise.all(steps.map((s) => runStep(s))); + // Launch all steps — each waits for its own deps internally. allSettled (not + // all) so a failing/cancelled branch doesn't abandon still-executing + // branches mid-unit: every branch settles (its in-flight unit completes and + // journals) before the flow reports its outcome. + const settled = await Promise.allSettled(steps.map((s) => runStep(s))); + const rejections = settled.filter( + (r): r is PromiseRejectedResult => r.status === "rejected", + ); + if (rejections.length > 0) { + // Prefer a REAL failure over a CancelledError so a genuine error isn't + // masked when cancellation raced in behind it. + const real = rejections.find((r) => !isCancelledError(r.reason)); + throw (real ?? rejections[0]!).reason; + } // Return last step's output (by array order). If skipped, return undefined. const lastOut = scope[steps[steps.length - 1]!.id]; @@ -316,17 +400,40 @@ async function executeFlow( // Control flow steps that manage their own template resolution. const SELF_RESOLVING_STEPS = new Set(["loop", "foreach", "subflow"]); +const hasOwn = (obj: Record, key: string) => + Object.prototype.hasOwnProperty.call(obj, key); + +/** The slice of the resume journal that belongs to one step's own synthetic + * iteration events (`#…`) — what a step sees as `ctx.journal`. */ +function sliceJournal( + journal: Record | undefined, + path: string, +): Record | undefined { + if (!journal) return undefined; + const prefix = `${path}#`; + let out: Record | undefined; + for (const key of Object.keys(journal)) { + if (key.startsWith(prefix)) (out ??= {})[key] = journal[key]; + } + return out; +} + async function executeStep( step: Step, scope: Record, - registry: StepRegistry, - runId: string, + exec: Exec, path: string, - emit: EmitFn, - workspace: SubflowResolver | undefined, - services: unknown, - paramOverrides?: Record>, ): Promise { + // Resume replay (RUN_CONTROL_SPEC §5): a step whose path has a journaled + // output replays it — zero cost, no side effects re-executed. Emitted as + // `step.replayed`, never a fake `step.end` (honest timings). The first path + // NOT in the journal executes live below. + if (exec.journal && hasOwn(exec.journal, path)) { + const output = exec.journal[path]; + await exec.emit({ type: "step.replayed", path, stepType: step.type, output }); + return output; + } + const maxRetries = step.options?.retry?.max ?? 0; const retryDelay = step.options?.retry?.delayMs ?? 0; @@ -335,13 +442,15 @@ async function executeStep( for (let attempt = 0; attempt <= maxRetries; attempt++) { try { if (attempt > 0) { - await emit({ + await exec.emit({ type: "step.retry", path, stepType: step.type, iteration: attempt, }); await sleep(retryDelay); + // Cooperative boundary: between retry attempts (§2.1). + await exec.controller?.checkpoint(); } const startTime = Date.now(); @@ -366,7 +475,7 @@ async function executeStep( : undefined : resolvedConfig; - await emit({ + await exec.emit({ type: "step.start", path, stepType: step.type, @@ -374,22 +483,11 @@ async function executeStep( }); // Execute based on step type - const output = await dispatchStep( - step, - resolvedConfig, - scope, - registry, - runId, - path, - emit, - workspace, - services, - paramOverrides, - ); + const output = await dispatchStep(step, resolvedConfig, scope, exec, path); const durationMs = Date.now() - startTime; - await emit({ + await exec.emit({ type: "step.end", path, stepType: step.type, @@ -399,6 +497,10 @@ async function executeStep( return output; } catch (err) { + // Cancellation is NOT the error path (§3): never retried, never + // diverted into onError — the branch stops at this boundary. + if (isCancelledError(err)) throw err; + lastError = err instanceof Error ? err : new Error(String(err)); if (attempt === maxRetries) { @@ -413,18 +515,13 @@ async function executeStep( const fallbackOutput = await executeStep( step.options.onError, errorScope, - registry, - runId, + exec, `${path}/onError`, - emit, - workspace, - services, - paramOverrides, ); return fallbackOutput; } catch (fallbackErr) { // Fallback itself failed - await emit({ + await exec.emit({ type: "step.error", path, stepType: step.type, @@ -437,7 +534,7 @@ async function executeStep( } } - await emit({ + await exec.emit({ type: "step.error", path, stepType: step.type, @@ -461,28 +558,23 @@ async function dispatchStep( step: Step, resolvedConfig: Record, scope: Record, - registry: StepRegistry, - runId: string, + exec: Exec, path: string, - emit: EmitFn, - workspace: SubflowResolver | undefined, - services: unknown, - paramOverrides?: Record>, ): Promise { // Handle core control flow steps specially switch (step.type) { case "loop": - return executeLoop(step, resolvedConfig, scope, registry, runId, path, emit, workspace, services, paramOverrides); + return executeLoop(step, scope, exec, path); case "foreach": - return executeForeach(step, scope, registry, runId, path, emit, workspace, services, paramOverrides); + return executeForeach(step, scope, exec, path); case "subflow": - return executeSubflow(step, scope, registry, runId, path, emit, workspace, services, paramOverrides); + return executeSubflow(step, scope, exec, path); default: { // Look up in registry - const def = registry[step.type]; + const def = exec.registry[step.type]; if (!def) { throw new Error(`Unknown step type: "${step.type}"`); } @@ -491,17 +583,32 @@ async function dispatchStep( // Default to {} when no config is provided in the YAML. const validConfig = def.input.parse(resolvedConfig ?? {}); + const stepJournal = sliceJournal(exec.journal, path); const ctx: StepContext = { - runId, + runId: exec.runId, path, scope, input: scope["input"], - emit, - services, - registry, + emit: exec.emit, + services: exec.services, + registry: exec.registry, + // Unit-scoped control: a parked `ctx.control.checkpoint()` releases + // this unit so the subtree can quiesce (§2.2 threading). + ...(exec.controller ? { control: exec.controller.forUnit() } : {}), + ...(stepJournal ? { journal: stepJournal } : {}), }; - return def.run(validConfig, ctx); + if (exec.controller) { + // checkpoint → beginUnit with no interleaving await, so a pause can + // never observe a false quiesce between the two. + await exec.controller.checkpoint(); + exec.controller.beginUnit(); + } + try { + return await def.run(validConfig, ctx); + } finally { + exec.controller?.endUnit(); + } } } } @@ -510,15 +617,9 @@ async function dispatchStep( async function executeLoop( step: Step, - _config: Record, scope: Record, - registry: StepRegistry, - runId: string, + exec: Exec, path: string, - emit: EmitFn, - workspace: SubflowResolver | undefined, - services: unknown, - paramOverrides?: Record>, ): Promise { // Resolve scalar config values (but not `until` or `body` which need per-iteration resolution) const maxIterations = resolveConfig(step.config["maxIterations"], scope) as number; @@ -530,37 +631,48 @@ async function executeLoop( let current: unknown = undefined; for (let i = 0; i < maxIterations; i++) { + // Cooperative boundary: between loop iterations (§2.1). + await exec.controller?.checkpoint(); + + const iterPath = `${path}#${i}`; + + // Resume replay of a completed iteration (§5): the `until` condition is + // still re-evaluated against the replayed `$current`, reconstructing the + // loop's original control flow. + if (exec.journal && hasOwn(exec.journal, iterPath)) { + current = exec.journal[iterPath]; + await exec.emit({ + type: "step.replayed", + path: iterPath, + stepType: rawBody.type, + output: current, + iteration: i, + }); + const done = resolveConfig(untilExpr, { ...scope, $current: current }); + if (done) return current; + continue; + } + // Make $current available in scope for template resolution const iterScope = { ...scope, $current: current }; // Resolve the body step's config with this iteration's scope const resolvedBodyConfig = resolveConfig(rawBody.config, iterScope) as Record; - await emit({ + await exec.emit({ type: "step.start", - path: `${path}#${i}`, + path: iterPath, stepType: rawBody.type, iteration: i, input: resolvedBodyConfig, }); const startTime = Date.now(); - current = await dispatchStep( - rawBody, - resolvedBodyConfig, - iterScope, - registry, - runId, - `${path}#${i}`, - emit, - workspace, - services, - paramOverrides, - ); + current = await dispatchStep(rawBody, resolvedBodyConfig, iterScope, exec, iterPath); - await emit({ + await exec.emit({ type: "step.end", - path: `${path}#${i}`, + path: iterPath, stepType: rawBody.type, output: current, durationMs: Date.now() - startTime, @@ -596,13 +708,8 @@ async function executeLoop( async function executeForeach( step: Step, scope: Record, - registry: StepRegistry, - runId: string, + exec: Exec, path: string, - emit: EmitFn, - workspace: SubflowResolver | undefined, - services: unknown, - paramOverrides?: Record>, ): Promise { // Resolve `items` once against the parent scope. const itemsResolved = resolveConfig(step.config["items"], scope); @@ -638,34 +745,43 @@ async function executeForeach( const results: unknown[] = []; for (let i = 0; i < items.length; i++) { + // Cooperative boundary: between foreach iterations (§2.1). + await exec.controller?.checkpoint(); + + const iterPath = `${path}#${i}`; + + // Resume replay of a completed iteration (§5): a failed iteration re-runs + // alone — completed ones replay by their `#i` paths. + if (exec.journal && hasOwn(exec.journal, iterPath)) { + const output = exec.journal[iterPath]; + await exec.emit({ + type: "step.replayed", + path: iterPath, + stepType: rawBody.type, + output, + iteration: i, + }); + results.push(output); + continue; + } + const iterScope = { ...scope, $current: items[i], $index: i }; const resolvedBodyConfig = resolveConfig(rawBody.config, iterScope) as Record; - await emit({ + await exec.emit({ type: "step.start", - path: `${path}#${i}`, + path: iterPath, stepType: rawBody.type, iteration: i, input: resolvedBodyConfig, }); const startTime = Date.now(); - const output = await dispatchStep( - rawBody, - resolvedBodyConfig, - iterScope, - registry, - runId, - `${path}#${i}`, - emit, - workspace, - services, - paramOverrides, - ); + const output = await dispatchStep(rawBody, resolvedBodyConfig, iterScope, exec, iterPath); - await emit({ + await exec.emit({ type: "step.end", - path: `${path}#${i}`, + path: iterPath, stepType: rawBody.type, output, durationMs: Date.now() - startTime, @@ -681,13 +797,8 @@ async function executeForeach( async function executeSubflow( step: Step, scope: Record, - registry: StepRegistry, - runId: string, + exec: Exec, path: string, - emit: EmitFn, - workspace: SubflowResolver | undefined, - services: unknown, - paramOverrides?: Record>, ): Promise { // Resolve workflow name, version, and input from parent scope const wfName = resolveConfig(step.config["workflow"], scope) as string; @@ -700,22 +811,22 @@ async function executeSubflow( throw new Error(`subflow step "${step.id}" requires a "workflow" config (workflow name)`); } - if (!workspace) { + if (!exec.workspace) { throw new Error( `subflow step "${step.id}" references workflow "${wfName}" but no workspace was provided to runWorkflow`, ); } const childFlow = version - ? await workspace.getWorkflowVersion(wfName, version) - : await workspace.getWorkflow(wfName); + ? await exec.workspace.getWorkflowVersion(wfName, version) + : await exec.workspace.getWorkflow(wfName); // Validate child flow input against its schema const validatedInput = childFlow.input.parse(childInput); // Thread `paramOverrides` (but NOT the entry-only flat `paramsOverride`) into // the child so a keyed override can reach knobs that live in this subflow. - return executeFlow(childFlow, validatedInput, registry, runId, path, emit, workspace, services, undefined, paramOverrides); + return executeFlow(childFlow, validatedInput, exec, path); } // ── Utilities ────────────────────────────────────────────────────────────── diff --git a/vein/src/steps/core/agent.ts b/vein/src/steps/core/agent.ts index 2473dc293..dbe723178 100644 --- a/vein/src/steps/core/agent.ts +++ b/vein/src/steps/core/agent.ts @@ -794,6 +794,15 @@ export default defineStep({ stopWhen, ...(providerOptions ? { providerOptions } : {}), ...(useSchema ? { output: Output.object({ schema: jsonSchema(cfg.schema) }) } : {}), + // Cooperative boundary BETWEEN tool calls (RUN_CONTROL_SPEC §4) — the + // single highest-value checkpoint in long agent sessions: a pause parks + // before the next LLM call starts (the in-flight one finishes and is + // journaled); a cancel stops the session here. `ctx.control` is the + // runner's unit-scoped view, so a parked agent counts as quiesced. + prepareStep: async () => { + await ctx?.control?.checkpoint(); + return undefined; + }, onStepFinish: (sf: any) => { // A length finish means the generation was TRUNCATED at the output // cap — a cut-off tool call never executes, so the loop dies with no diff --git a/vein/src/steps/lib/meta/run-workflow.ts b/vein/src/steps/lib/meta/run-workflow.ts index f90e8dbd5..adb2ffab5 100644 --- a/vein/src/steps/lib/meta/run-workflow.ts +++ b/vein/src/steps/lib/meta/run-workflow.ts @@ -24,6 +24,11 @@ export default defineStep({ }), output: z.any(), async run(cfg, ctx) { - return requireAuthoring(ctx.services).runWorkflow(cfg.name, cfg.input, cfg.params, cfg.version); + // parentRunId links the child run's controller under this run's — so + // cancelling/pausing this run reaches the children it launched + // (RUN_CONTROL_SPEC §2.2 tree linkage). + return requireAuthoring(ctx.services).runWorkflow(cfg.name, cfg.input, cfg.params, cfg.version, { + parentRunId: ctx.runId, + }); }, }); diff --git a/vein/src/store.ts b/vein/src/store.ts index 86e192d89..400d14b7c 100644 --- a/vein/src/store.ts +++ b/vein/src/store.ts @@ -4,9 +4,20 @@ import type { RunEvent, RunSummary } from "./core.js"; const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); -/** A run is terminal once its log records a `run.end` or `run.error`. */ +/** A run is terminal once its log records a `run.end`, `run.error`, or + * `run.cancelled` — though a later `run.resumed` REOPENS it (§5.2: a + * resumed run appends to the same log past its old terminal event). */ function isTerminal(event: RunEvent): boolean { - return event.type === "run.end" || event.type === "run.error"; + return ( + event.type === "run.end" || + event.type === "run.error" || + event.type === "run.cancelled" + ); +} + +/** A `run.resumed` marker reopens a log whose previous event was terminal. */ +function reopensRun(event: RunEvent): boolean { + return event.type === "run.resumed"; } /** @@ -28,12 +39,28 @@ function isTerminal(event: RunEvent): boolean { export async function* tailJsonl( file: string, isTerminal: (event: T) => boolean, - opts: { intervalMs?: number; signal?: AbortSignal } = {}, + opts: { + intervalMs?: number; + signal?: AbortSignal; + /** A later event that REOPENS a log whose previous event was terminal + * (a resumed run's `run.resumed`, RUN_CONTROL_SPEC §5.2). When set, a + * terminal event doesn't end the tail immediately: the tail scans + * ahead for a reopening event, and only closes at EOF (or, if + * `stillLive` says the producer is live again, keeps following). */ + reopens?: (event: T) => boolean; + /** Consulted at EOF after a terminal event when `reopens` is set: a live + * producer (a registered run controller) means a resume is in flight — + * keep following instead of closing. Default: close at EOF. */ + stillLive?: () => boolean; + } = {}, ): AsyncGenerator { const intervalMs = opts.intervalMs ?? 250; const signal = opts.signal; let offset = 0; let leftover = ""; + // Deferred-close mode (opts.reopens set): saw a terminal event, close at + // EOF unless a reopening event arrives first. + let sawTerminal = false; while (true) { if (signal?.aborted) return; @@ -66,11 +93,24 @@ export async function* tailJsonl( if (!line) continue; const event = JSON.parse(line) as T; yield event; - if (isTerminal(event)) return; + if (isTerminal(event)) { + if (!opts.reopens) return; + sawTerminal = true; + } else if (sawTerminal && opts.reopens?.(event)) { + sawTerminal = false; + } } } } + // After a terminal event: re-check for appended bytes immediately (no + // poll delay for the common completed-run tail); at EOF, close — unless + // the producer is live again (a resume re-attached), then keep following. + if (sawTerminal) { + if (chunk) continue; + if (!(opts.stillLive?.() ?? false)) return; + } + await sleep(intervalMs); } } @@ -159,27 +199,40 @@ export class FileRunStore implements RunStore { async *tailEvents( workflow: string, runId: string, - opts: { intervalMs?: number; signal?: AbortSignal } = {}, + opts: { intervalMs?: number; signal?: AbortSignal; stillLive?: () => boolean } = {}, ): AsyncGenerator { const file = join(this.runDir(workflow, runId), "events.jsonl"); - yield* tailJsonl(file, isTerminal, opts); + // `run.error`/`run.cancelled` are no longer unconditionally terminal: a + // later `run.resumed` reopens the stream (historical tails scan ahead; + // live tails consult `opts.stillLive` — the server's controllers map). + yield* tailJsonl(file, isTerminal, { ...opts, reopens: reopensRun }); } - /** Read events.jsonl for a specific run. */ + /** Read events.jsonl for a specific run. Tolerates a TORN TAIL: a process + * killed mid-append can leave a truncated final line (single-write + * atomicity is not guaranteed for large outputs) — it belongs to an + * incomplete unit by definition, so it is skipped, not fatal (§5.1). */ async getRunEvents(workflow: string, runId: string): Promise { + let raw: string; try { - const raw = await readFile( + raw = await readFile( join(this.runDir(workflow, runId), "events.jsonl"), "utf-8", ); - return raw - .trim() - .split("\n") - .filter(Boolean) - .map((line) => JSON.parse(line) as RunEvent); } catch { return []; } + const lines = raw.trim().split("\n").filter(Boolean); + const events: RunEvent[] = []; + for (let i = 0; i < lines.length; i++) { + try { + events.push(JSON.parse(lines[i]!) as RunEvent); + } catch (err) { + if (i === lines.length - 1) continue; // torn tail — skip + throw err; // corruption anywhere else is a real error + } + } + return events; } } diff --git a/vein/src/workspace.ts b/vein/src/workspace.ts index bc65373f2..d0f6b0aa5 100644 --- a/vein/src/workspace.ts +++ b/vein/src/workspace.ts @@ -232,6 +232,24 @@ export class WorkspaceManager { return readFile(join(dir, `${version}.yaml`), "utf-8"); } + /** Content hash of a workflow version's source (active version when + * omitted) — recorded on `run.start` so resume can refuse to replay a + * journal into a different DAG (RUN_CONTROL_SPEC §5). Null when the + * workflow/version is unknown: hash recording degrades gracefully for + * runs launched from a bare Flow object. */ + async getWorkflowHash(name: string, version?: string): Promise { + try { + const meta = await this.readWorkflowMetadata(name); + if (!meta) return null; + const v = version ?? meta.active; + const recorded = meta.versions[v]?.hash; + if (recorded) return recorded; + return contentHash(await this.getWorkflowSource(name, v)); + } catch { + return null; + } + } + private async loadFlowYaml(name: string, version: string): Promise { const dir = join(this.root, "workflows", name); const raw = await readFile(join(dir, `${version}.yaml`), "utf-8"); From fded9b5a18558693af9491834c3d9c035f5f86bf Mon Sep 17 00:00:00 2001 From: Evanfeenstra Date: Fri, 28 Aug 2026 13:44:38 -0700 Subject: [PATCH 2/5] =?UTF-8?q?vein/web:=20run=20control=20UI=20=E2=80=94?= =?UTF-8?q?=20Cancel/Pause/Resume=20topbar,=20re-run-from-here=20flyout?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Topbar: Cancel (confirm dialog states the subtree consequence) + Pause on a live run; Resume on a paused run and on dead runs (stale / error / cancelled — the stale badge finally has a purpose: it marks resumables). - Step flyout: "Re-run from here" on non-live runs — durable resume with from= (drill-aware). - runEpoch re-tails the SSE stream after a durable resume (the previous tail closed at the old terminal event). Co-Authored-By: Claude Fable 5 --- vein/web/src/api.ts | 28 +++++++++ vein/web/src/app.tsx | 76 ++++++++++++++++++++++- vein/web/src/components/StepRunFlyout.tsx | 11 ++++ vein/web/src/styles/components.css | 8 +++ 4 files changed, 121 insertions(+), 2 deletions(-) diff --git a/vein/web/src/api.ts b/vein/web/src/api.ts index 058c4efb4..60b388661 100644 --- a/vein/web/src/api.ts +++ b/vein/web/src/api.ts @@ -310,6 +310,34 @@ export const getRun = (workflow: string, runId: string) => export const getRunEvents = (workflow: string, runId: string) => fetchJSON(`/workflows/${workflow}/runs/${runId}/events`); +// ── Run control (RUN_CONTROL_SPEC) ───────────────────────────────────────── +// Cancel/pause act on the live run tree (nested runs included). Resume is +// dual-purpose: releases a paused run, or durably resumes a dead one +// ("stale"/error/cancelled) by replaying its journal — optionally forcing +// re-execution from a step path (`from`, the "re-run from here" gesture). + +export const cancelRun = (workflow: string, runId: string) => + fetchJSON<{ ok: boolean; state: string }>( + `/workflows/${workflow}/runs/${runId}/cancel`, + { method: "POST" }, + ); + +export const pauseRun = (workflow: string, runId: string) => + fetchJSON<{ ok: boolean; state: string; quiesced: boolean }>( + `/workflows/${workflow}/runs/${runId}/pause`, + { method: "POST" }, + ); + +export const resumeRun = (workflow: string, runId: string, from?: string) => + fetchJSON<{ ok: boolean; resumed: string }>( + `/workflows/${workflow}/runs/${runId}/resume`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(from ? { from } : {}), + }, + ); + // ── Promotions (promote a run output → a target workflow's param) ─────────── /** A declared promotion resolved against a specific run's output. `value` is diff --git a/vein/web/src/app.tsx b/vein/web/src/app.tsx index de5cff7f2..c970b5f31 100644 --- a/vein/web/src/app.tsx +++ b/vein/web/src/app.tsx @@ -63,6 +63,9 @@ export function App() { const [selectedRun, setSelectedRun] = useState(null); const [events, setEvents] = useState([]); const [running, setRunning] = useState(false); + // Bumped after a durable resume so the run-view effect re-tails the log + // (the prior tail closed at the old terminal event). + const [runEpoch, setRunEpoch] = useState(0); // True while an in-tab launched run is streaming live into `events`. Disarmed // when the user navigates into another run, so the stream can't clobber it. const liveStreamRef = useRef(false); @@ -331,7 +334,7 @@ export function App() { }) .catch(console.error); return () => ctrl.abort(); - }, [selectedRun]); + }, [selectedRun, runEpoch]); // Resolve the selected run's declared promotions (a winning value → a target // param). Drives the topbar Promote button + flyout. Empty unless the @@ -462,6 +465,52 @@ export function App() { setSelectedRun(runId); }, [selectedWf]); + // ── Run control (RUN_CONTROL_SPEC): cancel / pause / resume ────────────── + const selectedRunSummary = selectedRun ? runs.find((r) => r.runId === selectedRun) : undefined; + const runControlStatus = selectedRunSummary?.status; + // Live states come from the server's controller map; "stale"/"error"/ + // "cancelled" are dead-but-resumable (journal replay). + const runIsLive = + runControlStatus === "running" || runControlStatus === "pausing" || + runControlStatus === "paused" || runControlStatus === "cancelling"; + const runIsPaused = runControlStatus === "paused" || runControlStatus === "pausing"; + const runIsResumable = + runControlStatus === "stale" || runControlStatus === "error" || runControlStatus === "cancelled"; + + const handleCancelRun = useCallback(async () => { + if (!selectedWf || !selectedRun) return; + if (!confirm("Cancel this run? Any nested runs it launched are cancelled too — each stops at its next step boundary (the in-flight step finishes and is journaled).")) return; + try { await api.cancelRun(selectedWf, selectedRun); } catch (e) { alert(`Cancel failed: ${(e as Error).message}`); } + await refreshRuns(selectedWf); + }, [selectedWf, selectedRun, refreshRuns]); + + const handlePauseRun = useCallback(async () => { + if (!selectedWf || !selectedRun) return; + try { await api.pauseRun(selectedWf, selectedRun); } catch (e) { alert(`Pause failed: ${(e as Error).message}`); } + await refreshRuns(selectedWf); + }, [selectedWf, selectedRun, refreshRuns]); + + // Dual-purpose: releases a paused run, or durably resumes a dead one + // (stale / error / cancelled) by replaying its journal. + const handleResumeRun = useCallback(async () => { + if (!selectedWf || !selectedRun) return; + try { await api.resumeRun(selectedWf, selectedRun); } catch (e) { alert(`Resume failed: ${(e as Error).message}`); return; } + await refreshRuns(selectedWf); + setRunEpoch((n) => n + 1); // re-tail: the log continues past its old terminal + }, [selectedWf, selectedRun, refreshRuns]); + + // "Re-run from here" (§5.2 `from` invalidation): the chosen step, its + // dependents, and later loop iterations re-execute; upstream replays free. + const handleRerunFrom = useCallback(async (path: string) => { + if (!selectedWf || !selectedRun) return; + if (!confirm(`Re-run from "${path}"?\n\nThis step, everything downstream of it, and later iterations of an enclosing loop re-execute. Completed work upstream replays from the journal at zero cost.`)) return; + try { await api.resumeRun(selectedWf, selectedRun, path); } catch (e) { alert(`Re-run failed: ${(e as Error).message}`); return; } + setFlyoutStepId(null); + setFlyoutStepIndex(null); + await refreshRuns(selectedWf); + setRunEpoch((n) => n + 1); + }, [selectedWf, selectedRun, refreshRuns]); + const handleRun = useCallback(async () => { if (!selectedWf || !localSteps || localSteps.length === 0) return; const first = localSteps[0]!; @@ -772,6 +821,17 @@ export function App() { {isDirty && }
+ {/* Run control: cancel/pause a live run tree; resume a paused or + dead (stale/error/cancelled) one — RUN_CONTROL_SPEC §3–§5. */} + {selectedRun && runIsLive && runControlStatus !== "cancelling" && ( + + )} + {selectedRun && runControlStatus === "running" && ( + + )} + {selectedRun && (runIsPaused || runIsResumable) && ( + + )} {isDirty && !viewingOld && } {isRunView && promotions.length > 0 && (
)} + {props.onRerunFrom && ( +
+ Resume + +
+ )} {disp.start?.ts && (
Started diff --git a/vein/web/src/styles/components.css b/vein/web/src/styles/components.css index b5c69f3f0..267f24332 100644 --- a/vein/web/src/styles/components.css +++ b/vein/web/src/styles/components.css @@ -440,6 +440,14 @@ body.is-resizing-events * { .btn-primary:hover { background: var(--accent-strong); } +.btn-danger { + border-color: var(--danger, #b3423f); + color: var(--danger, #b3423f); +} +.btn-danger:hover { + background: var(--danger, #b3423f); + color: var(--bg); +} /* ─── dialog ─────────────────────────────────────────────────── */ From 2011208170a9bbda476a9755bbb247fb03a0ba55 Mon Sep 17 00:00:00 2001 From: Evanfeenstra Date: Fri, 28 Aug 2026 13:47:17 -0700 Subject: [PATCH 3/5] lab: wire run control through the optimizer tree + evolve-loop resume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - OptimizerCapability.run accepts parentRunId; evolve-loop and eval/optimize pass ctx.runId so generation/eval/reflect runs attach under the launching run's controller — cancelling or pausing an evolve run now reaches every nested run (RUN_CONTROL_SPEC §2.2). - Both loops checkpoint between generations (ctx.control, §2.1 code-step opt-in). - harvey/evolve-loop consumes ctx.journal on durable resume: journaled generations replay (state — best/sinceImprove/stop logic — is rebuilt from their synthetic #gen outputs, which now also carry the approach summary + digest excerpt so later briefings stay faithful); the loop continues from the first missing generation. Co-Authored-By: Claude Fable 5 --- mcp/src/lab/createLabVein.ts | 8 +++- mcp/src/lab/eval/steps/optimize.ts | 12 ++++-- mcp/src/lab/harvey/steps/evolve-loop.ts | 54 +++++++++++++++++++++++-- 3 files changed, 66 insertions(+), 8 deletions(-) diff --git a/mcp/src/lab/createLabVein.ts b/mcp/src/lab/createLabVein.ts index f370599a2..e56d1b6a1 100644 --- a/mcp/src/lab/createLabVein.ts +++ b/mcp/src/lab/createLabVein.ts @@ -31,7 +31,13 @@ export interface OptimizerCapability { run( name: string, input: unknown, - opts?: { paramOverrides?: Record> }, + opts?: { + paramOverrides?: Record>; + /** The calling step's `ctx.runId` — links the nested run's controller + * under the launching run's, so cancelling/pausing an evolve run + * reaches its generation/candidate runs (RUN_CONTROL_SPEC §2.2). */ + parentRunId?: string; + }, ): Promise; getParams(name: string): Promise>; } diff --git a/mcp/src/lab/eval/steps/optimize.ts b/mcp/src/lab/eval/steps/optimize.ts index 1a93a7a87..cf25f7fef 100644 --- a/mcp/src/lab/eval/steps/optimize.ts +++ b/mcp/src/lab/eval/steps/optimize.ts @@ -71,7 +71,7 @@ interface Optimizer { run( name: string, input: unknown, - opts?: { paramOverrides?: Record> }, + opts?: { paramOverrides?: Record>; parentRunId?: string }, ): Promise; getParams(name: string): Promise>; } @@ -212,6 +212,10 @@ export default defineStep({ let fromReflect: RunRef | undefined; for (let gen = 0; gen < cfg.maxGenerations; gen++) { + // Cooperative boundary between generations (RUN_CONTROL_SPEC §2.1 + // code-step opt-in): pause parks here; cancel stops the loop here. + await ctx.control?.checkpoint(); + const genStart = Date.now(); await emitGen(gen, { type: "step.start", @@ -224,7 +228,9 @@ export default defineStep({ // prompt is injected into all of them via the same paramOverrides. const paramOverrides = { [cfg.targetWorkflow]: { [cfg.promptParam]: candidate } }; const evalRuns = await mapLimit(dataset, cfg.concurrency, async (datum, i) => { - const run = await opt.run(cfg.evalWorkflow, datum ?? {}, { paramOverrides }); + // parentRunId: nested eval runs attach under this run's controller + // (cancel/pause the optimize run → its eval runs follow). + const run = await opt.run(cfg.evalWorkflow, datum ?? {}, { paramOverrides, parentRunId: ctx.runId }); if (run.status !== "success") { throw new Error(`eval run for "${labelFor(datum, i)}" failed: ${run.error?.message ?? "unknown"}`); } @@ -317,7 +323,7 @@ export default defineStep({ insight: r.insight, })), history, - }); + }, { parentRunId: ctx.runId }); if (reflectRun.status !== "success") { throw new Error(`reflect run failed: ${reflectRun.error?.message ?? "unknown"}`); } diff --git a/mcp/src/lab/harvey/steps/evolve-loop.ts b/mcp/src/lab/harvey/steps/evolve-loop.ts index a687f2120..12c09afc3 100644 --- a/mcp/src/lab/harvey/steps/evolve-loop.ts +++ b/mcp/src/lab/harvey/steps/evolve-loop.ts @@ -41,7 +41,7 @@ interface Optimizer { run( name: string, input: unknown, - opts?: { paramOverrides?: Record> }, + opts?: { paramOverrides?: Record>; parentRunId?: string }, ): Promise; } @@ -225,6 +225,43 @@ export default defineStep({ } as RunEvent); for (let gen = 0; gen < cfg.maxGenerations; gen++) { + // Cooperative boundary between generations (RUN_CONTROL_SPEC §2.1 + // code-step opt-in): pause parks here; cancel stops the loop here. + await ctx.control?.checkpoint(); + + // Durable resume (§5, iterative code steps): a generation whose + // synthetic `#gen` step.end is journaled replays — its run is NOT + // re-launched. State (best / sinceImprove / stop logic) is rebuilt + // from the journaled output so the loop continues where it left off. + const journaled = ctx.journal?.[`${ctx.path}#${gen}`] as AnyRec | undefined; + if (journaled) { + const passRate = num(journaled["passRate"]) ?? 0; + const entry: GenEntry = { + gen, + genRunId: String((journaled["runs"] as AnyRec[] | undefined)?.[0]?.["runId"] ?? ""), + version: typeof journaled["version"] === "string" ? (journaled["version"] as string) : undefined, + passRate, + summary: typeof journaled["summary"] === "string" ? (journaled["summary"] as string) : undefined, + digestText: typeof journaled["digestText"] === "string" ? (journaled["digestText"] as string) : undefined, + explore: journaled["directive"] === "explore", + }; + generations.push(entry); + consecutiveFailures = 0; + totalKnownCost += num(journaled["knownCost"]) ?? 0; + if (passRate > best.passRate + cfg.improveMargin) { + best = { gen, version: entry.version, passRate, digestText: entry.digestText ?? "" }; + sinceImprove = 0; + } else { + sinceImprove++; + } + await emitGen(gen, { type: "step.replayed", output: journaled }); + if (passRate >= cfg.stopPassRate) { + stopReason = `stopPassRate ${cfg.stopPassRate} reached`; + break; + } + continue; + } + const explore = sinceImprove >= cfg.exploreAfter; const briefing = composeBriefing({ baseWorkflow: cfg.baseWorkflow, @@ -246,9 +283,14 @@ export default defineStep({ const run = await opt.run( cfg.genWorkflow, { tasks: cfg.tasks, mission: cfg.mission, candidateName: cfg.candidateName, generation: gen, briefing }, - cfg.genParams && Object.keys(cfg.genParams).length - ? { paramOverrides: { [cfg.genWorkflow]: cfg.genParams } } - : undefined, + { + // Tree linkage: cancelling/pausing THIS run reaches the generation + // run (and its candidate runs) — RUN_CONTROL_SPEC §2.2. + parentRunId: ctx.runId, + ...(cfg.genParams && Object.keys(cfg.genParams).length + ? { paramOverrides: { [cfg.genWorkflow]: cfg.genParams } } + : {}), + }, ); if (run.status !== "success") { @@ -311,6 +353,10 @@ export default defineStep({ bestGen: best.gen, knownCost: Math.round((authorCost + produceCost) * 10000) / 10000, runs: [{ label: `generation ${gen}`, workflow: cfg.genWorkflow, runId: run.runId }], + // Carried so a durable resume can rebuild later generations' + // briefings (approach summaries + best digest) from the journal. + ...(entry.summary ? { summary: excerpt(entry.summary, 1200) } : {}), + ...(entry.digestText ? { digestText: excerpt(entry.digestText, 900) } : {}), }, }); From 2227e5f03d85bf449c77dbfce741a76d8e217075 Mon Sep 17 00:00:00 2001 From: Evanfeenstra Date: Fri, 28 Aug 2026 13:48:00 -0700 Subject: [PATCH 4/5] =?UTF-8?q?vein:=20docs=20=E2=80=94=20mark=20RUN=5FCON?= =?UTF-8?q?TROL=5FSPEC=20implemented,=20document=20run=20control=20in=20AG?= =?UTF-8?q?ENTS.md?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- vein/AGENTS.md | 29 ++++++++++++++++++++++++++--- vein/RUN_CONTROL_SPEC.md | 8 +++++++- 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/vein/AGENTS.md b/vein/AGENTS.md index ecc6ae535..ad8536d02 100644 --- a/vein/AGENTS.md +++ b/vein/AGENTS.md @@ -33,7 +33,9 @@ vein/ ├── src/ │ ├── core.ts # flow(), step(), defineStep(), services bag, all types │ ├── expr.ts # {{ }} template evaluator (recursive descent; whitelisted array methods + arrow lambdas) -│ ├── runner.ts # execution engine: DAG (topological), retry, onError, control flow +│ ├── runner.ts # execution engine: DAG (topological), retry, onError, control flow, journal replay +│ ├── run-control.ts # RunController: cooperative cancel/pause/resume for run TREES (RUN_CONTROL_SPEC.md) +│ ├── journal.ts # resume journal: step.end outputs → {path→output}; `from` invalidation │ ├── store.ts # RunStore interface + FileRunStore + MemoryRunStore + tailJsonl (shared append-only tail engine) │ ├── chat-store.ts # ChatStore interface + FileChatStore + MemoryChatStore (chats//: meta.json + messages.jsonl + events.jsonl) + truncateToolMessages │ ├── workspace.ts # WorkspaceManager: versioning, _metadata.json, YAML loading @@ -54,7 +56,7 @@ vein/ │ │ │ # create_workflow, run_workflow (threads ctx.services) │ │ ├── stepHelpers.ts # lsSteps / searchSteps / readStepSource (filesystem-style browser) │ │ └── schemaHelpers.ts # Zod → FieldDesc[] (for get_step schema rendering) -│ └── *.test.ts # 298 tests across 12 files +│ └── *.test.ts # 533 tests across 25 files └── web/ ├── package.json # preact, system-canvas, vite ├── vite.config.ts # preact preset, dev proxy to :3000 (/workflows, /steps, /chat, /health) @@ -91,7 +93,7 @@ vein/ # Engine cd vein npm install -npm test # 298 tests, ~330ms +npm test # 533 tests, ~1s npm run dev # starts Hono server on :3000 # Web UI (dev mode with HMR) @@ -416,6 +418,27 @@ services bag can override it, same as `http`/`secrets`). `streamRun(name, runId)` reattaches to the tail — so callers see the same `(onEvent, → RunResult)` interface as before. +- **Run control** (`RUN_CONTROL_SPEC.md`, `src/run-control.ts` + + `src/journal.ts`). Every launch site registers a `RunController` + (createVein's `trackRun` — superseding the old `activeRuns` set); nested + launches attach to the parent's controller via `parentRunId` (set by + meta/run-workflow + the lab's optimizer from `ctx.runId`), so + cancel/pause apply to WHOLE SUBTREES. All control is cooperative: the + runner awaits `checkpoint()` between DAG steps / loop+foreach iterations / + retry attempts; the agent step checkpoints between tool calls + (`prepareStep`); code steps with long loops opt in via + `ctx.control?.checkpoint()`. Endpoints: + `POST /workflows/:name/runs/:runId/{cancel,pause,resume}`. Cancel + finalizes honestly as `status: "cancelled"` (never the error path). + Durable resume replays the journal (`step.end` outputs keyed by path → + `step.replayed` events, zero cost) and re-executes from the first + incomplete path — valid for stale (crashed), error, and cancelled runs; + a successful run needs `from: ` ("re-run from here", which + drops the target + transitive dependents + later loop iterations). + `run.start` records the workflow content hash (resume refuses a changed + DAG unless forced) and per-run params. Iterative code steps consume + `ctx.journal` to resume completed iterations (harvey/evolve-loop does). + - **`RunStore.append/finalize`** take `(workflow, runId, ...)` — the workflow name is the first param. `MemoryRunStore` keys by `"workflow/runId"` internally; use `store.getEvents(wf, id)` diff --git a/vein/RUN_CONTROL_SPEC.md b/vein/RUN_CONTROL_SPEC.md index 928e97baa..5281aa6d1 100644 --- a/vein/RUN_CONTROL_SPEC.md +++ b/vein/RUN_CONTROL_SPEC.md @@ -5,7 +5,13 @@ run TREES the lab produces (an evolve run that launches generation runs that launch candidate runs), where today the only control is killing the server and orphaning every in-flight log as "stale". -Status: DRAFT. Nothing in this document is implemented. +Status: IMPLEMENTED (all three rungs). The runner/controller live in +`src/run-control.ts` + `src/runner.ts`, the journal in `src/journal.ts`, +the endpoints in `src/createVein.ts` +(`POST /workflows/:name/runs/:runId/{cancel,pause,resume}`), the UI in +`web/src/app.tsx`, and the lab linkage (optimizer `parentRunId`, +evolve-loop `ctx.journal` iteration resume) in `mcp/src/lab`. Tests: +`src/run-control.test.ts`. --- From 69f79feaf099160d4835c46de24fa2ab0b7e3a0f Mon Sep 17 00:00:00 2001 From: Evanfeenstra Date: Fri, 28 Aug 2026 13:48:50 -0700 Subject: [PATCH 5/5] vein: guard against concurrent resume requests double-launching one run Co-Authored-By: Claude Fable 5 --- vein/src/createVein.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/vein/src/createVein.ts b/vein/src/createVein.ts index d89478d14..c1e67c4d9 100644 --- a/vein/src/createVein.ts +++ b/vein/src/createVein.ts @@ -691,6 +691,11 @@ export async function createVein( } } + // Re-check liveness after the awaits above: two concurrent resume + // requests must not both launch onto the same log. + if (controllers.has(`${name}/${runId}`)) { + return c.json({ error: "Resume already in flight for this run" }, 409); + } launchDetached( flow, {