jcp/jcp-rs/crates/jcp-gateway/src/bin/jcp_gateway.rs

@@ -12,6 +12,14 @@
 //! `JCP_GATEWAY_RECEIPTS` (default `./jcp-gateway-receipts.cose`, length-
 //! prefixed envelopes) and a one-line summary per call goes to stderr.
 //!
+//! That file is also read back at startup, because SPEC §8.1 is REQUIRED of
+//! an enforcement point: it rebuilds its budget state by replaying its own
+//! verified receipt stream, so a restart does not hand back a ceiling the
+//! caller already spent, and the chain continues rather than restarting at
+//! `seq = 0` on top of itself. A stream that does not verify stops the
+//! gateway: it cannot know the spend, and serving as though the spend were
+//! zero is the overspend it is there to prevent.
+//!
 //! What the server writes that is not a response — a notification, or a
 //! request of its own in the middle of a call — goes straight to the
 //! client, and the client's answer to such a request straight back to the
@@ -133,6 +141,42 @@ impl Upstream for ChildServer {
     }
 }
 
+/// Read the length-prefixed receipt stream this gateway writes (see the
+/// append at the bottom of `main`): a little-endian `u32` length, then that
+/// many envelope bytes, repeated.
+///
+/// A missing file is an empty history — the first run. A **truncated tail is
+/// an error, not a shorter history**: the receipt whose bytes are missing
+/// carries a debit, and treating it as absent refills exactly that much
+/// budget. The operator who wants to start over moves the file aside
+/// deliberately.
+fn read_receipt_stream(path: &str) -> Result<Vec<Vec<u8>>, String> {
+    let bytes = match std::fs::read(path) {
+        Ok(b) => b,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+        Err(e) => return Err(format!("{path}: {e}")),
+    };
+    let mut out = Vec::new();
+    let mut i = 0usize;
+    while i < bytes.len() {
+        if i + 4 > bytes.len() {
+            return Err(format!("{path}: truncated length prefix at byte {i}"));
+        }
+        let len = u32::from_le_bytes([bytes[i], bytes[i + 1], bytes[i + 2], bytes[i + 3]]) as usize;
+        i += 4;
+        let end = i.checked_add(len).ok_or_else(|| format!("{path}: length overflows at byte {i}"))?;
+        if end > bytes.len() {
+            return Err(format!(
+                "{path}: record at byte {i} declares {len} bytes, {} present",
+                bytes.len() - i
+            ));
+        }
+        out.push(bytes[i..end].to_vec());
+        i = end;
+    }
+    Ok(out)
+}
+
 fn main() -> std::io::Result<()> {
     let args: Vec<String> = std::env::args().collect();
     let mut server = "mcp".to_string();
@@ -201,6 +245,30 @@ fn main() -> std::io::Result<()> {
     let receipts_path =
         std::env::var("JCP_GATEWAY_RECEIPTS").unwrap_or_else(|_| "./jcp-gateway-receipts.cose".into());
 
+    // SPEC §8.1: rebuild budget state from our own sealed stream before
+    // serving anything. Fail closed — an unreadable or unverifiable history
+    // means the spend is unknown, and starting from zero is the budget refill
+    // this replay exists to prevent.
+    let prior = match read_receipt_stream(&receipts_path) {
+        Ok(p) => p,
+        Err(e) => {
+            eprintln!("jcp-gateway: refusing to serve: cannot read the receipt stream: {e}");
+            std::process::exit(2);
+        }
+    };
+    let mut prev_receipts = match gw.resume(prior) {
+        Ok(n) => {
+            if n > 0 {
+                eprintln!("jcp-gateway: replayed {n} prior receipts from {receipts_path} (SPEC §8.1)");
+            }
+            n
+        }
+        Err(e) => {
+            eprintln!("jcp-gateway: refusing to serve: {receipts_path} is not a verified chain: {e}");
+            std::process::exit(2);
+        }
+    };
+
     // The client's lines are read on a thread of their own. A response of
     // the client's (an `id` and no `method`), answering a request the server
     // made in the middle of a call, goes straight to the server: the relay
@@ -225,7 +293,6 @@ fn main() -> std::io::Result<()> {
         }
     });
 
