Merge BP-3: an oversize answer is deferred to the body plane

An answer too large for one send becomes HuginnMindResponse::Deferred,
carrying only the reference's CultMeshCdnArtifactManifest (Q-BP2 A). The
client fetches the body over the session it already holds with
cultnet-rs fetch_content (Q-BP1 A) and gets bytes identical to the mind's
answer. Retention is in memory only: 256 MiB counted by distinct chunk,
a sliding 60 s TTL, and LRU eviction. ResponseTooLarge stays the backstop
above 64 MiB. Every echoed field is bounded at 256 bytes, so every reply
fits the window. A deferred envelope carries its answer's status.

Soul closed it after one fix batch. Named limit: there is no per-peer
budget, so a client can evict another client's body mid-fetch. That is
acceptable for trusted WireGuard clients. mind 91, daemon 27.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/huginn-daemon/src/bodies.rs

@@ -0,0 +1,288 @@
+//! The deferred bodies: answers too large for one send, held for the client to
+//! fetch chunk by chunk. Ephemeral transport state, never a document store and
+//! never world truth: it is lost when the process exits, and a client that
+//! finds a body gone asks the operation again.
+//!
+//! Identity is the manifest's `contentHash`, so the same bytes deferred twice
+//! are one body. Chunking, hashing and the manifest are `cultnet-rs`'s; what is
+//! owned here is only how long and how much is kept.
+//!
+//! A body expires when it has been untouched for the TTL, where retaining it
+//! again and serving any of its chunks both touch it. When the stored bytes
+//! exceed the budget the body touched longest ago is evicted first. A chunk
+//! that two bodies share is stored once and lives until the last of them goes.
+//! Time is an argument, so nothing here reads a clock; expiry therefore runs
+//! only when a frame arrives, and an idle daemon holds an expired body until
+//! the next one.
+//!
+//! Limit: there is no per-session or per-peer budget, so any client can evict
+//! another client's body mid-fetch: four 64 MiB reads flush the 256 MiB store,
+//! and the loser's next chunk is answered `found: false`. That is acceptable
+//! for the organ's trusted WireGuard clients, where an instance identity is
+//! declared attribution and not authentication.
+
+use std::cell::Cell;
+use std::collections::{BTreeSet, HashMap};
+use std::time::Duration;
+
+use chrono::{DateTime, Utc};
+use cultnet_rs::{CultMeshCdnArtifactManifest, CultMeshCdnChunk};
+
+/// The size of one chunk of a deferred body. Four fit one send with room for a
+/// control-plane answer beside them on the same session.
+pub const DEFERRED_CHUNK_BYTES: usize = 256 * 1024;
+/// The largest answer that is deferred rather than refused, measured on the
+/// encoded payload.
+pub const MAX_DEFERRED_BODY_BYTES: u64 = 64 * 1024 * 1024;
+/// What the daemon keeps across all deferred bodies.
+pub const DEFERRED_BUDGET_BYTES: u64 = 256 * 1024 * 1024;
+
+struct Chunk {
+    bytes: Vec<u8>,
+    /// The content hashes of the retained bodies that reference this chunk.
+    owners: BTreeSet<String>,
+}
+
+struct Body {
+    chunk_hashes: BTreeSet<String>,
+    touched_at: Cell<DateTime<Utc>>,
+    /// Orders touches when the clock does not: the largest is the newest.
+    touch_order: Cell<u64>,
+}
+
+pub struct DeferredBodies {
+    budget: u64,
+    ttl: chrono::Duration,
+    chunks: HashMap<String, Chunk>,
+    bodies: HashMap<String, Body>,
+    stored: u64,
+    touches: Cell<u64>,
+}
+
+impl DeferredBodies {
+    pub fn new(budget: u64, ttl: Duration) -> Self {
+        Self {
+            budget,
+            ttl: chrono::Duration::from_std(ttl).unwrap_or(chrono::Duration::MAX),
+            chunks: HashMap::new(),
+            bodies: HashMap::new(),
+            stored: 0,
+            touches: Cell::new(0),
+        }
+    }
+
+    /// Bytes held: each distinct chunk once.
+    #[cfg(test)]
+    pub fn stored_bytes(&self) -> u64 {
+        self.stored
+    }
+
+    #[cfg(test)]
+    pub fn body_count(&self) -> usize {
+        self.bodies.len()
+    }
+
+    /// Holds a body, or, when its bytes are already held, touches it and
+    /// stores nothing. Then evicts, least recently touched first, while the
+    /// stored bytes exceed the budget and another body could be evicted
+    /// instead: the body just retained is the last to go, and one that alone
+    /// exceeds the budget is kept whole.
+    pub fn retain(&mut self, manifest: &CultMeshCdnArtifactManifest, chunks: Vec<CultMeshCdnChunk>, now: DateTime<Utc>) {
+        let content_hash = manifest.content_hash.clone();
+        if let Some(body) = self.bodies.get(&content_hash) {
+            self.touch(body, now);
+            return;
+        }
+        let mut chunk_hashes = BTreeSet::new();
+        for chunk in chunks {
+            chunk_hashes.insert(chunk.chunk_hash.clone());
+            let held = self.chunks.entry(chunk.chunk_hash).or_insert_with(|| {
+                self.stored += chunk.payload.len() as u64;
+                Chunk { bytes: chunk.payload, owners: BTreeSet::new() }
+            });
+            held.owners.insert(content_hash.clone());
+        }
+        let body = Body { chunk_hashes, touched_at: Cell::new(now), touch_order: Cell::new(0) };
+        self.touch(&body, now);
+        self.bodies.insert(content_hash, body);
+        while self.stored > self.budget && self.bodies.len() > 1 {
+            let oldest = self
+                .bodies
+                .iter()
+                .min_by_key(|(_, body)| body.touch_order.get())
+                .map(|(hash, _)| hash.clone())
+                .expect("more than one body is held");
+            self.drop_body(&oldest);
+        }
+    }
+
+    /// One chunk's bytes by its normalised hash, touching every body that
+    /// references it. Serving a chunk is what keeps its body alive.
+    pub fn chunk(&self, hash: &str, now: DateTime<Utc>) -> Option<&[u8]> {
+        let chunk = self.chunks.get(hash)?;
+        for owner in &chunk.owners {
+            self.touch(&self.bodies[owner], now);
+        }
+        Some(&chunk.bytes)
+    }
+
+    /// Drops every body untouched for the TTL as of `now`. A clock that reads
+    /// earlier than a touch expires nothing.
+    pub fn expire(&mut self, now: DateTime<Utc>) {
+        let expired: Vec<String> = self
+            .bodies
+            .iter()
+            .filter(|(_, body)| now.signed_duration_since(body.touched_at.get()) >= self.ttl)
+            .map(|(hash, _)| hash.clone())
+            .collect();
+        for hash in expired {
+            self.drop_body(&hash);
+        }
+    }
+
+    fn touch(&self, body: &Body, now: DateTime<Utc>) {
+        body.touched_at.set(now);
+        self.touches.set(self.touches.get() + 1);
+        body.touch_order.set(self.touches.get());
+    }
+
+    fn drop_body(&mut self, content_hash: &str) {
+        let Some(body) = self.bodies.remove(content_hash) else { return };
+        for hash in body.chunk_hashes {
+            let chunk = self.chunks.get_mut(&hash).expect("a retained body's chunks are held");
+            chunk.owners.remove(content_hash);
+            if chunk.owners.is_empty() {
+                self.stored -= chunk.bytes.len() as u64;
+                self.chunks.remove(&hash);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use cultnet_rs::pack_content;
+
+    /// The three numbers a client author is told, spelled out: a chunk of
+    /// 256 KiB, a largest deferred body of 64 MiB, a budget of 256 MiB that
+    /// holds at least one whole body.
+    #[test]
+    fn the_deferral_constants_are_the_documented_numbers() {
+        assert_eq!(DEFERRED_CHUNK_BYTES, 262_144);
+        assert_eq!(MAX_DEFERRED_BODY_BYTES, 67_108_864);
+        assert_eq!(DEFERRED_BUDGET_BYTES, 268_435_456);
+    }
+
+    fn at(seconds: i64) -> DateTime<Utc> {
+        DateTime::from_timestamp(1_800_000_000 + seconds, 0).unwrap()
+    }
+
+    /// A body of `pieces` chunks of 4 bytes each, distinct per `tag`.
+    fn body(tag: u8, pieces: usize) -> (CultMeshCdnArtifactManifest, Vec<CultMeshCdnChunk>) {
+        let mut bytes = Vec::new();
+        for piece in 0..pieces {
+            bytes.extend_from_slice(&[tag, piece as u8, 7, 7]);
+        }
+        pack_content("t", "package", "", "", "", &bytes, 4).unwrap()
+    }
+
+    fn held(store: &DeferredBodies, manifest: &CultMeshCdnArtifactManifest, now: DateTime<Utc>) -> bool {
+        manifest.chunks.iter().all(|chunk| store.chunk(&chunk.chunk_hash, now).is_some())
+    }
+
+    const TTL: Duration = Duration::from_secs(60);
+
+    /// The same bytes deferred twice are one body, counted once; serving a
+    /// chunk slides the expiry; an untouched body is gone exactly at the TTL.
+    #[test]
+    fn retention_is_idempotent_and_sliding() {
+        let mut store = DeferredBodies::new(1_000, TTL);
+        let (manifest, chunks) = body(1, 3);
+        store.retain(&manifest, chunks.clone(), at(0));
+        store.retain(&manifest, chunks, at(1));
+        assert_eq!((store.body_count(), store.stored_bytes()), (1, 12), "one body, its bytes counted once");
+
+        // Retaining again at t=1 touched it, so it lives to t=61, not t=60.
+        store.expire(at(60));
+        assert_eq!(store.body_count(), 1, "the second retain refreshed it");
+        // Serving a chunk at t=60 touches it, so t=119 is inside a fresh TTL.
+        assert!(store.chunk(&manifest.chunks[0].chunk_hash, at(60)).is_some());
+        store.expire(at(119));
+        assert_eq!(store.body_count(), 1, "serving a chunk slid the expiry");
+        store.expire(at(120));
+        assert_eq!((store.body_count(), store.stored_bytes()), (0, 0), "untouched for the TTL is gone");
+
+        // An untouched body is gone at the TTL and not before it.
+        let (other, chunks) = body(2, 1);
+        store.retain(&other, chunks, at(200));
+        store.expire(at(259));
+        assert_eq!(store.body_count(), 1);
+        store.expire(at(260));
+        assert_eq!(store.body_count(), 0);
+    }
+
+    /// Time going backwards expires nothing.
+    #[test]
+    fn a_clock_behind_the_last_touch_expires_nothing() {
+        let mut store = DeferredBodies::new(1_000, TTL);
+        let (manifest, chunks) = body(1, 1);
+        store.retain(&manifest, chunks, at(100));
+        store.expire(at(0));
+        assert_eq!(store.body_count(), 1);
+    }
+
+    /// Over the budget the body touched longest ago goes first, not the
+    /// newest and not the first inserted, and a chunk two bodies share stays
+    /// while either lives.
+    #[test]
+    fn eviction_is_least_recently_touched_and_bounded_in_bytes() {
+        // Budget 12 bytes: three chunks of 4.
+        let mut store = DeferredBodies::new(12, TTL);
+        let (a, chunks_a) = body(1, 1);
+        let (b, chunks_b) = body(2, 1);
+        let (c, chunks_c) = body(3, 1);
+        store.retain(&a, chunks_a, at(0));
+        store.retain(&b, chunks_b, at(1));
+        // The oldest-inserted body is touched last, so insertion order and
+        // recency disagree about who goes.
+        assert!(store.chunk(&a.chunks[0].chunk_hash, at(2)).is_some());
+        store.retain(&c, chunks_c, at(3));
+        assert_eq!(store.stored_bytes(), 12);
+        let (d, chunks_d) = body(4, 1);
+        store.retain(&d, chunks_d, at(4));
+        assert_eq!(store.stored_bytes(), 12, "bounded by the budget");
+        assert!(!held(&store, &b, at(5)), "the least recently touched went");
+        assert!(held(&store, &a, at(5)) && held(&store, &c, at(5)) && held(&store, &d, at(5)));
+
+        // A shared chunk survives its first owner.
+        let mut store = DeferredBodies::new(12, TTL);
+        let (x, chunks_x) = body(5, 2); // chunks x0 x1
+        store.retain(&x, chunks_x, at(0));
+        let (y, mut chunks_y) = body(6, 1); // chunk y0, plus x1 shared
+        let mut manifest_y = y.clone();
+        manifest_y.content_hash = "shared-owner".into();
+        chunks_y.push(CultMeshCdnChunk {
+            chunk_hash: x.chunks[1].chunk_hash.clone(),
+            payload: vec![5, 1, 7, 7],
+        });
+        store.retain(&manifest_y, chunks_y, at(1));
+        assert_eq!(store.stored_bytes(), 12, "the shared chunk is stored once");
+        // Evict x (older): its private chunk goes, the shared one stays.
+        let (z, chunks_z) = body(7, 1);
+        store.retain(&z, chunks_z, at(2));
+        assert_eq!(store.body_count(), 2, "x was evicted");
+        assert!(store.chunk(&x.chunks[0].chunk_hash, at(3)).is_none(), "x's private chunk is gone");
+        assert!(store.chunk(&x.chunks[1].chunk_hash, at(3)).is_some(), "the shared chunk lives with y");
+    }
+
+    /// A body that alone exceeds the budget is kept whole rather than
+    /// evicting itself, so a manifest handed out is never a lie.
+    #[test]
+    fn a_body_over_the_budget_by_itself_is_kept() {
+        let mut store = DeferredBodies::new(4, TTL);
+        let (manifest, chunks) = body(1, 3);
+        store.retain(&manifest, chunks, at(0));
+        assert!(held(&store, &manifest, at(0)));
+    }
+}

crates/huginn-daemon/src/lib.rs

@@ -6,12 +6,15 @@
 //! request": it dispatches, it asks `Mind::require_instance` for every read
 //! that names an instance, and it compares nothing itself. `envelope` owns "is
 //! this a request at all": a message it cannot decode is answered with a typed
-//! failure and touches no mind. `serve` owns "which session gets which reply".
+//! failure and touches no mind. `serve` owns "which session gets which reply"
+//! and whether an answer is delivered whole, deferred or refused; `bodies`
+//! owns how long and how much of a deferred answer is held.
 //!
 //! Every admission rule, the instance check, the receipt, the derived status
 //! and the wire vocabulary are `huginn-mind`'s. The leaf and the store type
 //! are reached through `huginn_mind`, so one crate pins one revision of each.
 
+pub mod bodies;
 pub mod daemon;
 pub mod envelope;
 pub mod serve;

crates/huginn-mind/src/lib.rs

@@ -48,7 +48,7 @@ pub use receipt::{DocumentVersion, Faculty, HuginnCommitReceipt, PipelineProvena
 pub use refusal::MindRefusal;
 pub use store::{MindStore, OwnedRedbMessagePackBackingStore};
 pub use wire::{
-    HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA,
+    DeferredAnswer, HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA,
     MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus,
 };
 

crates/huginn-mind/src/refusal.rs

@@ -55,14 +55,15 @@ pub enum MindRefusal {
     FindingWithoutEvidence,
     UnknownInvariant { label: String },
     NotStewarded { repo: String },
-    /// The one refusal a mind never raises: the answer does not fit the
-    /// transport that would have carried it. It lives here because a refusal
-    /// rides the response schema and there is one of those, so a client reads
-    /// it the way it reads every other refusal rather than parsing a second
-    /// vocabulary. `bytes` is the encoded answer and `limit` is what one send
-    /// may carry, both so a caller can narrow its own request; nothing is
-    /// truncated or paginated on its behalf. The daemon owns it and the
-    /// transport's own bound decides it; no rule of the mind is involved.
+    /// The one refusal a mind never raises: the answer exceeds the largest
+    /// body the organ will deliver by any plane, one send or a deferred body.
+    /// It lives here because a refusal rides the response schema and there is
+    /// one of those, so a client reads it the way it reads every other refusal
+    /// rather than parsing a second vocabulary. `bytes` is the encoded answer
+    /// (the payload, not its envelope) and `limit` is the largest body the
+    /// daemon defers, both so a caller can narrow its own request; nothing is
+    /// truncated or paginated on its behalf. The daemon owns it and its
+    /// deferral bound decides it; no rule of the mind is involved.
     ResponseTooLarge { bytes: u64, limit: u64 },
     /// A selection the substrate or the organ's value door refused: `field`
     /// is the selection field the client sent, `value` the offending part.

crates/huginn-mind/src/wire.rs

@@ -5,7 +5,7 @@
 //! names a transport.
 
 use cultcache_rs::DatabaseEntry;
-use cultnet_rs::Selection;
+use cultnet_rs::{CultMeshCdnArtifactManifest, Selection};
 use epiphany_pipeline::{PIPELINE_SCHEMA_EPOCH, PipelineKind, PipelineRef, Slug};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
@@ -41,6 +41,18 @@ fn selection_ref(_: &mut schemars::SchemaGenerator) -> schemars::Schema {
     })
 }
 
+/// The deferred answer's manifest is CultMesh's own type, a positional array
+/// the C# reference defines (`CultMeshCdnArtifactManifest`, schema version
+/// `gamecult.mesh.cdn_artifact_manifest.v1`). CultLib publishes no JSON schema
+/// for it, so the schema says what the reference says: an array of its ten
+/// fields in order, and nothing here copies their meaning.
+fn manifest_array(_: &mut schemars::SchemaGenerator) -> schemars::Schema {
+    schemars::json_schema!({
+        "description": "gamecult.mesh.cdn_artifact_manifest.v1, the C# reference's CultMeshCdnArtifactManifest as its positional array: [artifactId, kind, version, contentHash, sizeBytes, mimeType, createdAtUtc, chunks[[chunkHash, offset, sizeBytes, recordKey]], tags, metadata].",
+        "type": "array"
+    })
+}
+
 /// Every operation a mind answers. `Admit` carries the batch whole because the
 /// batch already names the instance and the asker; a second `instance` beside
 /// it would be two declarations. Every read names the instance because ruling
@@ -91,13 +103,30 @@ pub enum HuginnMindResponse {
     View(Option<PipelineDocumentView>),
     Query(PipelineSelectionPage),
     Refused(MindRefusal),
+    /// The answer did not fit one send and is being held for the client to
+    /// fetch. Nothing in a mind produces this: the daemon's serve module does,
+    /// in place of an answer, and the answer it stands for is the named
+    /// MessagePack of the `HuginnMindResponse` the mind gave.
+    Deferred(DeferredAnswer),
+}
+
+/// Where the body of an answer that did not fit one send can be fetched: the
+/// reference's own manifest, whose `contentHash` is the SHA-256 of the answer's
+/// encoded bytes and whose chunks are asked for, one at a time, on the session
+/// this arrived on. Row-level facts come from asking again with a header
+/// projection; there is no summary.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+pub struct DeferredAnswer {
+    #[schemars(schema_with = "manifest_array")]
+    pub manifest: CultMeshCdnArtifactManifest,
 }
 
 impl HuginnMindResponse {
     /// The envelope's `status`, in the C# reference's vocabulary: `rejected`
     /// exactly for a refusal, `accepted` otherwise. `Conflict` and
     /// `AlreadyAdmitted` are answers about the mind, not refusals of the
-    /// request.
+    /// request. A `Deferred` is nominally `accepted`; the daemon sets the
+    /// envelope of a deferred answer to the status of the answer it stands for.
     pub fn status(&self) -> &'static str {
         match self {
             Self::Refused(_) | Self::Admit(PipelineAdmissionOutcome::Refused(_)) => "rejected",
@@ -176,6 +205,11 @@ mod tests {
                 edges: None,
             }),
             HuginnMindResponse::Refused(refusal),
+            HuginnMindResponse::Deferred(DeferredAnswer {
+                manifest: cultnet_rs::pack_content("huginn.mind_response", "package", "", "", "", b"body", 4)
+                    .unwrap()
+                    .0,
+            }),
         ]
     }
 

schemas/cultnet/huginn.mind_response.v1.schema.json

@@ -69,6 +69,19 @@
       "required": [
         "Refused"
       ]
+    },
+    {
+      "description": "The answer did not fit one send and is being held for the client to\nfetch. Nothing in a mind produces this: the daemon's serve module does,\nin place of an answer, and the answer it stands for is the named\nMessagePack of the `HuginnMindResponse` the mind gave.",
+      "type": "object",
+      "properties": {
+        "Deferred": {
+          "$ref": "#/$defs/DeferredAnswer"
+        }
+      },
+      "additionalProperties": false,
+      "required": [
+        "Deferred"
+      ]
     }
   ],
   "$defs": {
@@ -287,6 +300,19 @@
       "type": "string",
       "maxLength": 10
     },
