huginn-mind: import::plan lays out a moved campaign's replay, and the move tests replay through it
The verb and the tests share one definition of a replay batch: grouped by source receipt ordinal, the Moved dropped from its own batch, foreign referents listed, oversized or unordered receipts refused. Deletes the hand-built replay filter in move_tests.rs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
0d4cccc157
commit
d215a6fa03
3 changed files with 315 additions and 30 deletions
284
crates/huginn-mind/src/import.rs
Normal file
284
crates/huginn-mind/src/import.rs
Normal file
|
|
@ -0,0 +1,284 @@
|
|||
//! The planner of a campaign import: what the receiver of a moved campaign replays from the
|
||||
//! giver, and in what batches.
|
||||
//!
|
||||
//! `plan` is a function of the giver's views of the campaign's root and nothing else, so the
|
||||
//! `huginn import` verb and the move tests share one definition of a replay batch. It reads no
|
||||
//! mind and writes none: the verb sends what it returns, and the receiver's admission decides
|
||||
//! every write.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
|
||||
use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, ResolutionOutcome, Short, Slug};
|
||||
|
||||
use crate::admission::BATCH_MAX;
|
||||
use crate::docs::citations;
|
||||
use crate::query::{PipelineDocumentView, PipelineStatus};
|
||||
use crate::receipt::{IMPORT_EXTERNAL_MAX, ImportSource};
|
||||
use crate::rows::root_and_local;
|
||||
|
||||
/// A moved campaign's history as the receiver replays it: the Moved resolution that moved it, and
|
||||
/// one batch per source receipt in ascending source ordinal.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct ImportPlan {
|
||||
pub moved: PipelineRef,
|
||||
pub batches: Vec<ImportBatch>,
|
||||
}
|
||||
|
||||
/// One source receipt's documents under the root, minus the Moved resolution. `source` quotes the
|
||||
/// receipt with no externals: the caller declares the `foreign` referents the receiver does not hold.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct ImportBatch {
|
||||
pub source: ImportSource,
|
||||
pub documents: Vec<(PipelineRef, PipelineDocument)>,
|
||||
/// Referents the batch cites under another root, in id order.
|
||||
pub foreign: Vec<PipelineRef>,
|
||||
}
|
||||
|
||||
/// Why a campaign cannot be replayed from these views.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum ImportFault {
|
||||
/// No in-force Moved resolution closes the campaign.
|
||||
NotMoved,
|
||||
/// The campaign moved to another mind.
|
||||
MovedElsewhere { to: Slug },
|
||||
/// The source is the receiver.
|
||||
SourceIsReceiver,
|
||||
/// A source receipt wrote more documents under the root than one batch holds.
|
||||
BatchTooLarge { ordinal: u64, documents: usize },
|
||||
/// A batch cites more foreign referents than one import may declare.
|
||||
ExternalsTooMany { ordinal: u64, count: usize },
|
||||
/// A batch cites a document of its own root that no earlier receipt wrote.
|
||||
Unordered { ordinal: u64, id: Short },
|
||||
}
|
||||
|
||||
/// The plan for `receiver` to import `campaign` from `source`, given every document the source
|
||||
/// holds under the campaign's root.
|
||||
pub fn plan(campaign: &Slug, source: &Slug, receiver: &Slug, views: Vec<PipelineDocumentView>) -> Result<ImportPlan, ImportFault> {
|
||||
let under_root = |view: &&PipelineDocumentView| root_and_local(&view.id.id.0).0 == campaign.0;
|
||||
let (moved, to) = views
|
||||
.iter()
|
||||
.filter(under_root)
|
||||
.find_map(|view| match &view.document {
|
||||
PipelineDocument::Resolution(resolution) if view.status == PipelineStatus::InForce && resolution.subject.kind == PipelineKind::Campaign => {
|
||||
match &resolution.outcome {
|
||||
ResolutionOutcome::Moved { to } => Some((view.id.clone(), to.clone())),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
_ => None,
|
||||
})
|
||||
.ok_or(ImportFault::NotMoved)?;
|
||||
if to != *receiver {
|
||||
return Err(ImportFault::MovedElsewhere { to });
|
||||
}
|
||||
if source == receiver {
|
||||
return Err(ImportFault::SourceIsReceiver);
|
||||
}
|
||||
|
||||
let mut receipts: BTreeMap<u64, Vec<&PipelineDocumentView>> = BTreeMap::new();
|
||||
for view in views.iter().filter(under_root).filter(|view| view.id != moved) {
|
||||
receipts.entry(view.admission.ordinal).or_default().push(view);
|
||||
}
|
||||
let mut written: BTreeSet<&str> = BTreeSet::new();
|
||||
let mut batches = Vec::new();
|
||||
for (ordinal, mut group) in receipts {
|
||||
group.sort_by(|a, b| a.id.id.0.cmp(&b.id.id.0));
|
||||
if group.len() > BATCH_MAX {
|
||||
return Err(ImportFault::BatchTooLarge { ordinal, documents: group.len() });
|
||||
}
|
||||
let in_batch: BTreeSet<&str> = group.iter().map(|view| view.id.id.0.as_str()).collect();
|
||||
let mut foreign: BTreeMap<&str, PipelineRef> = BTreeMap::new();
|
||||
for view in &group {
|
||||
for (_, target) in cited(&view.document, receiver) {
|
||||
let key = target.id.0.as_str();
|
||||
if in_batch.contains(key) || written.contains(key) {
|
||||
continue;
|
||||
}
|
||||
if root_and_local(key).0 == campaign.0 {
|
||||
return Err(ImportFault::Unordered { ordinal, id: Short(key.to_string()) });
|
||||
}
|
||||
foreign.entry(key).or_insert_with(|| target.clone());
|
||||
}
|
||||
}
|
||||
if foreign.len() > IMPORT_EXTERNAL_MAX {
|
||||
return Err(ImportFault::ExternalsTooMany { ordinal, count: foreign.len() });
|
||||
}
|
||||
let facts = &group[0].admission;
|
||||
batches.push(ImportBatch {
|
||||
source: ImportSource {
|
||||
instance: source.clone(),
|
||||
receipt_id: Short(facts.receipt_id.clone()),
|
||||
ordinal,
|
||||
committed_at: Short(facts.admitted_at.clone()),
|
||||
faculty: facts.provenance.faculty,
|
||||
agent: facts.provenance.agent.clone(),
|
||||
session: facts.provenance.session.clone(),
|
||||
tool: facts.provenance.tool.clone(),
|
||||
external: Vec::new(),
|
||||
},
|
||||
documents: group.iter().map(|view| (view.id.clone(), view.document.clone())).collect(),
|
||||
foreign: foreign.into_values().collect(),
|
||||
});
|
||||
written.extend(in_batch);
|
||||
}
|
||||
Ok(ImportPlan { moved, batches })
|
||||
}
|
||||
|
||||
/// What admission requires a document's referents to be. A hand-off cites its documents on its
|
||||
/// source side only (`admission::references`), so one from another mind cites nothing here.
|
||||
fn cited(document: &PipelineDocument, receiver: &Slug) -> Vec<(crate::CitationRole, PipelineRef)> {
|
||||
match document {
|
||||
PipelineDocument::HandOff(hand_off) if hand_off.from_instance != *receiver => Vec::new(),
|
||||
_ => citations(document),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use eureka_pipeline::{PipelineKind as K, pipeline_key};
|
||||
|
||||
use super::*;
|
||||
use crate::fixtures::*;
|
||||
use crate::query::AdmissionFacts;
|
||||
use crate::receipt::{Faculty, PipelineProvenance};
|
||||
|
||||
const RECEIVER: &str = "heimdall";
|
||||
|
||||
fn view(document: PipelineDocument, ordinal: u64) -> PipelineDocumentView {
|
||||
let key = pipeline_key(&document).unwrap();
|
||||
let id = PipelineRef { kind: document.kind(), id: s(&key) };
|
||||
let faculty = if ordinal % 2 == 0 { Faculty::Hands } else { Faculty::Soul };
|
||||
let admission = AdmissionFacts {
|
||||
receipt_id: format!("receipt-{ordinal}"),
|
||||
admitted_at: format!("2026-11-08T16:{ordinal:02}:00Z"),
|
||||
provenance: PipelineProvenance { faculty, agent: s(&format!("agent-{ordinal}")), session: s(&format!("session-{ordinal}")), tool: s("huginn"), imported_from: None },
|
||||
ordinal,
|
||||
};
|
||||
PipelineDocumentView { id, document, admission, status: PipelineStatus::InForce }
|
||||
}
|
||||
|
||||
fn moved(to: &str, ordinal: u64) -> PipelineDocumentView {
|
||||
view(resolution(r(K::Campaign, &id("campaign", "self")), ResolutionOutcome::Moved { to: slug(to) }), ordinal)
|
||||
}
|
||||
|
||||
fn plan_of(views: Vec<PipelineDocumentView>) -> Result<ImportPlan, ImportFault> {
|
||||
plan(&slug(CAMPAIGN), &slug(INSTANCE), &slug(RECEIVER), views)
|
||||
}
|
||||
|
||||
fn q(label: &str) -> PipelineDocument {
|
||||
question(label, &["A", "B"], "A")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plan_groups_by_source_ordinal_and_drops_only_the_moved() {
|
||||
let fu = follow_up("FU-1", r(K::Question, &id("question", "Q1")));
|
||||
// The Moved receipt also wrote a follow-up under the root; views arrive in no order.
|
||||
let views = vec![moved(RECEIVER, 9), view(q("Q2"), 3), view(fu.clone(), 9), view(q("Q1"), 3), view(q("Q3"), 5)];
|
||||
let plan = plan_of(views).unwrap();
|
||||
assert_eq!(plan.moved.id.0, pipeline_key(&resolution(r(K::Campaign, &id("campaign", "self")), ResolutionOutcome::Moved { to: slug(RECEIVER) })).unwrap());
|
||||
let shape: Vec<(u64, Vec<String>)> =
|
||||
plan.batches.iter().map(|batch| (batch.source.ordinal, batch.documents.iter().map(|(id, _)| id.id.0.clone()).collect())).collect();
|
||||
assert_eq!(
|
||||
shape,
|
||||
vec![
|
||||
(3, vec![id("question", "Q1"), id("question", "Q2")]),
|
||||
(5, vec![id("question", "Q3")]),
|
||||
(9, vec![pipeline_key(&fu).unwrap()]),
|
||||
],
|
||||
"ascending ordinals, keys in order, the follow-up of the Moved receipt replays and the Moved does not"
|
||||
);
|
||||
let nine = &plan.batches[2].source;
|
||||
assert_eq!(
|
||||
nine,
|
||||
&ImportSource {
|
||||
instance: slug(INSTANCE),
|
||||
receipt_id: s("receipt-9"),
|
||||
ordinal: 9,
|
||||
committed_at: s("2026-11-08T16:09:00Z"),
|
||||
faculty: Faculty::Soul,
|
||||
agent: s("agent-9"),
|
||||
session: s("session-9"),
|
||||
tool: s("huginn"),
|
||||
external: vec![],
|
||||
}
|
||||
);
|
||||
assert!(plan.batches.iter().all(|batch| batch.foreign.is_empty()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plan_refuses_unmoved_moved_elsewhere_and_self() {
|
||||
let held = vec![view(q("Q1"), 3)];
|
||||
assert_eq!(plan_of(held.clone()), Err(ImportFault::NotMoved));
|
||||
let mut withdrawn_move = moved(RECEIVER, 4);
|
||||
withdrawn_move.status = PipelineStatus::Resolved {
|
||||
resolution: r(K::Resolution, &id("resolution", "x")),
|
||||
record: match resolution(r(K::Campaign, &id("campaign", "self")), withdrawn()) {
|
||||
PipelineDocument::Resolution(record) => record,
|
||||
_ => unreachable!(),
|
||||
},
|
||||
};
|
||||
assert_eq!(plan_of([held.clone(), vec![withdrawn_move]].concat()), Err(ImportFault::NotMoved), "only an in-force Moved moves a campaign");
|
||||
let elsewhere = [held.clone(), vec![moved("kiko", 4)]].concat();
|
||||
assert_eq!(plan_of(elsewhere), Err(ImportFault::MovedElsewhere { to: slug("kiko") }));
|
||||
let to_source = [held.clone(), vec![moved(RECEIVER, 4)]].concat();
|
||||
assert_eq!(plan(&slug(CAMPAIGN), &slug(RECEIVER), &slug(RECEIVER), to_source), Err(ImportFault::SourceIsReceiver));
|
||||
let other_root = view(
|
||||
resolution(r(K::Campaign, &format!("elsewhere:campaign:self")), ResolutionOutcome::Moved { to: slug(RECEIVER) }),
|
||||
4,
|
||||
);
|
||||
assert_eq!(plan_of([held, vec![other_root]].concat()), Err(ImportFault::NotMoved), "another root's move does not move this one");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plan_refuses_an_oversized_receipt_before_anything() {
|
||||
let mut views = vec![moved(RECEIVER, 100), view(q("Q0"), 2)];
|
||||
views.extend((0..=BATCH_MAX).map(|n| view(q(&format!("B{n}")), 7)));
|
||||
assert_eq!(plan_of(views.clone()), Err(ImportFault::BatchTooLarge { ordinal: 7, documents: BATCH_MAX + 1 }));
|
||||
views.pop();
|
||||
let plan = plan_of(views).unwrap();
|
||||
assert_eq!(plan.batches[1].documents.len(), BATCH_MAX, "exactly one batch's worth replays");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plan_splits_foreign_from_in_root_referents() {
|
||||
let q1 = r(K::Question, &id("question", "Q1"));
|
||||
let abroad = r(K::Ruling, "elsewhere:ruling:R9");
|
||||
let answered = |question: &PipelineRef, by: &PipelineRef| resolution(question.clone(), ResolutionOutcome::Answered { by: by.clone() });
|
||||
let views = vec![
|
||||
moved(RECEIVER, 20),
|
||||
view(q("Q1"), 3),
|
||||
view(answered(&q1, &abroad), 4),
|
||||
view(follow_up("FU-1", q1.clone()), 5),
|
||||
];
|
||||
let plan = plan_of(views.clone()).unwrap();
|
||||
assert_eq!(plan.batches[1].foreign, vec![abroad.clone()], "another root's ruling is foreign");
|
||||
assert!(plan.batches[0].foreign.is_empty() && plan.batches[2].foreign.is_empty(), "an in-root referent of an earlier receipt is not foreign");
|
||||
|
||||
// The same follow-up before the question it cites: no earlier receipt wrote it.
|
||||
let backwards = vec![moved(RECEIVER, 20), view(follow_up("FU-1", q1.clone()), 2), view(q("Q1"), 3)];
|
||||
assert_eq!(plan_of(backwards), Err(ImportFault::Unordered { ordinal: 2, id: s(&q1.id.0) }));
|
||||
|
||||
// One receipt carrying both sides is ordered by being whole.
|
||||
let together = vec![moved(RECEIVER, 20), view(follow_up("FU-1", q1.clone()), 3), view(q("Q1"), 3)];
|
||||
assert!(plan_of(together).is_ok());
|
||||
|
||||
// More foreign referents than an import may declare.
|
||||
let many = (0..=IMPORT_EXTERNAL_MAX).map(|n| format!("elsewhere:ruling:R{n}")).collect::<Vec<_>>();
|
||||
let mut spec = cut_spec("1", 1);
|
||||
spec.rulings = many.iter().map(|key| s(key)).collect();
|
||||
let mut views = vec![moved(RECEIVER, 20), view(PipelineDocument::CutSpec(spec.clone()), 6)];
|
||||
assert_eq!(plan_of(views.clone()), Err(ImportFault::ExternalsTooMany { ordinal: 6, count: IMPORT_EXTERNAL_MAX + 1 }));
|
||||
spec.rulings.pop();
|
||||
views[1] = view(PipelineDocument::CutSpec(spec), 6);
|
||||
assert_eq!(plan_of(views).unwrap().batches[0].foreign.len(), IMPORT_EXTERNAL_MAX, "the limit itself is declarable");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plan_ignores_a_hand_off_cited_from_another_mind() {
|
||||
let q9 = id("question", "Q9");
|
||||
let from_libby = view(hand_off("libby", RECEIVER, REPO, &[&q9]), 3);
|
||||
assert!(plan_of(vec![moved(RECEIVER, 20), from_libby]).unwrap().batches[0].foreign.is_empty(), "the receiving side cites nothing");
|
||||
let from_receiver = view(hand_off(RECEIVER, "libby", REPO, &[&q9]), 3);
|
||||
assert_eq!(plan_of(vec![moved(RECEIVER, 20), from_receiver]), Err(ImportFault::Unordered { ordinal: 3, id: s(&q9) }));
|
||||
}
|
||||
}
|
||||
|
|
@ -30,6 +30,7 @@ pub mod admission;
|
|||
mod docs;
|
||||
mod image;
|
||||
pub mod envelope;
|
||||
pub mod import;
|
||||
pub mod index;
|
||||
pub mod mind;
|
||||
pub mod persona;
|
||||
|
|
@ -54,6 +55,7 @@ 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};
|
||||
pub use import::{ImportBatch, ImportFault, ImportPlan, plan};
|
||||
pub use mind::{HuginnMindEpoch, Mind};
|
||||
pub use persona::{PersonaPutOutcome, PersonaStateView};
|
||||
pub use query::{
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ 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::receipt::{Faculty, IMPORT_EXTERNAL_MAX, ImportSource, PipelineProvenance};
|
||||
use crate::refusal::MindRefusal;
|
||||
use crate::store::test_stores::MemoryStore;
|
||||
|
||||
|
|
@ -157,36 +157,35 @@ 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<MemoryStore>, receiver: &mut Mind<MemoryStore>, through: u64) -> Vec<PipelineAdmissionOutcome> {
|
||||
let mut receipts: Vec<HuginnCommitReceipt> = 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<D> = 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;
|
||||
/// Every document the giver holds under the campaign's root, as the verb reads them: one cursor
|
||||
/// walk of the document projection.
|
||||
fn giver_views(giver: &Mind<MemoryStore>) -> Vec<crate::query::PipelineDocumentView> {
|
||||
let mut selection = Selection {
|
||||
fields: Some(vec![FieldPredicate { index: "root".into(), op: "any_of".into(), values: Some(vec![CAMPAIGN.into()]), number: None }]),
|
||||
projection: "document".into(),
|
||||
limit: Some(cultnet_rs::LIMIT_MAX),
|
||||
..Selection::default()
|
||||
};
|
||||
let mut views = Vec::new();
|
||||
loop {
|
||||
let page = giver.query(&selection).unwrap();
|
||||
let PipelinePageItems::Documents(found) = page.items else { panic!("the document projection answers documents") };
|
||||
views.extend(found);
|
||||
match page.next {
|
||||
Some(next) => selection.cursor = Some(next),
|
||||
None => return views,
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
/// Replays the giver's campaign into the receiver as `import::plan` lays it out: one import per
|
||||
/// source receipt, each naming its source.
|
||||
fn replay(giver: &Mind<MemoryStore>, receiver: &mut Mind<MemoryStore>) -> Vec<PipelineAdmissionOutcome> {
|
||||
let plan = crate::import::plan(&slug(CAMPAIGN), giver.instance(), receiver.instance(), giver_views(giver)).unwrap();
|
||||
plan.batches
|
||||
.into_iter()
|
||||
.map(|batch| import_as(receiver, batch.source, batch.documents.into_iter().map(|(_, document)| document).collect()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -390,7 +389,7 @@ fn replaying_a_whole_campaign_reproduces_it() {
|
|||
assert_eq!(moved_at, last + 1);
|
||||
|
||||
let mut mind = receiver();
|
||||
let outcomes = replay(&giver, &mut mind, last);
|
||||
let outcomes = replay(&giver, &mut mind);
|
||||
assert!(outcomes.len() >= 8, "{} receipts replayed", outcomes.len());
|
||||
for outcome in &outcomes {
|
||||
assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
|
||||
|
|
|
|||
Loading…
Reference in a new issue