jcp/jcp-rs/Cargo.lock

@@ -598,6 +598,16 @@ dependencies = [
  "nvml-wrapper",
 ]
 
+[[package]]
+name = "jcp-flow"
+version = "0.1.0"
+dependencies = [
+ "blake3",
+ "jcp-schema",
+ "serde",
+ "thiserror 2.0.18",
+]
+
 [[package]]
 name = "jcp-gateway"
 version = "0.1.0"

jcp/jcp-rs/Cargo.toml

@@ -22,6 +22,7 @@ members = [
     "crates/jcp-orchestrate",
     "crates/jcp-intercept",
     "crates/jcp-efficiency",
+    "crates/jcp-flow",
 ]
 
 [workspace.package]

jcp/jcp-rs/crates/jcp-flow/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "jcp-flow"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+description = "Propagated provenance for JCP — the witness scan and session store that fold a derived value's flow Label from the tool outputs it was computed from, so authorization can be conditioned on where a value CAME FROM, not just its literal bytes. An exact blake3 window index detects verbatim carry across tool calls; each verdict emits a content-addressed, offline-checkable Witness."
+
+[dependencies]
+jcp-schema = { workspace = true }
+blake3 = { workspace = true }
+serde = { workspace = true }
+thiserror = { workspace = true }
+
+[lints]
+workspace = true

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