+    "DeferredAnswer": {
+      "description": "Where the body of an answer that did not fit one send can be fetched: the\nreference's own manifest, whose `contentHash` is the SHA-256 of the answer's\nencoded bytes and whose chunks are asked for, one at a time, on the session\nthis arrived on. Row-level facts come from asking again with a header\nprojection; there is no summary.",
+      "type": "object",
+      "properties": {
+        "manifest": {
+          "description": "gamecult.mesh.cdn_artifact_manifest.v1, the C# reference's CultMeshCdnArtifactManifest as its positional array: [artifactId, kind, version, contentHash, sizeBytes, mimeType, createdAtUtc, chunks[[chunkHash, offset, sizeBytes, recordKey]], tags, metadata].",
+          "type": "array"
+        }
+      },
+      "required": [
+        "manifest"
+      ]
+    },
     "Deviation": {
       "type": "object",
       "properties": {
@@ -1202,7 +1228,7 @@
           ]
         },
         {
-          "description": "The one refusal a mind never raises: the answer does not fit the\ntransport that would have carried it. It lives here because a refusal\nrides the response schema and there is one of those, so a client reads\nit the way it reads every other refusal rather than parsing a second\nvocabulary. `bytes` is the encoded answer and `limit` is what one send\nmay carry, both so a caller can narrow its own request; nothing is\ntruncated or paginated on its behalf. The daemon owns it and the\ntransport's own bound decides it; no rule of the mind is involved.",
+          "description": "The one refusal a mind never raises: the answer exceeds the largest\nbody the organ will deliver by any plane, one send or a deferred body.\nIt lives here because a refusal rides the response schema and there is\none of those, so a client reads it the way it reads every other refusal\nrather than parsing a second vocabulary. `bytes` is the encoded answer\n(the payload, not its envelope) and `limit` is the largest body the\ndaemon defers, both so a caller can narrow its own request; nothing is\ntruncated or paginated on its behalf. The daemon owns it and its\ndeferral bound decides it; no rule of the mind is involved.",
           "type": "object",
           "properties": {
             "ResponseTooLarge": {