-    let mut prev_receipts = 0usize;
     for line in client_lines {
         match gw.relay(&line, &mut up)? {
             Relayed::Allowed { response, cost_uj, budget_remaining_uj } => {

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

@@ -72,7 +72,7 @@ use ed25519_dalek::SigningKey;
 use jcp_bridge::{mcp_call_frame, mcp_resource, mcp_result_frame};
 use jcp_flow::{FlowStore, Mode, Witness};
 use jcp_grant::issue;
-use jcp_receipt::{FlowSeal, ReceiptLog};
+use jcp_receipt::{FlowSeal, ReceiptLog, ResumeError};
 use jcp_runtime::Runtime;
 use jcp_schema::flow::{FlowCeiling, Readers};
 use jcp_schema::{Capability, Caveat, ContentId, Coverage, Did, Energy, FlowFacts, Grant, SignedGrant};
@@ -422,6 +422,41 @@ impl Gateway {
     }
 
     /// The hash-chained receipt log — verify with [`jcp_receipt::verify_log`].
+    /// Rebuild enforcement state from this gateway's own sealed receipt
+    /// stream, as SPEC §8.1 requires of an enforcement point that crashes:
+    /// "allow-receipts re-apply their debits to every grant they name,
+    /// ancestors included, with no external index."
+    ///
+    /// Two things are restored, and both matter:
+    ///
+    /// - **The budgets.** Every grant this gateway issues is deterministic —
+    ///   a fixed issuer key, `nonce: [0; 16]`, `expiry: 0` — so a restart
+    ///   mints byte-identical grants with identical ids. Without replaying the
+    ///   debits, each restart hands back the full `budget_uj`, and a caller
+    ///   can spend a ceiling again after every restart, without bound.
+    /// - **The chain.** The log would otherwise append `seq = 0, prev = 0…0`
+    ///   onto a stream that already holds a chain, so `verify_log` over the
+    ///   accumulated file fails at the first post-restart entry — the artifact
+    ///   recording the overspend would also be the one that fails to verify.
+    ///
+    /// Nothing is applied unless the whole stream verifies under this
+    /// gateway's steward key. Call it before serving; it refuses a log that
+    /// has already appended.
+    ///
+    /// # Errors
+    /// [`ResumeError`] from [`ReceiptLog::resume`].
+    pub fn resume(&mut self, prior: Vec<Vec<u8>>) -> Result<usize, ResumeError> {
+        // Verify and adopt first: a stream that does not verify never reaches
+        // the budget ledger.
+        let n = self.log.resume(prior)?;
+        let steward = self.log.steward().to_string();
+        let pubkey = self.log.steward_pubkey();
+        let spend: Vec<_> =
+            jcp_receipt::replay_spend(self.log.entries().iter().map(Vec::as_slice), &steward, &pubkey);
+        self.runtime.restore(spend);
+        Ok(n)
+    }
+
     pub fn receipts(&self) -> &[Vec<u8>] {
         self.log.entries()
     }
@@ -1391,6 +1426,92 @@ mod tests {
         Gateway::new("files", &[("read_file", 50_000_000), ("search", 50_000_000)], SigningKey::from_bytes(&[7u8; 32]))
     }
 
+    /// SPEC §8.1, the part that was missing: a restarted gateway replays its
+    /// own sealed stream, so a budget the caller already spent is not handed
+    /// back. Every grant here is deterministic — fixed issuer key,
+    /// `nonce: [0; 16]`, `expiry: 0` — so the restart mints an identical grant
+    /// id, which is exactly why the debit has to be replayed onto it.
+    #[test]
+    fn a_restart_does_not_refill_a_spent_budget() {
+        let steward = SigningKey::from_bytes(&[7u8; 32]);
+        let tools = [("read_file", 50_000_000u64)];
+        let mut up = MockServer { calls: 0 };
+
+        // The assertions below use the identity the runtime guarantees —
+        // `budget_remaining_uj = budget_uj - (prior_spend + this_call)` — so
+        // they turn on the PRIOR spend alone and never compare two measured
+        // energy figures to each other. Costs here are real meter readings and
+        // vary with host load; `remaining + cost` does not.
+        const BUDGET: u64 = 50_000_000;
+
+        let mut first = Gateway::new("files", &tools, steward.clone());
+        let Relayed::Allowed { cost_uj: c1, budget_remaining_uj: r1, .. } =
+            first.relay(&call("read_file", 1), &mut up).unwrap()
+        else {
+            panic!("expected Allowed")
+        };
+        assert_eq!(r1 + c1, BUDGET, "a fresh grant is charged only for this call");
+        let stream: Vec<Vec<u8>> = first.receipts().to_vec();
+        drop(first);
+
+        // The process restarts and replays its own stream.
+        let mut resumed = Gateway::new("files", &tools, steward.clone());
+        assert_eq!(resumed.resume(stream.clone()), Ok(stream.len()));
+        let Relayed::Allowed { cost_uj: c2, budget_remaining_uj: r2, .. } =
+            resumed.relay(&call("read_file", 2), &mut up).unwrap()
+        else {
+            panic!("expected Allowed")
+        };
+        assert_eq!(
+            r2 + c2,
+            BUDGET - c1,
+            "the replayed debit is still charged against the identical grant id"
+        );
+
+        // The chain continues across the restart rather than restarting on
+        // top of itself.
+        assert!(
+            jcp_receipt::verify_log(
+                resumed.receipts(),
+                &format!("did:jcp:gateway:{}", "files"),
+                &steward.verifying_key().to_bytes()
+            )
+            .is_ok(),
+            "the accumulated stream verifies end to end"
+        );
+
+        // The discriminating half: a gateway that does NOT replay starts the
+        // ceiling over, which is the defect this fixes.
+        let mut fresh = Gateway::new("files", &tools, steward);
+        let Relayed::Allowed { cost_uj: c3, budget_remaining_uj: r3, .. } =
+            fresh.relay(&call("read_file", 3), &mut up).unwrap()
+        else {
+            panic!("expected Allowed")
+        };
+        assert_eq!(
+            r3 + c3, BUDGET,
+            "without the replay the ceiling starts over: the prior spend is gone"
+        );
+        assert!(c1 > 0, "there was a prior spend for the replay to carry");
+    }
+
+    /// A stream that does not verify never reaches the budget ledger.
+    #[test]
+    fn resume_refuses_a_stream_that_does_not_verify() {
+        let steward = SigningKey::from_bytes(&[7u8; 32]);
+        let tools = [("read_file", 50_000_000u64)];
+        let mut up = MockServer { calls: 0 };
+
+        let mut g = Gateway::new("files", &tools, steward.clone());
+        g.relay(&call("read_file", 1), &mut up).unwrap();
+        let mut stream: Vec<Vec<u8>> = g.receipts().to_vec();
+        *stream[0].last_mut().expect("envelope bytes") ^= 1;
+
+        let mut restarted = Gateway::new("files", &tools, steward);
+        assert_eq!(restarted.resume(stream), Err(ResumeError::BrokenAt(0)));
+        assert!(restarted.receipts().is_empty(), "nothing was adopted");
+    }
+
     /// The caller receives the receipt for its OWN call. Until now the receipt
     /// existed only in the gateway's log — the enforcing party keeping an
     /// artifact about itself, which the subject could never present.

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

@@ -674,6 +674,20 @@ pub struct ReceiptLog {
     entries: Vec<Vec<u8>>,
 }
 
+/// Why a log would not adopt a prior chain ([`ReceiptLog::resume`]).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
+pub enum ResumeError {
+    /// The prior stream stops being a verified chain at this index — the same
+    /// index [`verify_log`] reports. Nothing was adopted.
+    #[error("prior receipt chain broken at index {0}")]
+    BrokenAt(usize),
+    /// This log has already sealed entries of its own. Adopting a history now
+    /// would renumber them, so the call is refused rather than silently
+    /// orphaning the given number of entries.
+    #[error("log already holds {0} sealed entries; resume before the first append")]
+    NotEmpty(usize),
+}
+
 /// The genesis predecessor: an all-zero content id, the "no prior receipt"
 /// sentinel the first entry's `prev` links to.
 fn genesis_prev() -> jcp_schema::ContentId {
@@ -747,6 +761,51 @@ impl ReceiptLog {
         env
     }
 
+    /// Adopt this steward's own prior receipt stream as the log's history, so
+    /// the next [`append`](Self::append) links onto the real predecessor
+    /// instead of restarting at genesis.
+    ///
+    /// This is the durability half of SPEC §8.1 — "an enforcement point that
+    /// crashes MUST be able to rebuild its budget state by replaying its own
+    /// verified receipt stream ... durability and audit are the same artifact".
+    /// Without it a restarted enforcement point appends `seq = 0, prev = 0…0`
+    /// onto a file that already holds a chain, and [`verify_log`] over the
+    /// accumulated stream fails at the first post-restart entry: the one
+    /// artifact that records the spend is also the one that will not verify.
+    ///
+    /// The stream is verified end to end *before* anything is adopted, so a
+    /// broken history cannot become this log's history. Pair it with
+    /// [`replay_spend`] to rebuild the budget side.
+    ///
+    /// # Errors
+    /// [`ResumeError::BrokenAt`] if the prior chain does not verify under this
+    /// log's steward key; [`ResumeError::NotEmpty`] if this log has already
+    /// appended.
+    pub fn resume(&mut self, prior: Vec<Vec<u8>>) -> Result<usize, ResumeError> {
+        if !self.entries.is_empty() {
+            return Err(ResumeError::NotEmpty(self.entries.len()));
+        }
+        let pubkey = self.steward_sk.verifying_key().to_bytes();
+        verify_log(&prior, &self.steward, &pubkey).map_err(ResumeError::BrokenAt)?;
+        self.head = prior
+            .last()
+            .map_or_else(genesis_prev, |e| jcp_schema::ContentId::of_bytes(e));
+        self.seq = prior.len() as u64;
+        self.entries = prior;
+        Ok(self.entries.len())
+    }
+
+    /// The steward id these entries are sealed under.
+    pub fn steward(&self) -> &str {
+        &self.steward
+    }
+
+    /// The steward's public key — what [`verify_log`] and [`replay_spend`]
+    /// check the stream against.
+    pub fn steward_pubkey(&self) -> [u8; 32] {
+        self.steward_sk.verifying_key().to_bytes()
+    }
+
     /// The sealed entries, in order.
     pub fn entries(&self) -> &[Vec<u8>] {
         &self.entries
@@ -1195,6 +1254,89 @@ mod tests {
         }
     }
 
+    // ---- SPEC §8.1: the stream is the write-ahead log ----------------
+
+    /// A restarted enforcement point adopts its own stream, so the chain
+    /// continues instead of restarting on top of itself.
+    #[test]
+    fn a_resumed_log_continues_the_chain() {
+        let (frame, grant, d) = scenario();
+        let steward = "did:example:steward";
+
+        let mut first_run = ReceiptLog::new(sk(3), steward);
+        first_run.append(&frame, &grant, &d);
+        first_run.append(&frame, &grant, &d);
+        let stream: Vec<Vec<u8>> = first_run.entries().to_vec();
+        let head = first_run.head();
+        drop(first_run);
+
+        let mut second_run = ReceiptLog::new(sk(3), steward);
+        assert_eq!(second_run.resume(stream), Ok(2), "our own chain verifies");
+        assert_eq!(second_run.head(), head, "it continues from the real head");
+        second_run.append(&frame, &grant, &d);
+
+        assert_eq!(
+            verify_log(second_run.entries(), steward, &sk(3).verifying_key().to_bytes()),
+            Ok(3),
+            "the accumulated stream verifies end to end across the restart"
+        );
+    }
+
+    /// What happened without the resume, and why it matters: the second run
+    /// appends `seq = 0, prev = 0…0` on top of a chain that already has two
+    /// links, so the accumulated file — the only record of the spend — stops
+    /// verifying at exactly that entry.
+    #[test]
+    fn a_log_that_does_not_resume_breaks_the_accumulated_stream() {
+        let (frame, grant, d) = scenario();
+        let steward = "did:example:steward";
+
+        let mut first_run = ReceiptLog::new(sk(3), steward);
+        first_run.append(&frame, &grant, &d);
+        first_run.append(&frame, &grant, &d);
+        let mut accumulated: Vec<Vec<u8>> = first_run.entries().to_vec();
+
+        let mut second_run = ReceiptLog::new(sk(3), steward);
+        accumulated.push(second_run.append(&frame, &grant, &d));
+
+        assert_eq!(
+            verify_log(&accumulated, steward, &sk(3).verifying_key().to_bytes()),
+            Err(2),
+            "the first post-restart entry breaks the chain"
+        );
+    }
+
+    /// A history that does not verify cannot become this log's history.
+    #[test]
+    fn resume_refuses_a_broken_chain() {
+        let (frame, grant, d) = scenario();
+        let steward = "did:example:steward";
+        let mut first = ReceiptLog::new(sk(3), steward);
+        first.append(&frame, &grant, &d);
+        first.append(&frame, &grant, &d);
+        let mut stream: Vec<Vec<u8>> = first.entries().to_vec();
+        *stream[1].last_mut().expect("envelope bytes") ^= 1;
+
+        let mut second = ReceiptLog::new(sk(3), steward);
+        assert_eq!(second.resume(stream), Err(ResumeError::BrokenAt(1)));
+        assert!(second.entries().is_empty(), "nothing was adopted");
+        assert_eq!(second.head(), genesis_prev(), "and the head did not move");
+    }
+
+    /// Adopting a history after appending would renumber what this log already
+    /// sealed, so it is refused.
+    #[test]
+    fn resume_refuses_a_log_that_already_appended() {
+        let (frame, grant, d) = scenario();
+        let steward = "did:example:steward";
+        let mut a = ReceiptLog::new(sk(3), steward);
+        let stream = vec![a.append(&frame, &grant, &d)];
+
+        let mut b = ReceiptLog::new(sk(3), steward);
+        b.append(&frame, &grant, &d);
+        assert_eq!(b.resume(stream), Err(ResumeError::NotEmpty(1)));
+    }
+
     /// The invariant a hash chain makes non-negotiable: erasure must not break
     /// the links. Nothing behind the tombstone is touched, so every prior
     /// entry's bytes — and therefore every `prev` — are exactly as they were.