@@ -0,0 +1,235 @@
+//! Propagated provenance for JCP — the witness scan and session store.
+//!
+//! The [flow lattice](jcp_schema::flow) says what a label *is* and how two labels
+//! fold. This crate computes labels from real bytes: as a run reads tool outputs
+//! and derives new values, the [`FlowStore`] tracks where each value's bytes came
+//! from and folds a derived value's label from its ancestors. The result is that
+//! authorization can be conditioned on **where a value came from**, not merely on
+//! its literal bytes — the gap JCP's one-bit `Trust` tag cannot express.
+//!
+//! Two things make this a reference monitor rather than a heuristic:
+//!
+//! - The carry detector is an **exact** `blake3` window index. Two values share a
+//!   fingerprint exactly when they share a verbatim [`SCAN_W`]-byte run — no
+//!   rolling-hash state to get wrong, and the same bytes fingerprint identically
+//!   on every platform.
+//! - Every verdict emits a self-contained [`Witness`]: the ancestor cut, the taint
+//!   edges, and the folded label, content-addressed. A receipt commits only its
+//!   `id()`; an auditor re-folds it offline to re-derive the verdict
+//!   ([`verify_witness`]).
+//!
+//! **Scope.** This crate is the propagation and its offline witness check. The
+//! runtime *gate* (does a label pass a grant's ceiling), the `jcp/3` receipt seal,
+//! and the gateway relay that drives them are later layers; see the FL-2 plan.
+//!
+//! **Honest residuals.** The scan sees verbatim carry, not meaning: base64,
+//! translation, or paraphrase defeat it, and soundness then rests on a registered
+//! sensitive term or the `HighWater` floor. Sub-`SCAN_W` unregistered secrets are
+//! invisible to the scan. `verify_witness` checks *consistency* (the cut folds to
+//! the sealed label), not *truth* against original bytes — that needs an opt-in
+//! disclosure tier. The lattice laws prove taint never decreases and a verdict is
+//! reproducible; none of them says prompt injection cannot succeed.
+
+pub mod store;
+pub mod witness;
+
+pub use store::{ambient_source, windows, FlowStore, Mode, ScanParams, SCAN_W};
+pub use witness::{
+    verify_witness, CutNode, Edge, EdgeKind, Witness, WitnessError, WITNESS_SCHEMA,
+};
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use jcp_schema::flow::{FlowCeiling, Integrity, Label, Origins};
+    use jcp_schema::{ContentId, Did};
+    use std::collections::BTreeSet;
+
+    fn channel(byte: u8) -> ContentId {
+        ContentId::of_bytes(&[byte; 8])
+    }
+    fn anchor() -> ContentId {
+        ContentId([0u8; 32])
+    }
+    fn no_declared() -> BTreeSet<ContentId> {
+        BTreeSet::new()
+    }
+
+    #[test]
+    fn windows_are_platform_independent_and_exact() {
+        // Same bytes ⇒ same fingerprint set, regardless of who computes them.
+        let a = windows(b"the quick brown fox jumps", SCAN_W);
+        let b = windows(b"the quick brown fox jumps", SCAN_W);
+        assert_eq!(a, b);
+        assert!(!a.is_empty());
+        // Fewer than W bytes ⇒ no windows.
+        assert!(windows(b"short", SCAN_W).is_empty());
+        // A 25-byte string has 25-16+1 = 10 windows.
+        assert_eq!(windows(b"the quick brown fox jumps", SCAN_W).len(), 10);
+    }
+
+    #[test]
+    fn l17_carry_of_at_least_w_bytes_yields_a_witnessed_edge() {
+        let mut store = FlowStore::new(64);
+        let secret = b"ATTACKER-CONTROLLED-PAYLOAD-abcdefghij"; // > 16 bytes
+        let (email_id, email_label) = store.admit_output(
+            secret,
+            channel(1),
+            Some(Did([9u8; 32])),
+            &Label::bottom(),
+            jcp_schema::flow::Readers::Public,
+            &[],
+            anchor(),
+        );
+        assert_eq!(email_label.integrity, Integrity::Untrusted);
+
+        // A "summary" that carries a verbatim run of the email.
+        let summary = b"Summary: ATTACKER-CONTROLLED-PAYLOAD-abcdefghij was mentioned.";
+        let (_id, label, witness) = store.label_payload(summary, &no_declared(), Mode::Explicit, anchor());
+
+        let edge = witness.edges.iter().find(|e| e.parent == email_id).expect("edge to the email");
+        assert_eq!(edge.kind, EdgeKind::Witnessed, "L17: verbatim carry is witnessed");
+        assert!(!edge.grams.is_empty(), "the overlap is proven by grams");
+        // The taint propagated: the summary is now untrusted and carries the origin.
+        assert_eq!(label.integrity, Integrity::Untrusted);
+        assert!(!label.origins.is_empty(), "summary carries the email's origin");
+        assert!(verify_witness(&witness).is_ok());
+    }
+
+    #[test]
+    fn sub_w_unregistered_carry_is_not_found() {
+        // The documented residual, pinned as a test so it can never be mistaken for
+        // a guarantee: a short verbatim token below the window floor, not registered,
+        // produces no witnessed edge.
+        let mut store = FlowStore::new(64);
+        let (_id, _l) = store.admit_output(b"code=A1B2C3", channel(1), None, &Label::bottom(),
+            jcp_schema::flow::Readers::Public, &[], anchor());
+        let (_id, label, witness) =
+            store.label_payload(b"the value A1B2C3 appears here", &no_declared(), Mode::Explicit, anchor());
+        assert!(witness.edges.is_empty(), "a sub-W token yields no scan edge");
+        assert!(label.is_bottom(), "so nothing propagates");
+    }
+
+    #[test]
+    fn l19_a_registered_sensitive_term_of_any_length_is_found() {
+        let mut store = FlowStore::new(64);
+        // An 11-char account number, registered by the boundary adapter.
+        let (src_id, _l) = store.admit_output(b"balance page", channel(2), None, &Label::bottom(),
+            jcp_schema::flow::Readers::Public, &[b"12345678901"], anchor());
+
+        let (_id, label, witness) =
+            store.label_payload(b"transfer from 12345678901 now", &no_declared(), Mode::Explicit, anchor());
+        let edge = witness.edges.iter().find(|e| e.parent == src_id).expect("edge to the source");
+        assert_eq!(edge.kind, EdgeKind::Witnessed, "L19: a registered term taints even below W");
+        assert_eq!(edge.witnessed_bytes, 11);
+        assert!(!label.is_bottom());
+        assert!(verify_witness(&witness).is_ok());
+    }
+
+    #[test]
+    fn l18_a_shared_window_only_ever_adds_a_parent() {
+        // Over-approximation soundness: two admitted values sharing a window both
+        // become parents of anything carrying it — a match is never missed, and a
+        // (spurious) collision would only add taint, never remove it.
+        let mut store = FlowStore::new(64);
+        let run = b"SHARED-VERBATIM-RUN-0123456789";
+        let (a, _) = store.admit_output(run, channel(1), None, &Label::bottom(),
+            jcp_schema::flow::Readers::Public, &[], anchor());
+        let (b, _) = store.admit_output(run, channel(2), None, &Label::bottom(),
+            jcp_schema::flow::Readers::Public, &[], anchor());
+        // Distinct channels ⇒ distinct content? No — identical bytes ⇒ same id.
+        assert_eq!(a, b, "identical bytes content-address identically");
+
+        // Two *different* payloads sharing one window.
+        let mut store = FlowStore::new(64);
+        let (p, _) = store.admit_output(b"prefix SHARED-VERBATIM-RUN-0123456789 alpha",
+            channel(1), None, &Label::bottom(), jcp_schema::flow::Readers::Public, &[], anchor());
+        let (q, _) = store.admit_output(b"beta SHARED-VERBATIM-RUN-0123456789 suffix",
+            channel(2), None, &Label::bottom(), jcp_schema::flow::Readers::Public, &[], anchor());
+        assert_ne!(p, q);
+        let (_id, _label, witness) = store.label_payload(
+            b"carries SHARED-VERBATIM-RUN-0123456789 from both", &no_declared(), Mode::Explicit, anchor());
+        let parents: BTreeSet<ContentId> = witness.edges.iter().map(|e| e.parent).collect();
+        assert!(parents.contains(&p) && parents.contains(&q), "both sharers are parents, never missed");
+    }
+
+    #[test]
+    fn l20_verify_refolds_the_cut_and_catches_tampering() {
+        let mut store = FlowStore::new(64);
+        store.admit_output(b"ATTACKER-CONTROLLED-PAYLOAD-abcdefghij", channel(1), Some(Did([9u8; 32])),
+            &Label::bottom(), jcp_schema::flow::Readers::Public, &[], anchor());
+        let (_id, _label, witness) = store.label_payload(
+            b"echo ATTACKER-CONTROLLED-PAYLOAD-abcdefghij", &no_declared(), Mode::Explicit, anchor());
+        assert!(verify_witness(&witness).is_ok(), "a well-formed witness verifies");
+
+        // Tamper the sealed label to look clean; the refold of the cut exposes it.
+        let mut forged = witness.clone();
+        forged.label = Label::bottom();
+        assert_eq!(verify_witness(&forged), Err(WitnessError::FoldMismatch));
+    }
+
+    #[test]
+    fn a_clean_payload_stays_bottom_and_is_byte_identical_ground() {
+        let mut store = FlowStore::new(64);
+        store.admit_principal(b"a principal instruction", anchor());
+        let (_id, label, witness) =
+            store.label_payload(b"an unrelated derived value", &no_declared(), Mode::Explicit, anchor());
+        assert!(label.is_bottom(), "no untrusted ancestor ⇒ ⊥");
+        assert!(witness.edges.is_empty());
+        assert!(verify_witness(&witness).is_ok());
+    }
+
+    #[test]
+    fn a_declared_but_absent_parent_fails_closed_to_top() {
+        let mut store = FlowStore::new(64);
+        let phantom = ContentId::of_bytes(b"a value never admitted");
+        let declared: BTreeSet<ContentId> = [phantom].into_iter().collect();
+        let (_id, label, witness) =
+            store.label_payload(b"derived from something we lost", &declared, Mode::Explicit, anchor());
+        assert_eq!(label, Label::top(), "a lost declared dependency folds to ⊤");
+        assert!(verify_witness(&witness).is_ok(), "the ⊤ fold is witness-consistent");
+    }
+
+    #[test]
+    fn the_attack_shape_a_ceiling_would_deny() {
+        // The store side of read_email → summarize → send_email: after summarizing
+        // an attacker email, the summary's label does not sit below a Principal-only
+        // ceiling, which is exactly what the runtime gate will reject.
+        let mut store = FlowStore::new(64);
+        store.admit_output(b"From attacker: EXFIL-THIS-SECRET-abcdefghij", channel(7),
+            Some(Did([3u8; 32])), &Label::bottom(), jcp_schema::flow::Readers::Public, &[], anchor());
+        let (_id, summary_label, _w) = store.label_payload(
+            b"The email said EXFIL-THIS-SECRET-abcdefghij", &no_declared(), Mode::Explicit, anchor());
+
+        // A send_email grant permitting only clean, principal-origin arguments.
+        let ceiling = FlowCeiling::new(Label {
+            integrity: Integrity::Principal,
+            origins: Origins::none(),
+            readers: jcp_schema::flow::Readers::Public,
+        });
+        assert!(!ceiling.admits(&summary_label), "the tainted summary must not pass the ceiling");
+        // And a bottom argument still would.
+        assert!(ceiling.admits(&Label::bottom()));
+    }
+
+    #[test]
+    fn high_water_floor_taints_a_laundered_payload_when_the_scan_cannot() {
+        // The negative companion to the carry test: base64/paraphrase defeats the
+        // window index, but HighWater's session floor still denies at the sink.
+        let mut store = FlowStore::new(64);
+        store.admit_output(b"ATTACKER-CONTROLLED-PAYLOAD-abcdefghij", channel(1),
+            Some(Did([9u8; 32])), &Label::bottom(), jcp_schema::flow::Readers::Public, &[], anchor());
+
+        // A laundered value sharing no window with the email.
+        let laundered = b"totally different words with no overlap whatsoever xyzzy";
+        let (_id, explicit_label, _w) =
+            store.label_payload(laundered, &no_declared(), Mode::Explicit, anchor());
+        assert!(explicit_label.is_bottom(), "the scan alone cannot see laundering (documented residual)");
+
+        // But HighWater joins the floor, which the untrusted read already raised.
+        let (_id, hw_label, witness) =
+            store.label_payload(laundered, &no_declared(), Mode::HighWater, anchor());
+        assert_eq!(hw_label.integrity, Integrity::Untrusted, "the floor still taints the sink");
+        assert!(witness.edges.iter().any(|e| e.kind == EdgeKind::Ambient));
+    }
+}

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

