jouleclaw/SPEC.md

@@ -942,6 +942,45 @@ Reference: [`crates/jouleclaw-graph`](./jouleclaw-rs/crates/jouleclaw-graph/)
 `jc:WorkerHandoffShape` at
 [`vocab/v1.shapes.ttl`](./vocab/v1.shapes.ttl).
 
+### 17.3 A node in flight survives the process
+
+The run journal records a node when it completes. A node whose
+process dies before it completes would otherwise leave no trace,
+and a restarted run would dispatch it again as if it had never
+begun, repeating any work it did outside the run: a message sent,
+a payment made, a record written.
+
+A consumer that dispatches a node SHOULD therefore record the
+dispatch (`RunEvent::NodeDispatched`) **before** the node's work
+begins, with the node's **replay rule**:
+
+- `safe`: running the node again is harmless, because it is a read
+  or its work is keyed so a repeat finds its first result.
+- `never` (the default): the node may have acted outside the run,
+  and nothing in the journal says whether it did.
+
+A dispatched node that is not yet recorded is **in flight**, and
+MUST NOT be dispatched again. After a restart, the consumer MUST
+mark every node still in flight as interrupted
+(`RunEvent::NodeInterrupted`) before dispatching anything, keeping
+the energy observed for the attempt as a lower bound on what it
+spent. An interrupted `safe` node is ready again, as a new attempt.
+An interrupted `never` node is **held**: whether its work took
+effect is not knowable from the journal, so it MUST NOT run again
+until it is settled, by recording what it did, by an approval to
+run it again (`RunEvent::RerunApproved`) that names who gave it,
+or by failing the run. It is not left to the model to decide: a
+model told only that a call "may have partly run" can repeat a
+write it cannot see.
+
+Until a run is recovered, a node in flight stays out of the ready
+set, so a consumer that omits recovery stalls rather than
+repeating work. This is the run-journal form of JCP SPEC §4.11.
+
+Reference: `Run::dispatch`, `Run::recover`, `Run::approve_rerun`,
+`Replay`, `NodeAttempt` in
+[`crates/jouleclaw-graph`](./jouleclaw-rs/crates/jouleclaw-graph/).
+
 ---
 
 ## 18. Mission orchestrator

jouleclaw/jouleclaw-rs/crates/jouleclaw-graph/src/lib.rs

@@ -21,6 +21,15 @@
 //!   graph; that is a separate concern from "did this step finish."
 //!   The brief excluded time-travel replay in v1; this is its
 //!   structural enforcement.
+//! - **A node in flight survives the process.** A consumer that
+//!   records a node's dispatch ([`Run::dispatch`]) before running it
+//!   leaves the node on record as in flight, so a process that dies in
+//!   the middle of it does not resume as if the node had never begun.
+//!   [`Run::recover`] marks such a node interrupted. A node dispatched as
+//!   [`Replay::Safe`] is then ready again; one dispatched as
+//!   [`Replay::Never`] (the default) is held until the consumer settles
+//!   it, because whether its work took effect is not in the journal
+//!   (SPEC §17.3).
 //! - **No human-in-the-loop checkpoints.** A pending HITL approval is
 //!   modelled by the consumer holding a node un-recorded; the graph
 //!   doesn't enforce a "waiting for approval" state in v1.
