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

@@ -9,9 +9,10 @@
 //! loop — which is the entire point: the inverse-scaling-law result means a more
 //! capable model is a more dangerous one, so the gate cannot be the model.
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 
 use jcp_grant::{verify, verify_chain, GrantError, RevocationSet};
+use jcp_schema::flow::{FlowCeiling, FlowFacts, Origins};
 use jcp_schema::{ContentId, DenyReason, Did, Energy, Frame, Grant, SignedGrant, SignedRevocation, Trust, Verdict};
 use serde::{Deserialize, Serialize};
 
@@ -65,9 +66,107 @@ pub struct Runtime {
     /// (open); once any root is configured for a resource, a chain whose root
     /// issuer isn't listed is denied `untrusted-root`.
     trust_roots: HashMap<ContentId, Vec<Did>>,
+    /// The enforcement-point taint floor (SPEC §4.10) — the flow analogue of
+    /// `trust_roots`. Default-open, so an unconfigured runtime behaves exactly as
+    /// before.
+    taint_floor: TaintFloor,
+    /// Joules spent against each grant's *tainted* sub-budget — the tighter wallet
+    /// for calls carrying untrusted provenance. Tracked separately from
+    /// `spent_uj`, which it never converts into.
+    spent_tainted_uj: HashMap<ContentId, u64>,
 }
 
 
+/// An enforcement-point-wide taint floor (SPEC §4.10): resources whose calls may
+/// not carry values from certain origins, regardless of any grant. The flow
+/// analogue of [`Runtime`]'s trust roots — an operator control that needs no new
+/// grants and cannot be attenuated away by a delegated ceiling.
+#[derive(Clone, Debug)]
+pub struct TaintFloor {
+    forbidden: BTreeMap<ContentId, Origins>,
+    default_forbidden: Origins,
+}
+
+impl Default for TaintFloor {
+    fn default() -> Self {
+        // Nothing forbidden by default: an empty origin mask intersects nothing,
+        // so an unconfigured floor is a no-op and every existing deployment is
+        // unaffected.
+        TaintFloor { forbidden: BTreeMap::new(), default_forbidden: Origins::none() }
+    }
+}
+
+impl TaintFloor {
+    /// Does this floor refuse a value with label `l` on `resource`? True when the
+    /// label's origins touch the resource's forbidden mask (or the default mask,
+    /// where the resource has none of its own).
+    fn refuses(&self, resource: &ContentId, l: &jcp_schema::Label) -> bool {
+        let mask = self.forbidden.get(resource).unwrap_or(&self.default_forbidden);
+        l.origins.intersects(mask)
+    }
+}
+
+/// The grant's effective flow ceiling: the [`FlowCeiling::attenuate`] fold over
+/// every [`FlowBelow`](jcp_schema::Caveat::FlowBelow) caveat it carries, so
+/// multiple ceilings compose to the strictest. `None` when the grant carries no
+/// flow caveat — the case that behaves exactly as before this feature existed.
+fn ceiling_of(g: &Grant) -> Option<FlowCeiling> {
+    g.caveats.iter().filter_map(|c| c.flow_ceiling()).fold(None, |acc, fc| {
+        Some(match acc {
+            Some(parent) => parent.attenuate(fc),
+            None => fc.clone(),
+        })
+    })
+}
+
+/// The flow gate (SPEC §4.10, check 6c). A value may exercise the grant only when
+/// its label sits at or below the grant's ceiling **and** the resource's taint
+/// floor does not refuse its origins.
+///
+/// The `facts` argument distinguishes three cases deliberately:
+/// - `None` — provenance was **not computed** (an old, un-migrated caller). With a
+///   ceiling present this **fails closed**; a ceiling-free grant is unaffected.
+/// - `Some(bottom)` — provenance computed and **clean**; passes any ceiling.
+/// - `Some(tainted)` — tested against the ceiling and the floor.
+fn flow_gate(
+    g: &Grant,
+    floor: &TaintFloor,
+    resource: &ContentId,
+    facts: Option<&FlowFacts>,
+) -> Option<DenyReason> {
+    // The enforcement-point floor applies whenever facts exist, independent of any
+    // grant ceiling — an operator refusal a delegated grant cannot widen.
+    if let Some(f) = facts {
+        if floor.refuses(resource, &f.label) {
+            return Some(DenyReason::TaintedFlow);
+        }
+    }
+    match (ceiling_of(g), facts) {
+        // No ceiling: today's behaviour, for any facts including None.
+        (None, _) => None,
+        // Ceiling present but provenance not computed: fail closed.
+        (Some(_), None) => Some(DenyReason::TaintedFlow),
+        (Some(ceiling), Some(f)) => {
+            if ceiling.admits(&f.label) {
+                None
+            } else {
+                Some(DenyReason::TaintedFlow)
+            }
+        }
+    }
+}
+
+/// The effective ceiling across a delegation chain — the attenuation fold of every
+/// link's ceiling, so a chain is at most as permissive as its strictest link.
+fn chain_ceiling(chain: &[SignedGrant]) -> Option<FlowCeiling> {
+    chain.iter().filter_map(|sg| ceiling_of(&sg.grant)).fold(None, |acc, fc| {
+        Some(match acc {
+            Some(parent) => parent.attenuate(&fc),
+            None => fc,
+        })
+    })
+}
+
 /// Whether every caveat carried inside the grant holds (SPEC §4.9).
 ///
 /// `cost_uj` is `None` in the two-phase metered path ([`Runtime::authorize`]),
