feat(kernel): add durable run cancellation

kernel/DESIGN.md

@@ -38,6 +38,12 @@ First entry of segment 1. Payload:
 (int, stamped per segment thereafter via `epoch.summary`), `created_by`
 (client identity string).
 
+### 1.1a `run.cancel.requested`
+Durable operator intent, appended before cancellation closes any work. Payload:
+`requested_by` (client identity string). Resume treats this entry as a one-way
+state transition: no new work may start, active attempts and waits close with
+`completionReason: canceled`, and exactly one terminal canceled fact follows.
+
 ### 1.2 `step.attempt.started`
 One per attempt. Payload:
 
@@ -330,7 +336,7 @@ kernel/
 │       └── registry.rs       # relayflowd.sqlite3 run index (rebuildable)
 └── relayflowd/               # the binary
     └── src/
-        ├── main.rs           # CLI: run <spec.json> | resume <run_id> | serve
+        ├── main.rs           # CLI: run | resume | cancel | serve
         ├── engine.rs         # drives machine.rs Actions against journal + executors
         ├── exec_det.rs       # deterministic steps: spawn, capture, timeout
         ├── server.rs         # unix socket, protocol v0 (§5), worker dispatch
@@ -367,6 +373,7 @@ Minimal verb set for gate 1:
 | `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.resume` | `{run_id}` → `{run_id, state}` | §3 memoized resume |
+| `run.cancel` | `{run_id}` → `{run_id, status, completion_reason}` | append durable intent, close active leases, and append the terminal canceled fact; repeated calls return the existing outcome |
 | `run.get` | `{run_id}` → `{status, steps, budget}` | snapshot for legibility |
 | `run.watch` | `{run_id}` → stream of `{event: "entry", data: Entry}` | every appended entry, pushed |
 | `worker.attach` | `{worker_id, step_types: ["llm","agent"], pins}` → `{}` | connection becomes a worker; agent workers **must** supply opaque initial workspace revisions/stream offsets (refused otherwise) and receive `step.dispatch` events with pins plus recovery context. A step whose declared surfaces no attached worker holds parks — it is not dispatched |

kernel/relayflowd-core/src/entry.rs

