From 511afa2f440924c83d54022f7f70a10c1092b426 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 05:16:05 +0200 Subject: [PATCH 001/523] RS-3 1: the edge list moves to its owner (docs.rs citations) docs.rs::citations and CitationRole are the one edge list: the fifteen roles, one per referring leaf field. Admission's A7 references() is a map over them, carrying the Missing classification by role and the hand-off source-side filter, with the same refusals in the same order. kind_of_id and outcome_references leave admission; the CitesResolvedDocument check reads outcome_citations. Co-Authored-By: Claude Sonnet 5 --- crates/huginn-mind/src/admission.rs | 102 +++++----------- crates/huginn-mind/src/docs.rs | 181 +++++++++++++++++++++++++++- crates/huginn-mind/src/lib.rs | 1 + 3 files changed, 209 insertions(+), 75 deletions(-) diff --git a/crates/huginn-mind/src/admission.rs b/crates/huginn-mind/src/admission.rs index 4734da2..8ea8f8a 100644 --- a/crates/huginn-mind/src/admission.rs +++ b/crates/huginn-mind/src/admission.rs @@ -37,7 +37,7 @@ use epiphany_pipeline::{ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; -use crate::docs::{Docs, Staged}; +use crate::docs::{CitationRole, Docs, Staged, citations, kind_of_id, outcome_citations}; use crate::mind::{HuginnMindEpoch, Mind, unavailable}; use crate::receipt::{self, CommitOutcome, PipelineProvenance}; use crate::refusal::MindRefusal; @@ -170,13 +170,6 @@ impl Mind { } } -/// The kind segment of a full id, for the one field typed `Short` whose -/// referent may be of any kind. -fn kind_of_id(id: &str) -> Option { - let name = id.split(':').nth(1)?; - PipelineKind::ALL.iter().copied().find(|kind| kind.name() == name) -} - /// A3 through the leaf, then A5. fn stage(envelope: CultCacheEnvelope, mind: &Slug) -> Result { validate_pipeline_write_envelope(&envelope)?; @@ -212,74 +205,35 @@ struct Reference { missing: Missing, } -fn reference(kind: PipelineKind, id: &str) -> Reference { - Reference { kind, id: id.to_string(), missing: Missing::Reference } -} - -fn referenced(pipeline_ref: &PipelineRef) -> Reference { - reference(pipeline_ref.kind, &pipeline_ref.id.0) -} - -fn outcome_references(outcome: &ResolutionOutcome) -> Vec { - match outcome { - ResolutionOutcome::Superseded { by } => by - .iter() - .map(|supersessor| Reference { kind: supersessor.kind, id: supersessor.id.0.clone(), missing: Missing::Supersessor }) - .collect(), - ResolutionOutcome::Answered { by } => vec![referenced(by)], - ResolutionOutcome::Fixed { by: Some(by), .. } => vec![referenced(by)], - ResolutionOutcome::Deferred { to } => vec![referenced(to)], - ResolutionOutcome::Fixed { by: None, .. } | ResolutionOutcome::Recorded { .. } | ResolutionOutcome::Withdrawn { .. } => { - vec![] - } - } -} - -/// A7's field list: every `PipelineRef` and every full-id `Short` field. -/// `Fixed.commit` and `ForeignRef` are syntax only. A hand-off's documents -/// are cited on the source side only; the receiving side imports them -/// afterwards (Cut 12). +/// A7's field list is `docs::citations`: every `PipelineRef` and every +/// full-id `Short` field, each carrying the refusal its absence earns. A +/// hand-off's documents are cited on the source side only, and each must read +/// as an id of a kind; the receiving side imports them afterwards (Cut 12). fn references(staged: &Staged, mind: &Slug) -> Result, MindRefusal> { - use PipelineDocument as D; - use PipelineKind as K; - let mut refs = Vec::new(); - match &staged.document { - D::Question(question) => refs.extend(question.raised_in.as_ref().map(referenced)), - D::Ruling(ruling) => refs.extend(ruling.answers.as_ref().map(|question| reference(K::Question, &question.0))), - D::CutSpec(spec) => { - refs.extend(spec.rulings.iter().map(|ruling| reference(K::Ruling, &ruling.0))); - refs.extend(spec.questions.iter().map(|question| reference(K::Question, &question.0))); + if let PipelineDocument::HandOff(hand_off) = &staged.document { + if hand_off.from_instance != *mind { + return Ok(Vec::new()); } - D::CutReport(report) => { - refs.push(Reference { - kind: K::CutSpec, - id: report.cut_spec.0.clone(), - missing: Missing::SpecOfReport(staged.key.clone()), - }); - refs.extend(report.forks.iter().map(|question| reference(K::Question, &question.0))); - } - D::Verdict(verdict) => { - refs.push(reference(K::CutReport, &verdict.cut_report.0)); - refs.extend(verdict.claims.iter().flat_map(|claim| claim.findings.iter()).map(|finding| reference(K::Finding, &finding.0))); - } - D::Finding(finding) => refs.push(reference(K::Verdict, &finding.verdict.0)), - D::FollowUp(follow_up) => refs.push(referenced(&follow_up.source)), - D::Resolution(resolution) => { - refs.push(referenced(&resolution.subject)); - refs.extend(outcome_references(&resolution.outcome)); - } - D::HandOff(hand_off) if hand_off.from_instance == *mind => { - for document in &hand_off.documents { - let kind = kind_of_id(&document.0).ok_or_else(|| PipelineRefusal::InvalidFormat { - field: "hand_off.documents".into(), - value: document.0.clone(), - })?; - refs.push(reference(kind, &document.0)); + if let Some(document) = hand_off.documents.iter().find(|document| kind_of_id(&document.0).is_none()) { + return Err(PipelineRefusal::InvalidFormat { + field: "hand_off.documents".into(), + value: document.0.clone(), } + .into()); } - D::Campaign(_) | D::Target(_) | D::Instance(_) | D::Stewardship(_) | D::HandOff(_) => {} } - Ok(refs) + Ok(citations(&staged.document) + .into_iter() + .map(|(role, target)| Reference { + kind: target.kind, + id: target.id.0, + missing: match role { + CitationRole::CutSpec => Missing::SpecOfReport(staged.key.clone()), + CitationRole::SupersededBy => Missing::Supersessor, + _ => Missing::Reference, + }, + }) + .collect()) } /// A7: every referent is in the batch or the image; the image ones are the @@ -652,9 +606,9 @@ fn resolution_rule(docs: &Docs, resolution: &PipelineResolution) -> Result<(), M later: later.into(), }); } - for referent in outcome_references(&resolution.outcome) { - if !docs.in_force(referent.kind, &referent.id) { - return Err(MindRefusal::CitesResolvedDocument { kind: referent.kind, id: referent.id }); + for (_, referent) in outcome_citations(&resolution.outcome) { + if !docs.in_force(referent.kind, &referent.id.0) { + return Err(MindRefusal::CitesResolvedDocument { kind: referent.kind, id: referent.id.0 }); } } match &resolution.outcome { diff --git a/crates/huginn-mind/src/docs.rs b/crates/huginn-mind/src/docs.rs index 8a21b2c..1f1b033 100644 --- a/crates/huginn-mind/src/docs.rs +++ b/crates/huginn-mind/src/docs.rs @@ -8,8 +8,10 @@ use cultcache_rs::CultCacheEnvelope; use epiphany_pipeline::{ OrgRepo, PipelineDocument, PipelineKind, PipelineRef, PipelineResolution, PipelineStewardship, ResolutionOutcome, - Slug, + Short, Slug, }; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; use crate::refusal::MindRefusal; @@ -262,3 +264,180 @@ fn later_than(base: &PipelineDocument, other: &PipelineDocument) -> bool { _ => false, } } + +/// The referring field of a citation edge, one per leaf field that names +/// another document. The wire spelling is the field's own name, so a client +/// reads an edge's role as the field it came from. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum CitationRole { + RaisedIn, + Answers, + Rulings, + Questions, + CutSpec, + Forks, + CutReport, + Findings, + Verdict, + Source, + Subject, + SupersededBy, + ResolvedBy, + DeferredTo, + Documents, +} + +impl CitationRole { + pub const ALL: [CitationRole; 15] = [ + Self::RaisedIn, + Self::Answers, + Self::Rulings, + Self::Questions, + Self::CutSpec, + Self::Forks, + Self::CutReport, + Self::Findings, + Self::Verdict, + Self::Source, + Self::Subject, + Self::SupersededBy, + Self::ResolvedBy, + Self::DeferredTo, + Self::Documents, + ]; + + /// The wire spelling, equal to the referring field's name. + pub fn name(self) -> &'static str { + match self { + Self::RaisedIn => "raised_in", + Self::Answers => "answers", + Self::Rulings => "rulings", + Self::Questions => "questions", + Self::CutSpec => "cut_spec", + Self::Forks => "forks", + Self::CutReport => "cut_report", + Self::Findings => "findings", + Self::Verdict => "verdict", + Self::Source => "source", + Self::Subject => "subject", + Self::SupersededBy => "superseded_by", + Self::ResolvedBy => "resolved_by", + Self::DeferredTo => "deferred_to", + Self::Documents => "documents", + } + } + + pub fn from_name(name: &str) -> Option { + Self::ALL.into_iter().find(|role| role.name() == name) + } + + /// The kind whose documents carry this field. + pub fn carrier(self) -> PipelineKind { + use PipelineKind as K; + match self { + Self::RaisedIn => K::Question, + Self::Answers => K::Ruling, + Self::Rulings | Self::Questions => K::CutSpec, + Self::CutSpec | Self::Forks => K::CutReport, + Self::CutReport | Self::Findings => K::Verdict, + Self::Verdict => K::Finding, + Self::Source => K::FollowUp, + Self::Subject | Self::SupersededBy | Self::ResolvedBy | Self::DeferredTo => K::Resolution, + Self::Documents => K::HandOff, + } + } + + /// The kinds a referent of this field may be: the one kind the field is + /// typed to, or every kind for the fields typed `PipelineRef` or a full + /// id of any kind. "Any" is spelled as all thirteen, never as none. + pub fn target_kinds(self) -> Vec { + use PipelineKind as K; + match self { + Self::Answers | Self::Questions | Self::Forks => vec![K::Question], + Self::Rulings => vec![K::Ruling], + Self::CutSpec => vec![K::CutSpec], + Self::CutReport => vec![K::CutReport], + Self::Findings => vec![K::Finding], + Self::Verdict => vec![K::Verdict], + Self::RaisedIn + | Self::Source + | Self::Subject + | Self::SupersededBy + | Self::ResolvedBy + | Self::DeferredTo + | Self::Documents => PipelineKind::ALL.to_vec(), + } + } +} + +/// The kind segment of a full id, for the fields typed `Short` whose referent +/// may be of any kind. +pub(crate) fn kind_of_id(id: &str) -> Option { + let name = id.split(':').nth(1)?; + PipelineKind::ALL.iter().copied().find(|kind| kind.name() == name) +} + +fn cited(kind: PipelineKind, id: &Short) -> PipelineRef { + PipelineRef { kind, id: id.clone() } +} + +/// What a resolution's outcome names, by the field that names it. +pub(crate) fn outcome_citations(outcome: &ResolutionOutcome) -> Vec<(CitationRole, PipelineRef)> { + match outcome { + ResolutionOutcome::Superseded { by } => by.iter().map(|by| (CitationRole::SupersededBy, by.clone())).collect(), + ResolutionOutcome::Answered { by } => vec![(CitationRole::ResolvedBy, by.clone())], + ResolutionOutcome::Fixed { by: Some(by), .. } => vec![(CitationRole::ResolvedBy, by.clone())], + ResolutionOutcome::Deferred { to } => vec![(CitationRole::DeferredTo, to.clone())], + ResolutionOutcome::Fixed { by: None, .. } | ResolutionOutcome::Recorded { .. } | ResolutionOutcome::Withdrawn { .. } => { + vec![] + } + } +} + +/// Every citation edge a document carries, derived from its leaf fields each +/// time it is asked and never stored: the one edge list. Admission's A7 reads +/// it to require the referents, and the read side's rows read it as their +/// references. `Fixed.commit`, `ForeignRef` precedents and the labels +/// (`depends_on`, `invariants`, `promise`) are not edges. A hand-off's +/// `documents` entry whose kind segment does not read cites nothing here; +/// admission refuses such an entry on the source side before it asks. +pub(crate) fn citations(document: &PipelineDocument) -> Vec<(CitationRole, PipelineRef)> { + use CitationRole as R; + use PipelineDocument as D; + use PipelineKind as K; + let mut edges = Vec::new(); + match document { + D::Question(question) => edges.extend(question.raised_in.iter().map(|to| (R::RaisedIn, to.clone()))), + D::Ruling(ruling) => edges.extend(ruling.answers.iter().map(|to| (R::Answers, cited(K::Question, to)))), + D::CutSpec(spec) => { + edges.extend(spec.rulings.iter().map(|to| (R::Rulings, cited(K::Ruling, to)))); + edges.extend(spec.questions.iter().map(|to| (R::Questions, cited(K::Question, to)))); + } + D::CutReport(report) => { + edges.push((R::CutSpec, cited(K::CutSpec, &report.cut_spec))); + edges.extend(report.forks.iter().map(|to| (R::Forks, cited(K::Question, to)))); + } + D::Verdict(verdict) => { + edges.push((R::CutReport, cited(K::CutReport, &verdict.cut_report))); + edges.extend( + verdict + .claims + .iter() + .flat_map(|claim| claim.findings.iter()) + .map(|to| (R::Findings, cited(K::Finding, to))), + ); + } + D::Finding(finding) => edges.push((R::Verdict, cited(K::Verdict, &finding.verdict))), + D::FollowUp(follow_up) => edges.push((R::Source, follow_up.source.clone())), + D::Resolution(resolution) => { + edges.push((R::Subject, resolution.subject.clone())); + edges.extend(outcome_citations(&resolution.outcome)); + } + D::HandOff(hand_off) => edges.extend( + hand_off.documents.iter().filter_map(|to| kind_of_id(&to.0).map(|kind| (R::Documents, cited(kind, to)))), + ), + D::Campaign(_) | D::Target(_) | D::Instance(_) | D::Stewardship(_) => {} + } + edges +} diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index 6c73109..778d652 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -35,6 +35,7 @@ pub mod wire; #[cfg(test)] pub(crate) mod fixtures; +pub use docs::CitationRole; pub use admission::{BATCH_MAX, PipelineAdmissionBatch, PipelineAdmissionOutcome}; pub use mind::{HuginnMindEpoch, Mind}; pub use query::{ From eb57bdba4041b6c30227d433c54f4f6a8f31619a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 05:56:37 +0200 Subject: [PATCH 002/523] RS-3 2: the read side is CultNet's selection over the organ's rows The four hand-written filters, the order and the cap are deleted: PipelineQuery, PipelineQueryPage, QUERY_LIMIT_MAX, Reader::matches, ordered, repo_matches. Mind::query(selection, semantic) validates through cultnet_rs::validate, closes the values door (rows.rs refuse_values, S6), takes the snapshot from the cursor's own asOf (an asOf beyond the head is CursorInvalid), builds Reader::at(as_of) so every derivation runs over the documents admitted at or before it, and evaluates through cultnet_rs::select. The page carries matched, asOf, the next cursor, headers or whole views by projection, and typed edges when the selection hops. rows.rs is the organ's vocabulary: SelectionRow (a Row), Vocabulary (a RowSet: thirteen schemas, eleven aliases, fifteen roles) and the value door. huginn-mind links cultnet-rs at the rev the daemon pins. Selection refusals map totally onto SelectionInvalid, CursorInvalid or Unavailable. The wire's Query carries CultLib's selection by $ref, and both published schemas are regenerated. The relations RS-2 unpinned are re-pinned as selections. Co-Authored-By: Claude Sonnet 5 --- Cargo.lock | 1 + crates/huginn-daemon/src/daemon.rs | 82 +- crates/huginn-daemon/src/serve.rs | 37 +- crates/huginn-mind/Cargo.toml | 8 +- crates/huginn-mind/src/docs.rs | 2 +- crates/huginn-mind/src/lib.rs | 10 +- crates/huginn-mind/src/mind.rs | 10 +- crates/huginn-mind/src/query.rs | 1109 +++++++++++++---- crates/huginn-mind/src/refusal.rs | 65 + crates/huginn-mind/src/rows.rs | 336 +++++ crates/huginn-mind/src/wire.rs | 37 +- .../huginn.mind_request.v1.schema.json | 99 +- .../huginn.mind_response.v1.schema.json | 812 +++++++++++- 13 files changed, 2223 insertions(+), 385 deletions(-) create mode 100644 crates/huginn-mind/src/rows.rs diff --git a/Cargo.lock b/Cargo.lock index b68e112..341aa15 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -468,6 +468,7 @@ dependencies = [ "anyhow", "chrono", "cultcache-rs", + "cultnet-rs", "epiphany-pipeline", "rmp-serde", "schemars", diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index 9e2e04c..a1abc28 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -85,7 +85,7 @@ impl> Daemon { Ok(view) => HuginnMindResponse::View(view), Err(refusal) => HuginnMindResponse::Refused(refusal), }, - HuginnMindRequest::Query { query, .. } => match self.mind.query(&query) { + HuginnMindRequest::Query { selection, semantic, .. } => match self.mind.query(&selection, semantic.as_ref()) { Ok(page) => HuginnMindResponse::Query(page), Err(refusal) => HuginnMindResponse::Refused(refusal), }, @@ -104,7 +104,8 @@ pub(crate) mod tests { StructuralDelta, Title, VerificationTest, }; use huginn_mind::wire::MindStatus; - use huginn_mind::{Faculty, PipelineAdmissionBatch, PipelineProvenance, PipelineQuery, PipelineStatus, SemanticQuery}; + use cultnet_rs::Selection; + use huginn_mind::{Faculty, PipelineAdmissionBatch, PipelineProvenance, PipelinePageItems, PipelineStatus, SemanticQuery}; use std::sync::{Arc, Mutex}; use tempfile::TempDir; @@ -375,19 +376,29 @@ pub(crate) mod tests { ); assert_eq!(daemon.runtime_id(), "huginn-yggdrasil"); - let query = PipelineQuery { kinds: vec![PipelineKind::Instance], ..PipelineQuery::default() }; - let HuginnMindResponse::Query(page) = - daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), query }, now()) - else { + let instances = Selection { schemas: Some(vec![PipelineKind::Instance.type_id().into()]), ..Selection::default() }; + let HuginnMindResponse::Query(page) = daemon.handle( + HuginnMindRequest::Query { instance: slug(INSTANCE), selection: instances.clone(), semantic: None }, + now(), + ) else { panic!("expected a page"); }; assert_eq!(page.matched, 1); - assert_eq!(page.items[0].admission.provenance, provenance()); - assert_eq!(page.items[0].status, PipelineStatus::InForce); + let PipelinePageItems::Headers(headers) = &page.items else { panic!("the default projection is a header") }; + assert_eq!(headers[0].admission.provenance, provenance()); + assert_eq!(headers[0].status, huginn_mind::PipelineStatusSummary::InForce); let id = writes[0].clone(); + let whole = Selection { projection: "document".into(), ..instances }; + let HuginnMindResponse::Query(page) = + daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection: whole, semantic: None }, now()) + else { + panic!("expected a page"); + }; + let PipelinePageItems::Documents(views) = &page.items else { panic!("the document projection is the view") }; + assert_eq!(views[0].status, PipelineStatus::InForce); let view = daemon.handle(HuginnMindRequest::View { instance: slug(INSTANCE), id: id.clone() }, now()); - assert_eq!(view, HuginnMindResponse::View(Some(page.items[0].clone()))); + assert_eq!(view, HuginnMindResponse::View(Some(views[0].clone()))); } /// Ruling 14 across the transport, on both sides: the mind refuses a read @@ -402,7 +413,7 @@ pub(crate) mod tests { }; let foreign = MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() }; - let query = HuginnMindRequest::Query { instance: slug(OTHER), query: PipelineQuery::default() }; + let query = HuginnMindRequest::Query { instance: slug(OTHER), selection: Selection::default(), semantic: None }; assert_eq!(daemon.handle(query, now()), HuginnMindResponse::Refused(foreign.clone())); let admit = HuginnMindRequest::Admit(batch(OTHER, vec![identity(OTHER)])); @@ -437,7 +448,7 @@ pub(crate) mod tests { assert_eq!(CASED.to_ascii_lowercase(), INSTANCE); for declared in [NEAR, PREFIXED, CASED] { let refusal = MindRefusal::ForeignInstance { declared: declared.into(), mind: INSTANCE.into() }; - let query = HuginnMindRequest::Query { instance: slug(declared), query: PipelineQuery::default() }; + let query = HuginnMindRequest::Query { instance: slug(declared), selection: Selection::default(), semantic: None }; assert_eq!(daemon.handle(query, now()), HuginnMindResponse::Refused(refusal.clone()), "{declared}"); let view = HuginnMindRequest::View { instance: slug(declared), id: writes[0].clone() }; assert_eq!(daemon.handle(view, now()), HuginnMindResponse::Refused(refusal.clone()), "{declared}"); @@ -472,7 +483,7 @@ pub(crate) mod tests { value: fullwidth.into(), }); - let query = HuginnMindRequest::Query { instance: slug(fullwidth), query: PipelineQuery::default() }; + let query = HuginnMindRequest::Query { instance: slug(fullwidth), selection: Selection::default(), semantic: None }; assert_eq!(daemon.handle(query, now()), HuginnMindResponse::Refused(expected.clone())); let view = HuginnMindRequest::View { @@ -508,7 +519,7 @@ pub(crate) mod tests { HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(expected.clone())) ); - let query = HuginnMindRequest::Query { instance: slug(fullwidth), query: PipelineQuery::default() }; + let query = HuginnMindRequest::Query { instance: slug(fullwidth), selection: Selection::default(), semantic: None }; assert_eq!(daemon.handle(query, now()), HuginnMindResponse::Refused(expected)); } @@ -535,14 +546,11 @@ pub(crate) mod tests { // A semantic query is unavailable until Cut 11 wires an index, and the // detail is the mind's own sentence. - let query = PipelineQuery { - semantic: Some(SemanticQuery { text: "who owns the state".into(), top_k: 4 }), - ..PipelineQuery::default() - }; + let semantic = Some(SemanticQuery { text: "who owns the state".into(), top_k: 4 }); let unwired = MindRefusal::Unavailable { detail: "semantic query: the index is not wired (Cut 11)".into() }; assert_eq!( - daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), query }, now()), + daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection: Selection::default(), semantic }, now()), HuginnMindResponse::Refused(unwired) ); @@ -553,11 +561,47 @@ pub(crate) mod tests { PipelineRef { kind: PipelineKind::Question, id: Short(format!("{CAMPAIGN}:question:Q1")) }; let present = HuginnMindRequest::View { instance: slug(INSTANCE), id: absent }; assert_eq!(daemon.handle(present, now()), HuginnMindResponse::View(None)); - let plain = HuginnMindRequest::Query { instance: slug(INSTANCE), query: PipelineQuery::default() }; + let plain = HuginnMindRequest::Query { instance: slug(INSTANCE), selection: Selection::default(), semantic: None }; let HuginnMindResponse::Query(page) = daemon.handle(plain, now()) else { panic!("expected a page") }; assert_eq!(page.matched, 1); } + /// A refusal of the selection crosses the dispatch as itself: typed, on + /// the field the client sent, and never swallowed into an empty page or + /// rewrapped as an unavailable store. A stale cursor is the organ's own + /// defect, so a valid walk never meets it. + #[test] + fn a_selection_refusal_crosses_the_dispatch_as_itself() { + let (_root, mut daemon) = seeded(); + let ask = |daemon: &mut Daemon, selection: Selection| { + daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection, semantic: None }, now()) + }; + let predicate = |index: &str, value: &str| Selection { + fields: Some(vec![cultnet_rs::FieldPredicate { + index: index.into(), + op: "any_of".into(), + values: Some(vec![value.into()]), + number: None, + }]), + ..Selection::default() + }; + let HuginnMindResponse::Refused(MindRefusal::SelectionInvalid { field, .. }) = + ask(&mut daemon, predicate("nonsense", "x")) + else { + panic!("an undeclared alias is the substrate's refusal"); + }; + assert_eq!(field, "fields[0].index"); + let HuginnMindResponse::Refused(MindRefusal::SelectionInvalid { field, value, .. }) = + ask(&mut daemon, predicate("repo", "a/b/c")) + else { + panic!("a value outside its domain is the organ's refusal"); + }; + assert_eq!((field.as_str(), value.as_deref()), ("fields[0].values", Some("a/b/c"))); + let garbled = Selection { cursor: Some("not a cursor".into()), ..Selection::default() }; + assert!(matches!(ask(&mut daemon, garbled), HuginnMindResponse::Refused(MindRefusal::CursorInvalid { .. }))); + assert_eq!(ask(&mut daemon, predicate("repo", "a/b/c")).status(), "rejected"); + } + /// Cut 11's seam: the index is handed every landed write, derived ones /// included, and can read each; an index that fails changes nothing. #[test] diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs index b097f02..888d5a2 100644 --- a/crates/huginn-daemon/src/serve.rs +++ b/crates/huginn-daemon/src/serve.rs @@ -318,7 +318,8 @@ mod tests { use base64::engine::general_purpose::STANDARD; use cultnet_rs::{CultMesh, CultMeshRudpSocketOptions}; use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, MindStatus}; - use huginn_mind::{MindRefusal, PipelineAdmissionOutcome, PipelineQuery}; + use cultnet_rs::{FieldPredicate, Selection}; + use huginn_mind::{MindRefusal, PipelineAdmissionOutcome, PipelinePageItems}; use std::sync::Arc; use std::sync::atomic::AtomicBool; @@ -408,7 +409,7 @@ mod tests { // A mind's refusal is an answer on the response schema, not a failure // of the envelope: the client decodes it as the typed refusal it is. - let read = HuginnMindRequest::Query { instance: slug(OTHER), query: PipelineQuery::default() }; + let read = HuginnMindRequest::Query { instance: slug(OTHER), selection: Selection::default(), semantic: None }; let reply = answer(&mut daemon, ®istry, encode_request("m-r", &read, None).unwrap(), now()); let CultNetMessage::OperationResponse { status, payload_schema, .. } = &reply else { panic!("expected an operation response, got {reply:?}"); @@ -483,7 +484,7 @@ mod tests { let requests = [ ("m-a", HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)]))), - ("m-b", HuginnMindRequest::Query { instance: slug(INSTANCE), query: PipelineQuery::default() }), + ("m-b", HuginnMindRequest::Query { instance: slug(INSTANCE), selection: Selection::default(), semantic: None }), ]; for (client, (message_id, request)) in clients.iter_mut().zip(requests.iter()) { client.send_schema_message(&encode_request(message_id, request, None).unwrap()).unwrap(); @@ -587,7 +588,17 @@ mod tests { assert!(client.connected()); let reads = [ - ("m-q", HuginnMindRequest::Query { instance: slug(INSTANCE), query: PipelineQuery::default() }), + // The whole documents: a header page is small by construction, so + // the transport's bound is measured on the projection that can + // exceed it. + ( + "m-q", + HuginnMindRequest::Query { + instance: slug(INSTANCE), + selection: Selection { projection: "document".into(), ..Selection::default() }, + semantic: None, + }, + ), ("m-v", HuginnMindRequest::View { instance: slug(INSTANCE), id: wide }), ]; let stopping = Arc::new(AtomicBool::new(false)); @@ -906,18 +917,24 @@ mod tests { /// One cut spec's own reference, by its cut label. fn spec_ref(daemon: &mut Daemon, cut: &str) -> PipelineRef { - let query = PipelineQuery { - kinds: vec![PipelineKind::CutSpec], - cut: Some(cut.into()), - ..PipelineQuery::default() + let selection = Selection { + schemas: Some(vec![PipelineKind::CutSpec.type_id().into()]), + fields: Some(vec![FieldPredicate { + index: "cut".into(), + op: "any_of".into(), + values: Some(vec![cut.into()]), + number: None, + }]), + ..Selection::default() }; let HuginnMindResponse::Query(page) = - daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), query }, now()) + daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection, semantic: None }, now()) else { panic!("expected a page"); }; assert_eq!(page.matched, 1, "one cut spec is labelled {cut}"); - page.items[0].id.clone() + let PipelinePageItems::Headers(headers) = &page.items else { panic!("expected headers") }; + headers[0].id.clone() } /// Ruling 15 as an order in the one place both happen. Both gates are shut diff --git a/crates/huginn-mind/Cargo.toml b/crates/huginn-mind/Cargo.toml index 64fea12..a2e60d4 100644 --- a/crates/huginn-mind/Cargo.toml +++ b/crates/huginn-mind/Cargo.toml @@ -8,12 +8,16 @@ publish.workspace = true # The organ's rule engine: storage, identity and admission of one instance's # mind. Document shape, bounds, formats and keys come from `epiphany-pipeline`; # the store comes from CultLib's Rust CultCache at the same rev Epiphany pins, -# so the leaf's wrappers and this crate's cache are one `cultcache-rs`. No -# network, no socket, no index: those are the daemon's (Cut 10, Cut 11). +# so the leaf's wrappers and this crate's cache are one `cultcache-rs`. The +# read side selects through CultNet's typed selection (`cultnet-rs`, the same +# rev the daemon pins): the selection's vocabulary and evaluator are the +# substrate's, this crate supplies the rows. No socket, no index: those are the +# daemon's (Cut 10, Cut 11). [dependencies] anyhow = "1" chrono = "0.4.44" cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultcache-rs" } +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultnet-rs" } epiphany-pipeline = { git = "https://github.com/GameCult/Epiphany.git", rev = "4891f2e59008efeb262197564eddaa75bc3c75ea", package = "epiphany-pipeline" } rmp-serde = "1" schemars = "1" diff --git a/crates/huginn-mind/src/docs.rs b/crates/huginn-mind/src/docs.rs index 1f1b033..ef4d723 100644 --- a/crates/huginn-mind/src/docs.rs +++ b/crates/huginn-mind/src/docs.rs @@ -36,7 +36,7 @@ pub(crate) struct Docs { pub(crate) batch: Vec, } -fn kind_of_type(type_id: &str) -> Option { +pub(crate) fn kind_of_type(type_id: &str) -> Option { PipelineKind::ALL.iter().copied().find(|kind| kind.type_id() == type_id) } diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index 778d652..14c5d1f 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -10,8 +10,9 @@ //! 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 typed queries -//! through the same `docs` the rules use; nothing is stored for it. +//! 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 wire vocabulary is here too, so the daemon and the client share one set //! of types. @@ -26,6 +27,7 @@ pub mod admission; mod docs; pub mod mind; +mod rows; pub mod query; pub mod receipt; pub mod refusal; @@ -39,8 +41,8 @@ pub use docs::CitationRole; pub use admission::{BATCH_MAX, PipelineAdmissionBatch, PipelineAdmissionOutcome}; pub use mind::{HuginnMindEpoch, Mind}; pub use query::{ - AdmissionFacts, PipelineDocumentView, PipelineQuery, PipelineQueryPage, PipelineStatus, QUERY_LIMIT_MAX, - SemanticQuery, + AdmissionFacts, PipelineDocumentSummary, PipelineDocumentView, PipelineEdge, PipelineFacts, PipelinePageItems, + PipelineSelectionPage, PipelineStatus, PipelineStatusSummary, SUMMARY_MAX_BYTES, SemanticQuery, }; pub use receipt::{DocumentVersion, Faculty, HuginnCommitReceipt, PipelineProvenance, RECEIPT_SCHEMA_VERSION}; pub use refusal::MindRefusal; diff --git a/crates/huginn-mind/src/mind.rs b/crates/huginn-mind/src/mind.rs index 5f644ba..8deed10 100644 --- a/crates/huginn-mind/src/mind.rs +++ b/crates/huginn-mind/src/mind.rs @@ -12,6 +12,7 @@ use std::path::{Path, PathBuf}; use cultcache_rs::{CultCache, CultCacheEnvelope, DatabaseEntry, OwnedRedbMessagePackBackingStore}; +use cultnet_rs::CursorKey; use epiphany_pipeline::{PIPELINE_SCHEMA_EPOCH, PipelineDocument, PipelineKind, Slug, register_pipeline_document_types}; use crate::receipt::HuginnCommitReceipt; @@ -96,6 +97,9 @@ pub struct Mind { store: S, cache: CultCache, image: Vec, + /// Mints and verifies this process's selection cursors. Never stored: a + /// cursor does not survive a reopen, which the substrate accepts. + cursor_key: CursorKey, } impl Mind { @@ -180,7 +184,7 @@ impl Mind { refuse_foreign_types(&raw)?; refuse_foreign_identity(&raw, instance)?; let (cache, image) = attach(store.clone())?; - Ok(Self { instance: instance.clone(), store, cache, image }) + Ok(Self { instance: instance.clone(), store, cache, image, cursor_key: CursorKey::random() }) } pub fn instance(&self) -> &Slug { @@ -208,6 +212,10 @@ impl Mind { &self.image } + pub(crate) fn cursor_key(&self) -> &CursorKey { + &self.cursor_key + } + pub fn is_empty(&self) -> bool { self.image.is_empty() } diff --git a/crates/huginn-mind/src/query.rs b/crates/huginn-mind/src/query.rs index 445b7a3..e657509 100644 --- a/crates/huginn-mind/src/query.rs +++ b/crates/huginn-mind/src/query.rs @@ -1,5 +1,5 @@ //! The read side of a mind: a document with the facts of its admission and -//! its status, and the four queries over them. +//! 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 @@ -10,31 +10,37 @@ //! in the image that no receipt wrote is an integrity fault and refuses; the //! store has one writer and the receipt is its proof. //! +//! `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 +//! projection to a header or a whole view, and the typing of the edges. +//! //! `semantic` is accepted by the type and refused typed until Cut 11 wires the //! index. No clock, no store handle, no network. use std::collections::BTreeMap; +use cultnet_rs::{Cursor, PROJECTION_DOCUMENT, Selection, select, validate}; use epiphany_pipeline::{ - Label, Line, OrgRepo, PipelineDocument, PipelineKind, PipelineRef, PipelineResolution, Short, Slug, + ClaimOutcome, CommitRange, Date, FindingConfidence, FindingOrigin, FindingSeverity, Label, Line, OrgRepo, + PipelineDocument, PipelineKind, PipelineRef, PipelineResolution, ResolutionOutcome, RulingAuthority, Sha, Short, + Slug, Title, }; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; -use crate::docs::{Docs, Held}; +use crate::docs::{CitationRole, Docs, Held, citations}; use crate::mind::Mind; -use crate::receipt::{self, Faculty, PipelineProvenance}; +use crate::receipt::{self, PipelineProvenance}; use crate::refusal::MindRefusal; +use crate::rows::{SelectionRow, Vocabulary, refuse_values}; use crate::store::MindStore; -/// The most views one query answers with. A query that matched more says so in -/// `matched`; the cap never truncates silently. -pub const QUERY_LIMIT_MAX: usize = 200; - /// What admission recorded about a document when it landed: the receipt that /// wrote it, the `now` that receipt was taken at, who asked, and the -/// receipt's ordinal. `ordinal` is display-only here until RS-3 makes it the -/// read side's order. +/// receipt's ordinal, which is the read side's order and its snapshot +/// position. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] pub struct AdmissionFacts { pub receipt_id: String, @@ -75,33 +81,195 @@ pub struct SemanticQuery { pub top_k: u32, } -/// The filters over one mind. Each is one rule, and an unset filter matches -/// everything. The mind is the instance, so there is no `instance` field: the -/// daemon picks the mind before it calls here. -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] -pub struct PipelineQuery { - /// The key's root segment: a campaign slug, or an instance for - /// stewardships, hand-offs and the identity document. - pub campaign: Option, - pub repo: Option, - pub cut: Option