jcp/SPEC.md

@@ -481,6 +481,38 @@ nothing is debited and nothing runs.
 A grant carrying no `FlowBelow` caveat is byte-identical to one minted before
 flow existed, and an enforcement point with no taint floor configured decides
 exactly as it did before. Conformance class **C25**.
+### 4.11 Durable spend: a restart is not a fresh budget (REQUIRED where a run can restart)
+
+A grant's budget is enforced against the spend an enforcement point has
+accumulated, and that accumulation is state. Where the enforcement point can
+restart — any long-running orchestrator, worker pool or gateway — an
+implementation that holds the spend only in memory gives every grant its full
+budget back on restart. A crash loop then spends a grant's budget repeatedly,
+and the ceiling that was meant to bound the work bounds one process lifetime
+instead.
+
+An implementation whose enforcement point can restart therefore **MUST** record
+every allowed decision durably, and on restart **MUST** re-establish the spend
+from that record before deciding anything new. Re-establishing it by **replaying
+the recorded decisions through the gate** is RECOMMENDED over restoring a stored
+total: a replay cannot produce a spend the gate would not have produced, and a
+stored total is one more figure that can be wrong or tampered with.
+
+**The decision is recorded before the action runs.** Recording it afterwards
+loses the debit for any action interrupted between the two — the grant was
+debited when the decision was made, so a record written after the fact reports a
+budget that has more left in it than it does.
+
+That ordering leaves one state the record cannot resolve. A recorded decision
+with no recorded outcome means the enforcement point stopped while the action
+was in flight, and **whether the action took effect is not knowable from the
+record** — only the party that was running it knows. An implementation **MUST
+NOT** silently retry such an action, and **MUST NOT** report it as completed or
+as failed. Its energy **MUST** stay debited, because the decision that debited it
+was made and sealed. The honest resolution is to surface it as unresolved,
+naming the party that held it, and let an operator or that party settle it:
+retrying risks performing paid work twice, discarding risks losing it, and
+choosing between those without asking is a decision the record does not support.
 
 ## 5. Energy (REQUIRED)
 

jcp/jcp-rs/crates/jcp-orchestrate/src/lib.rs

@@ -26,6 +26,32 @@
 //! decisions about a grant, not the absence of one — so the item is recorded as
 //! denied in the [`RunReport`] and never reaches a harness.
 //!
+//! ## Surviving the process
+//!
+//! "The whole run is an ordered ledger of sealed decisions" was true only for
+//! as long as the process lived. Every item's state, the sealed receipts, and —
+//! the part that matters — each harness's **spend** were held in memory, so a
+//! restart began with every budget whole. A crash loop could spend a harness's
+//! joule budget over and over, and a task that had already been paid for and
+//! run would be dispatched again.
+//!
+//! A run now emits a [`JournalEntry`] for every transition, through a
+//! [`Journal`] the caller supplies — [`MemoryJournal`] keeps the old behaviour,
+//! and a deployment that wants durability writes them somewhere that outlives
+//! the process. [`Orchestrator::resume`] rebuilds a run from those entries:
+//! item states, attempt counts, and the spend, **re-debited against the
+//! grants** so budgets are not restored by restarting.
+//!
+//! One case is deliberately left unresolved. The journal records a dispatch
+//! *before* the harness runs, because recording it afterwards would lose the
+//! debit that was already sealed. A `Dispatched` entry with no settlement
+//! therefore means the process died while a harness held the task, and whether
+//! the work happened is not knowable from here — only the harness knows. Such
+//! an item resumes as [`WorkState::Interrupted`] and is **not** re-dispatched:
+//! re-running it risks doing paid work twice, dropping it risks losing it, and
+//! choosing silently between those is not this crate's call. Its joules are
+//! counted either way, because the debit was sealed.
+//!
 //! Ingest, not compete: the compose-many-harnesses shape is the surface; JCP
 //! contributes the capability, the energy budget, and the sealed audit.
 