@@ -140,7 +239,7 @@ impl Runtime {
         };
 
         // Validity checks 1–6, in the same fixed order as `decide`.
-        if let Some(r) = self.check(frame, sg, caller, now) {
+        if let Some(r) = self.check(frame, sg, caller, now, None) {
             return deny(r);
         }
         // 6b. Caveats (§4.9).
@@ -220,6 +319,25 @@ impl Runtime {
         }
     }
 
+    /// Forbid values carrying any origin in `mask` from reaching `resource` (SPEC
+    /// §4.10) — the flow analogue of [`trust_root`](Self::trust_root). An
+    /// enforcement-point control that applies regardless of any grant's ceiling and
+    /// cannot be attenuated away.
+    pub fn forbid_flow(&mut self, resource: ContentId, mask: Origins) {
+        self.taint_floor.forbidden.insert(resource, mask);
+    }
+
+    /// Forbid `mask` on every resource without its own floor rule — a fail-closed
+    /// default, off unless set.
+    pub fn forbid_flow_by_default(&mut self, mask: Origins) {
+        self.taint_floor.default_forbidden = mask;
+    }
+
+    /// Joules spent against a grant's tainted sub-budget.
+    pub fn spent_tainted(&self, grant_id: &ContentId) -> u64 {
+        *self.spent_tainted_uj.get(grant_id).unwrap_or(&0)
+    }
+
     /// Decide a frame against the grant it references and a metered `cost`, for
     /// `caller` at time `now`. On `Allow`, debits the cost from the grant's
     /// budget. Pure and deterministic — the checks run in a fixed order and the
@@ -242,7 +360,7 @@ impl Runtime {
         };
 
         // Validity checks 1–6, in fixed order (first failure names the reason).
