Close Gate 1 deterministic crash-resume rung

.gitattributes

@@ -0,0 +1 @@
+kernel/vendor/** -whitespace

.gitignore

@@ -1,8 +1,12 @@
 node_modules/
 target/
+# vendored crate sources are immutable inputs to the build; nothing under
+# kernel/vendor may be swallowed by the target/ rule (cc ships src/target/)
+!kernel/vendor/**
 dist/
 *.log
 .DS_Store
 .agent-relay/
 .env
 .agentworkforce/
+.cargo-home/

docs/RFC-0001-everything-is-a-relayflow.md

@@ -205,7 +205,8 @@ The gates exist to be sold, not admired. The consumer list, in order of proof va
 11. **New: the verification split holds** — the kernel judges *completion* (`completionReason`), the evidence layer judges *quality* (review gates, owner adjudication). Garden merge gates are evidence-layer.
 12. **New: "Software Garden" is a product-level brand only.** Repos keep their names; no `factory` → `garden` rename.
 13. **New: the kernel vocabulary is closed; the surface is open.** Three step verbs (`run`/`llm`/`agent`) + four resident verbs (`on`/`human`/`dispatch`/`done`) are the whole kernel language. Everything else — integration helpers (generated from relayfile adapters), `f.memory` (relayhistory), auth-by-declaration (relayauth path scopes), `f.mcp`, and community **plugins** (herdr-model marketplace) — is surface that compiles to kernel primitives. A plugin contributes verbs, triggers, and gate predicates; it must state its preflight (covenant 2) and cannot touch the kernel. Full design: `docs/SURFACE.md`.
-14. **New: tenancy is option C — the kernel is tenant-unaware.** `tenant_id` never appears inside the kernel. Cloud is a **cell orchestrator** that provisions, wakes, and sleeps per-tenant `relayflowd` cells; a sleeping tenant costs storage only (the journal is a SQLite file). Self-host is running your own cell: the same binary, zero divergence, per-tenant isolation true by construction.
+14. **New: a relayflow compiles to an immutable, content-addressed bundle.** `flows build` produces a sealed artifact — canonical spec JSON, compiled TS dialect with pinned dependencies, helper/plugin lockfile, assets, the flow's preflight declaration, and a signature from its identity — addressed as `flow@sha256:…` and pushed to a bucket/registry. **Runs reference digests, never working trees.** What this buys, by construction: every journal records exactly which flow version produced it (provenance); triggers bind to digests, so a scheduled run is executable from the bucket by any cell with no checkout (the fix for the workerless/ephemeral-run failure class seen 2026-08-27); rollback is pointing at the previous digest; upgrades-at-epoch-boundaries (versioning policy) means "the next epoch opens on a new digest"; and gate 9's self-authoring ships a new digest through the Garden — an agent can propose a bundle but can never mutate a deployed one.
+15. **New: tenancy is option C — the kernel is tenant-unaware.** `tenant_id` never appears inside the kernel. Cloud is a **cell orchestrator** that provisions, wakes, and sleeps per-tenant `relayflowd` cells; a sleeping tenant costs storage only (the journal is a SQLite file). Self-host is running your own cell: the same binary, zero divergence, per-tenant isolation true by construction.
 
 
 ## 7. Open questions

docs/SURFACE.md

@@ -85,7 +85,11 @@ The herdr model: first-party helpers are just plugins that ship in the box; the
 - **Preflight is part of the contract:** a plugin declares what must be provable before a run using it starts (credentials present, server reachable, scope grantable). A plugin that can't state its preflight doesn't load. Covenant 2 extends to the ecosystem by construction.
 - Receipts, budget attribution, and identity scoping apply to plugin verbs exactly as to first-party ones — they come from the compile target, so a plugin can't opt out.
 
-## 4. Open surface questions (for gate-1 SDK work)
+## 4. Build: the immutable bundle
+
+`flows build` seals a flow into a content-addressed, immutable bundle: canonical spec JSON, compiled TS with pinned deps, helper/plugin lockfile, assets, preflight declaration, identity signature — `flow@sha256:…`, pushed to a bucket/registry. `flows deploy` points a trigger at a digest; `flows run flow@sha256:…` executes from the bucket on any cell, no checkout. Preflight runs at build time for everything build-provable and again at deploy time for environment facts (credentials, workers, MCP servers). The working tree is for authoring; **production only ever runs digests.**
+
+## 5. Open surface questions (for gate-1 SDK work)
 
 - `gate:` in YAML: tiny expression language (`length < 200`) vs named checks only. Leaning: a deliberately small expression grammar + named checks for everything else.
 - Are YAML helper verbs (`slack:`, `mcp:`) core spec vocabulary or compile-time expansion into `run`/effect steps? Leaning: expansion — the kernel spec stays seven words; helpers stay a surface concern.

kernel/.cargo/config.toml

@@ -0,0 +1,5 @@
+[source.crates-io]
+replace-with = "vendored-sources"
+
+[source.vendored-sources]
+directory = "vendor"

kernel/README.md

@@ -0,0 +1,15 @@
+# Relayflow kernel
+
+Run the kernel gate from this directory with plain Cargo commands:
+
+```sh
+cargo test --workspace
+cargo clippy --workspace -- -D warnings
+cargo fmt --check
+```
+
+`.cargo/config.toml` redirects the locked crates.io dependencies to the
+repo-local `vendor/` source. This keeps clean-checkout builds deterministic and
+avoids reliance on the machine's `CARGO_HOME`; update the source alongside
+`Cargo.lock` with
+`CARGO_HOME="$(pwd)/.cargo-home" cargo vendor --locked vendor`.

kernel/relayflowd-core/src/machine.rs

@@ -76,6 +76,7 @@ pub fn next_actions(state: &RunState, now_ms: i64) -> Vec<Action> {
         return complete_run_actions(state, RunCompletionReason::Success, None, now_ms);
     }
 
+    let mut timers = Vec::new();
     for spec in &state.spec.steps {
         let runtime = &state.steps[&spec.id];
         match runtime.state {
@@ -97,13 +98,17 @@ pub fn next_actions(state: &RunState, now_ms: i64) -> Vec<Action> {
                 ))];
             }
             StepState::Backoff { wake_at_ms, .. } => {
-                return vec![Action::ArmTimer { at_ms: wake_at_ms }];
+                timers.push(Action::ArmTimer { at_ms: wake_at_ms });
             }
             StepState::Runnable => return start_actions(state, spec, runtime.attempts + 1, now_ms),
             _ => {}
         }
     }
-    Vec::new()
+    timers.sort_by_key(|action| match action {
+        Action::ArmTimer { at_ms } => *at_ms,
+        _ => unreachable!("the timer collection contains only timers"),
+    });
+    timers
 }
 
 fn start_actions(state: &RunState, step: &StepSpec, attempt: u32, now_ms: i64) -> Vec<Action> {

kernel/relayflowd-core/src/machine/tests.rs

@@ -139,3 +139,54 @@ fn crashed_attempt_does_not_consume_an_iteration() {
     assert_eq!(payload.disposition, Disposition::Retry);
     assert!(payload.next_attempt_at_ms.is_some());
 }
+
+#[test]
+fn all_backing_off_steps_return_timers() {
+    let spec = crate::RunSpec::parse(&json!({
+        "steps": [
+            {
+                "id": "first",
+                "type": "deterministic",
+                "command": "false",
+                "max_iterations": 2,
+                "retry": {"initial_backoff_ms": 200, "max_backoff_ms": 200, "multiplier": 1, "jitter_percent": 0}
+            },
+            {
+                "id": "second",
+                "type": "deterministic",
+                "command": "false",
+                "max_iterations": 2,
+                "retry": {"initial_backoff_ms": 100, "max_backoff_ms": 100, "multiplier": 1, "jitter_percent": 0}
+            }
+        ]
+    }))
+    .unwrap();
+    let result = AttemptResult {
+        output: Value::Null,
+        budget: Budget::default(),
+        completed_by: "kernel".to_owned(),
+        end_pins: None,
+        effects: Vec::new(),
+        failure_reason: Some(CompletionReason::WorkerError),
+    };
+    let mut entries = Vec::new();
+    for step in &spec.steps {
+        entries.extend(
+            completion_actions("run", step, 1, 0, result.clone(), 1_000)
+                .into_iter()
+                .filter_map(|action| match action {
+                    Action::Append(entry) => Some(entry),
+                    _ => None,
+                }),
+        );
+    }
+
+    let state = RunState::fold("run", spec, &entries).unwrap();
+    assert_eq!(
+        next_actions(&state, 1_050),
+        vec![
+            Action::ArmTimer { at_ms: 1_100 },
+            Action::ArmTimer { at_ms: 1_200 },
+        ]
+    );
+}

kernel/relayflowd-core/src/spec.rs

@@ -49,7 +49,8 @@ impl RunSpec {
     /// here; every non-flattened struct denies unknown fields via serde.
     pub fn parse(value: &Value) -> Result<Self, SpecError> {
         reject_unknown_step_fields(value)?;
-        serde_json::from_value(value.clone()).map_err(|error| SpecError::Malformed(error.to_string()))
+        serde_json::from_value(value.clone())
+            .map_err(|error| SpecError::Malformed(error.to_string()))
     }
 
     pub fn validate(&self) -> Result<(), SpecError> {

kernel/relayflowd/src/engine.rs

@@ -16,6 +16,16 @@ use ulid::Ulid;
 
 use crate::{clock::WallClock, exec_det};
 
+#[derive(Debug, Clone, Default)]
+#[doc(hidden)]
+pub struct DriveOptions {
+    pub stop_after: Option<usize>,
+    /// Test/debug hook: pause immediately before this runnable step starts.
+    pub pause_before_step: Option<String>,
+    /// Test/debug hook: pause after every step is durable, before run completion.
+    pub pause_before_completion: bool,
+}
+
 pub struct Engine<C = WallClock> {
     data_dir: PathBuf,
     clock: C,
@@ -43,6 +53,23 @@ impl<C: Clock> Engine<C> {
         spec: RunSpec,
         created_by: &str,
         stop_after: Option<usize>,
+    ) -> Result<RunOutcome> {
+        self.start_with_options(
+            spec,
+            created_by,
+            DriveOptions {
+                stop_after,
+                ..DriveOptions::default()
+            },
+        )
+    }
+
+    #[doc(hidden)]
+    pub fn start_with_options(
+        &self,
+        spec: RunSpec,
+        created_by: &str,
+        options: DriveOptions,
     ) -> Result<RunOutcome> {
         spec.validate().context("invalid run spec")?;
         ensure_deterministic(&spec)?;
@@ -71,18 +98,35 @@ impl<C: Clock> Engine<C> {
         self.registry()?
             .register(&run_id, &path)
             .context("register run")?;
-        self.drive(journal, spec, stop_after)
+        self.drive(journal, spec, options)
     }
 
     pub fn resume(&self, run_id: &str, stop_after: Option<usize>) -> Result<RunOutcome> {
+        self.resume_with_options(
+            run_id,
+            DriveOptions {
+                stop_after,
+                ..DriveOptions::default()
+            },
+        )
+    }
+
+    #[doc(hidden)]
+    pub fn resume_with_options(&self, run_id: &str, options: DriveOptions) -> Result<RunOutcome> {
         let mut journal = self.open_run(run_id)?;
+        let registry = self.registry()?;
+        if registry.lookup(run_id)?.is_none() {
+            registry
+                .register(run_id, &self.run_path(run_id))
+                .context("repair missing run registry entry")?;
+        }
         let spec = journal.run_spec().context("read run spec")?;
         ensure_deterministic(&spec)?;
         let state = self.load_state(&journal, spec.clone())?;
         for action in recovery_actions(&state, self.clock.now_ms()) {
             self.persist_only(&mut journal, action)?;
         }
-        self.drive(journal, spec, stop_after)
+        self.drive(journal, spec, options)
     }
 
     pub fn snapshot(&self, run_id: &str) -> Result<RunSnapshot> {
@@ -107,15 +151,16 @@ impl<C: Clock> Engine<C> {
         &self,
         mut journal: SqliteJournal,
         spec: RunSpec,
-        stop_after: Option<usize>,
+        options: DriveOptions,
     ) -> Result<RunOutcome> {
         let initial_completed = self.load_state(&journal, spec.clone())?.completed_steps();
+        let mut pause_consumed = false;
         loop {
             let state = self.load_state(&journal, spec.clone())?;
             if let Some(reason) = state.completion {
                 return Ok(outcome_from_state(&state, reason));
             }
-            if stop_after.is_some_and(|limit| {
+            if options.stop_after.is_some_and(|limit| {
                 state.completed_steps().saturating_sub(initial_completed) >= limit
             }) {
                 self.registry()?
@@ -127,6 +172,10 @@ impl<C: Clock> Engine<C> {
                     completed_steps: state.completed_steps(),
                 });
             }
+            if !pause_consumed && should_pause(&state, &options) {
+                pause_consumed = true;
+                thread::sleep(Duration::from_secs(300));
+            }
 
             let actions = next_actions(&state, self.clock.now_ms());
             if actions.is_empty() {
@@ -163,7 +212,13 @@ impl<C: Clock> Engine<C> {
                     Action::Dispatch { worker_class, .. } => {
                         bail!("no {worker_class:?} worker is attached to the deterministic rung")
                     }
-                    Action::ArmTimer { at_ms } => self.wait_for_timer(&journal, at_ms)?,
+                    Action::ArmTimer { at_ms } => {
+                        // `next_actions` exposes every durable timer. This
+                        // synchronous rung sleeps only until the earliest one,
+                        // then reloads state before choosing more work.
+                        self.wait_for_timer(&journal, at_ms)?;
+                        break;
+                    }
                     Action::CompleteRun { reason } => {
                         self.registry()?
                             .set_status(journal.run_id(), "completed", None)?;
@@ -233,6 +288,22 @@ impl<C: Clock> Engine<C> {
     }
 }
 
+fn should_pause(state: &RunState, options: &DriveOptions) -> bool {
+    if options.pause_before_completion && state.all_steps_succeeded() {
+        return true;
+    }
+    let Some(step_id) = options.pause_before_step.as_deref() else {
+        return false;
+    };
+    state.spec.steps.iter().find_map(|step| {
+        matches!(
+            state.steps[&step.id].state,
+            relayflowd_core::StepState::Runnable
+        )
+        .then_some(step.id.as_str())
+    }) == Some(step_id)
+}
+
 fn ensure_deterministic(spec: &RunSpec) -> Result<()> {
     if let Some(step) = spec
         .steps
@@ -328,7 +399,7 @@ pub fn read_spec(path: &Path) -> Result<RunSpec> {
     let value: serde_json::Value = serde_json::from_slice(&bytes)
         .with_context(|| format!("parse run spec {}", path.display()))?;
     // Fail closed: unknown fields are an error, never a silently dropped gate.
-    let spec = RunSpec::parse(&value)
-        .with_context(|| format!("parse run spec {}", path.display()))?;
+    let spec =
+        RunSpec::parse(&value).with_context(|| format!("parse run spec {}", path.display()))?;
     Ok(spec)
 }

kernel/relayflowd/src/lib.rs

@@ -3,4 +3,4 @@ pub mod engine;
 pub mod exec_det;
 pub mod server;
 
-pub use engine::{Engine, RunOutcome, RunSnapshot, RunStatus};
+pub use engine::{DriveOptions, Engine, RunOutcome, RunSnapshot, RunStatus};

kernel/relayflowd/src/main.rs

@@ -2,7 +2,7 @@ use std::path::PathBuf;
 
 use anyhow::{Result, bail};
 use clap::{Parser, Subcommand};
-use relayflowd::{Engine, RunStatus, engine::read_spec, server};
+use relayflowd::{DriveOptions, Engine, RunStatus, engine::read_spec, server};
 
 #[derive(Debug, Parser)]
 #[command(
@@ -27,6 +27,12 @@ enum Command {
         /// Test/debug boundary: return after this many newly completed steps.
         #[arg(long, hide = true)]
         stop_after: Option<usize>,
+        /// Test/debug boundary: pause before the named runnable step.
+        #[arg(long, hide = true)]
+        pause_before_step: Option<String>,
+        /// Test/debug boundary: pause after all steps, before run completion.
+        #[arg(long, hide = true)]
+        pause_before_completion: bool,
     },
     /// Resume a run from its durable journal.
     Resume {
@@ -46,8 +52,18 @@ fn main() -> Result<()> {
             spec,
             created_by,
             stop_after,
+            pause_before_step,
+            pause_before_completion,
         } => {
-            let outcome = engine.start(read_spec(&spec)?, &created_by, stop_after)?;
+            let outcome = engine.start_with_options(
+                read_spec(&spec)?,
+                &created_by,
+                DriveOptions {
+                    stop_after,
+                    pause_before_step,
+                    pause_before_completion,
+                },
+            )?;
             println!("{}", serde_json::to_string(&outcome)?);
             if outcome.status == RunStatus::Failed {
                 bail!("run {} failed", outcome.run_id);