@@ -0,0 +1,353 @@
+//! The session store and the exact carry detector.
+//!
+//! The store holds every value a run has admitted — tool outputs, principal
+//! messages — each with its flow label, indexed by an **exact** window
+//! fingerprint. When a later value is computed, [`FlowStore::label_payload`] folds
+//! its label from the ancestors it verbatim-carries bytes from, plus any the
+//! caller declared, plus (optionally) a coarse session floor.
+//!
+//! The detector is an exact `blake3` window index, deliberately *not* a rolling
+//! hash: a rolling hash carries state between windows and a carry-arithmetic bug
+//! silently misses overlaps. Here each W-byte window is hashed independently, so
+//! two byte sequences share a fingerprint exactly when they share a verbatim
+//! W-byte run — provable, platform-independent, with no state to get wrong.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use jcp_schema::flow::{Integrity, Label, Origin, Origins, Readers};
+use jcp_schema::{ContentId, Did};
+use serde::{Deserialize, Serialize};
+
+use crate::witness::{CutNode, Edge, EdgeKind, Witness, WITNESS_SCHEMA};
+
+/// The detection floor: overlaps of at least this many verbatim bytes are found;
+/// shorter runs are not (a registered sensitive term closes the sub-window gap).
+pub const SCAN_W: usize = 16;
+
+/// Domain separation for the window hash, so a fingerprint here cannot collide by
+/// construction with any other `blake3` use in the system.
+const WINDOW_DOMAIN: &[u8] = b"jcp/flow/window/v1";
+
+/// The scan parameters pinned into every [`Witness`], so a replay uses the exact
+/// window size and the exact sensitive-term commitment the verdict was made under.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ScanParams {
+    pub w: u16,
+    /// Content address of the registered sensitive-term set at emission time.
+    pub sensitive_root: ContentId,
+}
+
+/// Every W-byte window of `bytes`, each hashed independently with a domain-keyed
+/// `blake3` and truncated to `u64`. Fewer than `w` bytes yields no windows — the
+/// sub-window gap the sensitive-term index exists to cover.
+pub fn windows(bytes: &[u8], w: usize) -> BTreeSet<u64> {
+    let mut out = BTreeSet::new();
+    if w == 0 || bytes.len() < w {
+        return out;
+    }
+    let key = *blake3::hash(WINDOW_DOMAIN).as_bytes();
+    for start in 0..=bytes.len() - w {
+        let digest = blake3::keyed_hash(&key, &bytes[start..start + w]);
+        let mut eight = [0u8; 8];
+        eight.copy_from_slice(&digest.as_bytes()[0..8]);
+        out.insert(u64::from_le_bytes(eight));
+    }
+    out
+}
+
+/// Naive verbatim substring search over bytes — the sub-window sensitive-term test.
+fn contains(haystack: &[u8], needle: &[u8]) -> bool {
+    if needle.is_empty() || needle.len() > haystack.len() {
+        return false;
+    }
+    haystack.windows(needle.len()).any(|window| window == needle)
+}
+
+/// Which mode [`FlowStore::label_payload`] folds in.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Mode {
+    /// Only explicit evidence: verbatim carry, registered terms, and declarations.
+    Explicit,
+    /// Adds the monotone session floor — everything after the first untrusted read
+    /// joins it. A blunt over-approximation of implicit flow; sound, not precise.
+    HighWater,
+}
+
+/// What the store knows about one admitted value.
+#[derive(Clone, Debug)]
+struct Admitted {
+    label: Label,
+    /// Receipt-log head at admission — the anchor a receipt replay uses to prove
+    /// this value existed before anything derived from it.
+    anchor: ContentId,
+    /// This value's own window fingerprints, kept so eviction can remove them
+    /// without rescanning.
+    windows: BTreeSet<u64>,
+}
+
+/// The session's flow state: labels, the exact window index, registered sensitive
+/// terms, and the coarse floor.
+pub struct FlowStore {
+    admitted: BTreeMap<ContentId, Admitted>,
+    /// fingerprint → the admitted ids that contain that window.
+    window_index: BTreeMap<u64, BTreeSet<ContentId>>,
+    /// registered short/exact secret → the source ids that carry it.
+    sensitive: BTreeMap<Vec<u8>, BTreeSet<ContentId>>,
+    /// FIFO admission order, for deterministic eviction.
+    order: Vec<ContentId>,
+    /// The monotone `HighWater` floor.
+    high_water: Label,
+    /// Receipt-log head at the last admission — the next witness's anchor.
+    head: ContentId,
+    cap: usize,
+    w: usize,
+}
+
+/// The sentinel parent id an `Ambient` edge points at (the session floor has no
+/// single source payload).
+pub fn ambient_source() -> ContentId {
+    ContentId::of_bytes(b"jcp/flow/high-water")
+}
+
+impl FlowStore {
+    /// A store holding at most `cap` values (FIFO eviction past that) with window
+    /// size [`SCAN_W`].
+    pub fn new(cap: usize) -> Self {
+        FlowStore {
+            admitted: BTreeMap::new(),
+            window_index: BTreeMap::new(),
+            sensitive: BTreeMap::new(),
+            order: Vec::new(),
+            high_water: Label::bottom(),
+            head: ContentId([0u8; 32]),
+            cap: cap.max(1),
+            w: SCAN_W,
+        }
+    }
+
+    pub fn head(&self) -> ContentId {
+        self.head
+    }
+
+    pub fn get(&self, id: &ContentId) -> Option<&Label> {
+        self.admitted.get(id).map(|entry| &entry.label)
+    }
+
+    /// The content address of the registered sensitive-term set — pinned into a
+    /// witness so a replay knows which terms were in force.
+    pub fn sensitive_root(&self) -> ContentId {
+        let terms: Vec<&Vec<u8>> = self.sensitive.keys().collect();
+        ContentId::of(&terms).expect("term set is canonically serializable")
+    }
+
+    fn params(&self) -> ScanParams {
+        ScanParams { w: self.w as u16, sensitive_root: self.sensitive_root() }
+    }
+
+    fn register_terms(&mut self, id: ContentId, terms: &[&[u8]]) {
+        for term in terms {
+            if term.is_empty() {
+                continue;
+            }
+            self.sensitive.entry(term.to_vec()).or_default().insert(id);
+        }
+    }
+
+    /// Insert a value with a fixed label, indexing its windows and advancing the
+    /// floor. Shared by the boundary admissions and by `label_payload`.
+    fn admit(&mut self, bytes: &[u8], label: Label, anchor: ContentId) -> ContentId {
+        let id = ContentId::of_bytes(bytes);
+        let value_windows = windows(bytes, self.w);
+        for fingerprint in &value_windows {
+            self.window_index.entry(*fingerprint).or_default().insert(id);
+        }
+        // A re-admission of identical bytes keeps the earlier (already-anchored)
+        // entry; its label is content-determined and equal.
+        if !self.admitted.contains_key(&id) {
+            self.order.push(id);
+        }
+        self.admitted.insert(id, Admitted { label: label.clone(), anchor, windows: value_windows });
+        self.high_water = self.high_water.join(&label);
+        self.evict_if_needed();
+        id
+    }
+
+    fn evict_if_needed(&mut self) {
+        while self.order.len() > self.cap {
+            let victim = self.order.remove(0);
+            if let Some(entry) = self.admitted.remove(&victim) {
+                for fingerprint in &entry.windows {
+                    if let Some(ids) = self.window_index.get_mut(fingerprint) {
+                        ids.remove(&victim);
+                        if ids.is_empty() {
+                            self.window_index.remove(fingerprint);
+                        }
+                    }
+                }
+            }
+            // An evicted id, if later declared, is named-but-absent ⇒ folded to ⊤
+            // (fail closed). Its sensitive registrations are left; a term still
+            // points at a gone id, which `label_payload` treats as absent ⇒ ⊤.
+        }
+    }
+
+    /// Boundary: admit a tool **result**. Its label is the channel origin (always
+    /// `Untrusted`, joined with the args that produced the call and the tool's
+    /// reader policy), and any adapter-declared sensitive terms are registered.
+    pub fn admit_output(
+        &mut self,
+        bytes: &[u8],
+        channel: ContentId,
+        principal: Option<Did>,
+        args: &Label,
+        readers: Readers,
+        terms: &[&[u8]],
+        anchor: ContentId,
+    ) -> (ContentId, Label) {
+        let origin = match principal {
+            Some(did) => Origin::from(channel, did),
+            None => Origin::channel(channel),
+        };
+        let channel_label = Label {
+            integrity: Integrity::Untrusted,
+            origins: Origins::one(origin),
+            readers,
+        };
+        let label = channel_label.join(args);
+        let id = self.admit(bytes, label.clone(), anchor);
+        self.register_terms(id, terms);
+        self.head = id;
+        (id, label)
+    }
+
+    /// Boundary: admit a **principal** message — a clean, bottom-labelled value.
+    pub fn admit_principal(&mut self, bytes: &[u8], anchor: ContentId) -> (ContentId, Label) {
+        let id = self.admit(bytes, Label::bottom(), anchor);
+        self.head = id;
+        (id, Label::bottom())
+    }
+
+    /// The propagation entry point: fold a payload's label from every ancestor it
+    /// verbatim-carries bytes from (`Witnessed`), every registered sensitive term
+    /// it contains (`Witnessed`), every caller-declared parent (`Declared`, or ⊤
+    /// if that parent is absent — fail closed), and — in `HighWater` mode — the
+    /// session floor (`Ambient`). Admits the payload under that label and returns a
+    /// self-contained [`Witness`].
+    pub fn label_payload(
+        &mut self,
+        bytes: &[u8],
+        declared: &BTreeSet<ContentId>,
+        mode: Mode,
+        anchor: ContentId,
+    ) -> (ContentId, Label, Witness) {
+        let subject_windows = windows(bytes, self.w);
+        let params = self.params();
+
+        // 1. Window-found parents and the grams that prove each overlap.
+        let mut grams_by_parent: BTreeMap<ContentId, BTreeSet<u64>> = BTreeMap::new();
+        for fingerprint in &subject_windows {
+            if let Some(ids) = self.window_index.get(fingerprint) {
+                for id in ids {
+                    grams_by_parent.entry(*id).or_default().insert(*fingerprint);
+                }
+            }
+        }
+
+        // 2. Sensitive-found parents and the longest matching term per parent.
+        let mut sensitive_bytes: BTreeMap<ContentId, usize> = BTreeMap::new();
+        for (term, sources) in &self.sensitive {
+            if contains(bytes, term) {
+                for source in sources {
+                    let entry = sensitive_bytes.entry(*source).or_insert(0);
+                    *entry = (*entry).max(term.len());
+                }
+            }
+        }
+
+        // 3. Assemble the cut and edges over the union of parents.
+        let mut parent_ids: BTreeSet<ContentId> = BTreeSet::new();
+        parent_ids.extend(grams_by_parent.keys().copied());
+        parent_ids.extend(sensitive_bytes.keys().copied());
+        parent_ids.extend(declared.iter().copied());
+
+        let mut cut: Vec<CutNode> = Vec::new();
+        let mut edges: Vec<Edge> = Vec::new();
+
+        for parent in &parent_ids {
+            match self.admitted.get(parent) {
+                Some(entry) => {
+                    cut.push(CutNode { id: *parent, label: entry.label.clone(), anchor: entry.anchor });
+                    if let Some(grams) = grams_by_parent.get(parent) {
+                        // Verbatim overlap: at least one W-window matched, so at
+                        // least W bytes are verbatim (a sound lower bound).
+                        edges.push(Edge {
+                            parent: *parent,
+                            kind: EdgeKind::Witnessed,
+                            witnessed_bytes: self.w as u64,
+                            grams: grams.clone(),
+                        });
+                    } else if let Some(bytes_len) = sensitive_bytes.get(parent) {
+                        // A registered term appears verbatim. Below the window floor
+                        // there are no grams; the byte count is the evidence.
+                        edges.push(Edge {
+                            parent: *parent,
+                            kind: EdgeKind::Witnessed,
+                            witnessed_bytes: *bytes_len as u64,
+                            grams: BTreeSet::new(),
+                        });
+                    } else {
+                        // Declared with no verbatim overlap: verifiable only via a
+                        // named artifact.
+                        edges.push(Edge {
+                            parent: *parent,
+                            kind: EdgeKind::Declared,
+                            witnessed_bytes: 0,
+                            grams: BTreeSet::new(),
+                        });
+                    }
+                }
+                None => {
+                    // Declared but absent (e.g. evicted): fail closed — fold ⊤ so
+                    // the subject inherits maximal taint rather than silently
+                    // dropping a dependency.
+                    cut.push(CutNode { id: *parent, label: Label::top(), anchor: ContentId([0u8; 32]) });
+                    edges.push(Edge {
+                        parent: *parent,
+                        kind: EdgeKind::Declared,
+                        witnessed_bytes: 0,
+                        grams: BTreeSet::new(),
+                    });
+                }
+            }
+        }
+
+        // 4. HighWater: join the coarse session floor as an Ambient edge.
+        if mode == Mode::HighWater && !self.high_water.is_bottom() {
+            let ambient = ambient_source();
+            cut.push(CutNode { id: ambient, label: self.high_water.clone(), anchor: self.head });
+            edges.push(Edge {
+                parent: ambient,
+                kind: EdgeKind::Ambient,
+                witnessed_bytes: 0,
+                grams: BTreeSet::new(),
+            });
+        }
+
+        let label = jcp_schema::flow::fold(
+            &cut.iter().map(|node| node.label.clone()).collect::<Vec<_>>(),
+        );
+
+        let id = self.admit(bytes, label.clone(), anchor);
+        self.head = id;
+
+        let witness = Witness {
+            schema: WITNESS_SCHEMA.to_string(),
+            params,
+            subject: id,
+            label: label.clone(),
+            cut,
+            edges,
+        };
+        (id, label, witness)
+    }
+}

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