-        if let Some(r) = self.check(frame, sg, caller, now) {
+        if let Some(r) = self.check(frame, sg, caller, now, None) {
             return deny(r);
         }
         // 6b. Caveats carried inside the grant (§4.9) — the open attenuation axis.
@@ -269,7 +387,14 @@ impl Runtime {
     /// Validity checks 1–6 shared by [`decide`](Self::decide) and
     /// [`authorize`](Self::authorize): everything except the cost-specific
     /// budget test. `None` means the frame may proceed to a budget decision.
-    fn check(&self, frame: &Frame, sg: &SignedGrant, caller: Did, now: u64) -> Option<DenyReason> {
+    fn check(
+        &self,
+        frame: &Frame,
+        sg: &SignedGrant,
+        caller: Did,
+        now: u64,
+        facts: Option<&FlowFacts>,
+    ) -> Option<DenyReason> {
         let g = &sg.grant;
         let gid = g.id();
         // 1. Untrusted output can never exercise control — anti-tool-poisoning.
@@ -307,6 +432,17 @@ impl Runtime {
         if !g.capability.actions.iter().any(|a| a == &frame.action) {
             return Some(DenyReason::ActionNotGranted);
         }
+        // 6c. Flow (SPEC §4.10): the argument's propagated provenance must sit at
+        //     or below the grant's ceiling, and the resource's taint floor must not
+        //     refuse its origins. Placed here on purpose — after capability (the
+        //     ceiling lives inside an authenticated grant, so judging flow before
+        //     the signature would seal a verdict over attacker-chosen policy) and
+        //     before caveats and budget (a tainted call must cost zero joules, the
+        //     same invariant tool-poisoning denial holds). A ceiling-free grant
+        //     with no facts reaches here unchanged.
+        if let Some(r) = flow_gate(g, &self.taint_floor, &frame.resource, facts) {
+            return Some(r);
+        }
         None
     }
 
@@ -321,7 +457,7 @@ impl Runtime {
         let g = &sg.grant;
         let gid = g.id();
         let remaining = g.budget_uj.saturating_sub(self.spent(&gid));
-        if let Some(r) = self.check(frame, sg, caller, now)
+        if let Some(r) = self.check(frame, sg, caller, now, None)
             .or_else(|| caveats_unsatisfied(g, frame, caller, None, now))
         {
             return Decision {
@@ -372,6 +508,109 @@ impl Runtime {
         }
     }
 
+    /// The flow-aware [`authorize`](Self::authorize): the same two-phase gate, but
+    /// the flow facts are checked (check 6c) so a tainted argument is denied
+    /// `TaintedFlow` **before** any budget is touched, and a call already at its
+    /// tainted sub-budget is denied `TaintBudgetExceeded` before the tool runs.
+    /// `authorize` itself delegates here conceptually with `facts = None`; a
+    /// gateway that computes provenance calls this with real facts.
+    pub fn authorize_flow(
+        &mut self,
+        frame: &Frame,
+        sg: &SignedGrant,
+        caller: Did,
+        now: u64,
+        facts: &FlowFacts,
+    ) -> Decision {
+        let g = &sg.grant;
+        let gid = g.id();
+        let remaining = g.budget_uj.saturating_sub(self.spent(&gid));
+        let deny = |r: DenyReason| Decision {
+            verdict: Verdict::Deny(r),
+            cost: Energy::measured(0),
+            budget_remaining_uj: remaining,
+        };
+        if let Some(r) = self
+            .check(frame, sg, caller, now, Some(facts))
+            .or_else(|| caveats_unsatisfied(g, frame, caller, None, now))
+        {
+            return deny(r);
+        }
+        if remaining == 0 {
+            return Decision {
+                verdict: Verdict::Deny(DenyReason::BudgetExceeded),
+                cost: Energy::measured(0),
+                budget_remaining_uj: 0,
+            };
+        }
+        // Tainted work gets the smaller wallet. Pre-checked at authorize (like the
+        // main budget's `remaining == 0` degradation) so a call that has exhausted
+        // its tainted sub-budget is denied before the upstream tool runs.
+        if !facts.is_bottom() {
+            if let Some(cap) = ceiling_of(g).and_then(|fc| fc.tainted_budget_uj) {
+                if self.spent_tainted(&gid) >= cap {
+                    return deny(DenyReason::TaintBudgetExceeded);
+                }
+            }
+        }
+        Decision { verdict: Verdict::Allow, cost: Energy::measured(0), budget_remaining_uj: remaining }
+    }
+
+    /// The flow-aware [`charge`](Self::charge): debit the measured tool joules as
+    /// usual, and — when the call was tainted — also accrue them against the
+    /// grant's tainted sub-budget, so the tighter wallet actually depletes. The
+    /// monitor's own evaluation cost (`facts.eval_uj`) is **not** folded in here;
+    /// it is reported and sealed separately so a call's energy keeps its meaning.
+    pub fn charge_flow(&mut self, grant: &Grant, measured: Energy, facts: &FlowFacts) -> Decision {
+        let decision = self.charge(grant, measured);
+        if !facts.is_bottom() {
+            let gid = grant.id();
+            let accrued = self.spent_tainted(&gid).saturating_add(measured.uj());
+            self.spent_tainted_uj.insert(gid, accrued);
+        }
+        decision
+    }
+
+    /// The flow-aware [`decide`](Self::decide): the one-shot path with flow facts
+    /// checked at 6c. A tainted argument is denied `TaintedFlow` before the budget
+    /// test, so — like a tool-poisoning denial — it debits zero joules.
+    pub fn decide_flow(
+        &mut self,
+        frame: &Frame,
+        sg: &SignedGrant,
+        caller: Did,
+        cost: Energy,
+        now: u64,
+        facts: &FlowFacts,
+    ) -> Decision {
+        let g = &sg.grant;
+        let gid = g.id();
+        let remaining = g.budget_uj.saturating_sub(self.spent(&gid));
+        let deny = |r: DenyReason| Decision {
+            verdict: Verdict::Deny(r),
+            cost,
+            budget_remaining_uj: remaining,
+        };
+        if let Some(r) = self.check(frame, sg, caller, now, Some(facts)) {
+            return deny(r);
+        }
+        if let Some(r) = caveats_unsatisfied(g, frame, caller, Some(cost.uj()), now) {
+            return deny(r);
+        }
+        let spent = self.spent(&gid);
+        let c = cost.uj();
+        if spent.saturating_add(c) > g.budget_uj {
+            return deny(DenyReason::BudgetExceeded);
+        }
+        let new_spent = spent + c;
+        self.spent_uj.insert(gid, new_spent);
+        if !facts.is_bottom() {
+            let accrued = self.spent_tainted(&gid).saturating_add(c);
+            self.spent_tainted_uj.insert(gid, accrued);
+        }
+        Decision { verdict: Verdict::Allow, cost, budget_remaining_uj: g.budget_uj - new_spent }
+    }
+
     /// Decide a frame against a full delegation chain, **root first, leaf
     /// last** — multi-hop enforcement (SPEC §4.1).
     ///
@@ -389,6 +628,35 @@ impl Runtime {
         caller: Did,
         cost: Energy,
         now: u64,
+    ) -> Decision {
+        // No flow facts: a chain whose links carry a flow ceiling fails closed
+        // (the un-migrated path), exactly as `decide` does with `check(.., None)`.
+        self.decide_chained_inner(frame, chain, caller, cost, now, None)
+    }
+
+    /// The flow-aware [`decide_chained`](Self::decide_chained): the chain path with
+    /// the argument's provenance checked against the chain's attenuated ceiling and
+    /// the resource's taint floor, denied before any link is debited.
+    pub fn decide_chained_flow(
+        &mut self,
+        frame: &Frame,
+        chain: &[SignedGrant],
+        caller: Did,
+        cost: Energy,
+        now: u64,
+        facts: &FlowFacts,
+    ) -> Decision {
+        self.decide_chained_inner(frame, chain, caller, cost, now, Some(facts))
+    }
+
+    fn decide_chained_inner(
+        &mut self,
+        frame: &Frame,
+        chain: &[SignedGrant],
+        caller: Did,
+        cost: Energy,
+        now: u64,
+        facts: Option<&FlowFacts>,
     ) -> Decision {
         let leaf = match chain.last() {
             Some(l) => l,
@@ -445,6 +713,27 @@ impl Runtime {
         if !leaf.grant.capability.actions.iter().any(|a| a == &frame.action) {
             return deny(DenyReason::ActionNotGranted);
         }
+        // 6c (chain form): flow against the chain's attenuated ceiling and the
+        // resource's taint floor. The ceiling is the meet over every link, so a
+        // chain is at most as permissive as its strictest hop, and a facts-less
+        // call under a ceilinged chain fails closed — mirroring the leaf gate.
+        {
+            let chained = chain_ceiling(chain);
+            if let Some(f) = facts {
+                if self.taint_floor.refuses(&frame.resource, &f.label) {
+                    return deny(DenyReason::TaintedFlow);
+                }
+            }
+            match (chained, facts) {
+                (None, _) => {}
+                (Some(_), None) => return deny(DenyReason::TaintedFlow),
+                (Some(ceiling), Some(f)) => {
+                    if !ceiling.admits(&f.label) {
+                        return deny(DenyReason::TaintedFlow);
+                    }
+                }
+            }
+        }
         // Energy conservation: the cost must fit every link's remaining budget.
         let c = cost.uj();
         for sg in chain {
@@ -459,6 +748,14 @@ impl Runtime {
             let s = self.spent(&gid) + c;
             self.spent_uj.insert(gid, s);
         }
+        // A tainted chained call also depletes each link's tainted sub-budget.
+        if facts.map(|f| !f.is_bottom()).unwrap_or(false) {
+            for sg in chain {
+                let gid = sg.grant.id();
+                let accrued = self.spent_tainted(&gid).saturating_add(c);
+                self.spent_tainted_uj.insert(gid, accrued);
+            }
+        }
         Decision {
             verdict: Verdict::Allow,
             cost,
@@ -918,4 +1215,199 @@ mod tests {
         let f = leaf_frame(&chain_a, r);
         assert_eq!(rt.decide_chained(&f, &[], sub, Energy::measured(1), 10).verdict, Verdict::Deny(DenyReason::ChainBroken));
     }
+
+    // ── flow gate (SPEC §4.10, check 6c) ─────────────────────────────────────
+
+    use jcp_schema::flow::{FlowCeiling, FlowFacts, Integrity, Label, Origin, Origins, Readers};
+    use jcp_schema::Provenance;
+
+    fn flow_grant(caveats: Vec<Caveat>) -> (SignedGrant, Did, ContentId) {
+        let issuer = sk(1);
+        let holder = sk(2);
+        let resource = ContentId::of_bytes(b"db");
+        let g = Grant {
+            issuer: did(&issuer),
+            subject: did(&holder),
+            capability: Capability::new(resource, ["read"]),
+            budget_uj: 100,
+            budget_funds: None,
+            caveats,
+            expiry: 1000,
+            parent: None,
+            delegate_depth: None,
+            nonce: [0; 16],
+        };
+        (issue(&issuer, g).unwrap(), did(&holder), resource)
+    }
+
+    /// A tainted argument: untrusted, carrying `channel`'s origin.
+    fn tainted(channel: u8) -> FlowFacts {
+        FlowFacts {
+            label: Label {
+                integrity: Integrity::Untrusted,
+                origins: Origins::one(Origin::channel(ContentId::of_bytes(&[channel; 4]))),
+                readers: Readers::Public,
+            },
+            witness_id: ContentId([0u8; 32]),
+            eval_uj: 5,
+            eval_prov: Provenance::ModelBased,
+        }
+    }
+
+    /// A ceiling that admits only principal-clean, origin-free values.
+    fn principal_only() -> FlowCeiling {
+        FlowCeiling::new(Label {
+            integrity: Integrity::Principal,
+            origins: Origins::none(),
+            readers: Readers::Public,
+        })
+    }
+
+    /// A permissive ceiling that admits any label but has a small tainted budget.
+    fn permissive_with_tainted_budget(budget: u64) -> FlowCeiling {
+        FlowCeiling { ceiling: Label::top(), tainted_budget_uj: Some(budget) }
+    }
+
+    #[test]
+    fn ceiling_free_grant_is_a_noop_for_any_facts() {
+        // Every grant that exists today is ceiling-free; it must behave identically
+        // whether facts are absent, clean, or tainted.
+        let (sg, holder, r) = flow_grant(Vec::new());
+        let f = frame(&sg, r, "read", Trust::Control);
+        let mut rt = Runtime::new();
+        assert!(rt.decide(&f, &sg, holder, Energy::measured(10), 10).is_allow());
+        let mut rt = Runtime::new();
+        assert!(rt.decide_flow(&f, &sg, holder, Energy::measured(10), 10, &FlowFacts::bottom()).is_allow());
+        let mut rt = Runtime::new();
+        assert!(rt.decide_flow(&f, &sg, holder, Energy::measured(10), 10, &tainted(3)).is_allow());
+    }
+
+    #[test]
+    fn a_bottom_flow_call_is_byte_identical_to_the_plain_path() {
+        let (sg, holder, r) = flow_grant(Vec::new());
+        let f = frame(&sg, r, "read", Trust::Control);
+        let mut plain = Runtime::new();
+        let mut flow = Runtime::new();
+        assert_eq!(
+            plain.decide(&f, &sg, holder, Energy::measured(30), 10),
+            flow.decide_flow(&f, &sg, holder, Energy::measured(30), 10, &FlowFacts::bottom()),
+        );
+        assert_eq!(plain.spent(&sg.grant.id()), flow.spent(&sg.grant.id()));
+    }
+
+    #[test]
+    fn clean_facts_pass_a_ceiling() {
+        let (sg, holder, r) = flow_grant(vec![Caveat::FlowBelow(principal_only())]);
+        let f = frame(&sg, r, "read", Trust::Control);
+        let mut rt = Runtime::new();
+        assert!(rt.decide_flow(&f, &sg, holder, Energy::measured(10), 10, &FlowFacts::bottom()).is_allow());
+    }
+
+    #[test]
+    fn a_tainted_argument_is_denied_by_a_ceiling_with_zero_debit() {
+        // The core property: a value from an untrusted origin cannot exercise a
+        // principal-only grant, and the denial precedes the budget so it costs no
+        // joules — the same invariant tool poisoning holds.
+        let (sg, holder, r) = flow_grant(vec![Caveat::FlowBelow(principal_only())]);
+        let f = frame(&sg, r, "read", Trust::Control);
+        let mut rt = Runtime::new();
+        let d = rt.decide_flow(&f, &sg, holder, Energy::measured(30), 10, &tainted(7));
+        assert_eq!(d.verdict, Verdict::Deny(DenyReason::TaintedFlow));
+        assert_eq!(rt.spent(&sg.grant.id()), 0, "a tainted deny debits nothing");
+    }
+
+    #[test]
+    fn a_ceiling_grant_on_the_old_path_fails_closed() {
+        // An un-migrated caller (no facts) handed a ceilinged grant is denied, never
+        // silently unenforced.
+        let (sg, holder, r) = flow_grant(vec![Caveat::FlowBelow(principal_only())]);
+        let f = frame(&sg, r, "read", Trust::Control);
+        let mut rt = Runtime::new();
+        assert_eq!(
+            rt.decide(&f, &sg, holder, Energy::measured(10), 10).verdict,
+            Verdict::Deny(DenyReason::TaintedFlow),
+        );
+        assert_eq!(rt.spent(&sg.grant.id()), 0);
+    }
+
+    #[test]
+    fn the_taint_floor_refuses_a_forbidden_origin_regardless_of_ceiling() {
+        // An enforcement-point refusal that no grant carries and none can widen.
+        let (sg, holder, r) = flow_grant(Vec::new()); // no ceiling at all
+        let f = frame(&sg, r, "read", Trust::Control);
+        let mut rt = Runtime::new();
+        let forbidden = Origins::one(Origin::channel(ContentId::of_bytes(&[7u8; 4])));
+        rt.forbid_flow(r, forbidden);
+        let d = rt.decide_flow(&f, &sg, holder, Energy::measured(10), 10, &tainted(7));
+        assert_eq!(d.verdict, Verdict::Deny(DenyReason::TaintedFlow));
+        // A value from a different origin is unaffected.
+        assert!(rt.decide_flow(&f, &sg, holder, Energy::measured(10), 10, &tainted(9)).is_allow());
+    }
+
+    #[test]
+    fn the_tainted_sub_budget_pre_checks_at_authorize() {
+        // A permissive ceiling admits the label, but the tighter tainted wallet
+        // denies once spent — before the upstream tool runs.
+        let (sg, holder, r) = flow_grant(vec![Caveat::FlowBelow(permissive_with_tainted_budget(20))]);
+        let f = frame(&sg, r, "read", Trust::Control);
+        let mut rt = Runtime::new();
+        // First tainted call is admitted and depletes the tainted wallet.
+        assert!(rt.authorize_flow(&f, &sg, holder, 10, &tainted(3)).is_allow());
+        rt.charge_flow(&sg.grant, Energy::measured(20), &tainted(3));
+        assert_eq!(rt.spent_tainted(&sg.grant.id()), 20);
+        // The wallet is now spent: the next tainted call is refused before running.
+        let d = rt.authorize_flow(&f, &sg, holder, 10, &tainted(3));
+        assert_eq!(d.verdict, Verdict::Deny(DenyReason::TaintBudgetExceeded));
+        // A clean call still passes — the sub-budget only bounds tainted work.
+        assert!(rt.authorize_flow(&f, &sg, holder, 10, &FlowFacts::bottom()).is_allow());
+    }
+
+    #[test]
+    fn charge_flow_only_depletes_the_tainted_wallet_for_tainted_calls() {
+        let (sg, _holder, _r) = flow_grant(Vec::new());
+        let mut rt = Runtime::new();
+        rt.charge_flow(&sg.grant, Energy::measured(15), &FlowFacts::bottom());
+        assert_eq!(rt.spent_tainted(&sg.grant.id()), 0, "a clean call spends no tainted budget");
+        rt.charge_flow(&sg.grant, Energy::measured(15), &tainted(1));
+        assert_eq!(rt.spent_tainted(&sg.grant.id()), 15, "a tainted call does");
+    }
+
+    #[test]
+    fn a_chained_grant_gates_flow_and_fails_closed_without_facts() {
+        // A delegated grant carrying a ceiling: the plain chain path fails closed,
+        // and the flow path denies a tainted argument.
+        let issuer = sk(1);
+        let holder = sk(2);
+        let resource = ContentId::of_bytes(b"db");
+        let root = Grant {
+            issuer: did(&issuer),
+            subject: did(&holder),
+            capability: Capability::new(resource, ["read"]),
+            budget_uj: 100,
+            budget_funds: None,
+            caveats: vec![Caveat::FlowBelow(principal_only())],
+            expiry: 1000,
+            parent: None,
+            delegate_depth: None,
+            nonce: [0; 16],
+        };
+        let root_sg = issue(&issuer, root).unwrap();
+        let f = frame(&root_sg, resource, "read", Trust::Control);
+
+        let mut rt = Runtime::new();
+        // Plain chain path: no facts under a ceiling ⇒ fail closed.
+        assert_eq!(
+            rt.decide_chained(&f, &[root_sg.clone()], did(&holder), Energy::measured(10), 10).verdict,
+            Verdict::Deny(DenyReason::TaintedFlow),
+        );
+        // Flow chain path: a tainted argument is denied.
+        assert_eq!(
+            rt.decide_chained_flow(&f, &[root_sg.clone()], did(&holder), Energy::measured(10), 10, &tainted(7)).verdict,
+            Verdict::Deny(DenyReason::TaintedFlow),
+        );
+        // A clean argument passes.
+        assert!(rt
+            .decide_chained_flow(&f, &[root_sg], did(&holder), Energy::measured(10), 10, &FlowFacts::bottom())
+            .is_allow());
+    }
 }

jcp/jcp-rs/crates/jcp-schema/src/flow.rs

@@ -370,6 +370,42 @@ pub fn fold(labels: &[Label]) -> Label {
     labels.iter().fold(Label::bottom(), |acc, l| acc.join(l))
 }
 
+/// What the runtime consumes at gate time: a value's folded [`Label`], a
+/// commitment to the [witness](../../jcp_flow) that justifies it, and the monitor's
+/// own metered cost of computing it (reported separately, never folded into the
+/// call's energy).
+///
+/// A `bottom` facts value means "provenance computed, and clean" — distinct from
+/// *no* facts (`None` at the gate), which fails closed against a ceiling. This is
+/// deliberately not a wire type: it travels in-process from the gateway to the
+/// runtime, and what gets sealed is the label and `witness_id`, via the receipt.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct FlowFacts {
+    pub label: Label,
+    pub witness_id: ContentId,
+    /// The monitor's own energy for computing the label (microjoules).
+    pub eval_uj: u64,
+    pub eval_prov: crate::Provenance,
+}
+
+impl FlowFacts {
+    /// Clean facts — a value with no untrusted provenance and no monitor cost.
+    pub fn bottom() -> Self {
+        FlowFacts {
+            label: Label::bottom(),
+            witness_id: ContentId([0u8; 32]),
+            eval_uj: 0,
+            eval_prov: crate::Provenance::Estimator,
+        }
+    }
+
+    /// Is this a clean (`⊥`) label? The gate treats bottom facts as "no ceiling
+    /// concern"; it is *not* the same as facts being absent.
+    pub fn is_bottom(&self) -> bool {
+        self.label.is_bottom()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

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

@@ -21,7 +21,7 @@ mod cbor;
 pub use cbor::{canonical_cbor, from_canonical_cbor};
 
 pub mod flow;
-pub use flow::{FlowCeiling, Integrity, Label, Origin, Origins, Readers};
+pub use flow::{FlowCeiling, FlowFacts, Integrity, Label, Origin, Origins, Readers};
 
 /// Errors from canonical serialization, content addressing, or DID parsing.
 #[derive(Debug, thiserror::Error)]
@@ -256,6 +256,13 @@ pub enum Caveat {
     /// `budget_uj`, which is cumulative: this bounds any single call, so a
     /// delegate cannot spend the whole remaining budget in one shot.
     MaxCostUj(u64),
+    /// A **flow ceiling** (SPEC §4.10): the greatest-tainted provenance label
+    /// whose values may exercise this grant. Unlike the other caveats, this one
+    /// cannot be decided from the frame alone — it needs the argument's propagated
+    /// label, which the runtime supplies as flow facts. So [`holds`](Self::holds)
+    /// always returns `true` for it (it never blocks the caveat path); the real
+    /// test is the runtime's flow gate, which reads this ceiling.
+    FlowBelow(flow::FlowCeiling),
 }
 
 impl Caveat {
@@ -285,6 +292,17 @@ impl Caveat {
             Caveat::NotBefore(t) => now >= *t,
             Caveat::NotAfter(t) => now <= *t,
             Caveat::MaxCostUj(max) => cost_uj <= *max,
+            // Flow is decided by the runtime gate against the argument's label,
+            // not from the frame — so as a caveat it never blocks.
+            Caveat::FlowBelow(_) => true,
+        }
+    }
+
+    /// The flow ceiling this caveat carries, if it is a [`Caveat::FlowBelow`].
+    pub fn flow_ceiling(&self) -> Option<&flow::FlowCeiling> {
+        match self {
+            Caveat::FlowBelow(ceiling) => Some(ceiling),
+            _ => None,
         }
     }
 }
@@ -480,6 +498,15 @@ pub enum DenyReason {
     /// resource (SPEC §7) — the enforcement point honors only the root keys it
     /// was configured with, `authorized_keys`-style.
     UntrustedRoot,
+    /// The argument's propagated provenance label does not sit at or below the
+    /// grant's flow ceiling, or the resource's taint floor refuses its origins
+    /// (SPEC §4.10). The one denial that is about *where a value came from* rather
+    /// than what it is.
+    TaintedFlow,
+    /// A tainted call would exceed the grant's smaller, once-tainted joule budget
+    /// (SPEC §4.10) — a tighter wallet for work that has touched an untrusted
+    /// source, distinct from the cumulative `BudgetExceeded`.
+    TaintBudgetExceeded,
 }
 
 impl DenyReason {
@@ -499,6 +526,8 @@ impl DenyReason {
             DenyReason::Revoked => "revoked",
             DenyReason::ChainBroken => "chain-broken",
             DenyReason::UntrustedRoot => "untrusted-root",
+            DenyReason::TaintedFlow => "tainted-flow",
+            DenyReason::TaintBudgetExceeded => "taint-budget-exceeded",
         }
     }
 }
@@ -627,6 +656,31 @@ mod tests {
         assert!(String::from_utf8_lossy(&bytes).contains("budget_funds"));
     }
 
+    #[test]
+    fn absent_flow_ceiling_does_not_change_the_encoding() {
+        // A grant with no FlowBelow caveat is byte-identical to one minted before
+        // the flow feature existed: adding a variant to an externally-tagged enum
+        // changes the encoding of no value that omits it.
+        let g = grant_without_funds();
+        let bytes = crate::cbor::canonical_cbor(&g).expect("canonical");
+        assert!(!String::from_utf8_lossy(&bytes).contains("FlowBelow"));
+        assert!(!String::from_utf8_lossy(&bytes).contains("caveats"), "no caveats ⇒ field omitted");
+        assert_eq!(g.id(), ContentId::of(&g).unwrap());
+    }
+
+    #[test]
+    fn a_flow_caveat_changes_identity_and_carries_its_ceiling() {
+        use crate::flow::{FlowCeiling, Label};
+        let plain = grant_without_funds();
+        let mut ceilinged = grant_without_funds();
+        ceilinged.caveats = vec![Caveat::FlowBelow(FlowCeiling::new(Label::bottom()))];
+        assert_ne!(plain.id(), ceilinged.id(), "a flow ceiling is part of authority");
+        // The ceiling round-trips through canonical CBOR.
+        let bytes = crate::cbor::canonical_cbor(&ceilinged).expect("canonical");
+        let back: Grant = crate::cbor::from_canonical_cbor(&bytes).expect("decode");
+        assert_eq!(back.caveats[0].flow_ceiling(), Some(&FlowCeiling::new(Label::bottom())));
+    }
+
     #[test]
     fn funds_currency_compare_is_case_insensitive() {
         let f = Funds::new("USD", 500);