@@ -279,6 +288,49 @@ fn reachable_from(g: &RunGraph, start: NodeId) -> HashSet<NodeId> {
 // Journal — append-only event log with blake3 hash chain
 // ─────────────────────────────────────────────────────────────────────
 
+/// Whether a node may run again after a crash interrupted it (SPEC §17.3).
+/// Declared when the node is dispatched, and recorded with the dispatch.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum Replay {
+    /// Running the node again is harmless: a read, or work keyed so a
+    /// repeat finds its first result. After a crash it is ready again, as a
+    /// new attempt.
+    Safe,
+    /// The node may have acted outside the run (sent a message, moved money,
+    /// written to another system), and nothing in the journal says whether it
+    /// did. After a crash it is held for the consumer to settle: record what
+    /// happened, approve a rerun, or fail the run. The default.
+    #[default]
+    Never,
+}
+
+/// Where a dispatched node stands (SPEC §17.3), derived from the journal.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum NodeAttempt {
+    /// Dispatched and not yet recorded: running, or, after a restart, not yet
+    /// found by [`Run::recover`].
+    InFlight { attempt: u32, replay: Replay },
+    /// In flight when its process stopped. Whether its work took effect is
+    /// not knowable from the journal. `observed_uj` is the energy observed for
+    /// the attempt before it stopped: a lower bound on what it spent.
+    Interrupted { attempt: u32, replay: Replay, observed_uj: u64, rerun_approved: bool },
+}
+
+impl NodeAttempt {
+    /// Whether the node may be dispatched again: interrupted, and either safe
+    /// to rerun or approved for one.
+    pub fn may_rerun(&self) -> bool {
+        matches!(self, NodeAttempt::Interrupted { replay: Replay::Safe, .. } | NodeAttempt::Interrupted { rerun_approved: true, .. })
+    }
+
+    fn attempt(&self) -> u32 {
+        match self {
+            NodeAttempt::InFlight { attempt, .. } | NodeAttempt::Interrupted { attempt, .. } => *attempt,
+        }
+    }
+}
+
 /// One event in a [`Journal`]. The variant set is exhaustive and
 /// versioned by the `jc_journal` schema; new variants are additive
 /// and must serialise to a JSON discriminant that older readers
@@ -299,6 +351,31 @@ pub enum RunEvent {
         output_hash: String,
         recorded_at: u64,
     },
+    /// A node's work is about to begin. Recorded before it begins, so a node
+    /// the process dies in is on record as in flight rather than absent
+    /// (SPEC §17.3). `attempt` counts from 1.
+    NodeDispatched {
+        node_id: NodeId,
+        attempt: u32,
+        replay: Replay,
+        dispatched_at: u64,
+    },
+    /// A node in flight when its process stopped, found by [`Run::recover`].
+    /// Its attempt may or may not have taken effect; `observed_uj` is the
+    /// energy observed for it before it stopped, a lower bound on its spend.
+    NodeInterrupted {
+        node_id: NodeId,
+        attempt: u32,
+        observed_uj: u64,
+        recovered_at: u64,
+    },
+    /// The consumer approved running an interrupted [`Replay::Never`] node
+    /// again, and says who approved it.
+    RerunApproved {
+        node_id: NodeId,
+        approved_by: String,
+        approved_at: u64,
+    },
     /// An explicit energy observation that does NOT complete a node.
     /// Useful when a per-token attributor, an external meter, or a
     /// retry attempt produces an energy reading the runtime should
