integration(v2): compose issue 132 stack for research migration

.github/workflows/cloud-runtime-artifact.yml

@@ -42,6 +42,20 @@ jobs:
         working-directory: kernel
         run: cargo build --locked --release -p relayflowd
 
+      - name: Install SDK dependencies
+        run: npm ci --prefix sdk
+
+      - name: Test SDK and type-level authoring contracts
+        working-directory: sdk
+        run: |
+          npm run build
+          npm run typecheck:tests
+          ./node_modules/.bin/vitest run \
+            tests/typed-output.test.ts \
+            tests/validate.test.ts \
+            tests/spec-parity.test.ts \
+            tests/deterministic-llm.test.ts
+
       - name: Build standalone flows CLI
         run: |
           cd surface

kernel/DESIGN.md

@@ -365,7 +365,7 @@ Minimal verb set for gate 1:
 | verb | params → result | purpose |
 |---|---|---|
 | `hello` | `{protocol: 0, client}` → `{protocol: 0, server}` | handshake; version mismatch is a hard error |
-| `run.start` | `{spec}` → `{run_id}` | validate spec (zero-agent flows are legal), create run file, append `run.spawned`, begin scheduling |
+| `run.start` | `{spec}` → `{run_id}` | validate spec (zero-agent flows are legal; invalid declarations return `invalid_spec` before storage), create run file, append `run.spawned`, begin scheduling |
 | `run.resume` | `{run_id}` → `{run_id, state}` | §3 memoized resume |
 | `run.get` | `{run_id}` → `{status, steps, budget}` | snapshot for legibility |
 | `run.watch` | `{run_id}` → stream of `{event: "entry", data: Entry}` | every appended entry, pushed |

kernel/relayflowd-core/src/lib.rs

@@ -10,6 +10,7 @@ pub mod event;
 pub mod journal;
 pub mod machine;
 pub mod retry;
+mod schema;
 pub mod spec;
 pub mod state;
 pub mod verify;

kernel/relayflowd-core/src/machine.rs

@@ -88,6 +88,15 @@ pub fn next_actions(state: &RunState, now_ms: i64) -> Vec<Action> {
         return Vec::new();
     }
     if let Some(failed_step_id) = state.failed_step() {
+        if state
+            .steps
+            .values()
+            .any(|runtime| matches!(runtime.state, StepState::Running { .. }))
+        {
+            // Drain already-started siblings before making run.completed
+            // terminal. No fresh work is elected once failure is inevitable.
+            return Vec::new();
+        }
         return complete_run_actions(
             state,
             RunCompletionReason::StepFailed,
@@ -99,15 +108,25 @@ 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 {
-            StepState::Backoff {
+    // Wake every retry whose deterministic timer is due before starting work.
+    // Recovery can put several crashed parallel lanes into the same zero-delay
+    // backoff; waking just one would let an already-runnable peer start and park
+    // the run while the other due lane remained asleep.
+    let due_waits = state
+        .spec
+        .steps
+        .iter()
+        .filter_map(|spec| {
+            let runtime = &state.steps[&spec.id];
+            let StepState::Backoff {
                 attempt,
                 wake_at_ms,
-            } if wake_at_ms <= now_ms => {
-                return vec![Action::Append(JournalEntry::new(
+            } = runtime.state
+            else {
+                return None;
+            };
+            (wake_at_ms <= now_ms).then(|| {
+                Action::Append(JournalEntry::new(
                     EntryType::WaitCompleted,
                     state.run_id.clone(),
                     Some(spec.id.clone()),
@@ -118,13 +137,31 @@ pub fn next_actions(state: &RunState, now_ms: i64) -> Vec<Action> {
                         completion_reason: WaitCompletionReason::TimerFired,
                         result: Value::Null,
                     },
-                ))];
-            }
-            StepState::Backoff { wake_at_ms, .. } => {
-                timers.push(Action::ArmTimer { at_ms: wake_at_ms });
-            }
-            StepState::Runnable => return start_actions(state, spec, runtime.attempts + 1, now_ms),
-            _ => {}
+                ))
+            })
+        })
+        .collect::<Vec<_>>();
+    if !due_waits.is_empty() {
+        return due_waits;
+    }
+
+    // A fold marks every dependency-free step Runnable at once. Preserve the
+    // authored spec order while emitting every journal-first start pair from
+    // that one state snapshot; dependencies unlocked by these executions are
+    // considered only after their completions are folded on the next pass.
+    let starts = parallel::runnable_batch(state)
+        .into_iter()
+        .flat_map(|(spec, runtime)| start_actions(state, spec, runtime.attempts + 1, now_ms))
+        .collect::<Vec<_>>();
+    if !starts.is_empty() {
+        return starts;
+    }
+
+    let mut timers = Vec::new();
+    for spec in &state.spec.steps {
+        let runtime = &state.steps[&spec.id];
+        if let StepState::Backoff { wake_at_ms, .. } = runtime.state {
+            timers.push(Action::ArmTimer { at_ms: wake_at_ms });
         }
     }
     timers.sort_by_key(|action| match action {
@@ -416,5 +453,9 @@ fn deterministic_ulid(
 mod recovery;
 pub use recovery::{abandonment_actions, recovery_actions, recovery_actions_filtered};
 
+mod parallel;
+
+#[cfg(test)]
+mod parallel_tests;
 #[cfg(test)]
 mod tests;

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

@@ -0,0 +1,70 @@
+//! Deterministic parallel batch selection.
+//!
+//! DAG independence is not enough for agent work: two steps can be topology-
+//! independent while mutating the same declared surface. The authored-order
+//! maximal batch therefore reserves each mutable surface once. An unfinished
+//! attempt in backoff or waiting retains its reservation just like a live
+//! lease, preventing a crash/retry from turning into a last-write-wins fork.
+
+use std::collections::BTreeSet;
+
+use crate::{
+    spec::{StepKind, StepSpec},
+    state::{RunState, StepRuntime, StepState},
+};
+
+pub(super) fn runnable_batch(state: &RunState) -> Vec<(&StepSpec, &StepRuntime)> {
+    let mut occupied = state
+        .spec
+        .steps
+        .iter()
+        .filter(|step| {
+            matches!(
+                state.steps[&step.id].state,
+                StepState::Running { .. }
+                    | StepState::Backoff { .. }
+                    | StepState::Waiting { .. }
+                    | StepState::NeedsHuman { .. }
+            )
+        })
+        .flat_map(surface_keys)
+        .collect::<BTreeSet<_>>();
+    let mut selected = Vec::new();
+    for step in &state.spec.steps {
+        let runtime = &state.steps[&step.id];
+        if runtime.state != StepState::Runnable {
+            continue;
+        }
+        let surfaces = surface_keys(step).collect::<Vec<_>>();
+        if surfaces.iter().any(|surface| occupied.contains(surface)) {
+            continue;
+        }
+        occupied.extend(surfaces);
+        selected.push((step, runtime));
+    }
+    selected
+}
+
+fn surface_keys(step: &StepSpec) -> impl Iterator<Item = String> + '_ {
+    let StepKind::Agent { surfaces, .. } = &step.kind else {
+        return Vec::new().into_iter();
+    };
+    surfaces
+        .workspace
+        .iter()
+        .map(|surface| format!("workspace:{}", surface.surface))
+        .chain(
+            surfaces
+                .streams
+                .iter()
+                .map(|surface| format!("stream:{}", surface.stream)),
+        )
+        .chain(
+            surfaces
+                .external
+                .iter()
+                .map(|surface| format!("external:{surface}")),
+        )
+        .collect::<Vec<_>>()
+        .into_iter()
+}