vein: run control — cancel, pause/resume, durable resume (RUN_CONTROL_SPEC)

mcp/src/lab/createLabVein.ts

@@ -31,7 +31,13 @@ export interface OptimizerCapability {
   run(
     name: string,
     input: unknown,
-    opts?: { paramOverrides?: Record<string, Record<string, unknown>> },
+    opts?: {
+      paramOverrides?: Record<string, Record<string, unknown>>;
+      /** 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<RunResult>;
   getParams(name: string): Promise<Record<string, unknown>>;
 }

mcp/src/lab/eval/steps/optimize.ts

@@ -71,7 +71,7 @@ interface Optimizer {
   run(
     name: string,
     input: unknown,
-    opts?: { paramOverrides?: Record<string, Record<string, unknown>> },
+    opts?: { paramOverrides?: Record<string, Record<string, unknown>>; parentRunId?: string },
   ): Promise<RunResultLike>;
   getParams(name: string): Promise<Record<string, unknown>>;
 }
@@ -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"}`);
         }

mcp/src/lab/harvey/steps/evolve-loop.ts

@@ -41,7 +41,7 @@ interface Optimizer {
   run(
     name: string,
     input: unknown,
-    opts?: { paramOverrides?: Record<string, Record<string, unknown>> },
+    opts?: { paramOverrides?: Record<string, Record<string, unknown>>; parentRunId?: string },
   ): Promise<RunResultLike>;
 }
 
@@ -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) } : {}),
         },
       });
 

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/<id>/: 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: <stepPath>` ("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)`

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`.
 
 ---
 
@@ -252,7 +258,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: <path>`). 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/<runId>/...`): a drafted memo written before the crash is
@@ -286,14 +327,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: <stepPath> }` (§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 +385,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

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",

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 };

vein/src/ai/prompts.ts

@@ -50,6 +50,15 @@ export interface AiDeps {
       promise: Promise<RunResult>;
     }) => 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 ──────────────────────────────────────────────────────────

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<string, unknown> | 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;