@@ -761,6 +838,16 @@ pub enum RunError {
     AlreadyRecorded(NodeId),
     #[error("run is no longer in progress (status: {0:?})")]
     NotInProgress(RunStatus),
+    /// The node is dispatched and not yet recorded.
+    #[error("node {0} is already in flight")]
+    InFlight(NodeId),
+    /// The node was interrupted and is held: it may have taken effect, so it
+    /// is settled before it runs again (SPEC §17.3).
+    #[error("node {0} was interrupted and is held until it is settled")]
+    Held(NodeId),
+    /// A rerun was approved for a node that is not held.
+    #[error("node {0} is not an interrupted node awaiting settlement")]
+    NotHeld(NodeId),
 }
 
 impl Run {
@@ -790,14 +877,20 @@ impl Run {
 
     /// Nodes that are now ready to dispatch — present in the graph, not
     /// yet completed, and whose merge strategy is satisfied. The entry
-    /// node is ready until it is recorded.
+    /// node is ready until it is recorded. A node in flight is not ready,
+    /// so it is not dispatched twice, and neither is an interrupted node
+    /// until it may run again ([`NodeAttempt::may_rerun`]).
     pub fn ready_nodes(&self) -> Vec<NodeId> {
         let done = self.completed_nodes();
+        let attempts = self.attempts();
         let mut out = Vec::new();
         for n in &self.graph.nodes {
             if done.contains(&n.id) {
                 continue;
             }
+            if attempts.get(&n.id).is_some_and(|a| !a.may_rerun()) {
+                continue;
+            }
             let preds = self.graph.predecessors(n.id);
             let ready = if preds.is_empty() {
                 // No predecessors → only the entry is ready by
@@ -832,7 +925,10 @@ impl Run {
         if self.completed_nodes().contains(&checkpoint.node_id) {
             return Err(RunError::AlreadyRecorded(checkpoint.node_id));
         }
-        if !self.ready_nodes().contains(&checkpoint.node_id) {
+        // A dispatched node records its result; an interrupted one is settled
+        // by recording what happened. Either was ready when it was dispatched.
+        let dispatched = self.attempt_state(checkpoint.node_id).is_some();
+        if !dispatched && !self.ready_nodes().contains(&checkpoint.node_id) {
             return Err(RunError::NotReady(checkpoint.node_id));
         }
         let event = RunEvent::NodeRecorded {
@@ -853,6 +949,128 @@ impl Run {
         Ok(())
     }
 
+    /// Record that a ready node's work is about to begin, and how it may be
+    /// treated if a crash interrupts it (SPEC §17.3). Call it before running
+    /// the node: the record is what lets a restarted process tell a node that
+    /// began from one that never did. Returns the attempt number, counting
+    /// from 1.
+    pub fn dispatch(&mut self, node_id: NodeId, replay: Replay, dispatched_at: u64) -> Result<u32, RunError> {
+        if self.status != RunStatus::InProgress {
+            return Err(RunError::NotInProgress(self.status));
+        }
+        if self.graph.node(node_id).is_none() {
+            return Err(RunError::UnknownNode(node_id));
+        }
+        if self.completed_nodes().contains(&node_id) {
+            return Err(RunError::AlreadyRecorded(node_id));
+        }
+        let prior = self.attempt_state(node_id);
+        match prior {
+            Some(NodeAttempt::InFlight { .. }) => return Err(RunError::InFlight(node_id)),
+            Some(a @ NodeAttempt::Interrupted { .. }) if !a.may_rerun() => return Err(RunError::Held(node_id)),
+            _ => {}
+        }
+        if !self.ready_nodes().contains(&node_id) {
+            return Err(RunError::NotReady(node_id));
+        }
+        let attempt = prior.map_or(1, |a| a.attempt() + 1);
+        self.journal.append(RunEvent::NodeDispatched { node_id, attempt, replay, dispatched_at });
+        Ok(attempt)
+    }
+
+    /// After a restart: mark every node still in flight as interrupted,
+    /// since the process that ran it is gone (SPEC §17.3). Call it once a
+    /// loaded or replayed run is in the new process, before dispatching
+    /// anything. Returns the nodes it marked. A node dispatched as
+    /// [`Replay::Safe`] is ready again; one dispatched as [`Replay::Never`]
+    /// is held until it is settled.
+    pub fn recover(&mut self, recovered_at: u64) -> Vec<NodeId> {
+        let mut marked = Vec::new();
+        if self.status != RunStatus::InProgress {
+            return marked;
+        }
+        let observed = self.observed_since_dispatch();
+        for (node_id, a) in self.attempts() {
+            if let NodeAttempt::InFlight { attempt, .. } = a {
+                let observed_uj = observed.get(&node_id).copied().unwrap_or(0);
+                self.journal.append(RunEvent::NodeInterrupted { node_id, attempt, observed_uj, recovered_at });
+                marked.push(node_id);
+            }
+        }
+        marked
+    }
+
+    /// Settle a held node by running it again: whoever decides records
+    /// that they did, so the rerun is attributable (SPEC §17.3).
+    pub fn approve_rerun(&mut self, node_id: NodeId, approved_by: impl Into<String>, approved_at: u64) -> Result<(), RunError> {
+        match self.attempt_state(node_id) {
+            Some(NodeAttempt::Interrupted { replay: Replay::Never, rerun_approved: false, .. }) => {
+                self.journal.append(RunEvent::RerunApproved { node_id, approved_by: approved_by.into(), approved_at });
+                Ok(())
+            }
+            _ => Err(RunError::NotHeld(node_id)),
+        }
+    }
+
+    /// Where a dispatched, unrecorded node stands, if it is one.
+    pub fn attempt_state(&self, node_id: NodeId) -> Option<NodeAttempt> {
+        self.attempts().remove(&node_id)
+    }
+
+    /// Nodes interrupted by a stopped process and held until settled.
+    pub fn held_nodes(&self) -> Vec<(NodeId, NodeAttempt)> {
+        self.attempts().into_iter().filter(|(_, a)| matches!(a, NodeAttempt::Interrupted { .. }) && !a.may_rerun()).collect()
+    }
+
+    /// The lifecycle of every dispatched node not yet recorded, by
+    /// replaying the dispatch events in the journal.
+    fn attempts(&self) -> std::collections::BTreeMap<NodeId, NodeAttempt> {
+        let mut out = std::collections::BTreeMap::new();
+        for entry in &self.journal.entries {
+            match &entry.event {
+                RunEvent::NodeDispatched { node_id, attempt, replay, .. } => {
+                    out.insert(*node_id, NodeAttempt::InFlight { attempt: *attempt, replay: *replay });
+                }
+                RunEvent::NodeRecorded { node_id, .. } => {
+                    out.remove(node_id);
+                }
+                RunEvent::NodeInterrupted { node_id, attempt, observed_uj, .. } => {
+                    let replay = match out.get(node_id) {
+                        Some(NodeAttempt::InFlight { replay, .. }) => *replay,
+                        _ => Replay::Never,
+                    };
+                    out.insert(*node_id, NodeAttempt::Interrupted { attempt: *attempt, replay, observed_uj: *observed_uj, rerun_approved: false });
+                }
+                RunEvent::RerunApproved { node_id, .. } => {
+                    if let Some(NodeAttempt::Interrupted { rerun_approved, .. }) = out.get_mut(node_id) {
+                        *rerun_approved = true;
+                    }
+                }
+                _ => {}
+            }
+        }
+        out
+    }
+
+    /// Energy observed for each node since its latest dispatch.
+    fn observed_since_dispatch(&self) -> std::collections::BTreeMap<NodeId, u64> {
+        let mut out = std::collections::BTreeMap::new();
+        for entry in &self.journal.entries {
+            match &entry.event {
+                RunEvent::NodeDispatched { node_id, .. } => {
+                    out.insert(*node_id, 0u64);
+                }
+                RunEvent::EnergyObserved { node_id, observed_uj, .. } => {
+                    if let Some(total) = out.get_mut(node_id) {
+                        *total = total.saturating_add(*observed_uj);
+                    }
+                }
+                _ => {}
+            }
+        }
+        out
+    }
+
     /// Mark the run as failed (consumer decides when; e.g. a verifier
     /// rejected a critical node's output). Appends a
     /// [`RunEvent::Failed`] to the journal.
@@ -989,6 +1207,9 @@ impl Run {
                 RunEvent::EnergyObserved { .. } => {}
                 RunEvent::HandoffRecorded { .. } => {}
                 RunEvent::FeedbackReported { .. } => {}
+                // The dispatch lifecycle is derived from the journal on demand
+                // (`Run::attempts`), so replay carries it without rebuilding it.
+                RunEvent::NodeDispatched { .. } | RunEvent::NodeInterrupted { .. } | RunEvent::RerunApproved { .. } => {}
             }
         }
         Ok(run)
@@ -1522,6 +1743,120 @@ mod tests {
         p
     }
 
+    /// A two-node run, nothing recorded: send (1), then report (2).
+    fn two(id: &str) -> Run {
+        let g = RunGraph {
+            entry: NodeId(1),
+            terminals: vec![NodeId(2)],
+            nodes: vec![n(1, "send"), n(2, "report")],
+            edges: vec![edge(1, 2)],
+        };
+        Run::start(g, id).unwrap()
+    }
+
+    /// The process dies: a new one has only the journal.
+    fn restart(run: &Run) -> Run {
+        Run::replay_from(run.graph.clone(), &run.journal.entries, run.run_id.clone()).unwrap()
+    }
+
+    /// A node the process died in is not run again as if it had never
+    /// begun. Without a dispatch record it used to be ready again after a
+    /// restart, so the next dispatch repeated its work silently. Dispatched
+    /// as `Never`, it is held: its energy so far is kept, and running it
+    /// again takes an attributable approval.
+    #[test]
+    fn an_interrupted_node_is_held_not_rerun() {
+        let mut run = two("held");
+        assert_eq!(run.dispatch(NodeId(1), Replay::Never, 10).unwrap(), 1);
+        run.observe_energy(NodeId(1), 40, Provenance::HwShunt);
+
+        let mut after = restart(&run);
+        assert!(!after.ready_nodes().contains(&NodeId(1)), "in flight, so not ready even before recovery");
+        assert_eq!(after.recover(20), vec![NodeId(1)]);
+        assert_eq!(
+            after.attempt_state(NodeId(1)),
+            Some(NodeAttempt::Interrupted { attempt: 1, replay: Replay::Never, observed_uj: 40, rerun_approved: false })
+        );
+        assert!(after.ready_nodes().is_empty());
+        assert_eq!(after.dispatch(NodeId(1), Replay::Never, 21), Err(RunError::Held(NodeId(1))));
+        assert_eq!(after.held_nodes().len(), 1);
+
+        after.approve_rerun(NodeId(1), "did:key:operator", 30).unwrap();
+        assert_eq!(after.ready_nodes(), vec![NodeId(1)]);
+        assert_eq!(after.dispatch(NodeId(1), Replay::Never, 31).unwrap(), 2);
+        after.record(ck(1, 90, Provenance::HwShunt)).unwrap();
+        assert_eq!(after.ready_nodes(), vec![NodeId(2)]);
+        assert!(after.held_nodes().is_empty());
+    }
+
+    /// A node dispatched as safe to rerun is ready again after a crash, as a
+    /// new attempt, and needs no approval.
+    #[test]
+    fn a_safe_node_is_ready_again_after_a_crash() {
+        let mut run = two("safe");
+        run.dispatch(NodeId(1), Replay::Safe, 10).unwrap();
+        let mut after = restart(&run);
+        after.recover(20);
+        assert_eq!(after.ready_nodes(), vec![NodeId(1)]);
+        assert_eq!(after.approve_rerun(NodeId(1), "anyone", 21), Err(RunError::NotHeld(NodeId(1))));
+        assert_eq!(after.dispatch(NodeId(1), Replay::Safe, 22).unwrap(), 2);
+    }
+
+    /// A node in flight is not ready, so it is not dispatched twice.
+    #[test]
+    fn an_in_flight_node_is_not_dispatched_twice() {
+        let mut run = two("twice");
+        run.dispatch(NodeId(1), Replay::Safe, 10).unwrap();
+        assert!(run.ready_nodes().is_empty());
+        assert_eq!(run.dispatch(NodeId(1), Replay::Safe, 11), Err(RunError::InFlight(NodeId(1))));
+        run.record(ck(1, 5, Provenance::HwShunt)).unwrap();
+        assert_eq!(run.ready_nodes(), vec![NodeId(2)]);
+        assert_eq!(run.attempt_state(NodeId(1)), None);
+    }
+
+    /// Recording what an interrupted node did settles it.
+    #[test]
+    fn recording_settles_an_interrupted_node() {
+        let mut run = two("settle");
+        run.dispatch(NodeId(1), Replay::Never, 10).unwrap();
+        let mut after = restart(&run);
+        after.recover(20);
+        after.record(ck(1, 70, Provenance::Estimator)).unwrap();
+        assert_eq!(after.ready_nodes(), vec![NodeId(2)]);
+        assert!(after.held_nodes().is_empty());
+    }
+
+    /// Loading a run through a store is a restart too. Until it is
+    /// recovered, the node stays in flight and out of `ready_nodes`, so a
+    /// consumer that forgets `recover` stalls rather than repeating work.
+    #[test]
+    fn a_run_loaded_from_a_store_keeps_its_node_in_flight() {
+        let dir = tmpdir("in-flight");
+        let mut store = FileRunStore::open(&dir).unwrap();
+        let mut run = two("stored");
+        run.dispatch(NodeId(1), Replay::Never, 10).unwrap();
+        store.save(&run);
+        let mut loaded = store.load("stored").unwrap();
+        assert!(loaded.ready_nodes().is_empty());
+        assert_eq!(loaded.recover(20), vec![NodeId(1)]);
+        assert_eq!(loaded.held_nodes().len(), 1);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    /// The dispatch lifecycle is in the hash chain and replays exactly.
+    #[test]
+    fn dispatch_events_are_chained_and_replay_exactly() {
+        let mut run = two("chain");
+        run.dispatch(NodeId(1), Replay::Never, 10).unwrap();
+        let mut after = restart(&run);
+        after.recover(20);
+        after.approve_rerun(NodeId(1), "did:key:operator", 30).unwrap();
+        assert!(after.journal.verify_chain().is_ok());
+        let again = restart(&after);
+        assert_eq!(again.journal, after.journal);
+        assert_eq!(again.attempt_state(NodeId(1)), after.attempt_state(NodeId(1)));
+    }
+
     fn linear_run(id: &str) -> Run {
         let g = RunGraph {
             entry: NodeId(1),

jouleclaw/jouleclaw-rs/crates/jouleclaw-stuck/src/lib.rs

@@ -511,6 +511,12 @@ fn last_observed_timestamp(entries: &[JournalEntry]) -> Option<u64> {
         let ts = match &e.event {
             RunEvent::NodeRecorded { recorded_at, .. } => Some(*recorded_at),
             RunEvent::HandoffRecorded { closed_at, .. } => Some(*closed_at),
+            // A node beginning, found interrupted, or approved for a rerun is
+            // activity too: a run whose last event is a long node's dispatch
+            // is working, not abandoned.
+            RunEvent::NodeDispatched { dispatched_at, .. } => Some(*dispatched_at),
+            RunEvent::NodeInterrupted { recovered_at, .. } => Some(*recovered_at),
+            RunEvent::RerunApproved { approved_at, .. } => Some(*approved_at),
             _ => None,
         };
         if let Some(t) = ts {
@@ -705,6 +711,24 @@ mod tests {
         assert_eq!(sigs.len(), 1);
     }
 
+    /// A node's dispatch is activity: a run whose last event is a long
+    /// node beginning is working, not abandoned.
+    #[test]
+    fn a_recent_dispatch_is_activity() {
+        let mut r = fresh_run("dispatching");
+        r.record_handoff(WorkerHandoffEvent {
+            worker: "w".into(),
+            closed_at: 1_700_000_000,
+            ..Default::default()
+        });
+        r.dispatch(NodeId(1), jouleclaw_graph::Replay::Never, 1_700_000_480).unwrap();
+        let d = AbandonmentDetector {
+            idle_grace_secs: 60,
+            now_secs: 1_700_000_500,
+        };
+        assert!(d.detect(&r.journal).is_empty(), "dispatched 20 s ago");
+    }
+
     #[test]
     fn abandonment_quiet_inside_grace_period() {
         let mut r = fresh_run("fresh");