@@ -0,0 +1,145 @@
+//! The witness — the content-addressed, offline-checkable record of *why* a value
+//! carries the label it does.
+//!
+//! A receipt commits only the witness's `id()`. An entitled auditor fetches the
+//! witness and rechecks it: re-fold the disclosed ancestor cut and confirm it
+//! reproduces the sealed label ([`verify_witness`]). That is what makes a
+//! provenance verdict replayable by a third party rather than a claim to trust.
+
+use std::collections::BTreeSet;
+
+use jcp_schema::{flow::Label, ContentId};
+use serde::{Deserialize, Serialize};
+
+/// How strongly a taint edge is justified — sealed, so a receipt shows whether a
+/// verdict rested on hard bytes or the conservative floor.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum EdgeKind {
+    /// A verbatim overlap the scan found between this value and an ancestor: the
+    /// grams prove it and a third party rechecks them. The strongest evidence.
+    Witnessed,
+    /// The caller (or a registered sensitive term) named this ancestor. Verifiable
+    /// only with the named artifact — a citation, an endorsement, or the term set.
+    Declared,
+    /// The coarse session floor (`HighWater` mode): everything after the first
+    /// untrusted read joins it. Sound but imprecise, and flagged as such in replay.
+    Ambient,
+}
+
+fn u64_is_zero(value: &u64) -> bool {
+    *value == 0
+}
+
+/// One taint edge: a derived value depends on an ancestor payload, and this says
+/// how that was established.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Edge {
+    /// The ancestor payload's content id — its key in the [`crate::FlowStore`] and
+    /// the cut node it justifies.
+    pub parent: ContentId,
+    pub kind: EdgeKind,
+    /// For a `Witnessed` edge, how many verbatim bytes were matched (informational;
+    /// the grams are the proof).
+    #[serde(skip_serializing_if = "u64_is_zero", default)]
+    pub witnessed_bytes: u64,
+    /// The shared window fingerprints proving the overlap. A verifier rechecks
+    /// these are a subset of both endpoints' window sets. Empty for `Declared` and
+    /// `Ambient` edges, and for a sub-window sensitive-term match.
+    #[serde(skip_serializing_if = "BTreeSet::is_empty", default)]
+    pub grams: BTreeSet<u64>,
+}
+
+/// A direct ancestor in the cut: its content id, the label it was admitted with,
+/// and the receipt-log head at admission time (the anchor a receipt replay uses to
+/// prove the read happened before this value existed).
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CutNode {
+    pub id: ContentId,
+    pub label: Label,
+    pub anchor: ContentId,
+}
+
+/// The minimal ancestor cut, its edges, and the folded label — self-contained and
+/// content-addressed. The one artifact a third party needs (beyond the receipt
+/// chain) to re-derive a flow verdict offline.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Witness {
+    /// Schema tag, `"flow/1"`.
+    pub schema: String,
+    /// The scan parameters in force when this witness was emitted — pinned so a
+    /// replay uses the exact window size and sensitive-term commitment.
+    pub params: crate::store::ScanParams,
+    /// The payload this witness explains (`== frame.payload_hash` of the call).
+    pub subject: ContentId,
+    /// The folded effective label — what the ceiling was tested against.
+    pub label: Label,
+    /// Each direct ancestor.
+    pub cut: Vec<CutNode>,
+    /// How each ancestor taints the subject.
+    pub edges: Vec<Edge>,
+}
+
+/// The witness schema tag.
+pub const WITNESS_SCHEMA: &str = "flow/1";
+
+impl Witness {
+    /// The content id a receipt commits to. Order-free because every field is
+    /// canonically serialized (`BTree`/`Vec` in insertion order the store fixes).
+    pub fn id(&self) -> ContentId {
+        ContentId::of(self).expect("witness is canonically serializable")
+    }
+}
+
+/// Why a witness failed to verify.
+#[derive(Debug, thiserror::Error, PartialEq, Eq)]
+pub enum WitnessError {
+    #[error("witness schema is not {WITNESS_SCHEMA}")]
+    BadSchema,
+    /// Re-folding the cut did not reproduce the sealed label — the witness does not
+    /// justify its own verdict (violates the fold law L20).
+    #[error("re-folded label does not match the sealed label")]
+    FoldMismatch,
+    /// An edge names a parent that is not in the cut.
+    #[error("edge parent {0:?} is not in the cut")]
+    DanglingEdge(ContentId),
+    /// A `Witnessed` edge carries no evidence (no grams and zero bytes).
+    #[error("witnessed edge to {0:?} carries no evidence")]
+    EmptyWitnessedEdge(ContentId),
+}
+
+/// Verify a witness against itself: the schema is right, every edge points at a
+/// cut node, every `Witnessed` edge carries evidence, and — the load-bearing check
+/// (law L20) — re-folding the cut's labels reproduces the sealed label.
+///
+/// This is the *consistency* tier: it proves the witness is internally coherent and
+/// that its verdict follows from its own cut. It does **not** re-derive the grams
+/// from the original bytes; that truth tier requires disclosing the endpoint bytes
+/// and is a separate, opt-in check (see the crate residuals). Consistency is enough
+/// to catch a witness that was tampered to soften its own verdict, because the
+/// label is refolded from the cut rather than trusted.
+pub fn verify_witness(witness: &Witness) -> Result<(), WitnessError> {
+    if witness.schema != WITNESS_SCHEMA {
+        return Err(WitnessError::BadSchema);
+    }
+
+    let cut_ids: BTreeSet<&ContentId> = witness.cut.iter().map(|node| &node.id).collect();
+    for edge in &witness.edges {
+        if !cut_ids.contains(&edge.parent) {
+            return Err(WitnessError::DanglingEdge(edge.parent));
+        }
+        if edge.kind == EdgeKind::Witnessed && edge.grams.is_empty() && edge.witnessed_bytes == 0 {
+            return Err(WitnessError::EmptyWitnessedEdge(edge.parent));
+        }
+    }
+
+    // L20 — re-fold the cut. Join is commutative and associative, so any order
+    // reproduces the same label; a tampered `label` field is caught here.
+    let refolded = jcp_schema::flow::fold(
+        &witness.cut.iter().map(|node| node.label.clone()).collect::<Vec<_>>(),
+    );
+    if refolded != witness.label {
+        return Err(WitnessError::FoldMismatch);
+    }
+    Ok(())
+}