huginn-mind: refuse an unkeyable document as a document, and prove the real mind reads back

Admission wrapped every prepare error in Unavailable, so a key the leaf
could not compose reached the client as an unavailable organ. One private
prepare(document, cache) now keys through the leaf first, mapped to
MindRefusal::Document, then encodes; Mind::admit and the derived-write loop
both use it. Unavailable stays for store, cache-encode and decode faults.

Tests: an_unkeyable_document_is_refused_as_a_document (store and receipts
unchanged), prepare_classifies_a_key_refusal_as_a_document_and_encodes_the_rest
(the derived path cannot reach an unkeyable resolution after the per-kind
bound, so the helper is pinned directly), and the #[ignore]
snapshot_admits_and_reads_back, which opens a scratch copy of the live mind
named by HUGINN_MIND_SNAPSHOT, views every stored document back unchanged,
resolves the six stuck findings and withdraws one long-keyed resolution.

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

crates/huginn-mind/src/admission.rs

@@ -29,7 +29,7 @@
 use std::collections::BTreeSet;
 
 use chrono::{DateTime, Utc};
-use cultcache_rs::CultCacheEnvelope;
+use cultcache_rs::{CultCache, CultCacheEnvelope};
 use epiphany_pipeline::{
     ClaimOutcome, FindingConfidence, Line, PipelineDocument, PipelineKind, PipelineRef, PipelineRefusal,
     PipelineResolution, PipelineStewardship, ResolutionOutcome, RulingAuthority, Short, Slug, pipeline_key,
@@ -67,6 +67,14 @@ pub enum PipelineAdmissionOutcome {
     Conflict { identities: Vec<PipelineRef> },
 }
 
+/// Keys a document through the leaf, then encodes it. A key the leaf cannot
+/// compose is the document's fault and is refused as a document; only the
+/// cache's encode fault is the organ's.
+fn prepare(document: &PipelineDocument, cache: &CultCache) -> Result<CultCacheEnvelope, MindRefusal> {
+    pipeline_key(document).map_err(MindRefusal::Document)?;
+    document.prepare(cache).map_err(unavailable)
+}
+
 impl<S: MindStore> Mind<S> {
     /// Validates and prepares every document through the leaf, then admits
     /// the envelopes.
@@ -76,9 +84,9 @@ impl<S: MindStore> Mind<S> {
             if let Err(refusal) = document.validate() {
                 return PipelineAdmissionOutcome::Refused(MindRefusal::Document(refusal));
             }
-            match document.prepare(self.cache()) {
+            match prepare(document, self.cache()) {
                 Ok(envelope) => envelopes.push(envelope),
-                Err(error) => return PipelineAdmissionOutcome::Refused(unavailable(error)),
+                Err(refusal) => return PipelineAdmissionOutcome::Refused(refusal),
             }
         }
         self.admit_prepared(&batch.instance, batch.provenance, envelopes, now)
@@ -133,7 +141,7 @@ impl<S: MindStore> Mind<S> {
         let originals = docs.batch.len();
         for document in derive(&docs, &mind) {
             document.validate()?;
-            let envelope = document.prepare(self.cache()).map_err(unavailable)?;
+            let envelope = prepare(&document, self.cache())?;
             docs.push(stage(envelope, &mind)?)?;
         }
         for staged in &docs.batch[originals..] {
@@ -684,6 +692,8 @@ fn refuse_collisions(docs: &Docs) -> Result<(), MindRefusal> {
 mod tests {
     use super::*;
     use crate::fixtures::*;
+    use crate::fixtures::prepare;
+    use crate::mind::schema_cache;
     use crate::receipt::{DocumentVersion, Faculty, HuginnCommitReceipt};
     use crate::store::test_stores::{MemoryStore, RefusingStore, SwapCommand};
     use cultcache_rs::DatabaseEntry;
@@ -1671,6 +1681,53 @@ mod tests {
         );
     }
 
+    /// refusals-typed: a document the leaf cannot key is the client's fault.
+    /// The finding's local is 64 bytes of label under `cut-1.s1.`; the
+    /// resolution's subject is a resolution whose own local leaves no room for
+    /// a third nesting. Both pass `validate` and fail only at the key.
+    fn unkeyable() -> Vec<D> {
+        let wide = D::Finding(finding("1", 1, &"w".repeat(64), FindingConfidence::Confirmed));
+        let deep_subject = id("resolution", &format!("{}.{}", "x".repeat(60), "y".repeat(40)));
+        vec![wide, resolution(r(K::Resolution, &deep_subject), withdrawn())]
+    }
+
+    #[test]
+    fn an_unkeyable_document_is_refused_as_a_document() {
+        let store = RefusingStore::new();
+        let mut mind = opened(store.clone(), INSTANCE);
+        seed(&mut mind);
+        let before = store.rows();
+        let receipts = receipt_count(&mind);
+        for document in unkeyable() {
+            document.validate().unwrap();
+            match refusal(admit(&mut mind, vec![document])) {
+                MindRefusal::Document(PipelineRefusal::InvalidFormat { field, .. }) => {
+                    assert!(field.ends_with(".key"), "{field}")
+                }
+                other => panic!("{other:?}"),
+            }
+            assert_eq!(store.rows(), before);
+            assert_eq!(receipt_count(&mind), receipts);
+        }
+    }
+
+    /// The derived path (`admit_steps`) shares the helper. No derivable
+    /// document can fail to key after the leaf's per-kind bound (a derived
+    /// resolution is depth one or two over a subject of at most 64 bytes), so
+    /// the helper is pinned directly.
+    #[test]
+    fn prepare_classifies_a_key_refusal_as_a_document_and_encodes_the_rest() {
+        let cache = schema_cache().unwrap();
+        let keyable = question("Q1", &["A", "B"], "A");
+        assert_eq!(super::prepare(&keyable, &cache).unwrap(), prepare(&keyable));
+        for document in unkeyable() {
+            assert!(matches!(
+                super::prepare(&document, &cache),
+                Err(MindRefusal::Document(PipelineRefusal::InvalidFormat { .. }))
+            ));
+        }
+    }
+
     #[test]
     fn a_ruling_answering_a_question_derives_the_answered_resolution_atomically() {
         let mut mind = seeded();

crates/huginn-mind/src/mind.rs

@@ -355,6 +355,7 @@ fn snapshot(cache: &CultCache) -> Vec<CultCacheEnvelope> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::admission::PipelineAdmissionOutcome;
     use crate::fixtures::{INSTANCE, epoch, instance, prepare, slug};
     use crate::store::test_stores::MemoryStore;
     use cultcache_rs::CacheBackingStore;
@@ -778,4 +779,71 @@ mod tests {
             }))
         );
     }
+
+    /// stored-documents-valid and every-subject-resolvable, against the real
+    /// mind: every stored document views back unchanged; the six findings the
+    /// 64-byte bound left unresolvable take a resolution; and a stored
+    /// resolution whose withdrawal key is over 64 bytes is withdrawn. Run it
+    /// on a scratch copy of the state root only: it writes.
+    #[test]
+    #[ignore = "reads and writes the scratch copy of a live state root named by HUGINN_MIND_SNAPSHOT"]
+    fn snapshot_admits_and_reads_back() {
+        use crate::fixtures::{admit, committed, r, resolution, withdrawn};
+        use epiphany_pipeline::{PipelineRef, ResolutionOutcome, Short, pipeline_key};
+
+        let root = std::env::var("HUGINN_MIND_SNAPSHOT").expect("HUGINN_MIND_SNAPSHOT names a scratch copy of the state root");
+        let mut mind = Mind::open(Path::new(&root), &slug("eureka")).unwrap();
+
+        let mut viewed = 0;
+        let mut resolutions = Vec::new();
+        for envelope in mind.envelopes() {
+            let Some(kind) = PipelineKind::ALL.iter().copied().find(|kind| kind.type_id() == envelope.r#type) else { continue };
+            let stored = PipelineDocument::decode(envelope).unwrap();
+            let view = mind
+                .view(&PipelineRef { kind, id: Short(envelope.key.clone()) })
+                .unwrap()
+                .unwrap_or_else(|| panic!("{} does not view", envelope.key));
+            assert_eq!(view.document, stored, "{}", envelope.key);
+            if let PipelineDocument::Resolution(held) = &stored
+                && !matches!(held.outcome, epiphany_pipeline::ResolutionOutcome::Withdrawn { .. })
+                && !envelope.key.contains(":resolution:resolution.")
+            {
+                resolutions.push(envelope.key.clone());
+            }
+            viewed += 1;
+        }
+        assert!(viewed > 0, "the snapshot held no pipeline document");
+        println!("viewed {viewed} stored documents unchanged");
+
+        for local in [
+            "cut-bifrost-retire-alarm.s2.verb-default-accepts-malformed",
+            "cut-ops-notice-deploy.s2.pinned-bifrost-predates-retry",
+            "cut-bifrost-notice-retry.s1.unknown-test-ignores-backoff",
+            "cut-bifrost-notice-retry.s2.closed-unknown-retry-unpinned",
+            "cut-bifrost-notice-retry.s2.flapping-clock-retries-every-tick",
+            "cut-idunn-topology-lock.s2.boot-reconcile-skipped-on-contention",
+        ] {
+            let subject = r(PipelineKind::Finding, &format!("idunn-watchdog:finding:{local}"));
+            let outcome = admit(&mut mind, vec![resolution(subject, ResolutionOutcome::Recorded { reason: "snapshot read-back".into() })]);
+            let (_, writes) = committed(outcome);
+            assert_eq!(writes.len(), 1, "{local}");
+        }
+
+        let mut tried = 0;
+        let mut withdrew = None;
+        for key in resolutions {
+            let withdrawal = resolution(r(PipelineKind::Resolution, &key), withdrawn());
+            let withdrawal_key = pipeline_key(&withdrawal).unwrap();
+            if withdrawal_key.rsplit(':').next().unwrap().len() <= 64 {
+                continue;
+            }
+            tried += 1;
+            if let PipelineAdmissionOutcome::Committed { .. } = admit(&mut mind, vec![withdrawal]) {
+                withdrew = Some(withdrawal_key);
+                break;
+            }
+        }
+        let withdrew = withdrew.unwrap_or_else(|| panic!("no stored resolution with a withdrawal key over 64 bytes withdrew; tried {tried}"));
+        println!("withdrew a resolution under a {}-byte key", withdrew.len());
+    }
 }