diff --git a/crates/eureka-state/src/bin/huginn/main.rs b/crates/eureka-state/src/bin/huginn/main.rs index 2bab473..711d64f 100644 --- a/crates/eureka-state/src/bin/huginn/main.rs +++ b/crates/eureka-state/src/bin/huginn/main.rs @@ -485,7 +485,7 @@ fn admit( } } let sent = documents.clone(); - let provenance = PipelineProvenance { faculty, agent: Short(agent), session: Short(session), tool: Short(TOOL.to_string()) }; + let provenance = PipelineProvenance { faculty, agent: Short(agent), session: Short(session), tool: Short(TOOL.to_string()), imported_from: None }; let request = |instance: &Slug| { HuginnMindRequest::Admit(PipelineAdmissionBatch { instance: instance.clone(), provenance, documents }) }; @@ -625,7 +625,7 @@ fn persona_put( if !state.is_object() { return Err(Trouble::invalid("the persona document is one JSON object, the whole gamecult.persona_state.v0")); } - let provenance = PipelineProvenance { faculty, agent: Short(agent), session: Short(session), tool: Short(TOOL.to_string()) }; + let provenance = PipelineProvenance { faculty, agent: Short(agent), session: Short(session), tool: Short(TOOL.to_string()), imported_from: None }; match settings.client()?.persona_put(provenance, state, expected_updated_at).map_err(Trouble::from)? { HuginnMindResponse::PersonaPut(PersonaPutOutcome::Committed { receipt_id }) => { say(format_args!("Committed {receipt_id}")); diff --git a/crates/eureka-state/src/bin/huginn/near.rs b/crates/eureka-state/src/bin/huginn/near.rs index 0fd525b..248a682 100644 --- a/crates/eureka-state/src/bin/huginn/near.rs +++ b/crates/eureka-state/src/bin/huginn/near.rs @@ -82,6 +82,7 @@ mod tests { agent: Short("a".into()), session: Short("s".into()), tool: Short("huginn".into()), + imported_from: None, }, ordinal: 1, }, diff --git a/crates/eureka-state/src/main.rs b/crates/eureka-state/src/main.rs index 19b311b..51ca41c 100644 --- a/crates/eureka-state/src/main.rs +++ b/crates/eureka-state/src/main.rs @@ -279,6 +279,7 @@ impl EurekaState { agent: input.agent, session: input.session, tool: Short(TOOL.into()), + imported_from: None, }, documents, }) @@ -319,6 +320,7 @@ impl EurekaState { agent: input.agent, session: input.session, tool: Short(TOOL.into()), + imported_from: None, }; Self::replied( self.ask(move |client| client.persona_put(provenance, input.state, input.expected_updated_at)).await, diff --git a/crates/eureka-state/tests/cli.rs b/crates/eureka-state/tests/cli.rs index 0cfad0e..87e97f5 100644 --- a/crates/eureka-state/tests/cli.rs +++ b/crates/eureka-state/tests/cli.rs @@ -519,6 +519,7 @@ fn view_of(document: &PipelineDocument) -> PipelineDocumentView { agent: Short("a".into()), session: Short("s".into()), tool: Short("huginn".into()), + imported_from: None, }, ordinal: 1, }, diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs index 66a304d..2a042ec 100644 --- a/crates/eureka-state/tests/common/mod.rs +++ b/crates/eureka-state/tests/common/mod.rs @@ -45,6 +45,7 @@ pub fn provenance() -> PipelineProvenance { agent: Short("claude".into()), session: Short("session-1".into()), tool: Short("admit".into()), + imported_from: None, } } diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index 39d111b..25d32a5 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -338,6 +338,7 @@ pub(crate) mod tests { agent: Short("claude".into()), session: Short("session-1".into()), tool: Short("admit".into()), + imported_from: None, } } diff --git a/crates/huginn-mind/src/admission.rs b/crates/huginn-mind/src/admission.rs index b7e9ac3..ad37a64 100644 --- a/crates/huginn-mind/src/admission.rs +++ b/crates/huginn-mind/src/admission.rs @@ -6,18 +6,22 @@ //! first failing step is the outcome: //! //! - A1 the batch names this mind, and attributes itself to a named agent and -//! session (`refuse_provenance`); A2 one to sixty-four envelopes; +//! session (`refuse_provenance`); a batch that names a source (`imported_from`) +//! replays another mind's receipt, never this mind's own; A2 one to sixty-four +//! envelopes; //! - A3 every envelope passes the leaf's bounds, formats and key //! recomputation; A4 identities are unique in the batch; A5 every document //! carrying an instance names this mind; //! - A6 an empty mind's first batch carries exactly one `instance` document, //! and a batch carrying one derives the epoch record; //! - A7 every reference names a document present in the image or the batch, -//! looked up by `(kind, id)`; the ones resolved in the image become the -//! receipt's strong reads; +//! looked up by `(kind, id)`, or, in an import, listed as external and held +//! by neither; the ones resolved in the image become the receipt's strong +//! reads; //! - the derived writes (a ruling's `Answered` resolution, a hand-off's -//! stewardship withdrawal or assignment) are appended and pass A3-A7; -//! - A8 the per-kind rules, below; A9 an exact replay answers with the stored +//! stewardship withdrawal or assignment) are appended and pass A3-A7; an +//! import derives none, the source's riding in the batch as written; +//! - A8 the per-kind rules, below, after the two that concern a move; A9 an exact replay answers with the stored //! receipt, after validation and never before it (ruling 20); A10 no write //! collides with the image; A11 the commit. //! @@ -40,9 +44,10 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use crate::docs::{CitationRole, Docs, Staged, citations, kind_of_id, line_of, outcome_citations}; +use crate::rows::root_and_local; use crate::mind::{HuginnMindEpoch, Mind, unavailable}; use crate::persona::{self, HuginnPersonaEntry, PersonaPutOutcome}; -use crate::receipt::{self, CommitOutcome, PipelineProvenance}; +use crate::receipt::{self, CommitOutcome, IMPORT_EXTERNAL_MAX, ImportSource, PipelineProvenance}; use crate::refusal::{MindRefusal, RunId}; use crate::store::MindStore; @@ -119,6 +124,10 @@ impl Mind { // A1 self.require_instance(instance)?; refuse_provenance(&provenance)?; + let import = provenance.imported_from.clone(); + if import.as_ref().is_some_and(|source| source.instance == *self.instance()) { + return Err(MindRefusal::ImportFromSelf); + } // A2 if envelopes.is_empty() || envelopes.len() > BATCH_MAX { return Err(MindRefusal::BatchSize { actual: envelopes.len() as u32 }); @@ -136,19 +145,25 @@ impl Mind { return Err(MindRefusal::MissingIdentity); } // A7 + let external: &[PipelineRef] = import.as_ref().map_or(&[], |source| source.external.as_slice()); + if let Some(held) = external.iter().find(|referent| docs.find(referent.kind, &referent.id.0).is_some()) { + return Err(MindRefusal::ExternalPresent { id: held.id.0.clone() }); + } let mut strong = BTreeSet::new(); for staged in &docs.batch { - resolve(&docs, staged, &mind, &mut strong)?; + resolve(&docs, staged, &mind, external, &mut strong)?; } - // Derived writes pass A3-A7 themselves. + // Derived writes pass A3-A7 themselves. An import derives none: the source's rode in its batch. let originals = docs.batch.len(); - for document in derive(&docs, &mind) { - document.validate()?; - let envelope = prepare(&document, self.cache())?; - docs.push(stage(envelope, &mind)?)?; + if import.is_none() { + for document in derive(&docs, &mind) { + document.validate()?; + let envelope = prepare(&document, self.cache())?; + docs.push(stage(envelope, &mind)?)?; + } } for staged in &docs.batch[originals..] { - resolve(&docs, staged, &mind, &mut strong)?; + resolve(&docs, staged, &mind, external, &mut strong)?; } // The pins: cited image bytes, and every write. let strong_reads = strong @@ -162,7 +177,7 @@ impl Mind { let candidate = receipt::candidate(&mind, provenance, &strong_reads, &writes, receipt::head(self)? + 1, now)?; // A8 for staged in &docs.batch { - check(&docs, staged, &mind)?; + check(&docs, staged, &mind, import.as_ref())?; } // A9 if let Some(existing) = receipt::replay(self, &candidate)? { @@ -280,11 +295,33 @@ fn refuse_provenance(provenance: &PipelineProvenance) -> Result<(), MindRefusal> matches!(c, '\u{061C}' | '\u{200E}' | '\u{200F}' | '\u{202A}'..='\u{202E}' | '\u{2066}'..='\u{2069}') }; let control = |c: char| c.is_control() || matches!(c, '\u{2028}' | '\u{2029}'); - for (field, value) in [ + let mut texts = vec![ ("provenance.agent", &provenance.agent), ("provenance.session", &provenance.session), ("provenance.tool", &provenance.tool), - ] { + ]; + if let Some(source) = &provenance.imported_from { + source.instance.validate_slug().map_err(|refusal| relocate(refusal, "provenance.imported_from.instance"))?; + if source.external.len() > IMPORT_EXTERNAL_MAX { + return Err(PipelineRefusal::FieldBound { + field: "provenance.imported_from.external".into(), + limit: IMPORT_EXTERNAL_MAX as u32, + actual: source.external.len() as u32, + } + .into()); + } + for referent in &source.external { + referent.validate_ref().map_err(|refusal| relocate(refusal, "provenance.imported_from.external"))?; + } + texts.extend([ + ("provenance.imported_from.receipt_id", &source.receipt_id), + ("provenance.imported_from.committed_at", &source.committed_at), + ("provenance.imported_from.agent", &source.agent), + ("provenance.imported_from.session", &source.session), + ("provenance.imported_from.tool", &source.tool), + ]); + } + for (field, value) in texts { let text = value.0.as_str(); if !text.chars().any(char::is_alphanumeric) || text.contains(control) || text.contains(bidi) { return Err(PipelineRefusal::InvalidFormat { field: field.into(), value: text.into() }.into()); @@ -296,6 +333,16 @@ fn refuse_provenance(provenance: &PipelineProvenance) -> Result<(), MindRefusal> Ok(()) } +/// The leaf's refusal of a bare value, named at the provenance field that held it. +fn relocate(refusal: PipelineRefusal, field: &str) -> MindRefusal { + match refusal { + PipelineRefusal::InvalidFormat { value, .. } => PipelineRefusal::InvalidFormat { field: field.into(), value }, + PipelineRefusal::FieldBound { limit, actual, .. } => PipelineRefusal::FieldBound { field: field.into(), limit, actual }, + other => other, + } + .into() +} + /// A3 through the leaf, then A5. fn stage(envelope: CultCacheEnvelope, mind: &Slug) -> Result { validate_pipeline_write_envelope(&envelope)?; @@ -339,7 +386,7 @@ struct Reference { /// 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). +/// as an id of a kind. fn references(staged: &Staged, mind: &Slug) -> Result, MindRefusal> { if let PipelineDocument::HandOff(hand_off) = &staged.document { if hand_off.from_instance != *mind { @@ -367,9 +414,15 @@ fn references(staged: &Staged, mind: &Slug) -> Result, MindRefusa .collect()) } -/// A7: every referent is in the batch or the image; the image ones are the -/// receipt's strong reads. -fn resolve(docs: &Docs<'_>, staged: &Staged, mind: &Slug, strong: &mut BTreeSet<(String, String)>) -> Result<(), MindRefusal> { +/// A7: every referent is in the batch or the image, or is an import's declared external; the image +/// ones are the receipt's strong reads. +fn resolve( + docs: &Docs<'_>, + staged: &Staged, + mind: &Slug, + external: &[PipelineRef], + strong: &mut BTreeSet<(String, String)>, +) -> Result<(), MindRefusal> { for reference in references(staged, mind)? { if docs.in_batch(reference.kind, &reference.id).is_some() { continue; @@ -378,6 +431,9 @@ fn resolve(docs: &Docs<'_>, staged: &Staged, mind: &Slug, strong: &mut BTreeSet< strong.insert((reference.kind.type_id().to_string(), reference.id)); continue; } + if is_external(external, reference.kind, &reference.id) { + continue; + } return Err(match reference.missing { Missing::Reference => MindRefusal::MissingReference { kind: reference.kind, id: reference.id }, Missing::SpecOfReport(report) => MindRefusal::CutReportWithoutSpec { report }, @@ -447,11 +503,31 @@ fn derive(docs: &Docs<'_>, mind: &Slug) -> Vec { derived } -/// A8: the per-kind rules. -fn check(docs: &Docs<'_>, staged: &Staged, mind: &Slug) -> Result<(), MindRefusal> { +fn is_external(external: &[PipelineRef], kind: PipelineKind, id: &str) -> bool { + external.iter().any(|referent| referent.kind == kind && referent.id.0 == id) +} + +/// A8: the per-kind rules, after the two that concern a move: a Moved resolution is final, and a +/// campaign it closed takes no write under its root. An import's listed externals stand in force +/// wherever a rule looks a referent up, and no rule reads their content. +fn check(docs: &Docs<'_>, staged: &Staged, mind: &Slug, import: Option<&ImportSource>) -> Result<(), MindRefusal> { use PipelineDocument as D; use PipelineKind as K; let key = staged.key.as_str(); + let external: &[PipelineRef] = import.map_or(&[], |source| source.external.as_slice()); + if let D::Resolution(resolution) = &staged.document + && matches!(resolution.outcome, ResolutionOutcome::Withdrawn { .. }) + && resolution.subject.kind == K::Resolution + && matches!( + docs.find(K::Resolution, &resolution.subject.id.0), + Some(D::Resolution(subject)) if matches!(subject.outcome, ResolutionOutcome::Moved { .. }) + ) + { + return Err(MindRefusal::MoveIsFinal); + } + if let Some((moved, PipelineResolution { outcome: ResolutionOutcome::Moved { to }, .. })) = docs.moved_campaign(root_and_local(key).0) { + return Err(MindRefusal::CampaignMoved { campaign: root_and_local(moved).0.into(), to: to.0.clone() }); + } match &staged.document { D::Campaign(campaign) => { if campaign.repos.is_empty() { @@ -485,6 +561,9 @@ fn check(docs: &Docs<'_>, staged: &Staged, mind: &Slug) -> Result<(), MindRefusa } let Some(question_id) = &ruling.answers else { return Ok(()) }; let Some(D::Question(question)) = docs.find(K::Question, &question_id.0) else { + if is_external(external, K::Question, &question_id.0) { + return Ok(()); + } return Err(MindRefusal::MissingReference { kind: K::Question, id: question_id.0.clone() }); }; let own = |resolution: &PipelineResolution| { @@ -532,6 +611,9 @@ fn check(docs: &Docs<'_>, staged: &Staged, mind: &Slug) -> Result<(), MindRefusa } D::CutReport(report) => { let Some(D::CutSpec(spec)) = docs.find(K::CutSpec, &report.cut_spec.0) else { + if is_external(external, K::CutSpec, &report.cut_spec.0) { + return Ok(()); + } return Err(MindRefusal::CutReportWithoutSpec { report: key.into() }); }; if !docs.in_force(K::CutSpec, &report.cut_spec.0) { @@ -550,6 +632,9 @@ fn check(docs: &Docs<'_>, staged: &Staged, mind: &Slug) -> Result<(), MindRefusa } D::Verdict(verdict) => { let Some(D::CutReport(report)) = docs.find(K::CutReport, &verdict.cut_report.0) else { + if is_external(external, K::CutReport, &verdict.cut_report.0) { + return Ok(()); + } return Err(MindRefusal::MissingReference { kind: K::CutReport, id: verdict.cut_report.0.clone() }); }; for claim in &verdict.claims { @@ -594,7 +679,18 @@ fn check(docs: &Docs<'_>, staged: &Staged, mind: &Slug) -> Result<(), MindRefusa } Ok(()) } - D::Resolution(resolution) => resolution_rule(docs, resolution), + D::Resolution(resolution) => { + resolution_rule(docs, resolution)?; + match &resolution.outcome { + ResolutionOutcome::Moved { to } if to == mind => Err(MindRefusal::MoveToSelf), + // The giver's resolution: the receiver replays the history the move froze, never the move. + ResolutionOutcome::Moved { .. } if import.is_some() => Err(MindRefusal::IncompatibleResolution { + subject_kind: resolution.subject.kind, + outcome: outcome_name(&resolution.outcome).into(), + }), + _ => Ok(()), + } + } D::Stewardship(stewardship) => { let expected = docs.latest_stewardship(&stewardship.instance, &stewardship.repo, Some(stewardship)) + 1; if stewardship.sequence != expected { @@ -710,6 +806,7 @@ fn outcome_name(outcome: &ResolutionOutcome) -> &'static str { ResolutionOutcome::Recorded { .. } => "Recorded", ResolutionOutcome::Withdrawn { .. } => "Withdrawn", ResolutionOutcome::Slept { .. } => "Slept", + ResolutionOutcome::Moved { .. } => "Moved", } } @@ -758,6 +855,7 @@ fn matrix(subject: PipelineKind, outcome: &ResolutionOutcome) -> bool { (K::Thought, O::Answered { by }) => by.kind == K::FollowUp, (K::Thought, O::Recorded { .. }) => true, (K::Resolution, O::Withdrawn { .. }) => true, + (K::Campaign, O::Moved { .. }) => true, _ => false, } } @@ -951,7 +1049,7 @@ mod tests { fn admit_signed(agent: &str, session: &str, tool: &str) -> (Mind, PipelineAdmissionOutcome) { let mut mind = opened(MemoryStore::new(), INSTANCE); - let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: s(agent), session: s(session), tool: s(tool) }; + let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: s(agent), session: s(session), tool: s(tool), imported_from: None }; let batch = PipelineAdmissionBatch { instance: slug(INSTANCE), provenance, documents: vec![instance(INSTANCE), stewardship(INSTANCE, REPO)] }; let outcome = mind.admit(batch, now()); (mind, outcome) @@ -1073,7 +1171,7 @@ mod tests { D::Campaign(campaign).validate().is_ok() }; let mine_accepts = |text: &str| { - let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: s(text), session: s("s"), tool: s("t") }; + let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: s(text), session: s("s"), tool: s("t"), imported_from: None }; refuse_provenance(&provenance).is_ok() }; for text in [ diff --git a/crates/huginn-mind/src/docs.rs b/crates/huginn-mind/src/docs.rs index 8198eb4..da817ca 100644 --- a/crates/huginn-mind/src/docs.rs +++ b/crates/huginn-mind/src/docs.rs @@ -15,6 +15,7 @@ use serde::{Deserialize, Serialize}; use crate::image::Image; use crate::refusal::MindRefusal; +use crate::rows::root_and_local; /// A batch document after A3 and A5: decoded, keyed, and its envelope. pub(crate) struct Staged { @@ -203,9 +204,36 @@ impl<'i> Docs<'i> { } /// The resolution that closes the document, if any: the views' status, - /// derived where admission derives whether there is one at all. + /// derived where admission derives whether there is one at all. A document + /// with no in-force closer of its own reads closed by the Moved resolution of + /// the campaign it is rooted at, the Moved resolution itself excepted. The + /// fallback is the status's alone: a rule asks `in_force`, which does not + /// have it, so a moved campaign's own closures keep their standing. pub(crate) fn closing_resolution(&self, kind: PipelineKind, id: &str) -> Option<(&str, &PipelineResolution)> { - self.closing_resolution_unless(kind, id, |_| false) + self.closing_resolution_unless(kind, id, |_| false).or_else(|| { + let moved = self.moved_campaign(root_and_local(id).0)?; + (kind != PipelineKind::Resolution || id != moved.0).then_some(moved) + }) + } + + /// The Moved resolution in force over the campaign rooted at `root`, among the held documents + /// visible at `as_of`: the one fact that freezes a root and closes everything under it. The + /// batch is not asked, so a Moved and the writes it lands beside are one receipt. + pub(crate) fn moved_campaign(&self, root: &str) -> Option<(&'i str, &'i PipelineResolution)> { + let image: &'i Image = self.image; + image + .of_kind(PipelineKind::Campaign) + .filter(|(key, _)| root_and_local(key).0 == root) + .flat_map(|(key, _)| image.closers_of(PipelineKind::Campaign, key)) + .filter(|(_, row)| self.visible(*row)) + .find_map(|(key, row)| match &image.row(row).view.document { + PipelineDocument::Resolution(resolution) + if matches!(resolution.outcome, ResolutionOutcome::Moved { .. }) && self.in_force(PipelineKind::Resolution, key) => + { + Some((key, resolution)) + } + _ => None, + }) } /// Every in-force stewardship of `(mind, repo)`, ignoring the withdrawal @@ -534,7 +562,10 @@ pub(crate) fn outcome_citations(outcome: &ResolutionOutcome) -> Vec<(CitationRol ResolutionOutcome::Fixed { by: Some(by), .. } => vec![(CitationRole::ResolvedBy, by.clone())], ResolutionOutcome::Deferred { to } => vec![(CitationRole::DeferredTo, to.clone())], ResolutionOutcome::Slept { day_through, .. } => day_through.iter().map(|memory| (CitationRole::DayThrough, memory.clone())).collect(), - ResolutionOutcome::Fixed { by: None, .. } | ResolutionOutcome::Recorded { .. } | ResolutionOutcome::Withdrawn { .. } => { + ResolutionOutcome::Fixed { by: None, .. } + | ResolutionOutcome::Recorded { .. } + | ResolutionOutcome::Withdrawn { .. } + | ResolutionOutcome::Moved { .. } => { vec![] } } diff --git a/crates/huginn-mind/src/fixtures.rs b/crates/huginn-mind/src/fixtures.rs index f80064c..7620c15 100644 --- a/crates/huginn-mind/src/fixtures.rs +++ b/crates/huginn-mind/src/fixtures.rs @@ -439,7 +439,7 @@ pub(crate) fn idea(pass: &str, label: &str, text: &str) -> PipelineDocument { } pub(crate) fn provenance(faculty: Faculty) -> PipelineProvenance { - PipelineProvenance { faculty, agent: s("claude"), session: s("session-1"), tool: s("admit") } + PipelineProvenance { faculty, agent: s("claude"), session: s("session-1"), tool: s("admit"), imported_from: None } } pub(crate) fn now() -> DateTime { diff --git a/crates/huginn-mind/src/image.rs b/crates/huginn-mind/src/image.rs index 9d4c71b..6d2819f 100644 --- a/crates/huginn-mind/src/image.rs +++ b/crates/huginn-mind/src/image.rs @@ -16,7 +16,7 @@ use std::collections::{BTreeMap, BTreeSet, HashMap}; use cultcache_rs::DatabaseEntry; use cultnet_rs::Selection; -use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, PipelineResolution, Short}; +use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, PipelineResolution, ResolutionOutcome, Short}; use crate::docs::{Docs, citations, kind_of_id, kind_of_type}; use crate::mind::Envelopes; @@ -201,6 +201,22 @@ impl Image { current = subject; } } + // A Moved closes every document under its campaign's root, none of which it names. + let moved: Vec = (first..self.rows.len() as u32) + .filter_map(|id| match &self.row(id).view.document { + PipelineDocument::Resolution(resolution) + if resolution.subject.kind == PipelineKind::Campaign && matches!(resolution.outcome, ResolutionOutcome::Moved { .. }) => + { + Some(root_and_local(&resolution.subject.id.0).0.to_string()) + } + _ => None, + }) + .collect(); + for root in &moved { + for kind in PipelineKind::ALL { + touched.extend(self.posting(*kind, Alias::Root.name(), root).into_iter().flatten().copied()); + } + } for id in touched { self.rebuild(id, id >= first, facts.ordinal); } @@ -368,6 +384,7 @@ impl Image { agent: Short("claude".into()), session: Short("session-1".into()), tool: Short("admit".into()), + imported_from: None, }, ordinal: n as u64 + 1, }; diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index a57e5d1..fa6b840 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -48,6 +48,9 @@ pub(crate) mod fixtures; #[cfg(test)] mod read_tests; +#[cfg(test)] +mod move_tests; + pub use docs::CitationRole; pub use index::{INDEX_TEXT_MAX_BYTES, INDEX_TEXT_VERSION, IndexEntry, index_text}; pub use admission::{BATCH_MAX, PipelineAdmissionBatch, PipelineAdmissionOutcome}; @@ -57,7 +60,7 @@ pub use query::{ AdmissionFacts, PipelineDocumentSummary, PipelineDocumentView, PipelineEdge, PipelineFacts, PipelinePageItems, Members, PipelineSelectionPage, PipelineStatus, PipelineStatusSummary, Ranked, RootTouched, SUMMARY_MAX_BYTES, SemanticQuery, }; -pub use receipt::{DocumentVersion, Faculty, HuginnCommitReceipt, PipelineProvenance, RECEIPT_SCHEMA_VERSION}; +pub use receipt::{DocumentVersion, Faculty, HuginnCommitReceipt, IMPORT_EXTERNAL_MAX, ImportSource, PipelineProvenance, RECEIPT_SCHEMA_VERSION}; pub use refusal::{MindRefusal, RunId}; pub use store::{MindStore, OwnedRedbMessagePackBackingStore}; pub use wire::{ diff --git a/crates/huginn-mind/src/move_tests.rs b/crates/huginn-mind/src/move_tests.rs new file mode 100644 index 0000000..2fb2768 --- /dev/null +++ b/crates/huginn-mind/src/move_tests.rs @@ -0,0 +1,484 @@ +//! A campaign moving between minds: the giver's Moved resolution freezes its root and closes +//! everything under it, and the receiver admits the history as an import that names its source. +//! Each test fails when the rule it names breaks, and reads behaviour through the doors a caller +//! uses: admission, `view`, `query`. + +use cultnet_rs::{FieldPredicate, Selection}; +use eureka_pipeline::{ + ClaimOutcome, FindingConfidence, PipelineDocument as D, PipelineKind as K, PipelineRef, ResolutionOutcome, pipeline_key, +}; + +use crate::admission::{PipelineAdmissionBatch, PipelineAdmissionOutcome}; +use crate::fixtures::*; +use crate::mind::Mind; +use crate::query::{PipelinePageItems, PipelineStatus, Reader}; +use crate::receipt::{Faculty, HuginnCommitReceipt, IMPORT_EXTERNAL_MAX, ImportSource, PipelineProvenance}; +use crate::refusal::MindRefusal; +use crate::store::test_stores::MemoryStore; + +const RECEIVER: &str = "heimdall"; +const OTHER_ROOT: &str = "other-campaign"; + +fn campaign_ref() -> PipelineRef { + r(K::Campaign, &id("campaign", "self")) +} + +fn moved_to(mind: &str) -> D { + resolution(campaign_ref(), ResolutionOutcome::Moved { to: slug(mind) }) +} + +fn moved_key(mind: &str) -> String { + pipeline_key(&moved_to(mind)).unwrap() +} + +fn answering(campaign: &str, label: &str, question: &str) -> D { + let mut ruling = ruling(label); + ruling.campaign = slug(campaign); + ruling.answers = Some(s(question)); + ruling.choice = Some(l("A")); + D::Ruling(ruling) +} + +/// A question of another campaign, which this mind holds without holding the campaign. +fn other_question(label: &str) -> D { + let mut question = question(label, &["A", "B"], "A"); + let D::Question(value) = &mut question else { unreachable!() }; + value.campaign = slug(OTHER_ROOT); + question +} + +fn other_follow_up(label: &str, question: &str) -> D { + let mut follow_up = follow_up(label, r(K::Question, &format!("{OTHER_ROOT}:question:{question}"))); + let D::FollowUp(value) = &mut follow_up else { unreachable!() }; + value.campaign = slug(OTHER_ROOT); + follow_up +} + +fn view_of(mind: &Mind, kind: K, key: &str) -> crate::query::PipelineDocumentView { + mind.view(&r(kind, key)).unwrap().unwrap_or_else(|| panic!("{key} is not held")) +} + +fn head(mind: &Mind) -> u64 { + mind.image().unwrap().head +} + +fn moved_by(status: &PipelineStatus, moved: &str) -> bool { + matches!(status, PipelineStatus::Resolved { resolution, record } + if resolution.id.0 == moved && matches!(record.outcome, ResolutionOutcome::Moved { .. })) +} + +fn open_follow_ups(mind: &Mind) -> Vec { + let in_force = FieldPredicate { index: "in_force".into(), op: "any_of".into(), values: Some(vec!["true".into()]), number: None }; + let selection = Selection { + schemas: Some(vec![K::FollowUp.type_id().to_string()]), + fields: Some(vec![in_force]), + projection: "document".into(), + ..Selection::default() + }; + match mind.query(&selection).unwrap().items { + PipelinePageItems::Documents(found) => found.into_iter().map(|view| view.id.id.0).collect(), + PipelinePageItems::Headers(_) => panic!("the document projection answers documents"), + } +} + +fn receiver() -> Mind { + let mut mind = opened(MemoryStore::new(), RECEIVER); + committed(admit(&mut mind, vec![instance(RECEIVER), stewardship(RECEIVER, REPO)])); + mind +} + +fn source_of(instance: &str, external: Vec) -> ImportSource { + ImportSource { + instance: slug(instance), + receipt_id: s("mind-commit-source"), + ordinal: 5, + committed_at: s("2026-11-08T16:20:05Z"), + faculty: Faculty::Imagination, + agent: s("imagination"), + session: s("source-session"), + tool: s("huginn"), + external, + } +} + +fn import_as(mind: &mut Mind, source: ImportSource, documents: Vec) -> PipelineAdmissionOutcome { + let provenance = PipelineProvenance { + faculty: Faculty::SelfFaculty, + agent: s("heimdall-self"), + session: s("import-1"), + tool: s("huginn"), + imported_from: Some(source), + }; + let batch = PipelineAdmissionBatch { instance: mind.instance().clone(), provenance, documents }; + mind.admit(batch, now()) +} + +fn import(mind: &mut Mind, documents: Vec, external: Vec) -> PipelineAdmissionOutcome { + import_as(mind, source_of(INSTANCE, external), documents) +} + +/// The giver of a whole campaign: a question and the ruling that answers it, a spec, its report, a +/// falsifying verdict with its finding, the finding fixed, and two follow-ups one of which is +/// recorded. Returns the mind and the ordinal of its last receipt before any move. +fn giver() -> (Mind, u64) { + let mut mind = seeded(); + let q1 = id("question", "Q1"); + let finding = id("finding", "cut-1.s1.F1"); + committed(admit(&mut mind, vec![question("Q1", &["A", "B"], "A"), question("Q2", &["A", "B"], "A"), follow_up("FU-1", r(K::Question, &q1))])); + committed(admit(&mut mind, vec![answering(CAMPAIGN, "R1", &q1)])); + committed(admit(&mut mind, vec![D::CutSpec(cut_spec("1", 1))])); + committed(admit(&mut mind, vec![D::CutReport(cut_report("1", 1))])); + committed(admit( + &mut mind, + vec![ + verdict("1", 1, vec![claim(ClaimOutcome::Falsified, &[&finding], Some("P1"), &["M1"])]), + D::Finding(finding_doc()), + ], + )); + committed(admit(&mut mind, vec![resolution(r(K::Finding, &finding), ResolutionOutcome::Fixed { commit: sha(), by: None })])); + committed(admit(&mut mind, vec![follow_up("FU-2", r(K::Question, &q1))])); + committed(admit(&mut mind, vec![resolution(r(K::FollowUp, &id("follow_up", "FU-2")), ResolutionOutcome::Recorded { reason: "noted".into() })])); + let head = head(&mind); + (mind, head) +} + +fn finding_doc() -> eureka_pipeline::PipelineFinding { + finding("1", 1, "F1", FindingConfidence::Confirmed) +} + +fn root_of(key: &str) -> &str { + key.split(':').next().unwrap() +} + +/// Replays the giver's receipts up to `through`, in ordinal order, each filtered to the campaign's +/// root, into the receiver as imports that name their source receipt. +fn replay(giver: &Mind, receiver: &mut Mind, through: u64) -> Vec { + let mut receipts: Vec = giver.receipts().unwrap(); + receipts.sort_by_key(|receipt| receipt.ordinal); + let mut outcomes = Vec::new(); + for receipt in receipts.into_iter().filter(|receipt| receipt.ordinal <= through) { + let documents: Vec = receipt + .writes + .iter() + .filter(|write| crate::docs::kind_of_type(&write.document_type).is_some() && root_of(&write.document_key) == CAMPAIGN) + .map(|write| giver.get(crate::docs::kind_of_type(&write.document_type).unwrap(), &write.document_key).unwrap().unwrap()) + .collect(); + if documents.is_empty() { + continue; + } + let source = ImportSource { + instance: giver.instance().clone(), + receipt_id: s(&receipt.receipt_id), + ordinal: receipt.ordinal, + committed_at: s(&receipt.committed_at), + faculty: receipt.provenance.faculty, + agent: receipt.provenance.agent.clone(), + session: receipt.provenance.session.clone(), + tool: receipt.provenance.tool.clone(), + external: vec![], + }; + outcomes.push(import_as(receiver, source, documents)); + } + outcomes +} + +#[test] +fn moved_campaign_refuses_every_later_write_under_its_root() { + let mut mind = seeded(); + let q1 = id("question", "Q1"); + let q0 = id("question", "Q0"); + committed(admit(&mut mind, vec![question("Q1", &["A", "B"], "A"), question("Q0", &["A", "B"], "A")])); + // A ruling of another root answers a question of this one: the derived resolution is rooted here. + let (_, writes) = committed(admit(&mut mind, vec![answering(OTHER_ROOT, "R0", &q0)])); + assert_eq!(writes.len(), 2, "before the move, the ruling and its derived Answered land: {writes:?}"); + + committed(admit(&mut mind, vec![moved_to(RECEIVER)])); + let frozen_at = head(&mind); + let frozen = |outcome: PipelineAdmissionOutcome| { + assert_eq!(refusal(outcome), MindRefusal::CampaignMoved { campaign: CAMPAIGN.into(), to: RECEIVER.into() }); + }; + frozen(admit(&mut mind, vec![follow_up("FU-1", r(K::Question, &q1))])); + frozen(admit(&mut mind, vec![resolution(r(K::Question, &q1), withdrawn())])); + frozen(admit(&mut mind, vec![answering(OTHER_ROOT, "R1", &q1)])); + frozen(admit(&mut mind, vec![moved_to(RECEIVER)])); + frozen(admit(&mut mind, vec![moved_to("kiko")])); + assert_eq!(head(&mind), frozen_at, "a refused write leaves nothing behind"); + + committed(admit(&mut mind, vec![other_question("Q5")])); + assert_eq!(head(&mind), frozen_at + 1, "a write under another root lands"); +} + +#[test] +fn moved_rows_read_resolved_by_the_move_and_keep_their_own_closer() { + let (mut mind, _) = giver(); + let moved = moved_key(RECEIVER); + let finding = id("finding", "cut-1.s1.F1"); + let fixed = id("resolution", "finding.cut-1.s1.F1.n1"); + committed(admit(&mut mind, vec![other_question("Q5")])); + committed(admit(&mut mind, vec![moved_to(RECEIVER)])); + + for (kind, key) in [ + (K::Question, id("question", "Q2")), + (K::FollowUp, id("follow_up", "FU-1")), + (K::Campaign, id("campaign", "self")), + (K::Ruling, id("ruling", "R1")), + ] { + let status = view_of(&mind, kind, &key).status; + assert!(moved_by(&status, &moved), "{key} reads {status:?}"); + } + let answered = view_of(&mind, K::Question, &id("question", "Q1")).status; + assert!(matches!(&answered, PipelineStatus::Resolved { record, .. } if matches!(record.outcome, ResolutionOutcome::Answered { .. })), "{answered:?}"); + let PipelineStatus::Resolved { resolution, record } = view_of(&mind, K::Finding, &finding).status else { + panic!("the fixed finding stays resolved") + }; + assert_eq!(resolution.id.0, fixed, "its own closer, not the move"); + assert!(matches!(record.outcome, ResolutionOutcome::Fixed { .. })); + let recorded = view_of(&mind, K::FollowUp, &id("follow_up", "FU-2")).status; + assert!(matches!(&recorded, PipelineStatus::Resolved { record, .. } if matches!(record.outcome, ResolutionOutcome::Recorded { .. })), "{recorded:?}"); + + assert_eq!(view_of(&mind, K::Resolution, &moved).status, PipelineStatus::InForce, "the Moved resolution stands"); + assert_eq!(view_of(&mind, K::Question, &format!("{OTHER_ROOT}:question:Q5")).status, PipelineStatus::InForce); + assert_eq!(view_of(&mind, K::Instance, &format!("{INSTANCE}:instance:self")).status, PipelineStatus::InForce); +} + +#[test] +fn a_read_before_the_move_still_sees_in_force() { + let (mut mind, _) = giver(); + let moved = moved_key(RECEIVER); + let fixed = r(K::Finding, &id("finding", "cut-1.s1.F1")); + committed(admit(&mut mind, vec![moved_to(RECEIVER)])); + let landed = head(&mind); + + let before = Reader::at(&mind, landed - 1).unwrap(); + let follow_up = r(K::FollowUp, &id("follow_up", "FU-1")); + assert_eq!(before.view(&follow_up).unwrap().unwrap().status, PipelineStatus::InForce); + assert_eq!(before.view(&r(K::Campaign, &id("campaign", "self"))).unwrap().unwrap().status, PipelineStatus::InForce); + assert!(before.view(&r(K::Resolution, &moved)).unwrap().is_none(), "the Moved resolution did not exist yet"); + assert!(matches!(before.view(&fixed).unwrap().unwrap().status, PipelineStatus::Resolved { .. }), "a closer of its own reads as it did"); + assert!(moved_by(&Reader::at(&mind, landed).unwrap().view(&follow_up).unwrap().unwrap().status, &moved)); + assert!(moved_by(&mind.view(&follow_up).unwrap().unwrap().status, &moved)); +} + +#[test] +fn in_force_follow_ups_leave_a_moved_campaign() { + let mut mind = seeded(); + committed(admit(&mut mind, vec![question("Q1", &["A", "B"], "A"), follow_up("FU-1", r(K::Question, &id("question", "Q1")))])); + committed(admit(&mut mind, vec![other_question("Q5"), other_follow_up("FU-9", "Q5")])); + let mut both = open_follow_ups(&mind); + both.sort(); + assert_eq!(both, vec![id("follow_up", "FU-1"), format!("{OTHER_ROOT}:follow_up:FU-9")]); + + committed(admit(&mut mind, vec![moved_to(RECEIVER)])); + assert_eq!(open_follow_ups(&mind), vec![format!("{OTHER_ROOT}:follow_up:FU-9")], "the moved root's leads left, the other root's stayed"); +} + +#[test] +fn moved_only_closes_a_campaign_is_final_and_never_to_self() { + let mut mind = seeded(); + let q1 = id("question", "Q1"); + committed(admit(&mut mind, vec![question("Q1", &["A", "B"], "A"), follow_up("FU-1", r(K::Question, &q1))])); + let to = |mind: &str| ResolutionOutcome::Moved { to: slug(mind) }; + for (kind, key) in [(K::FollowUp, id("follow_up", "FU-1")), (K::Question, q1.clone())] { + assert_eq!( + refusal(admit(&mut mind, vec![resolution(r(kind, &key), to(RECEIVER))])), + MindRefusal::IncompatibleResolution { subject_kind: kind, outcome: "Moved".into() } + ); + } + assert_eq!(refusal(admit(&mut mind, vec![resolution(campaign_ref(), to(INSTANCE))])), MindRefusal::MoveToSelf); + assert!(mind.view(&r(K::Resolution, &moved_key(INSTANCE))).unwrap().is_none(), "nothing was written"); + + committed(admit(&mut mind, vec![moved_to(RECEIVER)])); + let withdrawal = resolution(r(K::Resolution, &moved_key(RECEIVER)), withdrawn()); + assert_eq!(refusal(admit(&mut mind, vec![withdrawal])), MindRefusal::MoveIsFinal); + assert!(moved_by(&view_of(&mind, K::Question, &q1).status, &moved_key(RECEIVER)), "the move still closes the campaign"); +} + +#[test] +fn import_replays_a_receipt_without_derivation() { + let mut mind = receiver(); + // A ruling of another root that answered this question in the giver: the derived Answered is + // rooted at the question's campaign, so a replay of the ruling's root does not carry it. The + // receiver does not derive it back. + let q1 = id("question", "Q1"); + let ruling = answering(OTHER_ROOT, "R1", &q1); + let before = head(&mind); + let (_, writes) = committed(import(&mut mind, vec![ruling.clone()], vec![r(K::Question, &q1)])); + assert_eq!(writes, vec![r(K::Ruling, &pipeline_key(&ruling).unwrap())], "only what the source wrote"); + assert_eq!(head(&mind), before + 1); + assert!(mind.view(&r(K::Resolution, &id("resolution", "question.Q1.n1"))).unwrap().is_none(), "no second resolution"); + + // The same shape through admission derives it, which is what import mode forgoes. + let mut plain = receiver(); + committed(admit(&mut plain, vec![question("Q1", &["A", "B"], "A")])); + let (_, writes) = committed(admit(&mut plain, vec![ruling])); + assert_eq!(writes.len(), 2, "an ordinary admission derives the Answered: {writes:?}"); +} + +#[test] +fn import_admits_a_declared_external_and_refuses_an_undeclared_one() { + let mut mind = receiver(); + committed(admit(&mut mind, vec![campaign(&[REPO]), question("Q1", &["A", "B"], "A"), follow_up("FU-1", r(K::Question, &id("question", "Q1")))])); + let report = r(K::CutReport, &format!("{OTHER_ROOT}:cut_report:cut-3.h1")); + let fixed = resolution( + r(K::FollowUp, &id("follow_up", "FU-1")), + ResolutionOutcome::Fixed { commit: sha(), by: Some(report.clone()) }, + ); + assert_eq!( + refusal(import(&mut mind, vec![fixed.clone()], vec![])), + MindRefusal::MissingReference { kind: K::CutReport, id: report.id.0.clone() } + ); + let (_, writes) = committed(import(&mut mind, vec![fixed.clone()], vec![report.clone()])); + assert_eq!(writes.len(), 1); + assert!(matches!(view_of(&mind, K::FollowUp, &id("follow_up", "FU-1")).status, PipelineStatus::Resolved { .. })); + assert!(mind.view(&report).unwrap().is_none(), "an external is stored nowhere"); + + // Outside an import, the same document is refused: externals are an import's alone. + let mut plain = receiver(); + committed(admit(&mut plain, vec![campaign(&[REPO]), question("Q1", &["A", "B"], "A"), follow_up("FU-1", r(K::Question, &id("question", "Q1")))])); + assert!(matches!(refusal(admit(&mut plain, vec![fixed])), MindRefusal::MissingReference { .. })); +} + +#[test] +fn import_refuses_a_present_external_a_moved_outcome_and_its_own_instance() { + let mut mind = receiver(); + committed(admit(&mut mind, vec![campaign(&[REPO]), question("Q1", &["A", "B"], "A")])); + let held = r(K::Question, &id("question", "Q1")); + let stray = follow_up("FU-1", held.clone()); + // Held in the image. + assert_eq!( + refusal(import(&mut mind, vec![stray.clone()], vec![held.clone()])), + MindRefusal::ExternalPresent { id: held.id.0.clone() } + ); + // Carried by the batch. + let carried = r(K::FollowUp, &id("follow_up", "FU-1")); + assert_eq!( + refusal(import(&mut mind, vec![stray], vec![carried.clone()])), + MindRefusal::ExternalPresent { id: carried.id.0.clone() } + ); + // The giver's resolution is never replayed. + assert_eq!( + refusal(import(&mut mind, vec![moved_to("kiko")], vec![])), + MindRefusal::IncompatibleResolution { subject_kind: K::Campaign, outcome: "Moved".into() } + ); + // A source that is this mind. + let own = import_as(&mut mind, source_of(RECEIVER, vec![]), vec![follow_up("FU-1", held)]); + assert_eq!(refusal(own), MindRefusal::ImportFromSelf); +} + +#[test] +fn replaying_a_whole_campaign_reproduces_it() { + let (mut giver, last) = giver(); + committed(admit(&mut giver, vec![moved_to(RECEIVER)])); + let moved_at = head(&giver); + assert_eq!(moved_at, last + 1); + + let mut mind = receiver(); + let outcomes = replay(&giver, &mut mind, last); + assert!(outcomes.len() >= 8, "{} receipts replayed", outcomes.len()); + for outcome in &outcomes { + assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}"); + } + + let before_the_move = Reader::at(&giver, last).unwrap(); + let mut compared = 0; + for kind in K::ALL { + for (key, _) in giver.image().unwrap().of_kind(*kind) { + if root_of(key) != CAMPAIGN || *kind == K::Resolution && key == moved_key(RECEIVER) { + continue; + } + let at = before_the_move.view(&r(*kind, key)).unwrap().unwrap(); + let there = view_of(&mind, *kind, key); + assert_eq!(there.document, at.document, "{key}"); + assert_eq!(there.status, at.status, "{key} reads as it did before the move"); + let source = there.admission.provenance.imported_from.expect("every replayed document names its source"); + assert_eq!(source.instance.0, INSTANCE); + assert_eq!(source.ordinal, at.admission.ordinal, "{key} names the giver's receipt"); + assert_eq!(source.receipt_id.0, at.admission.receipt_id); + compared += 1; + } + } + let held: usize = K::ALL.iter().map(|kind| mind.image().unwrap().of_kind(*kind).filter(|(key, _)| root_of(key) == CAMPAIGN).count()).sum(); + assert_eq!(compared, held, "the receiver holds exactly the documents compared, and the giver's all of them but the move"); + assert!(compared >= 14, "{compared} documents compared"); + assert!( + mind.image().unwrap().of_kind(K::Resolution).all(|(key, _)| key != moved_key(RECEIVER)), + "the move itself is not replayed" + ); + + // The receiver now keeps the campaign: its leads are hers and the giver's are gone. + assert_eq!(open_follow_ups(&mind), vec![id("follow_up", "FU-1")]); + assert!(open_follow_ups(&giver).is_empty()); + + // The source survives a restart: it is on the stored receipt. + let reopened = Mind::open_with(mind.store().clone(), &slug(RECEIVER)).unwrap(); + let key = id("question", "Q1"); + let source = reopened.view(&r(K::Question, &key)).unwrap().unwrap().admission.provenance.imported_from; + assert_eq!(source.map(|source| source.instance.0), Some(INSTANCE.to_string())); +} + +#[test] +fn an_import_replayed_twice_is_already_admitted() { + let mut mind = receiver(); + let documents = vec![campaign(&[REPO]), target(1, &[INVARIANT])]; + let (receipt_id, _) = committed(import(&mut mind, documents.clone(), vec![])); + let at = head(&mind); + assert_eq!(import(&mut mind, documents, vec![]), PipelineAdmissionOutcome::AlreadyAdmitted { receipt_id }); + assert_eq!(head(&mind), at, "the replay wrote nothing"); +} + +#[test] +fn receipts_without_imported_from_keep_their_bytes_and_digest() { + #[derive(serde::Serialize, serde::Deserialize)] + struct Before { + faculty: Faculty, + agent: eureka_pipeline::Short, + session: eureka_pipeline::Short, + tool: eureka_pipeline::Short, + } + let before = Before { faculty: Faculty::Hands, agent: s("claude"), session: s("session-1"), tool: s("admit") }; + let bytes = rmp_serde::to_vec_named(&before).unwrap(); + assert_eq!(rmp_serde::to_vec_named(&provenance(Faculty::Hands)).unwrap(), bytes, "a non-import provenance serialises as it did"); + let read: PipelineProvenance = rmp_serde::from_slice(&bytes).unwrap(); + assert_eq!(read, provenance(Faculty::Hands), "and a stored one reads back with no source"); + + // Who wrote it never enters the digest: the same batch imported and plain is one receipt. + let mut plain = receiver(); + let mut imported = receiver(); + let documents = vec![campaign(&[REPO])]; + let (plain_id, _) = committed(admit(&mut plain, documents.clone())); + let (imported_id, _) = committed(import(&mut imported, documents, vec![])); + assert_eq!(plain_id, imported_id); +} + +#[test] +fn an_imports_source_is_bounded_like_the_rest_of_provenance() { + let mut mind = receiver(); + let documents = vec![campaign(&[REPO])]; + let field = |outcome: PipelineAdmissionOutcome| match refusal(outcome) { + MindRefusal::Document(eureka_pipeline::PipelineRefusal::InvalidFormat { field, .. }) + | MindRefusal::Document(eureka_pipeline::PipelineRefusal::FieldBound { field, .. }) => field, + other => panic!("{other:?}"), + }; + let with = |edit: &dyn Fn(&mut ImportSource)| { + let mut source = source_of(INSTANCE, vec![]); + edit(&mut source); + source + }; + let bad = [ + ("provenance.imported_from.instance", with(&|source| source.instance = slug("Not A Slug!"))), + ("provenance.imported_from.agent", with(&|source| source.agent = s(" "))), + ("provenance.imported_from.session", with(&|source| source.session = s("two\nlines"))), + ("provenance.imported_from.receipt_id", with(&|source| source.receipt_id = s(&"x".repeat(201)))), + ("provenance.imported_from.external", with(&|source| source.external = vec![r(K::Question, "not an id")])), + ( + "provenance.imported_from.external", + with(&|source| source.external = (0..=IMPORT_EXTERNAL_MAX).map(|n| r(K::Question, &id("question", &format!("Q{n}")))).collect()), + ), + ]; + for (named, source) in bad { + assert_eq!(field(import_as(&mut mind, source, documents.clone())), named); + assert_eq!(head(&mind), 1, "{named}: a refused import wrote"); + } + let exact = with(&|source| source.external = (0..IMPORT_EXTERNAL_MAX).map(|n| r(K::Question, &id("question", &format!("Q{n}")))).collect()); + committed(import_as(&mut mind, exact, documents)); +} diff --git a/crates/huginn-mind/src/oracle.rs b/crates/huginn-mind/src/oracle.rs index 55b3b5d..4066d94 100644 --- a/crates/huginn-mind/src/oracle.rs +++ b/crates/huginn-mind/src/oracle.rs @@ -110,8 +110,25 @@ fn closing<'a>(visible: &[&'a Doc], kind: PipelineKind, key: &str) -> Option<(&' }) } +/// The Moved resolution in force over the campaign `key` is rooted at, unless `key` is that resolution. +fn moved_over<'a>(visible: &[&'a Doc], kind: PipelineKind, key: &str) -> Option<(&'a str, &'a PipelineResolution)> { + let root = key.split(':').next(); + visible.iter().find_map(|doc| match &doc.document { + PipelineDocument::Resolution(resolution) + if matches!(resolution.outcome, ResolutionOutcome::Moved { .. }) + && resolution.subject.kind == PipelineKind::Campaign + && resolution.subject.id.0.split(':').next() == root + && closing(visible, PipelineKind::Resolution, &doc.key).is_none() + && !(kind == PipelineKind::Resolution && key == doc.key) => + { + Some((doc.key.as_str(), resolution)) + } + _ => None, + }) +} + fn view_of(visible: &[&Doc], doc: &Doc) -> PipelineDocumentView { - let status = match closing(visible, doc.kind, &doc.key) { + let status = match closing(visible, doc.kind, &doc.key).or_else(|| moved_over(visible, doc.kind, &doc.key)) { Some((key, record)) => PipelineStatus::Resolved { resolution: PipelineRef { kind: PipelineKind::Resolution, id: Short(key.to_string()) }, record: record.clone(), @@ -278,6 +295,7 @@ pub fn synth_history(store: &S, instance: &Slug, factor: u32, spre agent: Short("synth".into()), session: Short("synth".into()), tool: Short("synth".into()), + imported_from: None, }; let now = Utc.with_ymd_and_hms(2026, 10, 1, 12, 0, 0).unwrap(); let mut pending: Vec> = Vec::new(); diff --git a/crates/huginn-mind/src/query.rs b/crates/huginn-mind/src/query.rs index 670e345..cb4e79d 100644 --- a/crates/huginn-mind/src/query.rs +++ b/crates/huginn-mind/src/query.rs @@ -2499,6 +2499,7 @@ mod tests { agent: s(&long("a", 200)), session: s(&long("s", 200)), tool: s(&long("t", 200)), + imported_from: None, }; let batch = crate::admission::PipelineAdmissionBatch { instance: slug(INSTANCE), diff --git a/crates/huginn-mind/src/read_tests.rs b/crates/huginn-mind/src/read_tests.rs index 14837df..aa14861 100644 --- a/crates/huginn-mind/src/read_tests.rs +++ b/crates/huginn-mind/src/read_tests.rs @@ -250,6 +250,31 @@ fn views_and_status_agree_with_the_oracle_at_every_snapshot() { assert_eq!(reopened.persona().unwrap(), mind.persona().unwrap()); } +/// A Moved resolution closes every document under its campaign's root and names none of them: the +/// held image's fan-out and the oracle's scan are two derivations of the same status, at every snapshot. +#[test] +fn a_moved_campaign_reads_the_same_held_and_fresh_at_every_snapshot() { + let mut mind = history_mind(); + let moved = resolution(r(K::Campaign, &id("campaign", "self")), eureka_pipeline::ResolutionOutcome::Moved { to: slug("heimdall") }); + committed(admit(&mut mind, vec![moved])); + held_equals_fresh(&mind); + let head = mind.image().unwrap().head; + let every = oracle_ids(&mind, head).unwrap(); + for as_of in 1..=head { + let reader = Reader::at(&mind, as_of).unwrap(); + for id in &every { + assert_eq!(reader.view(id).unwrap(), oracle_view(&mind, id, as_of).unwrap(), "{} at {as_of}", id.id.0); + } + } + let closed = mind.view(&r(K::Question, &id("question", "Q2"))).unwrap().unwrap().status; + assert!(matches!(closed, crate::query::PipelineStatus::Resolved { .. }), "the history's open question reads closed: {closed:?}"); + for selection in corpus(&mind) { + for as_of in [head - 1, head] { + let _ = agree(&mind, &selection, as_of); + } + } +} + #[test] fn a_restart_answers_what_the_held_image_answered() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/huginn-mind/src/receipt.rs b/crates/huginn-mind/src/receipt.rs index 4201958..e929cd4 100644 --- a/crates/huginn-mind/src/receipt.rs +++ b/crates/huginn-mind/src/receipt.rs @@ -50,8 +50,33 @@ pub struct PipelineProvenance { pub agent: eureka_pipeline::Short, pub session: eureka_pipeline::Short, pub tool: eureka_pipeline::Short, + /// Set when the batch replays another mind's receipt: the source it quotes. Absent on every + /// other admission, and then absent from the stored receipt too. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub imported_from: Option, } +/// The receipt a batch replays and the mind it came from: a quotation of the source receipt, which +/// stays the primary record. `faculty`, `agent`, `session` and `tool` are the source's, the batch's +/// own being the importer's. `external` lists the referents the batch cites that this mind does not +/// hold and does not import: each is treated as in force and nothing reads its content. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +pub struct ImportSource { + pub instance: eureka_pipeline::Slug, + pub receipt_id: eureka_pipeline::Short, + pub ordinal: u64, + pub committed_at: eureka_pipeline::Short, + pub faculty: Faculty, + pub agent: eureka_pipeline::Short, + pub session: eureka_pipeline::Short, + pub tool: eureka_pipeline::Short, + #[schemars(extend("maxItems" = 256))] + pub external: Vec, +} + +/// The most external referents one import may declare. +pub const IMPORT_EXTERNAL_MAX: usize = 256; + /// The exact bytes of one stored document, as read or as written. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] pub struct DocumentVersion { diff --git a/crates/huginn-mind/src/refusal.rs b/crates/huginn-mind/src/refusal.rs index da447df..5613547 100644 --- a/crates/huginn-mind/src/refusal.rs +++ b/crates/huginn-mind/src/refusal.rs @@ -58,6 +58,16 @@ pub enum MindRefusal { FindingWithoutEvidence, UnknownInvariant { label: String }, NotStewarded { repo: String }, + /// `campaign` moved to mind `to`: nothing is written under its root here any more, whoever asks. + CampaignMoved { campaign: String, to: String }, + /// A campaign moves to another mind, never to the one that gives it up. + MoveToSelf, + /// A Moved resolution is final; withdrawing it would leave two minds keeping the campaign. + MoveIsFinal, + /// An import names this mind as its source. + ImportFromSelf, + /// An import lists `id` as external, but this mind (or the batch) holds it. + ExternalPresent { id: String }, /// The persona document fails the published schema: `path` is the /// document location (a key the schema does not declare is `*`) and /// `message` the schema keyword that failed. Never the offending value. diff --git a/crates/huginn-mind/src/rows.rs b/crates/huginn-mind/src/rows.rs index 425eac3..ed5f344 100644 --- a/crates/huginn-mind/src/rows.rs +++ b/crates/huginn-mind/src/rows.rs @@ -64,7 +64,7 @@ pub(crate) enum Alias { } const FACULTIES: [&str; 7] = ["SelfFaculty", "Imagination", "Hands", "Soul", "Life", "Eyes", "Operator"]; -const OUTCOMES: [&str; 7] = ["Superseded", "Answered", "Fixed", "Deferred", "Recorded", "Withdrawn", "Slept"]; +const OUTCOMES: [&str; 8] = ["Superseded", "Answered", "Fixed", "Deferred", "Recorded", "Withdrawn", "Slept", "Moved"]; const BOOLEANS: [&str; 2] = ["true", "false"]; const MEMORY_ORIGINS: [&str; 5] = ["Room", "Direct", "Wake", "SelfRun", "Sleep"]; /// The one value a presence alias answers with; a row without it answers nothing. @@ -200,6 +200,7 @@ fn outcome_name(outcome: &ResolutionOutcome) -> &'static str { ResolutionOutcome::Recorded { .. } => "Recorded", ResolutionOutcome::Withdrawn { .. } => "Withdrawn", ResolutionOutcome::Slept { .. } => "Slept", + ResolutionOutcome::Moved { .. } => "Moved", } } diff --git a/crates/huginn-mind/tests/read_floor.rs b/crates/huginn-mind/tests/read_floor.rs index b030376..046bab1 100644 --- a/crates/huginn-mind/tests/read_floor.rs +++ b/crates/huginn-mind/tests/read_floor.rs @@ -154,7 +154,7 @@ fn read_floor() { tempo_vote_withdrawn: false, urgency_permille: Some(1), }); - let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: Short("floor".into()), session: Short("floor".into()), tool: Short("floor".into()) }; + let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: Short("floor".into()), session: Short("floor".into()), tool: Short("floor".into()), imported_from: None }; let started = Instant::now(); let outcome = mind.admit(PipelineAdmissionBatch { instance: instance.clone(), provenance, documents: vec![pulse] }, Utc::now()); let admit = started.elapsed(); diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index 5b5aa96..54f5f06 100644 --- a/crates/mind-body/src/persona.rs +++ b/crates/mind-body/src/persona.rs @@ -2639,6 +2639,7 @@ pub(crate) fn outcome_words(outcome: &ResolutionOutcome) -> String { ResolutionOutcome::Fixed { commit, .. } => format!("fixed by {}", commit.0), ResolutionOutcome::Deferred { to } => format!("deferred to {}", to.id.0), ResolutionOutcome::Slept { line, .. } => format!("slept: {}", line.0), + ResolutionOutcome::Moved { to } => format!("moved to {}", to.0), } } diff --git a/crates/mind-body/src/queue.rs b/crates/mind-body/src/queue.rs index 8533ed3..45d41cb 100644 --- a/crates/mind-body/src/queue.rs +++ b/crates/mind-body/src/queue.rs @@ -121,7 +121,7 @@ impl MindPort for HuginnMind<'_> { fn admit(&self, agent: &str, session: &str, documents: Vec) -> Result { let batch = PipelineAdmissionBatch { instance: self.instance().clone(), - provenance: PipelineProvenance { faculty: Faculty::SelfFaculty, agent: agent.into(), session: session.into(), tool: "mind-body".into() }, + provenance: PipelineProvenance { faculty: Faculty::SelfFaculty, agent: agent.into(), session: session.into(), tool: "mind-body".into(), imported_from: None }, documents, }; match (self.call)(HuginnMindRequest::Admit(batch)).map_err(|error| anyhow!("the mind could not be asked: {}", client_error_code(&error)))? { diff --git a/crates/mind-body/src/testkit.rs b/crates/mind-body/src/testkit.rs index 33e9aa9..ec04a2a 100644 --- a/crates/mind-body/src/testkit.rs +++ b/crates/mind-body/src/testkit.rs @@ -149,7 +149,7 @@ impl TestMind { self.now.set(self.now.get() + Duration::seconds(1)); let batch = PipelineAdmissionBatch { instance: self.instance.clone(), - provenance: PipelineProvenance { faculty, agent: "test".into(), session: "test".into(), tool: "test".into() }, + provenance: PipelineProvenance { faculty, agent: "test".into(), session: "test".into(), tool: "test".into(), imported_from: None }, documents, }; self.mind.borrow_mut().admit(batch, self.now.get()) @@ -231,7 +231,7 @@ impl MindPort for TestMind { self.now.set(self.now.get() + Duration::seconds(1)); let batch = PipelineAdmissionBatch { instance: self.instance.clone(), - provenance: PipelineProvenance { faculty: Faculty::SelfFaculty, agent: agent.into(), session: session.into(), tool: "mind-body".into() }, + provenance: PipelineProvenance { faculty: Faculty::SelfFaculty, agent: agent.into(), session: session.into(), tool: "mind-body".into(), imported_from: None }, documents, }; Ok(self.mind.borrow_mut().admit(batch, self.now.get())) diff --git a/crates/mind-body/tests/self_run.rs b/crates/mind-body/tests/self_run.rs index fd2e7c9..194e246 100644 --- a/crates/mind-body/tests/self_run.rs +++ b/crates/mind-body/tests/self_run.rs @@ -198,7 +198,7 @@ mod lane { let mut mind = Mind::open(state, &slug).unwrap(); let batch = PipelineAdmissionBatch { instance: slug.clone(), - provenance: PipelineProvenance { faculty: Faculty::Hands, agent: "test".into(), session: "test".into(), tool: "test".into() }, + provenance: PipelineProvenance { faculty: Faculty::Hands, agent: "test".into(), session: "test".into(), tool: "test".into(), imported_from: None }, documents, }; assert!(matches!(mind.admit(batch, Utc::now()), PipelineAdmissionOutcome::Committed { .. })); diff --git a/schemas/cultnet/huginn.mind_request.v1.schema.json b/schemas/cultnet/huginn.mind_request.v1.schema.json index a67e159..8f6f080 100644 --- a/schemas/cultnet/huginn.mind_request.v1.schema.json +++ b/schemas/cultnet/huginn.mind_request.v1.schema.json @@ -551,6 +551,56 @@ "type": "string", "maxLength": 40 }, + "ImportSource": { + "description": "The receipt a batch replays and the mind it came from: a quotation of the source receipt, which\nstays the primary record. `faculty`, `agent`, `session` and `tool` are the source's, the batch's\nown being the importer's. `external` lists the referents the batch cites that this mind does not\nhold and does not import: each is treated as in force and nothing reads its content.", + "type": "object", + "properties": { + "agent": { + "$ref": "#/$defs/Short" + }, + "committed_at": { + "$ref": "#/$defs/Short" + }, + "external": { + "type": "array", + "items": { + "$ref": "#/$defs/PipelineRef" + }, + "maxItems": 256 + }, + "faculty": { + "$ref": "#/$defs/Faculty" + }, + "instance": { + "$ref": "#/$defs/Slug" + }, + "ordinal": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "receipt_id": { + "$ref": "#/$defs/Short" + }, + "session": { + "$ref": "#/$defs/Short" + }, + "tool": { + "$ref": "#/$defs/Short" + } + }, + "required": [ + "instance", + "receipt_id", + "ordinal", + "committed_at", + "faculty", + "agent", + "session", + "tool", + "external" + ] + }, "Label": { "description": "A key label: `[A-Za-z0-9_-]{1,64}`.", "type": "string", @@ -1701,6 +1751,17 @@ "faculty": { "$ref": "#/$defs/Faculty" }, + "imported_from": { + "description": "Set when the batch replays another mind's receipt: the source it quotes. Absent on every\nother admission, and then absent from the stored receipt too.", + "anyOf": [ + { + "$ref": "#/$defs/ImportSource" + }, + { + "type": "null" + } + ] + }, "session": { "$ref": "#/$defs/Short" }, @@ -2640,6 +2701,27 @@ "required": [ "Slept" ] + }, + { + "description": "A campaign's history left this mind for `to`, the mind that now keeps it. Final: the campaign and every\ndocument under its root read resolved by this and take no further writes here.", + "type": "object", + "properties": { + "Moved": { + "type": "object", + "properties": { + "to": { + "$ref": "#/$defs/Slug" + } + }, + "required": [ + "to" + ] + } + }, + "additionalProperties": false, + "required": [ + "Moved" + ] } ] }, diff --git a/schemas/cultnet/huginn.mind_response.v1.schema.json b/schemas/cultnet/huginn.mind_response.v1.schema.json index 33a5915..3df0115 100644 --- a/schemas/cultnet/huginn.mind_response.v1.schema.json +++ b/schemas/cultnet/huginn.mind_response.v1.schema.json @@ -558,6 +558,56 @@ "type": "string", "maxLength": 40 }, + "ImportSource": { + "description": "The receipt a batch replays and the mind it came from: a quotation of the source receipt, which\nstays the primary record. `faculty`, `agent`, `session` and `tool` are the source's, the batch's\nown being the importer's. `external` lists the referents the batch cites that this mind does not\nhold and does not import: each is treated as in force and nothing reads its content.", + "type": "object", + "properties": { + "agent": { + "$ref": "#/$defs/Short" + }, + "committed_at": { + "$ref": "#/$defs/Short" + }, + "external": { + "type": "array", + "items": { + "$ref": "#/$defs/PipelineRef" + }, + "maxItems": 256 + }, + "faculty": { + "$ref": "#/$defs/Faculty" + }, + "instance": { + "$ref": "#/$defs/Slug" + }, + "ordinal": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "receipt_id": { + "$ref": "#/$defs/Short" + }, + "session": { + "$ref": "#/$defs/Short" + }, + "tool": { + "$ref": "#/$defs/Short" + } + }, + "required": [ + "instance", + "receipt_id", + "ordinal", + "committed_at", + "faculty", + "agent", + "session", + "tool", + "external" + ] + }, "IndexStatus": { "description": "How far the semantic index has caught up with the mind. The mind decides\nnone of it: the daemon's index worker reports it, in memory, and a restart\nrecomputes it. `pending` counts documents not yet known to be in the index.", "oneOf": [ @@ -1586,6 +1636,67 @@ "NotStewarded" ] }, + { + "description": "`campaign` moved to mind `to`: nothing is written under its root here any more, whoever asks.", + "type": "object", + "properties": { + "CampaignMoved": { + "type": "object", + "properties": { + "campaign": { + "type": "string" + }, + "to": { + "type": "string" + } + }, + "required": [ + "campaign", + "to" + ] + } + }, + "additionalProperties": false, + "required": [ + "CampaignMoved" + ] + }, + { + "description": "A campaign moves to another mind, never to the one that gives it up.", + "type": "string", + "const": "MoveToSelf" + }, + { + "description": "A Moved resolution is final; withdrawing it would leave two minds keeping the campaign.", + "type": "string", + "const": "MoveIsFinal" + }, + { + "description": "An import names this mind as its source.", + "type": "string", + "const": "ImportFromSelf" + }, + { + "description": "An import lists `id` as external, but this mind (or the batch) holds it.", + "type": "object", + "properties": { + "ExternalPresent": { + "type": "object", + "properties": { + "id": { + "type": "string" + } + }, + "required": [ + "id" + ] + } + }, + "additionalProperties": false, + "required": [ + "ExternalPresent" + ] + }, { "description": "The persona document fails the published schema: `path` is the\ndocument location (a key the schema does not declare is `*`) and\n`message` the schema keyword that failed. Never the offending value.", "type": "object", @@ -3996,6 +4107,17 @@ "faculty": { "$ref": "#/$defs/Faculty" }, + "imported_from": { + "description": "Set when the batch replays another mind's receipt: the source it quotes. Absent on every\nother admission, and then absent from the stored receipt too.", + "anyOf": [ + { + "$ref": "#/$defs/ImportSource" + }, + { + "type": "null" + } + ] + }, "session": { "$ref": "#/$defs/Short" }, @@ -5357,6 +5479,27 @@ "required": [ "Slept" ] + }, + { + "description": "A campaign's history left this mind for `to`, the mind that now keeps it. Final: the campaign and every\ndocument under its root read resolved by this and take no further writes here.", + "type": "object", + "properties": { + "Moved": { + "type": "object", + "properties": { + "to": { + "$ref": "#/$defs/Slug" + } + }, + "required": [ + "to" + ] + } + }, + "additionalProperties": false, + "required": [ + "Moved" + ] } ] },