@@ -42,6 +68,15 @@ use jcp_schema::{
 /// Errors registering a harness or running the loop.
 #[derive(Debug, thiserror::Error)]
 pub enum OrchestrateError {
+    /// A journal entry names a task no `Submitted` entry introduced: the
+    /// journal is truncated at the front or out of order.
+    #[error("journal names task `{0}`, which was never submitted")]
+    UnknownTask(String),
+    /// A journal entry names a harness that is not registered. Replay cannot
+    /// reproduce the grant the decision was made against, so it refuses rather
+    /// than guess.
+    #[error("journal names harness `{0}`, which is not registered")]
+    UnknownHarness(String),
     /// Minting a harness grant failed.
     #[error("grant: {0}")]
     Grant(#[from] jcp_grant::GrantError),
@@ -128,6 +163,89 @@ pub fn harness_resource(id: &str) -> ContentId {
     ContentId::of_bytes(format!("harness:{id}").as_bytes())
 }
 
+// ───────────────────────────── the journal ─────────────────────────────
+
+/// One transition in a run, in the order it happened.
+///
+/// Enough to rebuild a run: which item went to which harness on which attempt
+/// and at what cost, and how each one settled. The sealed receipt travels with
+/// the dispatch so a resumed run's ledger is the same ledger, not a summary of
+/// one.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum JournalEntry {
+    /// An item was submitted.
+    Submitted {
+        /// The task's id.
+        task_id: String,
+        /// Its kind, for a resumed run to re-route it.
+        kind: String,
+        /// Its payload.
+        payload: Vec<u8>,
+    },
+    /// A dispatch decision was sealed. Written **before** the harness runs.
+    ///
+    /// `allowed: false` is a sealed denial — the harness was out of budget or
+    /// out of scope — and the loop failed over to the next candidate. Only an
+    /// allowed dispatch debits.
+    Dispatched {
+        /// The task's id.
+        task_id: String,
+        /// The harness the decision was about.
+        harness: String,
+        /// Which attempt this was.
+        attempt: u32,
+        /// The joules the decision authorised, in µJ.
+        cost_uj: u64,
+        /// Whether the decision allowed the dispatch.
+        allowed: bool,
+        /// The sealed JCR-1 receipt for the decision.
+        receipt: Vec<u8>,
+    },
+    /// An item reached a terminal state, or went back to `Pending` for a retry.
+    Settled {
+        /// The task's id.
+        task_id: String,
+        /// Where it landed.
+        state: WorkState,
+    },
+}
+
+/// Where a run's [`JournalEntry`] stream goes.
+///
+/// A deployment that wants a run to survive its process implements this over
+/// something durable. The orchestrator never reads back through this trait —
+/// [`Orchestrator::resume`] takes the entries directly — so an implementation
+/// only has to append, and may append asynchronously as long as it does not
+/// reorder.
+pub trait Journal {
+    /// Record one transition.
+    fn append(&mut self, entry: JournalEntry);
+}
+
+/// The default journal: entries in memory, which is where a run's record lived
+/// before there was a journal at all.
+#[derive(Clone, Debug, Default)]
+pub struct MemoryJournal {
+    entries: Vec<JournalEntry>,
+}
+
+impl MemoryJournal {
+    /// The entries recorded so far, in order.
+    pub fn entries(&self) -> &[JournalEntry] {
+        &self.entries
+    }
+    /// Take the entries, leaving the journal empty.
+    pub fn into_entries(self) -> Vec<JournalEntry> {
+        self.entries
+    }
+}
+
+impl Journal for MemoryJournal {
+    fn append(&mut self, entry: JournalEntry) {
+        self.entries.push(entry);
+    }
+}
+
 // ───────────────────────────── work state ─────────────────────────────
 
 /// Where a work item is in its lifecycle.
@@ -141,6 +259,22 @@ pub enum WorkState {
     Denied(String),
     /// Ran but failed terminally, or exhausted its retries (carries the reason).
     Failed(String),
+    /// A dispatch was sealed and debited, and no settlement followed: the
+    /// process died while a harness held the task.
+    ///
+    /// Whether the work happened is not knowable from the journal — the
+    /// decision is recorded before the harness runs, because recording it after
+    /// would lose a debit that was already sealed. Carries the harness that held
+    /// it and the attempt number, so an operator can ask that harness. The
+    /// loop does not re-dispatch it: re-running risks paying twice for work
+    /// already done, dropping it risks losing the item, and choosing between
+    /// those without asking the harness is not a decision this crate can make.
+    Interrupted {
+        /// The harness that was holding the task.
+        harness: String,
+        /// Which attempt was in flight.
+        attempt: u32,
+    },
 }
 
 struct WorkItem {
@@ -158,12 +292,30 @@ pub struct RunReport {
     pub denied: usize,
     /// Items that failed or exhausted retries.
     pub failed: usize,
+    /// Items whose dispatch was sealed and never settled — the process died
+    /// while a harness held them. Neither done nor failed: unresolved, and
+    /// counted separately so a resumed run cannot report them as either.
+    pub interrupted: usize,
     /// Total joules authorised across the run, in µJ.
     pub total_uj: u64,
     /// Sealed decision receipts (the audit ledger length).
     pub receipts: usize,
 }
 
+/// What a [`Orchestrator::replay`] rebuilt.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ResumeReport {
+    /// Items restored.
+    pub items: usize,
+    /// Dispatch decisions put back through the gate.
+    pub replayed_dispatches: usize,
+    /// Items left [`WorkState::Interrupted`].
+    pub interrupted: usize,
+    /// Joules the replay re-debited across every grant, in µJ. Equal to what
+    /// the original run had spent at the point the journal ends.
+    pub spend_uj: u64,
+}
+
 // ───────────────────────────── the orchestrator ─────────────────────────────
 
 /// The control plane: a fleet of harnesses, each capability-scoped and
@@ -183,6 +335,7 @@ pub struct Orchestrator {
     receipts: Vec<Vec<u8>>,
     max_attempts: u32,
     now: u64,
+    journal: Box<dyn Journal>,
 }
 
 impl Orchestrator {
@@ -206,9 +359,22 @@ impl Orchestrator {
             receipts: Vec::new(),
             max_attempts: max_attempts.max(1),
             now: 0,
+            journal: Box::new(MemoryJournal::default()),
         }
     }
 
+    /// Send this run's transitions to `journal` instead of the default
+    /// in-memory one.
+    ///
+    /// A deployment that wants the run to survive its process gives a journal
+    /// backed by something that does. Set before [`Orchestrator::submit`]: a
+    /// journal installed later has not seen the entries already written, and
+    /// resuming from it would rebuild a run missing its earlier items.
+    pub fn with_journal(mut self, journal: Box<dyn Journal>) -> Self {
+        self.journal = journal;
+        self
+    }
+
     /// Register a harness with the task `kinds` it may handle and a `budget_uj`
     /// joule ceiling. Mints the capability-scoped grant.
     pub fn register(
@@ -242,9 +408,15 @@ impl Orchestrator {
 
     /// Submit a work item.
     pub fn submit(&mut self, task: Task) {
+        self.journal.append(JournalEntry::Submitted {
+            task_id: task.id.clone(),
+            kind: task.kind.clone(),
+            payload: task.payload.clone(),
+        });
         self.items.push(WorkItem { task, state: WorkState::Pending, attempts: 0 });
     }
 
+
     /// Run the reconcile loop until no item is pending, returning the report.
     /// Each iteration dispatches one pending item; a retried item goes back to
     /// pending and is bounded by `max_attempts`, so the loop always terminates.
@@ -252,6 +424,9 @@ impl Orchestrator {
         while let Some(i) = self.items.iter().position(|w| w.state == WorkState::Pending) {
             self.dispatch(i);
         }
+        // `Interrupted` items are not `Pending`, so the loop above never offers
+        // them to a harness. That is the point: only the harness that held one
+        // knows whether the work happened.
         self.report()
     }
 
@@ -279,7 +454,12 @@ impl Orchestrator {
             // decide against: the item is refused before any harness sees it.
             // This seals no receipt — JCP seals decisions about a grant, not the
             // absence of one.
-            self.items[i].state = WorkState::Denied("no harness is granted this kind".into());
+            let state = WorkState::Denied("no harness is granted this kind".into());
+            self.journal.append(JournalEntry::Settled {
+                task_id: task.id.clone(),
+                state: state.clone(),
+            });
+            self.items[i].state = state;
             return;
         }
 
@@ -302,6 +482,17 @@ impl Orchestrator {
             let decision = self.runtime.decide(&frame, &grant, self.caller, Energy::estimator(cost), self.now);
             let receipt = seal(&frame, &grant.grant, &decision, &self.steward_sk, &self.steward)
                 .expect("seal dispatch decision");
+            // Journalled before the harness runs. After would be wrong: the
+            // decision has already debited the grant, and a process that dies
+            // between the two would leave joules spent with no record of it.
+            self.journal.append(JournalEntry::Dispatched {
+                task_id: task.id.clone(),
+                harness: id.clone(),
+                attempt: self.items[i].attempts,
+                cost_uj: cost,
+                allowed: decision.is_allow(),
+                receipt: receipt.clone(),
+            });
             self.receipts.push(receipt);
             self.now += 1;
 
@@ -314,7 +505,7 @@ impl Orchestrator {
             }
 
             let outcome = self.harnesses[&id].run(&task);
-            self.items[i].state = match outcome.status {
+            let state = match outcome.status {
                 Status::Done => WorkState::Done,
                 Status::Retry => {
                     if self.items[i].attempts < self.max_attempts {
@@ -325,11 +516,18 @@ impl Orchestrator {
                 }
                 Status::Failed => WorkState::Failed(outcome.output),
             };
+            self.journal.append(JournalEntry::Settled {
+                task_id: task.id.clone(),
+                state: state.clone(),
+            });
+            self.items[i].state = state;
             return;
         }
 
         // Every granted harness refused — each refusal is sealed in the ledger.
-        self.items[i].state = WorkState::Denied(format!("gate: {last_reason}"));
+        let state = WorkState::Denied(format!("gate: {last_reason}"));
+        self.journal.append(JournalEntry::Settled { task_id: task.id.clone(), state: state.clone() });
+        self.items[i].state = state;
     }
 
     fn report(&self) -> RunReport {
@@ -339,6 +537,7 @@ impl Orchestrator {
                 WorkState::Done => r.done += 1,
                 WorkState::Denied(_) => r.denied += 1,
                 WorkState::Failed(_) => r.failed += 1,
+                WorkState::Interrupted { .. } => r.interrupted += 1,
                 WorkState::Pending => {}
             }
         }
@@ -346,6 +545,100 @@ impl Orchestrator {
         r
     }
 
+    /// Rebuild this run from a journal, before calling [`Orchestrator::run`].
+    ///
+    /// Register the same harnesses, with the same ids, kinds and budgets, then
+    /// replay. The grants are minted deterministically (a fixed nonce), so an
+    /// identical registration reproduces identical grant ids and the journal's
+    /// decisions land against the same grants they were made against.
+    ///
+    /// **Spend is re-derived, not asserted.** Each allowed dispatch is put back
+    /// through [`Runtime::decide`] in order, so the budgets end where the
+    /// original run left them — the point of the whole exercise, since a fresh
+    /// process otherwise starts with every budget whole and a crash loop could
+    /// spend a harness's budget over and over. Replaying through the gate rather
+    /// than writing a total also means a resumed run cannot be given a spend the
+    /// gate would not have produced.
+    ///
+    /// A dispatch with no following settlement resumes as
+    /// [`WorkState::Interrupted`]: its joules are counted, and it is not offered
+    /// to a harness again. See the module doc for why that is left to an
+    /// operator.
+    ///
+    /// Replaying does not re-journal: the entries are already the record.
+    pub fn replay(&mut self, entries: &[JournalEntry]) -> Result<ResumeReport, OrchestrateError> {
+        let mut report = ResumeReport::default();
+        // The dispatch each task is waiting on a settlement for, if any.
+        let mut in_flight: BTreeMap<String, (String, u32)> = BTreeMap::new();
+
+        for entry in entries {
+            match entry {
+                JournalEntry::Submitted { task_id, kind, payload } => {
+                    self.items.push(WorkItem {
+                        task: Task::new(task_id.clone(), kind.clone(), payload.clone()),
+                        state: WorkState::Pending,
+                        attempts: 0,
+                    });
+                    report.items += 1;
+                }
+                JournalEntry::Dispatched { task_id, harness, attempt, cost_uj, allowed, receipt } => {
+                    let Some(i) = self.items.iter().position(|w| w.task.id == *task_id) else {
+                        return Err(OrchestrateError::UnknownTask(task_id.clone()));
+                    };
+                    let Some(grant) = self.grants.get(harness).cloned() else {
+                        return Err(OrchestrateError::UnknownHarness(harness.clone()));
+                    };
+                    self.items[i].attempts = *attempt;
+                    self.receipts.push(receipt.clone());
+                    self.now += 1;
+                    if *allowed {
+                        // Through the gate, so the debit is the gate's and not a
+                        // figure this function made up.
+                        let task = self.items[i].task.clone();
+                        let frame = Frame {
+                            kind: FrameKind::AgentTask,
+                            trust: Trust::Control,
+                            grant_id: grant.grant.id(),
+                            action: task.kind.clone(),
+                            resource: harness_resource(harness),
+                            payload_hash: ContentId::of_bytes(&task.payload),
+                        };
+                        self.runtime.decide(
+                            &frame,
+                            &grant,
+                            self.caller,
+                            Energy::estimator(*cost_uj),
+                            self.now,
+                        );
+                        in_flight.insert(task_id.clone(), (harness.clone(), *attempt));
+                    }
+                    report.replayed_dispatches += 1;
+                }
+                JournalEntry::Settled { task_id, state } => {
+                    let Some(i) = self.items.iter().position(|w| w.task.id == *task_id) else {
+                        return Err(OrchestrateError::UnknownTask(task_id.clone()));
+                    };
+                    self.items[i].state = state.clone();
+                    in_flight.remove(task_id);
+                }
+            }
+        }
+
+        for (task_id, (harness, attempt)) in in_flight {
+            if let Some(i) = self.items.iter().position(|w| w.task.id == task_id) {
+                self.items[i].state = WorkState::Interrupted { harness, attempt };
+                report.interrupted += 1;
+            }
+        }
+        report.spend_uj = self
+            .order
+            .iter()
+            .filter_map(|id| self.grants.get(id))
+            .map(|g| self.runtime.spent(&g.grant.id()))
+            .sum();
+        Ok(report)
+    }
+
     /// The state of a submitted item by id, if present.
     pub fn state_of(&self, task_id: &str) -> Option<&WorkState> {
         self.items.iter().find(|w| w.task.id == task_id).map(|w| &w.state)
@@ -546,4 +839,271 @@ mod tests {
         let dup = o.register(Box::new(Fixed { id: "a".into(), kinds: vec!["x".into()], cost: 1 }), &["x"], 10);
         assert!(matches!(dup, Err(OrchestrateError::DuplicateHarness(_))));
     }
+    // ─── surviving the process ───────────────────────────────────────
+
+    /// A journal that hands its entries back, standing in for one that outlives
+    /// the process. `Rc<RefCell<..>>` so the test can read it while the
+    /// orchestrator owns the `Box<dyn Journal>`.
+    #[derive(Clone, Default)]
+    struct SharedJournal(std::rc::Rc<std::cell::RefCell<Vec<JournalEntry>>>);
+    impl Journal for SharedJournal {
+        fn append(&mut self, entry: JournalEntry) {
+            self.0.borrow_mut().push(entry);
+        }
+    }
+    impl SharedJournal {
+        fn entries(&self) -> Vec<JournalEntry> {
+            self.0.borrow().clone()
+        }
+    }
+
+    fn fixed(id: &str, kind: &str, cost: u64) -> Box<dyn Harness> {
+        Box::new(Fixed { id: id.into(), kinds: vec![kind.into()], cost })
+    }
+
+    #[test]
+    fn a_resumed_run_does_not_restore_the_budget_it_already_spent() {
+        // The hole. Spend lived in the in-memory `Runtime`, so a fresh process
+        // began with every budget whole: a crash loop could spend a harness's
+        // joule budget over and over.
+        let journal = SharedJournal::default();
+        let mut first = orch().with_journal(Box::new(journal.clone()));
+        first.register(fixed("h", "t", 300), &["t"], 1_000).unwrap();
+        for n in 0..3 {
+            first.submit(Task::new(format!("task-{n}"), "t", b"p".to_vec()));
+        }
+        let report = first.run();
+        assert_eq!((report.done, report.denied), (3, 0));
+        let spent = first.spent_uj("h");
+        assert_eq!(spent, 900, "three dispatches at 300 µJ");
+
+        // A new process, same registration, replaying the journal.
+        let mut second = orch();
+        second.register(fixed("h", "t", 300), &["t"], 1_000).unwrap();
+        assert_eq!(second.spent_uj("h"), 0, "premise: a fresh runtime starts whole");
+        let resumed = second.replay(&journal.entries()).expect("replay");
+
+        assert_eq!(resumed.items, 3);
+        assert_eq!(resumed.replayed_dispatches, 3);
+        assert_eq!(resumed.spend_uj, spent, "the replay must land on the original spend");
+        assert_eq!(second.spent_uj("h"), 900, "the budget was restored by restarting");
+
+        // And the replayed state is the finished state, so nothing re-runs.
+        assert_eq!(second.state_of("task-0"), Some(&WorkState::Done));
+        let again = second.run();
+        assert_eq!(again.done, 3);
+        assert_eq!(second.spent_uj("h"), 900, "a resumed run re-dispatched completed work");
+    }
+
+    #[test]
+    fn a_resumed_budget_still_denies_what_it_can_no_longer_afford() {
+        // The spend is not bookkeeping: it has to bind the gate afterwards. A
+        // harness with 1 000 µJ that has spent 900 can afford one more 300 µJ
+        // dispatch only if the restart gave its budget back.
+        let journal = SharedJournal::default();
+        let mut first = orch().with_journal(Box::new(journal.clone()));
+        first.register(fixed("h", "t", 300), &["t"], 1_000).unwrap();
+        for n in 0..3 {
+            first.submit(Task::new(format!("task-{n}"), "t", b"p".to_vec()));
+        }
+        first.run();
+
+        let mut second = orch();
+        second.register(fixed("h", "t", 300), &["t"], 1_000).unwrap();
+        second.replay(&journal.entries()).expect("replay");
+        // One more task, on the resumed budget: 900 spent of 1 000, so a 300 µJ
+        // dispatch does not fit and the gate must refuse it.
+        second.submit(Task::new("task-3", "t", b"p".to_vec()));
+        let report = second.run();
+        assert_eq!(report.denied, 1, "the resumed budget admitted work it could not pay for");
+        assert!(matches!(second.state_of("task-3"), Some(WorkState::Denied(_))));
+    }
+
+    #[test]
+    fn a_dispatch_with_no_settlement_resumes_unresolved_and_is_not_re_run() {
+        // The process died while a harness held the task. Whether the work
+        // happened is not knowable from the journal, so the item is neither
+        // done nor failed, and the loop must not quietly do it again.
+        let journal = SharedJournal::default();
+        let mut first = orch().with_journal(Box::new(journal.clone()));
+        first.register(fixed("h", "t", 300), &["t"], 1_000).unwrap();
+        first.submit(Task::new("task-0", "t", b"p".to_vec()));
+        first.run();
+
+        // Truncate the journal after the dispatch: a crash between the sealed
+        // decision and the outcome.
+        let mut entries = journal.entries();
+        let cut = entries
+            .iter()
+            .position(|e| matches!(e, JournalEntry::Dispatched { .. }))
+            .expect("a dispatch");
+        entries.truncate(cut + 1);
+
+        let runs = std::rc::Rc::new(Cell::new(0u32));
+        let counted = runs.clone();
+        struct Counting {
+            runs: std::rc::Rc<Cell<u32>>,
+        }
+        impl Harness for Counting {
+            fn id(&self) -> &str {
+                "h"
+            }
+            fn handles(&self, kind: &str) -> bool {
+                kind == "t"
+            }
+            fn estimate(&self, _t: &Task) -> u64 {
+                300
+            }
+            fn run(&self, _t: &Task) -> HarnessOutcome {
+                self.runs.set(self.runs.get() + 1);
+                HarnessOutcome::done(300)
+            }
+        }
+
+        let mut second = orch();
+        second.register(Box::new(Counting { runs: counted }), &["t"], 1_000).unwrap();
+        let resumed = second.replay(&entries).expect("replay");
+
+        assert_eq!(resumed.interrupted, 1);
+        assert_eq!(
+            second.state_of("task-0"),
+            Some(&WorkState::Interrupted { harness: "h".into(), attempt: 1 })
+        );
+        // Its joules are counted: the debit was sealed before the harness ran.
+        assert_eq!(second.spent_uj("h"), 300, "an interrupted dispatch's debit was dropped");
+
+        let report = second.run();
+        assert_eq!(runs.get(), 0, "an interrupted item was silently re-run");
+        assert_eq!(report.interrupted, 1);
+        assert_eq!((report.done, report.failed, report.denied), (0, 0, 0), "{report:?}");
+    }
+
+    #[test]
+    fn a_sealed_denial_is_replayed_without_debiting() {
+        // A denied dispatch seals a receipt and spends nothing. The replay must
+        // put the receipt back in the ledger and leave the budget alone.
+        let journal = SharedJournal::default();
+        let mut first = orch().with_journal(Box::new(journal.clone()));
+        // Budget for one dispatch; the second is denied and fails over nowhere.
+        first.register(fixed("h", "t", 600), &["t"], 1_000).unwrap();
+        first.submit(Task::new("task-0", "t", b"p".to_vec()));
+        first.submit(Task::new("task-1", "t", b"p".to_vec()));
+        let report = first.run();
+        assert_eq!((report.done, report.denied), (1, 1), "{report:?}");
+        let denials = journal
+            .entries()
+            .iter()
+            .filter(|e| matches!(e, JournalEntry::Dispatched { allowed: false, .. }))
+            .count();
+        assert_eq!(denials, 1, "one sealed denial");
+
+        let mut second = orch();
+        second.register(fixed("h", "t", 600), &["t"], 1_000).unwrap();
+        let resumed = second.replay(&journal.entries()).expect("replay");
+        assert_eq!(resumed.replayed_dispatches, 2, "both decisions, allow and deny");
+        assert_eq!(second.spent_uj("h"), 600, "only the allowed dispatch debited; a denial costs nothing");
+        assert_eq!(second.receipts().len(), 2, "the ledger is the same ledger");
+    }
+
+    #[test]
+    fn a_journal_naming_something_unregistered_is_refused_not_guessed() {
+        let journal = SharedJournal::default();
+        let mut first = orch().with_journal(Box::new(journal.clone()));
+        first.register(fixed("h", "t", 300), &["t"], 1_000).unwrap();
+        first.submit(Task::new("task-0", "t", b"p".to_vec()));
+        first.run();
+        let entries = journal.entries();
+
+        // A harness that is not registered: the grant its decision was made
+        // against cannot be reproduced, so the replay refuses.
+        let mut other = orch();
+        other.register(fixed("different", "t", 300), &["t"], 1_000).unwrap();
+        assert!(matches!(
+            other.replay(&entries),
+            Err(OrchestrateError::UnknownHarness(h)) if h == "h"
+        ));
+
+        // A journal truncated at the front, so a dispatch names a task that was
+        // never submitted.
+        let mut second = orch();
+        second.register(fixed("h", "t", 300), &["t"], 1_000).unwrap();
+        let headless: Vec<JournalEntry> = entries
+            .iter()
+            .filter(|e| !matches!(e, JournalEntry::Submitted { .. }))
+            .cloned()
+            .collect();
+        assert!(matches!(
+            second.replay(&headless),
+            Err(OrchestrateError::UnknownTask(t)) if t == "task-0"
+        ));
+    }
+
+    #[test]
+    fn the_dispatch_is_journalled_before_the_harness_runs() {
+        // The ordering the debit depends on. The decision debits the grant, so a
+        // process that dies while the harness holds the task must still leave
+        // the dispatch on the record — otherwise the joules are spent and
+        // nothing says so, and a resumed run both loses the debit and re-runs
+        // the task as `Pending`.
+        //
+        // Simulated by a harness that dies where a process would.
+        struct Dying;
+        impl Harness for Dying {
+            fn id(&self) -> &str {
+                "h"
+            }
+            fn handles(&self, kind: &str) -> bool {
+                kind == "t"
+            }
+            fn estimate(&self, _t: &Task) -> u64 {
+                300
+            }
+            fn run(&self, _t: &Task) -> HarnessOutcome {
+                panic!("the process died while the harness held the task");
+            }
+        }
+
+        let journal = SharedJournal::default();
+        let mut o = orch().with_journal(Box::new(journal.clone()));
+        o.register(Box::new(Dying), &["t"], 1_000).unwrap();
+        o.submit(Task::new("task-0", "t", b"p".to_vec()));
+
+        let hook = std::panic::take_hook();
+        std::panic::set_hook(Box::new(|_| {}));
+        let died = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| o.run())).is_err();
+        std::panic::set_hook(hook);
+        assert!(died, "premise: the harness takes the run down with it");
+
+        // The decision is on the record even though no outcome ever was.
+        let entries = journal.entries();
+        let dispatched = entries
+            .iter()
+            .filter(|e| matches!(e, JournalEntry::Dispatched { allowed: true, .. }))
+            .count();
+        assert_eq!(dispatched, 1, "the debited decision was not journalled: {entries:?}");
+        assert!(
+            !entries.iter().any(|e| matches!(e, JournalEntry::Settled { .. })),
+            "nothing settled, so the ordering is what put the dispatch on the record"
+        );
+
+        // And that is enough to resume it as unresolved rather than as pending
+        // work whose joules nobody spent.
+        let mut next = orch();
+        next.register(Box::new(Dying), &["t"], 1_000).unwrap();
+        let resumed = next.replay(&entries).expect("replay");
+        assert_eq!(resumed.interrupted, 1);
+        assert_eq!(next.spent_uj("h"), 300, "the debit survived the crash");
+    }
+
+    #[test]
+    fn the_default_journal_keeps_what_a_run_always_kept() {
+        // No journal installed: the entries go to the in-memory default, and the
+        // run behaves exactly as before.
+        let mut o = orch();
+        o.register(fixed("h", "t", 300), &["t"], 1_000).unwrap();
+        o.submit(Task::new("task-0", "t", b"p".to_vec()));
+        let report = o.run();
+        assert_eq!((report.done, report.interrupted), (1, 0));
+        assert_eq!(o.receipts().len(), 1);
+    }
 }

jouledesk/jouledesk-rs/crates/jouledesk-github/src/lib.rs

@@ -267,6 +267,10 @@ pub fn seal(work: &Work, intent: &str) -> (String, bool) {
         TagOutcome::Done => "completed".to_string(),
         TagOutcome::Denied(r) => format!("denied ({r})"),
         TagOutcome::Failed(r) => format!("failed ({r})"),
+        // Not "failed": the harness had it when the run stopped, so whether it
+        // took effect is unknown (JCP §4.11). The string names the harness,
+        // because that is who can settle it.
+        TagOutcome::Interrupted { harness } => format!("interrupted (held by {harness})"),
     };
     (outcome_str, verified)
 }

jouledesk/jouledesk-rs/crates/jouledesk/examples/channel.rs

@@ -45,6 +45,9 @@ fn main() {
             Some(TagOutcome::Done) => "✓ done, sealed".to_string(),
             Some(TagOutcome::Denied(r)) => format!("✗ refused ({r})"),
             Some(TagOutcome::Failed(r)) => format!("✗ failed ({r})"),
+            Some(TagOutcome::Interrupted { harness }) => {
+                format!("⏸ interrupted — {harness} had it; outcome unknown")
+            }
             None => "· not addressed to @joule".to_string(),
         };
         println!("  {member:>5} │ {msg}\n        └─ {verdict}");

jouledesk/jouledesk-rs/crates/jouledesk/src/lib.rs

@@ -13,7 +13,9 @@
 
 use ed25519_dalek::SigningKey;
 use jcp_efficiency::{Efficiency, EfficiencyLog, Totals};
-use jcp_orchestrate::{Harness, HarnessOutcome, Orchestrator, Task, WorkState};
+use jcp_orchestrate::{
+    Harness, HarnessOutcome, Journal, JournalEntry, Orchestrator, ResumeReport, Task, WorkState,
+};
 use jcp_schema::{Did, Energy};
 
 // Re-exported so a consumer can build a [`Skill`]'s outcome without depending on
@@ -43,6 +45,19 @@ pub enum TagOutcome {
     Denied(String),
     /// Ran but failed terminally, or exhausted its retries. Sealed.
     Failed(String),
+    /// The dispatch was decided and debited, and the process stopped before any
+    /// outcome was recorded (`jcp_orchestrate::WorkState::Interrupted`).
+    ///
+    /// Neither done nor failed: whether the skill's work took effect is not
+    /// knowable from here — only the harness that held it knows. Carries the
+    /// harness so a person can ask it. JCP §4.11 forbids reporting such a task
+    /// as completed *or* as failed, and it is deliberately **not** sealed as an
+    /// efficiency outcome: a `Failed` receipt asserts the spend bought no
+    /// result, which is precisely what nobody can say yet.
+    Interrupted {
+        /// The harness that was holding the task.
+        harness: String,
+    },
 }
 
 /// Parse a channel mention into a `(kind, intent)` tag — the surface binding's job
@@ -99,6 +114,40 @@ impl Harness for Skill {
     }
 }
 
+/// What a work item's state means for the person who tagged the teammate.
+///
+/// One place decides it, so `tag` and [`Teammate::outcome_of`] cannot drift —
+/// and so a test can reach the mapping without going through a dispatch.
+fn outcome_for(state: Option<&WorkState>) -> TagOutcome {
+    match state {
+        Some(WorkState::Done) => TagOutcome::Done,
+        Some(WorkState::Denied(r)) => TagOutcome::Denied(r.clone()),
+        Some(WorkState::Failed(r)) => TagOutcome::Failed(r.clone()),
+        // Named, not swallowed. This used to fall into the catch-all below and
+        // be reported as `Failed("no terminal state")` — and then sealed as
+        // waste, which asserts the spend bought no result. For an interrupted
+        // task that is exactly the unknown (JCP §4.11).
+        Some(WorkState::Interrupted { harness, .. }) => {
+            TagOutcome::Interrupted { harness: harness.clone() }
+        }
+        Some(WorkState::Pending) | None => TagOutcome::Failed("no terminal state".into()),
+    }
+}
+
+/// Whether an outcome seals an `efficiency/1` receipt.
+///
+/// `Done` is a completed unit and `Failed` is spend that bought no result
+/// (waste) — both are findings about what the energy achieved. `Denied` ran
+/// nothing. `Interrupted` may have run everything, so sealing it either way
+/// would state as fact something nobody can say yet (JCP §4.11); its joules stay
+/// debited on the grant by the dispatch decision that incurred them.
+///
+/// A named predicate rather than an inline `matches!`, so a test reads the rule
+/// the code uses instead of restating it.
+fn seals_efficiency(outcome: &TagOutcome) -> bool {
+    matches!(outcome, TagOutcome::Done | TagOutcome::Failed(_))
+}
+
 /// A persistent agent teammate bound to a channel (SPEC §2/§3).
 pub struct Teammate {
     orch: Orchestrator,
@@ -130,6 +179,39 @@ impl Teammate {
         }
     }
 
+    /// Record this teammate's dispatches through `journal`, so a run survives
+    /// the process (JCP §4.11).
+    ///
+    /// Without one, every state the teammate holds — including each skill's
+    /// spend — lives only as long as the process, and a restart hands every
+    /// skill its whole joule budget back. Set before granting skills or tagging:
+    /// a journal installed later has not seen what came before, and replaying it
+    /// would rebuild a teammate missing its earlier work.
+    pub fn with_journal(mut self, journal: Box<dyn Journal>) -> Self {
+        self.orch = self.orch.with_journal(journal);
+        self
+    }
+
+    /// Rebuild this teammate from a journal, before tagging again.
+    ///
+    /// Grant the same skills, with the same ids, kinds and budgets, then replay.
+    /// The spend is re-derived through the gate rather than restored as a total,
+    /// so the budgets land where the previous run left them. A dispatch the
+    /// journal records with no outcome resumes as
+    /// [`TagOutcome::Interrupted`](crate::TagOutcome::Interrupted) and is not
+    /// run again.
+    pub fn replay(&mut self, entries: &[JournalEntry]) -> Result<ResumeReport, String> {
+        self.orch.replay(entries).map_err(|e| e.to_string())
+    }
+
+    /// What a tag's work item means now, by the id `tag` returned work for.
+    ///
+    /// The one way to observe an interrupted tag: `tag` runs to a terminal
+    /// state in one process, so only a replayed teammate has one.
+    pub fn outcome_of(&self, tag_id: &str) -> Option<TagOutcome> {
+        self.orch.state_of(tag_id).map(|s| outcome_for(Some(s)))
+    }
+
     /// The teammate's persistent identity (SPEC §2).
     pub fn identity(&self) -> Did {
         self.identity
@@ -178,17 +260,12 @@ impl Teammate {
         let id = format!("tag-{}", self.n);
         self.orch.submit(Task::new(id.clone(), kind, intent.as_bytes().to_vec()));
         let _ = self.orch.run();
-        let outcome = match self.orch.state_of(&id) {
-            Some(WorkState::Done) => TagOutcome::Done,
-            Some(WorkState::Denied(r)) => TagOutcome::Denied(r.clone()),
-            Some(WorkState::Failed(r)) => TagOutcome::Failed(r.clone()),
-            _ => TagOutcome::Failed("no terminal state".into()),
-        };
+        let outcome = outcome_for(self.orch.state_of(&id));
         // Seal an efficiency/1 receipt for work that ran (§6): a `Done` is a completed
         // unit, a `Failed` is spend that bought no result (waste). The energy is the
         // committed estimate in the dispatch's decision receipt — incurred whether the
         // task completed or failed. A tag refused at routing ran nothing and is skipped.
-        if matches!(outcome, TagOutcome::Done | TagOutcome::Failed(_)) {
+        if seals_efficiency(&outcome) {
             let cost = self
                 .orch
                 .receipts()
@@ -441,4 +518,65 @@ mod tests {
         );
         assert_eq!(t.efficiency_ledger().len(), 3, "one efficiency/1 per task that ran");
     }
+
+    /// An interrupted tag is reported as interrupted and sealed as neither.
+    ///
+    /// End to end, through a replay — which is the only way an interrupted state
+    /// reaches the teammate: `tag` runs to a terminal state in one process.
+    /// Truncating the journal after the dispatch is the crash this models.
+    #[test]
+    fn an_interrupted_tag_is_neither_done_nor_waste() {
+        #[derive(Clone, Default)]
+        struct Shared(std::rc::Rc<std::cell::RefCell<Vec<JournalEntry>>>);
+        impl Journal for Shared {
+            fn append(&mut self, e: JournalEntry) {
+                self.0.borrow_mut().push(e);
+            }
+        }
+
+        let journal = Shared::default();
+        let mut first = Teammate::new().with_journal(Box::new(journal.clone()));
+        first
+            .grant_skill(
+                Skill::new("summarize", &["summarize"], 1_000, |_| SkillOutcome::done(1_000)),
+                100_000,
+            )
+            .expect("grant");
+        assert_eq!(first.tag("summarize", "the Q3 thread"), TagOutcome::Done);
+
+        // Cut the journal where a process would die: after the dispatch was
+        // decided and debited, before any outcome was recorded.
+        let mut entries = journal.0.borrow().clone();
+        let cut = entries
+            .iter()
+            .position(|e| matches!(e, JournalEntry::Dispatched { .. }))
+            .expect("a dispatch");
+        entries.truncate(cut + 1);
+
+        let mut resumed = Teammate::new();
+        resumed
+            .grant_skill(
+                Skill::new("summarize", &["summarize"], 1_000, |_| SkillOutcome::done(1_000)),
+                100_000,
+            )
+            .expect("grant");
+        let report = resumed.replay(&entries).expect("replay");
+        assert_eq!(report.interrupted, 1);
+
+        // Reported as interrupted, naming who can settle it — not as failed.
+        let out = resumed.outcome_of("tag-1").expect("the tag is known");
+        assert_eq!(out, TagOutcome::Interrupted { harness: "summarize".into() });
+        assert!(!matches!(out, TagOutcome::Failed(_)), "reported as failed");
+
+        // And sealed as neither a completed unit nor waste.
+        assert!(
+            !seals_efficiency(&out),
+            "an interrupted tag seals an efficiency receipt, asserting what its energy achieved"
+        );
+        // The rule still holds for the outcomes that do seal, so the predicate
+        // is not simply refusing everything.
+        assert!(seals_efficiency(&TagOutcome::Done));
+        assert!(seals_efficiency(&TagOutcome::Failed("x".into())));
+        assert!(!seals_efficiency(&TagOutcome::Denied("x".into())));
+    }
 }

jouledesk/jouledesk-rs/crates/jouledesk/src/surface.rs

@@ -113,6 +113,12 @@ fn reply_text(who: &str, outcome: &TagOutcome, verified: Result<usize, usize>) -
         TagOutcome::Done => format!("✅ **@joule** completed it, @{who} — decision sealed.\n\n{ledger}"),
         TagOutcome::Denied(r) => format!("🚫 **@joule** can't do that, @{who}: {r}"),
         TagOutcome::Failed(r) => format!("⚠️ **@joule** tried but failed, @{who}: {r}\n\n{ledger}"),
+        // Says what is and is not known, rather than picking one. Reporting
+        // "failed" here would be a guess, and so would "done".
+        TagOutcome::Interrupted { harness } => format!(
+            "⏸️ **@joule** was interrupted, @{who} — `{harness}` had it when the run stopped, so whether \
+             it took effect is not known yet. Its energy is accounted either way.\n\n{ledger}"
+        ),
     }
 }
 
@@ -156,6 +162,34 @@ mod tests {
     use super::*;
     use crate::{Skill, SkillOutcome};
 
+    /// An interrupted tag is reported as neither done nor failed.
+    ///
+    /// JCP §4.11: an implementation MUST NOT report an interrupted action as
+    /// completed or as failed. Before `TagOutcome::Interrupted` existed, such a
+    /// task fell into `Teammate::tag`'s catch-all and came out
+    /// `Failed("no terminal state")` — a reply that tells a person the work did
+    /// not happen, when nobody knows whether it did.
+    #[test]
+    fn an_interrupted_tag_says_neither_done_nor_failed() {
+        let out = TagOutcome::Interrupted { harness: "summarize".into() };
+        let text = reply_text("ana", &out, Ok(1));
+        assert!(text.contains("interrupted"), "{text}");
+        assert!(text.contains("summarize"), "the reply must name who can settle it: {text}");
+        assert!(
+            !text.to_lowercase().contains("failed") && !text.to_lowercase().contains("completed"),
+            "an interrupted tag was reported as done or failed: {text}"
+        );
+        // And it renders on every surface, so no binding falls back to a
+        // different story for the same outcome.
+        for surface in [Surface::Slack, Surface::GitHub, Surface::Discord, Surface::Matrix, Surface::Generic] {
+            let v = render_reply(surface, "ana", &out, Ok(1));
+            assert!(
+                serde_json::to_string(&v).unwrap().contains("interrupted"),
+                "{surface:?} did not render it"
+            );
+        }
+    }
+
     fn team() -> Teammate {
         let mut t = Teammate::new();
         t.grant_skill(