kernel: dispatch runnable steps in parallel

kernel/relayflowd-core/src/machine.rs

@@ -10,7 +10,7 @@ use crate::{
         StepCompletedPayload, WaitCompletedPayload, WaitCompletionReason,
     },
     retry::backoff_delay_ms,
-    spec::{AgentSurfaces, RecoveryMode, StepKind, StepSpec, StepType},
+    spec::{AgentSurfaces, RecoveryMode, StepKind, StepSpec, StepType, workspace_surfaces_equal},
     state::{RunState, StepState},
     verify::verify,
 };
@@ -95,6 +95,15 @@ pub fn next_actions(state: &RunState, now_ms: i64) -> Vec<Action> {
         return cancel_run_actions(state, now_ms);
     }
     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,
@@ -106,15 +115,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()),
@@ -125,13 +144,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 {
@@ -168,7 +205,7 @@ pub(crate) fn carried_pins(chain: Option<&Pins>, surfaces: &AgentSurfaces) -> Pi
                 chain
                     .workspace
                     .iter()
-                    .find(|pin| pin.surface == declared.surface)
+                    .find(|pin| workspace_surfaces_equal(&pin.surface, &declared.surface))
                     .cloned()
             })
             .collect(),
@@ -423,5 +460,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,123 @@
+//! 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 crate::{
+    spec::{StepKind, StepSpec, path_surface_identity},
+    state::{RunState, StepRuntime, StepState},
+};
+
+#[derive(Clone, PartialEq, Eq)]
+enum SurfaceIdentity {
+    Opaque(String),
+    Path {
+        kind: PathSurfaceKind,
+        namespace: String,
+        components: Vec<String>,
+    },
+}
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum PathSurfaceKind {
+    Workspace,
+    External,
+}
+
+impl SurfaceIdentity {
+    fn conflicts(&self, other: &Self) -> bool {
+        match (self, other) {
+            (Self::Opaque(left), Self::Opaque(right)) => left == right,
+            (
+                Self::Path {
+                    kind: left_kind,
+                    namespace: left_namespace,
+                    components: left,
+                },
+                Self::Path {
+                    kind: right_kind,
+                    namespace: right_namespace,
+                    components: right,
+                },
+            ) => {
+                left_kind == right_kind
+                    && left_namespace == right_namespace
+                    && (left.starts_with(right) || right.starts_with(left))
+            }
+            _ => false,
+        }
+    }
+}
+
+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::<Vec<_>>();
+    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.iter().any(|held| surface.conflicts(held)))
+        {
+            continue;
+        }
+        occupied.extend(surfaces);
+        selected.push((step, runtime));
+    }
+    selected
+}
+
+fn surface_keys(step: &StepSpec) -> impl Iterator<Item = SurfaceIdentity> + '_ {
+    let StepKind::Agent { surfaces, .. } = &step.kind else {
+        return Vec::new().into_iter();
+    };
+    surfaces
+        .workspace
+        .iter()
+        .map(|surface| {
+            let (namespace, components) = path_surface_identity(&surface.surface)
+                .expect("validated specs have canonical workspace surfaces");
+            SurfaceIdentity::Path {
+                kind: PathSurfaceKind::Workspace,
+                namespace,
+                components,
+            }
+        })
+        .chain(
+            surfaces
+                .streams
+                .iter()
+                .map(|surface| SurfaceIdentity::Opaque(format!("stream:{}", surface.stream))),
+        )
+        .chain(surfaces.external.iter().map(|surface| {
+            let (namespace, components) = path_surface_identity(surface)
+                .expect("validated specs have canonical external surfaces");
+            SurfaceIdentity::Path {
+                kind: PathSurfaceKind::External,
+                namespace,
+                components,
+            }
+        }))
+        .collect::<Vec<_>>()
+        .into_iter()
+}