diff --git a/jcp/SPEC.md b/jcp/SPEC.md index 765d85001..abf33f0fa 100644 --- a/jcp/SPEC.md +++ b/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) diff --git a/jcp/jcp-rs/crates/jcp-orchestrate/src/lib.rs b/jcp/jcp-rs/crates/jcp-orchestrate/src/lib.rs index b9c29d64e..416f3c59d 100644 --- a/jcp/jcp-rs/crates/jcp-orchestrate/src/lib.rs +++ b/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, + }, + /// 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, + }, + /// 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, +} + +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 { + 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>, max_attempts: u32, now: u64, + journal: Box, } 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) -> 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 { + let mut report = ResumeReport::default(); + // The dispatch each task is waiting on a settlement for, if any. + let mut in_flight: BTreeMap = 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>` so the test can read it while the + /// orchestrator owns the `Box`. + #[derive(Clone, Default)] + struct SharedJournal(std::rc::Rc>>); + impl Journal for SharedJournal { + fn append(&mut self, entry: JournalEntry) { + self.0.borrow_mut().push(entry); + } + } + impl SharedJournal { + fn entries(&self) -> Vec { + self.0.borrow().clone() + } + } + + fn fixed(id: &str, kind: &str, cost: u64) -> Box { + 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>, + } + 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 = 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); + } } diff --git a/jouledesk/jouledesk-rs/crates/jouledesk-github/src/lib.rs b/jouledesk/jouledesk-rs/crates/jouledesk-github/src/lib.rs index 68becd665..8bd589e48 100644 --- a/jouledesk/jouledesk-rs/crates/jouledesk-github/src/lib.rs +++ b/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) } diff --git a/jouledesk/jouledesk-rs/crates/jouledesk/examples/channel.rs b/jouledesk/jouledesk-rs/crates/jouledesk/examples/channel.rs index 583249a9c..c12d96017 100644 --- a/jouledesk/jouledesk-rs/crates/jouledesk/examples/channel.rs +++ b/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}"); diff --git a/jouledesk/jouledesk-rs/crates/jouledesk/src/lib.rs b/jouledesk/jouledesk-rs/crates/jouledesk/src/lib.rs index 585cc11f6..154581e89 100644 --- a/jouledesk/jouledesk-rs/crates/jouledesk/src/lib.rs +++ b/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) -> 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 { + 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 { + 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>>); + 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()))); + } } diff --git a/jouledesk/jouledesk-rs/crates/jouledesk/src/surface.rs b/jouledesk/jouledesk-rs/crates/jouledesk/src/surface.rs index 70117747d..c514f2dfc 100644 --- a/jouledesk/jouledesk-rs/crates/jouledesk/src/surface.rs +++ b/jouledesk/jouledesk-rs/crates/jouledesk/src/surface.rs @@ -113,6 +113,12 @@ fn reply_text(who: &str, outcome: &TagOutcome, verified: Result) - 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(