@@ -9,6 +9,8 @@ use crate::spec::{RecoveryMode, StepType};
 pub enum EntryType {
     #[serde(rename = "run.spawned")]
     RunSpawned,
+    #[serde(rename = "run.cancel.requested")]
+    RunCancelRequested,
     #[serde(rename = "event.received")]
     EventReceived,
     #[serde(rename = "subscription.registered")]
@@ -53,6 +55,7 @@ impl EntryType {
     pub fn as_str(self) -> &'static str {
         match self {
             Self::RunSpawned => "run.spawned",
+            Self::RunCancelRequested => "run.cancel.requested",
             Self::EventReceived => "event.received",
             Self::SubscriptionRegistered => "subscription.registered",
             Self::SubscriptionMatched => "subscription.matched",
@@ -75,6 +78,7 @@ impl EntryType {
     pub fn parse(value: &str) -> Option<Self> {
         Some(match value {
             "run.spawned" => Self::RunSpawned,
+            "run.cancel.requested" => Self::RunCancelRequested,
             "event.received" => Self::EventReceived,
             "subscription.registered" => Self::SubscriptionRegistered,
             "subscription.matched" => Self::SubscriptionMatched,
@@ -140,6 +144,11 @@ pub struct RunSpawnedPayload {
     pub created_by: String,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct RunCancelRequestedPayload {
+    pub requested_by: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct AttemptStartedPayload {
     pub step_type: StepType,

kernel/relayflowd-core/src/lib.rs

@@ -21,6 +21,7 @@ pub use journal::{Journal, JournalError, MemoryJournal};
 pub use machine::{
     Action, AttemptResult, RecoveryInstruction, abandonment_actions, carried_pins_for,
     completion_actions, next_actions, recovery_actions, recovery_actions_filtered,
+    request_cancel_action,
 };
 pub use spec::*;
 pub use state::{RunState, StateError, StepRuntime, StepState};

kernel/relayflowd-core/src/machine.rs

@@ -17,6 +17,10 @@ use crate::{
 
 const LEASE_DURATION_MS: i64 = 30_000;
 
+mod cancel;
+use cancel::cancel_run_actions;
+pub use cancel::request_cancel_action;
+
 #[derive(Debug, Clone, PartialEq)]
 pub enum Action {
     Append(JournalEntry),
@@ -87,6 +91,9 @@ pub fn next_actions(state: &RunState, now_ms: i64) -> Vec<Action> {
     if state.completion.is_some() {
         return Vec::new();
     }
+    if state.cancel_requested.is_some() {
+        return cancel_run_actions(state, now_ms);
+    }
     if let Some(failed_step_id) = state.failed_step() {
         return complete_run_actions(
             state,

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

@@ -0,0 +1,100 @@
+use serde_json::Value;
+
+use super::{Action, complete_run_actions, retry_wait_id};
+use crate::{
+    Budget, CompletionReason, Disposition, EntryType, JournalEntry, RunCancelRequestedPayload,
+    RunCompletionReason, RunState, StepCompletedPayload, StepState, WaitCompletedPayload,
+    WaitCompletionReason,
+};
+
+/// Persist cancellation intent before any work is closed. Repeating the
+/// request is a no-op both while cancellation is in progress and after the
+/// terminal fact exists.
+pub fn request_cancel_action(
+    state: &RunState,
+    requested_by: impl Into<String>,
+    now_ms: i64,
+) -> Option<Action> {
+    if state.completion.is_some() || state.cancel_requested.is_some() {
+        return None;
+    }
+    Some(Action::Append(JournalEntry::new(
+        EntryType::RunCancelRequested,
+        state.run_id.clone(),
+        None,
+        None,
+        now_ms,
+        RunCancelRequestedPayload {
+            requested_by: requested_by.into(),
+        },
+    )))
+}
+
+pub(super) fn cancel_run_actions(state: &RunState, now_ms: i64) -> Vec<Action> {
+    let mut actions = Vec::new();
+    for step in &state.spec.steps {
+        let runtime = &state.steps[&step.id];
+        match &runtime.state {
+            StepState::Running { attempt, .. } => {
+                actions.push(Action::Append(JournalEntry::new(
+                    EntryType::StepCompleted,
+                    state.run_id.clone(),
+                    Some(step.id.clone()),
+                    Some(*attempt),
+                    now_ms,
+                    StepCompletedPayload {
+                        completion_reason: CompletionReason::Canceled,
+                        disposition: Disposition::StepDone,
+                        output: Value::Null,
+                        verification: None,
+                        end_pins: None,
+                        effects: Vec::new(),
+                        trajectory_tail: None,
+                        budget: Budget::default(),
+                        completed_by: "kernel".to_owned(),
+                        next_attempt_at_ms: None,
+                    },
+                )));
+            }
+            StepState::Waiting { wait_id } | StepState::NeedsHuman { wait_id } => {
+                actions.push(Action::Append(JournalEntry::new(
+                    EntryType::WaitCompleted,
+                    state.run_id.clone(),
+                    Some(step.id.clone()),
+                    Some(runtime.attempts),
+                    now_ms,
+                    WaitCompletedPayload {
+                        wait_id: wait_id.clone(),
+                        completion_reason: WaitCompletionReason::Canceled,
+                        result: Value::Null,
+                    },
+                )));
+            }
+            StepState::Backoff {
+                attempt,
+                wake_at_ms,
+            } => {
+                actions.push(Action::Append(JournalEntry::new(
+                    EntryType::WaitCompleted,
+                    state.run_id.clone(),
+                    Some(step.id.clone()),
+                    Some(*attempt),
+                    now_ms,
+                    WaitCompletedPayload {
+                        wait_id: retry_wait_id(&state.run_id, &step.id, *attempt, *wake_at_ms),
+                        completion_reason: WaitCompletionReason::Canceled,
+                        result: Value::Null,
+                    },
+                )));
+            }
+            StepState::Pending | StepState::Runnable | StepState::Done { .. } => {}
+        }
+    }
+    actions.extend(complete_run_actions(
+        state,
+        RunCompletionReason::Canceled,
+        None,
+        now_ms,
+    ));
+    actions
+}

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

@@ -27,6 +27,12 @@ pub fn recovery_actions_filtered(
     now_ms: i64,
     lease_is_active: &dyn Fn(&str, u32) -> bool,
 ) -> Vec<Action> {
+    // Durable cancellation intent wins over crash classification. The cancel
+    // path closes the same lease as canceled; recovery must not get there
+    // first and rewrite the reason merely because the process restarted.
+    if state.cancel_requested.is_some() {
+        return Vec::new();
+    }
     let mut actions = Vec::new();
     for spec in &state.spec.steps {
         let runtime = &state.steps[&spec.id];

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

@@ -1,7 +1,7 @@
 use serde_json::json;
 
 use super::*;
-use crate::{entry::AttemptStartedPayload, state::RunState};
+use crate::{Clock, SimClock, entry::AttemptStartedPayload, state::RunState};
 
 fn retrying_spec() -> crate::RunSpec {
     serde_json::from_value(json!({
@@ -133,6 +133,91 @@ fn successful_memo_is_never_scheduled_again() {
     );
 }
 
+#[test]
+fn cancel_request_closes_the_active_lease_before_the_terminal_fact() {
+    let clock = SimClock::new(10);
+    let spec = crate::RunSpec::parse(&json!({
+        "steps": [{"id": "model", "type": "llm", "prompt": "answer"}]
+    }))
+    .unwrap();
+    let fresh = RunState::fold("run", spec.clone(), &[]).unwrap();
+    let Action::Append(started) = next_actions(&fresh, clock.now_ms()).remove(0) else {
+        panic!("the attempt lease must be durable");
+    };
+    let running = RunState::fold("run", spec.clone(), std::slice::from_ref(&started)).unwrap();
+    clock.advance(10);
+    let Action::Append(requested) =
+        request_cancel_action(&running, "operator", clock.now_ms()).unwrap()
+    else {
+        panic!("cancel must first persist its request");
+    };
+    assert_eq!(requested.entry_type, EntryType::RunCancelRequested);
+
+    let canceling = RunState::fold("run", spec, &[started, requested]).unwrap();
+    clock.advance(10);
+    let actions = next_actions(&canceling, clock.now_ms());
+    let Action::Append(closed) = &actions[0] else {
+        panic!("the active attempt must be closed first");
+    };
+    let closed: StepCompletedPayload = serde_json::from_value(closed.payload.clone()).unwrap();
+    assert_eq!(closed.completion_reason, CompletionReason::Canceled);
+    assert_eq!(closed.disposition, Disposition::StepDone);
+    let Action::Append(terminal) = &actions[1] else {
+        panic!("the run fact must follow the lease closure");
+    };
+    assert_eq!(terminal.entry_type, EntryType::RunCompleted);
+    let terminal: RunCompletedPayload = serde_json::from_value(terminal.payload.clone()).unwrap();
+    assert_eq!(terminal.completion_reason, RunCompletionReason::Canceled);
+}
+
+#[test]
+fn repeated_cancel_request_is_idempotent() {
+    let spec = retrying_spec();
+    let state = RunState::fold("run", spec.clone(), &[]).unwrap();
+    let Action::Append(requested) = request_cancel_action(&state, "operator", 10).unwrap() else {
+        panic!();
+    };
+    let canceling = RunState::fold("run", spec.clone(), std::slice::from_ref(&requested)).unwrap();
+    assert!(request_cancel_action(&canceling, "operator", 11).is_none());
+    let terminal_entries = next_actions(&canceling, 12)
+        .into_iter()
+        .filter_map(|action| match action {
+            Action::Append(entry) => Some(entry),
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+    let terminal = RunState::fold("run", spec, &[requested, terminal_entries[0].clone()]).unwrap();
+    assert!(request_cancel_action(&terminal, "operator", 13).is_none());
+}
+
+#[test]
+fn durable_cancel_request_outranks_crash_recovery() {
+    let clock = SimClock::new(10);
+    let spec = crate::RunSpec::parse(&json!({
+        "steps": [{"id": "model", "type": "llm", "prompt": "answer"}]
+    }))
+    .unwrap();
+    let fresh = RunState::fold("run", spec.clone(), &[]).unwrap();
+    let Action::Append(started) = next_actions(&fresh, clock.now_ms()).remove(0) else {
+        panic!("the attempt lease must be durable");
+    };
+    let running = RunState::fold("run", spec.clone(), std::slice::from_ref(&started)).unwrap();
+    clock.advance(10);
+    let Action::Append(requested) =
+        request_cancel_action(&running, "operator", clock.now_ms()).unwrap()
+    else {
+        panic!("the cancel request must be durable");
+    };
+    let canceling = RunState::fold("run", spec, &[started, requested]).unwrap();
+
+    assert!(recovery_actions(&canceling, clock.now_ms()).is_empty());
+    let Action::Append(closed) = &next_actions(&canceling, clock.now_ms())[0] else {
+        panic!("cancellation must close the active lease");
+    };
+    let closed: StepCompletedPayload = serde_json::from_value(closed.payload.clone()).unwrap();
+    assert_eq!(closed.completion_reason, CompletionReason::Canceled);
+}
+
 #[test]
 fn crashed_attempt_does_not_consume_an_iteration() {
     // max_iterations 2: crash attempt 1, verification-fail the replacement
@@ -146,7 +231,7 @@ fn crashed_attempt_does_not_consume_an_iteration() {
 
     // kill -9 between steps: attempt 1 is Running with no result. Recovery
     // must record the dead attempt as a retry, not a consumed iteration.
-    let state = RunState::fold("run", spec.clone(), &[started.clone()]).unwrap();
+    let state = RunState::fold("run", spec.clone(), std::slice::from_ref(&started)).unwrap();
     assert_eq!(state.steps["hello"].semantic_executions, 0);
     let recovery = recovery_actions(&state, 1_000);
     let Action::Append(crashed) = &recovery[0] else {
@@ -276,7 +361,7 @@ fn reset_recovery_dispatches_the_original_pinned_revision() {
     let spec = agent_spec("reset");
     let pinned = workspace_pins("rev-clean");
     let started = started_agent(&spec, pinned.clone());
-    let running = RunState::fold("run", spec.clone(), &[started.clone()]).unwrap();
+    let running = RunState::fold("run", spec.clone(), std::slice::from_ref(&started)).unwrap();
     let recovered = recovery_actions(&running, 20);
     let entries = vec![
         started,
@@ -311,7 +396,7 @@ fn inspect_recovery_injects_the_dirty_pin_completion_reason_and_tail() {
     let clean = workspace_pins("rev-clean");
     let dirty = workspace_pins("rev-dirty");
     let started = started_agent(&spec, clean);
-    let running = RunState::fold("run", spec.clone(), &[started.clone()]).unwrap();
+    let running = RunState::fold("run", spec.clone(), std::slice::from_ref(&started)).unwrap();
     let result = AttemptResult {
         output: Value::Null,
         budget: Budget::default(),
@@ -359,7 +444,7 @@ fn inspect_recovery_injects_the_dirty_pin_completion_reason_and_tail() {
 fn manual_recovery_parks_needs_human_and_never_redispatches() {
     let spec = agent_spec("manual");
     let started = started_agent(&spec, workspace_pins("rev-clean"));
-    let running = RunState::fold("run", spec.clone(), &[started.clone()]).unwrap();
+    let running = RunState::fold("run", spec.clone(), std::slice::from_ref(&started)).unwrap();
     let recovered = recovery_actions(&running, 20);
     let Action::Append(wait) = &recovered[1] else {
         panic!(

kernel/relayflowd-core/src/state.rs

@@ -6,8 +6,8 @@ use thiserror::Error;
 use crate::{
     entry::{
         Budget, CompletionReason, Disposition, EntryType, EpochSummaryPayload, JournalEntry, Pins,
-        RunCompletedPayload, RunCompletionReason, SleepUntilPayload, StepCompletedPayload,
-        WaitCompletedPayload, WaitCompletionReason,
+        RunCancelRequestedPayload, RunCompletedPayload, RunCompletionReason, SleepUntilPayload,
+        StepCompletedPayload, WaitCompletedPayload, WaitCompletionReason,
     },
     spec::{RunSpec, StepKind, StepType},
 };
@@ -67,6 +67,9 @@ pub struct RunState {
     pub memo: BTreeMap<String, Value>,
     pub budget: Budget,
     pub completion: Option<RunCompletionReason>,
+    /// Durable cancellation intent. Once present, scheduling can only close
+    /// live work and append the terminal canceled fact.
+    pub cancel_requested: Option<RunCancelRequestedPayload>,
     /// Appendix A rule 6 chain head: the last successful agent completion.
     pub current_pins: Option<Pins>,
 }
@@ -103,6 +106,7 @@ impl RunState {
             memo: BTreeMap::new(),
             budget: Budget::default(),
             completion: None,
+            cancel_requested: None,
             current_pins: None,
         };
 
@@ -164,6 +168,9 @@ impl RunState {
                     let payload: RunCompletedPayload = decode(entry)?;
                     state.completion = Some(payload.completion_reason);
                 }
+                EntryType::RunCancelRequested => {
+                    state.cancel_requested = Some(decode(entry)?);
+                }
                 EntryType::RunSpawned
                 | EntryType::EventReceived
                 | EntryType::SubscriptionRegistered