read-floor: a hop-only cites starts from the target's citers; reads racing admissions on the wire answer their snapshot; the persona receipt is pinned held and across a restart; FLOOR_SAMPLES; header comments restated

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

crates/huginn-daemon/src/serve.rs

@@ -792,6 +792,164 @@ mod tests {
         assert_eq!(documents(&mut daemon), 1, "the admission that crossed the wire landed");
     }
 
+    fn question_doc(n: usize) -> huginn_mind::eureka_pipeline::PipelineDocument {
+        use huginn_mind::eureka_pipeline::{Date, PipelineDocument, PipelineQuestion, QuestionOption, Title};
+        PipelineDocument::Question(PipelineQuestion {
+            campaign: slug(crate::daemon::tests::CAMPAIGN),
+            label: format!("Q{n}").as_str().into(),
+            title: Title("Who owns the state?".into()),
+            question: "Who owns the state?".into(),
+            options: vec![
+                QuestionOption { label: "A".into(), text: "the organ".into() },
+                QuestionOption { label: "B".into(), text: "the client".into() },
+            ],
+            recommended: "A".into(),
+            depends: vec![],
+            raised_in: None,
+            asked_on: Date("2026-09-16".into()),
+        })
+    }
+
+    fn resolution_of(subject: PipelineRef, outcome: huginn_mind::eureka_pipeline::ResolutionOutcome) -> huginn_mind::eureka_pipeline::PipelineDocument {
+        huginn_mind::eureka_pipeline::PipelineDocument::Resolution(huginn_mind::eureka_pipeline::PipelineResolution {
+            subject,
+            sequence: 1,
+            outcome,
+            rationale: "Resolved.".into(),
+            resolved_on: huginn_mind::eureka_pipeline::Date("2026-09-16".into()),
+        })
+    }
+
+    /// One client session asks `request` and decodes the answer.
+    fn ask(client: &mut Client, message_id: &str, request: &HuginnMindRequest) -> HuginnMindResponse {
+        let reply = exchange(client, &encode_request(message_id, request, None).unwrap()).unwrap();
+        decode_response(&reply).unwrap().1.unwrap_or_else(|failure| panic!("{message_id}: {failure:?}"))
+    }
+
+    fn connect(endpoint: &str, name: &str) -> Client {
+        let mut client = CultMesh::create_rudp_client_for_endpoint(
+            format!("eureka-state-{name}"),
+            CULTNET_OPERATION_CONNECTION_ID,
+            endpoint,
+            CultMeshRudpSocketOptions::default(),
+        )
+        .unwrap();
+        client.connect(Vec::new()).unwrap();
+        for _ in 0..1000 {
+            let _ = client.receive_once();
+            if client.connected() {
+                return client;
+            }
+            std::thread::sleep(Duration::from_millis(2));
+        }
+        panic!("{name} never connected");
+    }
+
+    /// One client admits batches that close, withdraw and re-open records while three others read
+    /// the same mind: fresh queries, cursor walks across commits, a hop and whoami. The loop is
+    /// serial, so every answer is one snapshot's; at quiesce each answer must equal what a fresh
+    /// derivation from the store gives at the snapshot it names, and no client sees the head go
+    /// backwards. The test pins the behaviour, not the mechanism.
+    #[test]
+    fn reads_racing_admissions_on_the_wire_answer_their_snapshot() {
+        use huginn_mind::eureka_pipeline::{PipelineKind as K, ResolutionOutcome};
+        let root = tempfile::tempdir().unwrap();
+        let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
+        let seeded = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, crate::daemon::tests::campaign_seed())), now()).answered();
+        assert!(matches!(seeded, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{seeded:?}");
+        let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap();
+        let endpoint = format!("rudp://{}", hub.local_addr().unwrap());
+        let registry = schema_registry().unwrap();
+        let stopping = Arc::new(AtomicBool::new(false));
+        let loop_stopping = Arc::clone(&stopping);
+        let serving = std::thread::spawn(move || {
+            let result = run(&mut daemon, &mut hub, &registry, &loop_stopping, &ServeOptions::default());
+            (result, daemon)
+        });
+
+        let done = Arc::new(AtomicBool::new(false));
+        let writer = {
+            let (endpoint, done) = (endpoint.clone(), Arc::clone(&done));
+            std::thread::spawn(move || {
+                let mut client = connect(&endpoint, "writer");
+                let admit = |client: &mut Client, n: usize, documents| {
+                    let answer = ask(client, &format!("w-{n}"), &HuginnMindRequest::Admit(batch(INSTANCE, documents)));
+                    assert!(matches!(answer, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{answer:?}");
+                };
+                let withdrawn = || ResolutionOutcome::Withdrawn { reason: "race".into() };
+                let mut n = 0;
+                for cut in 0..14 {
+                    let spec = PipelineRef { kind: K::Question, id: format!("{}:question:Q{cut}", crate::daemon::tests::CAMPAIGN).as_str().into() };
+                    admit(&mut client, n, vec![question_doc(cut)]);
+                    admit(&mut client, n + 1, vec![resolution_of(spec, withdrawn())]);
+                    if cut % 2 == 0 {
+                        let closing = PipelineRef { kind: K::Resolution, id: format!("{}:resolution:question.Q{cut}.n1", crate::daemon::tests::CAMPAIGN).as_str().into() };
+                        admit(&mut client, n + 2, vec![resolution_of(closing, withdrawn())]);
+                    }
+                    n += 3;
+                }
+                done.store(true, Ordering::Relaxed);
+            })
+        };
+        let readers: Vec<_> = (0..3)
+            .map(|index| {
+                let (endpoint, done) = (endpoint.clone(), Arc::clone(&done));
+                std::thread::spawn(move || {
+                    let mut client = connect(&endpoint, &format!("reader-{index}"));
+                    let of_specs = |extra: Selection| Selection { schemas: Some(vec![K::Question.type_id().into()]), ..extra };
+                    let in_force = FieldPredicate { index: "in_force".into(), op: "any_of".into(), values: Some(vec!["true".into()]), number: None };
+                    let mut recorded: Vec<(Selection, huginn_mind::PipelineSelectionPage)> = Vec::new();
+                    let mut head = 0;
+                    let mut round = 0;
+                    while !done.load(Ordering::Relaxed) || round < 3 {
+                        round += 1;
+                        let query = |client: &mut Client, selection: &Selection| {
+                            match ask(client, &format!("r-{index}-{round}"), &HuginnMindRequest::Query { instance: slug(INSTANCE), selection: selection.clone(), semantic: None }) {
+                                HuginnMindResponse::Query(page) => page,
+                                other => panic!("expected a page, got {other:?}"),
+                            }
+                        };
+                        let fresh = of_specs(Selection { fields: Some(vec![in_force.clone()]), limit: Some(5), ..Selection::default() });
+                        let page = query(&mut client, &fresh);
+                        assert!(page.as_of >= head, "a client's head never goes backwards");
+                        head = page.as_of;
+                        recorded.push((fresh, page));
+                        let mut walking = of_specs(Selection { limit: Some(1), ..Selection::default() });
+                        for _ in 0..3 {
+                            let page = query(&mut client, &walking);
+                            let next = page.next.clone();
+                            recorded.push((walking.clone(), page));
+                            match next {
+                                Some(cursor) => walking.cursor = Some(cursor),
+                                None => break,
+                            }
+                        }
+                        let hop = Selection { cited: Some(cultnet_rs::Incoming { role: "subject".into(), exists: true }), ..Selection::default() };
+                        let page = query(&mut client, &hop);
+                        recorded.push((hop, page));
+                        assert!(matches!(ask(&mut client, "whoami", &HuginnMindRequest::Whoami), HuginnMindResponse::Whoami(_)));
+                    }
+                    recorded
+                })
+            })
+            .collect();
+
+        writer.join().unwrap();
+        let recorded: Vec<_> = readers.into_iter().flat_map(|reader| reader.join().unwrap()).collect();
+        stopping.store(true, Ordering::Relaxed);
+        let (result, daemon) = serving.join().unwrap();
+        result.unwrap();
+        let mind = daemon.mind();
+        let final_head = mind.query(&Selection::default()).unwrap().as_of;
+        let mut snapshots = std::collections::BTreeSet::new();
+        for (selection, page) in &recorded {
+            snapshots.insert(page.as_of);
+            assert_eq!(huginn_mind::oracle::oracle_page(mind, selection, page.as_of).as_ref(), Ok(page), "{selection:?} at {}", page.as_of);
+        }
+        assert!(snapshots.len() > 1, "the readers saw more than one snapshot: {snapshots:?}");
+        assert!(snapshots.iter().all(|as_of| *as_of <= final_head));
+    }
+
     fn encoded_len(message: &CultNetMessage) -> u64 {
         encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0).unwrap().len() as u64
     }

crates/huginn-mind/src/image.rs

@@ -276,7 +276,8 @@ impl Image {
         if hop && !self.edges_in_target {
             return (0..self.rows.len() as u32).collect();
         }
-        let base = self.base_candidates(selection, as_of);
+        let hop_only_cites = selection.cites.is_some() && selection.cited.is_none();
+        let base = if hop_only_cites { BTreeSet::new() } else { self.base_candidates(selection, as_of) };
         let mut ids: BTreeSet<u32> = BTreeSet::new();
         let citers_of = |key: &str, role: Option<&str>| -> Vec<u32> {
             self.citers
@@ -291,8 +292,9 @@ impl Image {
             (None, None) => ids = base,
             (Some(cites), None) => {
                 let key = cites.target.record_key.as_str();
-                let citing: BTreeSet<u32> = citers_of(key, cites.role.as_deref()).into_iter().collect();
-                ids.extend(base.intersection(&citing));
+                // A row that passes the hop cites the target, so the citers are the superset; the kinds,
+                // keys and fields only the substrate decides.
+                ids.extend(citers_of(key, cites.role.as_deref()));
                 ids.extend(self.any_kind(key));
             }
             (cites, Some(cited)) => {

crates/huginn-mind/src/lib.rs

@@ -10,9 +10,10 @@
 //! cross-document rule, every derived write, and the replay check, in one
 //! ordered path that `admit`, the daemon's sink and the hand-off all share.
 //!
-//! The read side derives status, joins the receipts and answers one typed
-//! selection (CultNet's, over rows this crate supplies) through the same
-//! `docs` the rules use; nothing is stored for it.
+//! The read side answers one typed selection (CultNet's, over rows this crate
+//! supplies) from the mind's held image: a fold of the receipts, held by `Mind`,
+//! advanced per commit, rebuilt on restart and on a lost swap. Status is derived
+//! through the same `docs` the rules use; nothing is stored for it.
 //!
 //! The wire vocabulary and its operation envelope (`envelope`) are here too,
 //! so the daemon and the client share one set

crates/huginn-mind/src/persona.rs

@@ -611,6 +611,31 @@ mod tests {
     /// Status answers in the door's own terms: a stored persona the door
     /// cannot read is `Unreadable`, not the `Absent` of a mind with none, while
     /// `PersonaGet` answers `Unavailable` for the same bytes.
+    /// The persona's receipt is the newest put that wrote it, advanced with each put and the same
+    /// after a restart.
+    #[test]
+    fn the_persona_receipt_is_the_newest_put_held_and_after_a_restart() {
+        let store = MemoryStore::new();
+        let mut mind = seeded_over(&store);
+        committed(put(&mut mind, persona(INSTANCE, T1), None));
+        let first = mind.persona().unwrap().unwrap().receipt_id;
+        committed(put(&mut mind, persona(INSTANCE, T2), Some(T1)));
+        let second = mind.persona().unwrap().unwrap().receipt_id;
+        assert_ne!(first, second);
+        let newest = mind
+            .receipts()
+            .unwrap()
+            .into_iter()
+            .filter(|receipt| receipt.writes.iter().any(|write| write.document_type == HuginnPersonaEntry::TYPE))
+            .max_by_key(|receipt| receipt.ordinal)
+            .unwrap()
+            .receipt_id;
+        assert_eq!(second, newest);
+        let reopened = Mind::open_with(store, &slug(INSTANCE)).unwrap();
+        assert_eq!(reopened.persona().unwrap().unwrap().receipt_id, second);
+        assert_eq!(reopened.persona_receipt, mind.persona_receipt);
+    }
+
     #[test]
     fn status_says_unreadable_where_the_door_says_unavailable() {
         let yggdrasil = slug(INSTANCE);

crates/huginn-mind/src/query.rs

@@ -1,9 +1,12 @@
 //! The read side of a mind: a document with the facts of its admission and
 //! its status, and one typed selection over them.
 //!
-//! Nothing here is stored. Status is derived per call through the same
-//! `docs::Docs` the admission rules use, so a withdrawn resolution reopens its
-//! subject for a reader exactly when it reopens it for a rule. The facts come
+//! Nothing derived here is stored in a document. The rows come from the mind's
+//! held `image::Image`, a fold of the receipts; status is derived through the
+//! same `docs::Docs` the admission rules use, at the head when the image was
+//! advanced and at an old snapshot when a reader asks, so a withdrawn
+//! resolution reopens its subject for a reader exactly when it reopens it for
+//! a rule. The facts come
 //! from the commit receipts, from their `writes` and from nothing else: a
 //! document's `admitted_at` is the `now` its batch was admitted at
 //! (`committed_at`), never the store's `stored_at` stamp. A pipeline document
@@ -13,7 +16,8 @@
 //! `query` is CultNet's selection over rows this crate supplies (`rows.rs`).
 //! The substrate owns the filter, the order, the hop, the cursor and the
 //! paging; this module owns the snapshot (`Reader::at`: every derivation runs
-//! over the documents admitted at or before the page's `asOf`), the
+//! over the documents admitted at or before the page's `asOf`, and the image
+//! only narrows which of its rows the substrate is handed), the
 //! projection to a header or a whole view, and the typing of the edges.
 //!
 //! `rank` is the semantic read: the daemon's index supplies candidate ids and
@@ -503,8 +507,9 @@ impl<S: MindStore> Mind<S> {
     /// stored key, type and payload bytes, and views back unchanged through the
     /// read path `view` uses; every pipeline write a receipt names is stored,
     /// and every stored document has its receipt. Returns how many documents
-    /// it read back; an empty store is a fault. One `Reader` serves every view,
-    /// because `Reader::at` is the expensive build. Admits nothing. A fault
+    /// it read back; an empty store is a fault. The held image is first set beside a
+    /// fresh fold of the store and must equal it, so every upgrade proves the held
+    /// state against the store. One pass over the mind, O(N). Admits nothing. A fault
     /// names the document's key and what failed, never its content.
     pub fn read_back(&self) -> Result<usize, MindRefusal> {
         self.read_back_through(|reader, id| reader.view(id))

crates/huginn-mind/tests/read_floor.rs

@@ -6,7 +6,7 @@
 //! copied to a temp dir and never opened in place. `FLOOR_SCALE` (default 1)
 //! grows the copy with synthetic history to that multiple of its documents;
 //! `FLOOR_ROOTS` is `spread` (each hundred units under its own root) or
-//! `same` (one root grows). `FLOOR_INSTANCE` defaults to `eureka`.
+//! `same` (one root grows). `FLOOR_INSTANCE` defaults to `eureka`; `FLOOR_SAMPLES` (default 7) is how many times each read is timed, for a base too slow to sample 7 times.
 //!
 //! cargo test --release -p huginn-mind --features test-support --test read_floor -- --ignored --nocapture
 #![cfg(feature = "test-support")]
@@ -47,8 +47,9 @@ fn median(mut samples: Vec<Duration>) -> Duration {
 }
 
 fn time<T>(mut read: impl FnMut() -> T) -> Duration {
+    let samples: usize = env("FLOOR_SAMPLES", "7").parse().unwrap();
     median(
-        (0..7)
+        (0..samples)
             .map(|_| {
                 let started = Instant::now();
                 std::hint::black_box(read());