Merge pull request 'huginn import: a steward pulls a moved campaign into her own mind' (#16) from hands/eureka-body-campaign-import-verb into main
This commit is contained in:
commit
7c586e39b5
10 changed files with 1069 additions and 42 deletions
|
|
@ -39,6 +39,20 @@ impl Settings {
|
|||
self.client.as_ref().map_err(|detail| Trouble::Misconfigured(detail.clone()))
|
||||
}
|
||||
|
||||
/// A client for another mind, read from: `instance` at `endpoint` (`rudp://<ip>:<port>`), both
|
||||
/// given as arguments. It refuses this process's own instance before any call, so the one
|
||||
/// mind a verb writes is never also its source. Every request through it declares `instance`,
|
||||
/// so an endpoint that holds a different mind answers ForeignInstance.
|
||||
pub fn source(&self, instance: &str, endpoint: &str) -> Result<HuginnClient, Trouble> {
|
||||
let slug = Slug(instance.to_string());
|
||||
slug.validate_slug().map_err(|refusal| Trouble::invalid(format!("--from is not a valid instance name: {refusal:?}")))?;
|
||||
if self.instance.as_deref() == Some(instance) {
|
||||
return Err(Trouble::invalid(format!("--from {instance} is this process's own instance (EUREKA_INSTANCE)")));
|
||||
}
|
||||
let address = address(endpoint).ok_or_else(|| Trouble::invalid(format!("--from-endpoint is not rudp://<ip>:<port>: {endpoint}")))?;
|
||||
Ok(HuginnClient::new(address, slug, TIMEOUT))
|
||||
}
|
||||
|
||||
/// One wire call for the configured instance. The request is built from
|
||||
/// that instance, never from anything the caller sent. No retry.
|
||||
pub fn ask(&self, request: impl FnOnce(&Slug) -> HuginnMindRequest) -> Result<HuginnMindResponse, Trouble> {
|
||||
|
|
@ -78,9 +92,11 @@ fn client(instance: Result<Option<String>, String>, endpoint: Result<Option<Stri
|
|||
let slug = Slug(instance);
|
||||
slug.validate_slug().map_err(|refusal| format!("EUREKA_INSTANCE is not a valid instance name: {refusal:?}"))?;
|
||||
let endpoint = endpoint?.ok_or("HUGINN_ENDPOINT is not set")?;
|
||||
let address = endpoint
|
||||
.strip_prefix("rudp://")
|
||||
.and_then(|address| address.parse::<SocketAddr>().ok())
|
||||
.ok_or_else(|| format!("HUGINN_ENDPOINT is not rudp://<ip>:<port>: {endpoint}"))?;
|
||||
let address = address(&endpoint).ok_or_else(|| format!("HUGINN_ENDPOINT is not rudp://<ip>:<port>: {endpoint}"))?;
|
||||
Ok(HuginnClient::new(address, slug, TIMEOUT))
|
||||
}
|
||||
|
||||
/// The socket address of a `rudp://<ip>:<port>` endpoint.
|
||||
fn address(endpoint: &str) -> Option<SocketAddr> {
|
||||
endpoint.strip_prefix("rudp://").and_then(|address| address.parse::<SocketAddr>().ok())
|
||||
}
|
||||
|
|
|
|||
410
crates/eureka-state/src/bin/huginn/import.rs
Normal file
410
crates/eureka-state/src/bin/huginn/import.rs
Normal file
|
|
@ -0,0 +1,410 @@
|
|||
//! `huginn import <campaign> --from <instance> --from-endpoint <rudp>`: the receiver's Self replays
|
||||
//! a moved campaign's receipts from the giver's mind into her own.
|
||||
//!
|
||||
//! The verb decides only what to send and in what order. `huginn_mind::plan` lays the replay out
|
||||
//! from one cursor walk of the giver; the receiver's walk of her own root says what she already
|
||||
//! holds, so a rerun after any stop resumes at the first unheld batch. Every write is the
|
||||
//! receiver daemon's `Admit`, one per source receipt, and her admission decides it. The giver is
|
||||
//! sent `Query` alone. Nothing is stored here: progress is what the receiver holds, and one
|
||||
//! verification walk of her root at the end compares every planned document.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
|
||||
use cultnet_rs::{LIMIT_MAX, PROJECTION_DOCUMENT, Selection};
|
||||
use eureka_state::hold;
|
||||
use huginn_mind::eureka_pipeline::{PipelineDocument, PipelineRef, Short, Slug};
|
||||
use huginn_mind::{
|
||||
Faculty, HuginnMindRequest, HuginnMindResponse, ImportBatch, ImportFault, ImportPlan, PipelineAdmissionBatch, PipelineAdmissionOutcome,
|
||||
PipelineDocumentSummary, PipelineDocumentView, PipelinePageItems, PipelineProvenance, PipelineSelectionPage, plan,
|
||||
};
|
||||
|
||||
use crate::config::{Settings, TOOL};
|
||||
use crate::trouble::Trouble;
|
||||
use crate::{differences, say, to_value, unexpected};
|
||||
|
||||
/// The arguments of the verb, as given.
|
||||
pub struct Flags {
|
||||
pub campaign: String,
|
||||
pub from: String,
|
||||
pub from_endpoint: String,
|
||||
pub faculty: Faculty,
|
||||
pub agent: String,
|
||||
pub session: String,
|
||||
}
|
||||
|
||||
/// Why the verb stops: the mind said no (exit 1, the text printed) or a trouble (its own exit).
|
||||
enum Stop {
|
||||
No(String),
|
||||
Trouble(Trouble),
|
||||
}
|
||||
|
||||
impl From<Trouble> for Stop {
|
||||
fn from(trouble: Trouble) -> Self {
|
||||
Self::Trouble(trouble)
|
||||
}
|
||||
}
|
||||
|
||||
type Step<T> = Result<T, Stop>;
|
||||
|
||||
pub fn run(settings: &Settings, flags: Flags) -> Result<u8, Trouble> {
|
||||
match import(settings, flags) {
|
||||
Ok(()) => Ok(0),
|
||||
Err(Stop::No(text)) => {
|
||||
say(text);
|
||||
Ok(1)
|
||||
}
|
||||
Err(Stop::Trouble(trouble)) => Err(trouble),
|
||||
}
|
||||
}
|
||||
|
||||
fn import(settings: &Settings, flags: Flags) -> Step<()> {
|
||||
let campaign = Slug(flags.campaign.clone());
|
||||
campaign.validate_slug().map_err(|refusal| Trouble::invalid(format!("the campaign is not a valid name: {refusal:?}")))?;
|
||||
let giver = settings.source(&flags.from, &flags.from_endpoint)?;
|
||||
let receiver = settings.client()?.instance().clone();
|
||||
|
||||
let views = documents(&|selection| {
|
||||
giver.call(HuginnMindRequest::Query { instance: giver.instance().clone(), selection, semantic: None }).map_err(Trouble::from)
|
||||
}, &campaign)?;
|
||||
let plan = plan(&campaign, giver.instance(), &receiver, views).map_err(fault)?;
|
||||
|
||||
let ask_receiver = |selection: Selection| {
|
||||
settings.ask(|instance| HuginnMindRequest::Query { instance: instance.clone(), selection, semantic: None })
|
||||
};
|
||||
let held = resume(&plan, &headers(&ask_receiver, &campaign)?, giver.instance())?;
|
||||
let (receipts, total) = (plan.batches.len(), plan.batches.iter().map(|batch| batch.documents.len()).sum::<usize>());
|
||||
say(format_args!(
|
||||
"plan {campaign} from {giver}: {receipts} receipts {total} documents, {held} held, moved by {moved}",
|
||||
campaign = flags.campaign,
|
||||
giver = flags.from,
|
||||
held = held.iter().filter(|held| **held).count(),
|
||||
moved = plan.moved.id.0,
|
||||
));
|
||||
|
||||
for (index, (batch, held)) in plan.batches.iter().zip(&held).enumerate() {
|
||||
if *held {
|
||||
continue;
|
||||
}
|
||||
let external = unheld(&ask_receiver, &batch.foreign)?;
|
||||
let count = external.len();
|
||||
send(settings, &flags, batch, external)?;
|
||||
say(format_args!("imported ordinal {} {} documents {count} external ({}/{receipts})", batch.source.ordinal, batch.documents.len(), index + 1));
|
||||
}
|
||||
|
||||
verify(&documents(&|selection| ask_receiver(selection), &campaign)?, &plan, giver.instance())?;
|
||||
say(format_args!("imported {} from {}: {receipts} receipts {total} documents, verified", flags.campaign, flags.from));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// A planner fault as the verb ends: the mind's no for who the campaign moved to, an integrity
|
||||
/// trouble for a history that cannot be replayed whole.
|
||||
fn fault(fault: ImportFault) -> Stop {
|
||||
match fault {
|
||||
ImportFault::NotMoved | ImportFault::MovedElsewhere { .. } | ImportFault::SourceIsReceiver => Stop::No(format!("Refused {fault:?}")),
|
||||
ImportFault::BatchTooLarge { .. } | ImportFault::ExternalsTooMany { .. } | ImportFault::Unordered { .. } => {
|
||||
Stop::Trouble(Trouble::Integrity(format!("{fault:?}; nothing was written")))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// One batch, admitted as the importer's own with the source's receipt in `imported_from`.
|
||||
fn send(settings: &Settings, flags: &Flags, batch: &ImportBatch, external: Vec<PipelineRef>) -> Step<()> {
|
||||
let mut source = batch.source.clone();
|
||||
source.external = external;
|
||||
let provenance = PipelineProvenance {
|
||||
faculty: flags.faculty,
|
||||
agent: Short(flags.agent.clone()),
|
||||
session: Short(flags.session.clone()),
|
||||
tool: Short(TOOL.to_string()),
|
||||
imported_from: Some(source),
|
||||
};
|
||||
let documents = batch.documents.iter().map(|(_, document)| document.clone()).collect();
|
||||
let ordinal = batch.source.ordinal;
|
||||
match settings.ask(|instance| HuginnMindRequest::Admit(PipelineAdmissionBatch { instance: instance.clone(), provenance, documents }))? {
|
||||
HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. } | PipelineAdmissionOutcome::AlreadyAdmitted { .. }) => Ok(()),
|
||||
HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(refusal)) | HuginnMindResponse::Refused(refusal) => {
|
||||
Err(Stop::No(format!("Refused {refusal:?} at source ordinal {ordinal}")))
|
||||
}
|
||||
HuginnMindResponse::Admit(PipelineAdmissionOutcome::Conflict { identities }) => Err(Stop::No(format!(
|
||||
"Conflict at source ordinal {ordinal}: {}",
|
||||
identities.iter().map(|identity| identity.id.0.as_str()).collect::<Vec<_>>().join(", ")
|
||||
))),
|
||||
other => Err(unexpected(&other).into()),
|
||||
}
|
||||
}
|
||||
|
||||
/// The foreign referents the receiver does not hold: exactly the ones the batch declares external,
|
||||
/// since declaring one she holds is refused.
|
||||
fn unheld(ask: &dyn Fn(Selection) -> Result<HuginnMindResponse, Trouble>, foreign: &[PipelineRef]) -> Step<Vec<PipelineRef>> {
|
||||
let mut held = BTreeSet::new();
|
||||
for chunk in foreign.chunks(LIMIT_MAX as usize) {
|
||||
let selection = Selection {
|
||||
keys: Some(chunk.iter().map(|referent| referent.id.0.clone()).collect()),
|
||||
limit: Some(chunk.len() as u32),
|
||||
..Selection::default()
|
||||
};
|
||||
for page in pages(ask, selection)? {
|
||||
let PipelinePageItems::Headers(found) = page.items else { return Err(Trouble::Internal("a header query answered documents".into()).into()) };
|
||||
held.extend(found.into_iter().map(|header| header.id.id.0));
|
||||
}
|
||||
}
|
||||
Ok(foreign.iter().filter(|referent| !held.contains(&referent.id.0)).cloned().collect())
|
||||
}
|
||||
|
||||
/// Every page of a selection, to the end, each as of the first page's snapshot.
|
||||
fn pages(ask: &dyn Fn(Selection) -> Result<HuginnMindResponse, Trouble>, mut selection: Selection) -> Step<Vec<PipelineSelectionPage>> {
|
||||
let mut found = Vec::new();
|
||||
loop {
|
||||
match ask(selection.clone())? {
|
||||
HuginnMindResponse::Query(page) => {
|
||||
let next = page.next.clone();
|
||||
found.push(page);
|
||||
match next {
|
||||
Some(next) => selection.cursor = Some(next),
|
||||
None => return Ok(found),
|
||||
}
|
||||
}
|
||||
HuginnMindResponse::Refused(refusal) => return Err(Stop::No(format!("Refused {refusal:?}"))),
|
||||
other => return Err(unexpected(&other).into()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn root_walk(campaign: &Slug, projection: &str) -> Selection {
|
||||
Selection {
|
||||
projection: projection.to_string(),
|
||||
limit: Some(LIMIT_MAX),
|
||||
..hold::any_of(Selection::default(), "root", &[&campaign.0])
|
||||
}
|
||||
}
|
||||
|
||||
/// Every document under the campaign's root, whole.
|
||||
fn documents(ask: &dyn Fn(Selection) -> Result<HuginnMindResponse, Trouble>, campaign: &Slug) -> Step<Vec<PipelineDocumentView>> {
|
||||
let mut views = Vec::new();
|
||||
for page in pages(ask, root_walk(campaign, PROJECTION_DOCUMENT))? {
|
||||
let PipelinePageItems::Documents(found) = page.items else { return Err(Trouble::Internal("a document query answered headers".into()).into()) };
|
||||
views.extend(found);
|
||||
}
|
||||
Ok(views)
|
||||
}
|
||||
|
||||
/// Every document under the campaign's root, as a header.
|
||||
fn headers(ask: &dyn Fn(Selection) -> Result<HuginnMindResponse, Trouble>, campaign: &Slug) -> Step<Vec<PipelineDocumentSummary>> {
|
||||
let mut headers = Vec::new();
|
||||
for page in pages(ask, root_walk(campaign, &Selection::default().projection))? {
|
||||
let PipelinePageItems::Headers(found) = page.items else { return Err(Trouble::Internal("a header query answered documents".into()).into()) };
|
||||
headers.extend(found);
|
||||
}
|
||||
Ok(headers)
|
||||
}
|
||||
|
||||
/// Which batches the receiver already holds whole. A document under the root that is not one of
|
||||
/// the plan's, imported from the plan's receipt of the source, is her own record and refuses the
|
||||
/// import; a batch held in part cannot be resumed, only repaired.
|
||||
fn resume(plan: &ImportPlan, root: &[PipelineDocumentSummary], source: &Slug) -> Step<Vec<bool>> {
|
||||
let planned: BTreeMap<&str, (usize, &str)> = plan
|
||||
.batches
|
||||
.iter()
|
||||
.enumerate()
|
||||
.flat_map(|(index, batch)| batch.documents.iter().map(move |(id, _)| (id.id.0.as_str(), (index, batch.source.receipt_id.0.as_str()))))
|
||||
.collect();
|
||||
let mut held: Vec<BTreeSet<&str>> = vec![BTreeSet::new(); plan.batches.len()];
|
||||
let mut own = Vec::new();
|
||||
for header in root {
|
||||
let key = header.id.id.0.as_str();
|
||||
let imported = header.admission.provenance.imported_from.as_ref();
|
||||
match planned.get(key) {
|
||||
Some((index, receipt)) if imported.is_some_and(|from| from.instance == *source && from.receipt_id.0 == *receipt) => {
|
||||
held[*index].insert(key);
|
||||
}
|
||||
_ => own.push(key),
|
||||
}
|
||||
}
|
||||
if !own.is_empty() {
|
||||
return Err(Stop::No(format!("Refused the receiver already holds records under the root that are not imported from this source: {}; nothing was written", own.join(", "))));
|
||||
}
|
||||
plan.batches
|
||||
.iter()
|
||||
.zip(&held)
|
||||
.map(|(batch, held)| match held.len() {
|
||||
0 => Ok(false),
|
||||
count if count == batch.documents.len() => Ok(true),
|
||||
count => Err(Stop::Trouble(Trouble::Integrity(format!(
|
||||
"source ordinal {} is held in part ({count} of {} documents); repair it by hand, a rerun cannot",
|
||||
batch.source.ordinal,
|
||||
batch.documents.len()
|
||||
)))),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The receiver's root against the plan: every planned document equal and imported from its
|
||||
/// source receipt, and nothing else under the root.
|
||||
fn verify(root: &[PipelineDocumentView], plan: &ImportPlan, source: &Slug) -> Step<()> {
|
||||
let held: BTreeMap<&str, &PipelineDocumentView> = root.iter().map(|view| (view.id.id.0.as_str(), view)).collect();
|
||||
let mut differing = Vec::new();
|
||||
for batch in &plan.batches {
|
||||
for (id, document) in &batch.documents {
|
||||
let key = id.id.0.as_str();
|
||||
let Some(view) = held.get(key) else {
|
||||
differing.push(format!("{key}: missing from the receiver"));
|
||||
continue;
|
||||
};
|
||||
let mut what = Vec::new();
|
||||
differences(&to_value(document), &to_value(&view.document), "", &mut what);
|
||||
let imported = view.admission.provenance.imported_from.as_ref();
|
||||
if imported.is_none_or(|from| from.instance != *source || from.receipt_id != batch.source.receipt_id) {
|
||||
what.push("imported_from".to_string());
|
||||
}
|
||||
if !what.is_empty() {
|
||||
differing.push(format!("{key}: {}", if what.iter().all(|path| path.is_empty()) { "the document".to_string() } else { what.join(", ") }));
|
||||
}
|
||||
}
|
||||
}
|
||||
let planned: BTreeSet<&str> = plan.batches.iter().flat_map(|batch| batch.documents.iter().map(|(id, _)| id.id.0.as_str())).collect();
|
||||
differing.extend(held.keys().filter(|key| !planned.contains(*key)).map(|key| format!("{key}: not in the plan")));
|
||||
if differing.is_empty() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(Stop::Trouble(Trouble::ReadBackDiffers(format!("the receiver's root differs from the giver's: {}", differing.join("; ")))))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use huginn_mind::eureka_pipeline::{Date, PipelineKind, PipelineRuling, RulingAuthority};
|
||||
use huginn_mind::{AdmissionFacts, ImportSource, PipelineFacts, PipelineStatus, PipelineStatusSummary};
|
||||
|
||||
use super::*;
|
||||
|
||||
fn slug(text: &str) -> Slug {
|
||||
Slug(text.to_string())
|
||||
}
|
||||
|
||||
fn id(local: &str) -> PipelineRef {
|
||||
PipelineRef { kind: PipelineKind::Ruling, id: Short(format!("c:ruling:{local}")) }
|
||||
}
|
||||
|
||||
fn batch(ordinal: u64, locals: &[&str]) -> ImportBatch {
|
||||
let source = ImportSource {
|
||||
instance: slug("eureka"),
|
||||
receipt_id: Short(format!("receipt-{ordinal}")),
|
||||
ordinal,
|
||||
committed_at: Short("2026-11-08T16:00:00Z".into()),
|
||||
faculty: Faculty::Hands,
|
||||
agent: Short("a".into()),
|
||||
session: Short("s".into()),
|
||||
tool: Short("huginn".into()),
|
||||
external: vec![],
|
||||
};
|
||||
ImportBatch { source, documents: locals.iter().map(|local| (id(local), question(local))).collect(), foreign: vec![] }
|
||||
}
|
||||
|
||||
fn question(local: &str) -> PipelineDocument {
|
||||
PipelineDocument::Ruling(PipelineRuling {
|
||||
campaign: slug("c"),
|
||||
label: local.into(),
|
||||
title: "A ruling".into(),
|
||||
answers: None,
|
||||
choice: None,
|
||||
ruling: "It stands.".into(),
|
||||
operator_quote: None,
|
||||
ruled_on: Date("2026-11-08".into()),
|
||||
precedents: vec![],
|
||||
authority: RulingAuthority::Standing,
|
||||
})
|
||||
}
|
||||
|
||||
fn plan_of(batches: Vec<ImportBatch>) -> ImportPlan {
|
||||
ImportPlan { moved: id("moved"), batches }
|
||||
}
|
||||
|
||||
fn header(local: &str, from: Option<(&str, u64)>) -> PipelineDocumentSummary {
|
||||
let imported_from = from.map(|(instance, ordinal)| ImportSource { instance: slug(instance), ..batch(ordinal, &[]).source });
|
||||
PipelineDocumentSummary {
|
||||
id: id(local),
|
||||
admission: AdmissionFacts {
|
||||
receipt_id: "mine".into(),
|
||||
admitted_at: "2026-11-08T17:00:00Z".into(),
|
||||
provenance: PipelineProvenance { faculty: Faculty::Hands, agent: Short("me".into()), session: Short("s".into()), tool: Short("huginn".into()), imported_from },
|
||||
ordinal: 1,
|
||||
},
|
||||
status: PipelineStatusSummary::InForce,
|
||||
facts: PipelineFacts::Campaign { title: "t".into(), repos: vec![] },
|
||||
}
|
||||
}
|
||||
|
||||
fn two() -> ImportPlan {
|
||||
plan_of(vec![batch(3, &["a", "b"]), batch(5, &["c"])])
|
||||
}
|
||||
|
||||
fn stopped(step: Step<Vec<bool>>) -> String {
|
||||
match step {
|
||||
Err(Stop::No(text)) => format!("no: {text}"),
|
||||
Err(Stop::Trouble(trouble)) => format!("{trouble}"),
|
||||
Ok(held) => panic!("{held:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_rerun_resumes_at_the_first_unheld_batch() {
|
||||
let source = slug("eureka");
|
||||
assert_eq!(resume(&two(), &[], &source).ok().unwrap(), vec![false, false]);
|
||||
let first = [header("a", Some(("eureka", 3))), header("b", Some(("eureka", 3)))];
|
||||
assert_eq!(resume(&two(), &first, &source).ok().unwrap(), vec![true, false], "the held receipt is skipped");
|
||||
let both = [first.as_slice(), &[header("c", Some(("eureka", 5)))]].concat();
|
||||
assert_eq!(resume(&two(), &both, &source).ok().unwrap(), vec![true, true]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_receipt_held_in_part_is_an_integrity_fault() {
|
||||
let one_of_two = [header("a", Some(("eureka", 3)))];
|
||||
let said = stopped(resume(&two(), &one_of_two, &slug("eureka")));
|
||||
assert!(said.starts_with("Integrity: source ordinal 3 is held in part (1 of 2"), "{said}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_receivers_own_records_under_the_root_refuse_the_import() {
|
||||
let source = slug("eureka");
|
||||
for own in [
|
||||
header("zz", None),
|
||||
header("a", None),
|
||||
header("a", Some(("kiko", 3))),
|
||||
header("a", Some(("eureka", 5))),
|
||||
] {
|
||||
let said = stopped(resume(&two(), &[own], &source));
|
||||
assert!(said.starts_with("no: Refused the receiver already holds records under the root"), "{said}");
|
||||
}
|
||||
}
|
||||
|
||||
fn viewed(local: &str, document: PipelineDocument, from: Option<(&str, u64)>) -> PipelineDocumentView {
|
||||
let held = header(local, from);
|
||||
PipelineDocumentView { id: held.id, document, admission: held.admission, status: PipelineStatus::InForce }
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_verification_walk_names_every_difference() {
|
||||
let source = slug("eureka");
|
||||
let plan = plan_of(vec![batch(3, &["a", "b"]), batch(5, &["c"])]);
|
||||
let ours = |local: &str, ordinal| viewed(local, question(local), Some(("eureka", ordinal)));
|
||||
let right = vec![ours("a", 3), ours("b", 3), ours("c", 5)];
|
||||
assert!(verify(&right, &plan, &source).is_ok());
|
||||
|
||||
let mut changed = question("b");
|
||||
if let PipelineDocument::Ruling(ruling) = &mut changed {
|
||||
ruling.title = "Reworded".into();
|
||||
}
|
||||
let wrong = vec![ours("a", 3), viewed("b", changed, Some(("eureka", 3))), viewed("c", question("c"), Some(("eureka", 3))), ours("x", 5)];
|
||||
let Err(Stop::Trouble(trouble)) = verify(&wrong, &plan, &source) else { panic!("a difference is exit 5") };
|
||||
assert_eq!(trouble.exit(), 5);
|
||||
let said = trouble.to_string();
|
||||
for named in ["c:ruling:b: value.title", "c:ruling:c: imported_from", "c:ruling:x: not in the plan"] {
|
||||
assert!(said.contains(named), "{named} in {said}");
|
||||
}
|
||||
assert!(!said.contains("c:ruling:a:"), "{said}");
|
||||
|
||||
let missing = vec![ours("a", 3), ours("b", 3)];
|
||||
let Err(Stop::Trouble(trouble)) = verify(&missing, &plan, &source) else { panic!() };
|
||||
assert!(trouble.to_string().contains("c:ruling:c: missing from the receiver"), "{trouble}");
|
||||
}
|
||||
}
|
||||
|
|
@ -13,11 +13,13 @@
|
|||
//! 1 the mind said no (a refusal, a conflict, an absent document). Troubles go
|
||||
//! to stderr as `<Code>: <detail>`: 2 the call is wrong, 3 Unavailable, 4 no
|
||||
//! answer and a retry cannot help, 5 the read-back after a committed admission
|
||||
//! differs from what was sent, 6 `admit` held the batch once (`admit_hold`).
|
||||
//! differs from what was sent or the verification walk of `import` finds a difference, 6 `admit` held the batch once (`admit_hold`). `import` also
|
||||
//! exits 1 when the campaign is not moved to this mind or this mind holds records of its own under it, and 4 when the history cannot be replayed whole.
|
||||
|
||||
mod admit_hold;
|
||||
mod config;
|
||||
mod door;
|
||||
mod import;
|
||||
mod near;
|
||||
mod recall;
|
||||
mod recipes;
|
||||
|
|
@ -60,8 +62,10 @@ The file is a JSON array of 1 to 64 documents, each {\"kind\": <kind>, \"value\"
|
|||
Every key the schema does not know is refused (exit 2) with its path; nothing is sent. \
|
||||
After Committed or AlreadyAdmitted the batch is read back and compared with what was sent: `read back <n> equal`, or exit 5 with the ids and field paths that differ. A batch whose rulings or questions have in-force neighbours is held once: exit 6, the neighbours printed with a token, nothing sent. Run the same command with --seen <token> to admit it; a batch or --agent that changed needs a new token (a wrong one is exit 2).";
|
||||
|
||||
const IMPORT_HELP: &str = "Reads the giver's mind (it sends Query alone) and admits its receipts for the campaign into this one, one batch per source receipt, in order, each naming the giver's receipt in imported_from. It refuses unless the campaign is moved to this mind (exit 1), and when this mind already holds records under the campaign that it did not import from the giver (exit 1). A rerun after any stop resumes at the first batch this mind does not hold; it prints `imported <campaign> from <instance>: ... verified` only after comparing every document it holds with the giver's (exit 5 if one differs). Until it prints imported, this mind writes nothing else under the campaign; a write that collides stops the import with the refusal (exit 1). Exit 3 Unavailable (rerun), 4 the history cannot be replayed whole (a receipt over 64 documents, too many external referents, a receipt held in part). Only a Self run can run it: a Persona's traffic to a peer's huginn port is fenced.";
|
||||
|
||||
#[derive(Parser, Debug)]
|
||||
#[command(name = "huginn", about = "Huginn's mind: whoami, view, query, recipes, persona, admit, recall, schema", arg_required_else_help = true)]
|
||||
#[command(name = "huginn", about = "Huginn's mind: whoami, view, query, recipes, persona, admit, import, recall, schema", arg_required_else_help = true)]
|
||||
struct Cli {
|
||||
#[command(subcommand)]
|
||||
command: Command,
|
||||
|
|
@ -128,6 +132,26 @@ enum Command {
|
|||
},
|
||||
/// The JSON Schema of a document kind or of a selection; no name lists them.
|
||||
Schema { name: Option<String> },
|
||||
/// Replay a moved campaign's history from the giver's mind into this one.
|
||||
#[command(after_help = IMPORT_HELP)]
|
||||
Import {
|
||||
/// The campaign's name: the root of its ids.
|
||||
campaign: String,
|
||||
/// The giver's instance.
|
||||
#[arg(long = "from")]
|
||||
from: String,
|
||||
/// The giver's daemon, rudp://<ip>:<port>.
|
||||
#[arg(long)]
|
||||
from_endpoint: String,
|
||||
#[arg(long, value_parser = faculty)]
|
||||
faculty: Faculty,
|
||||
/// The importing agent's name.
|
||||
#[arg(long)]
|
||||
agent: String,
|
||||
/// The importing session.
|
||||
#[arg(long)]
|
||||
session: String,
|
||||
},
|
||||
/// Recall into a dispatched brief.
|
||||
Recall {
|
||||
#[command(subcommand)]
|
||||
|
|
@ -260,6 +284,9 @@ fn run(settings: &Settings, command: Command) -> Result<u8, Trouble> {
|
|||
query(settings, door::selection_value(&ask.selection)?, ask.semantic, ask.page.cursor, ask.page.documents)
|
||||
}
|
||||
Command::Admit { faculty, agent, session, seen, file } => admit(settings, faculty, agent, session, seen, &file),
|
||||
Command::Import { campaign, from, from_endpoint, faculty, agent, session } => {
|
||||
import::run(settings, import::Flags { campaign, from, from_endpoint, faculty, agent, session })
|
||||
}
|
||||
Command::Persona { command: PersonaCommand::Get } => persona_get(settings),
|
||||
Command::Persona { command: PersonaCommand::Put { faculty, agent, session, expected_updated_at, file } } => {
|
||||
persona_put(settings, faculty, agent, session, expected_updated_at, &file)
|
||||
|
|
|
|||
|
|
@ -2,7 +2,8 @@
|
|||
//!
|
||||
//! Exit codes: 2 the call is wrong (usage, InvalidInput, TooLarge,
|
||||
//! Misconfigured); 3 Unavailable (retry once); 4 no answer and a retry cannot
|
||||
//! help (Rejected, Unencodable, Internal); 5 ReadBackDiffers. Exit 0 and 1 are
|
||||
//! help (Rejected, Unencodable, Internal, Integrity); 5 ReadBackDiffers, the read-back after an
|
||||
//! admission or the verification walk of an import. Exit 0 and 1 are
|
||||
//! answers (the mind said yes, the mind said no) and are not troubles.
|
||||
|
||||
use std::fmt;
|
||||
|
|
@ -19,6 +20,8 @@ pub enum Trouble {
|
|||
Rejected(String),
|
||||
Unencodable(String),
|
||||
Internal(String),
|
||||
/// What an import found cannot be replayed whole, however often it is rerun.
|
||||
Integrity(String),
|
||||
ReadBackDiffers(String),
|
||||
}
|
||||
|
||||
|
|
@ -32,6 +35,7 @@ impl Trouble {
|
|||
Self::Rejected(_) => "Rejected",
|
||||
Self::Unencodable(_) => "Unencodable",
|
||||
Self::Internal(_) => "Internal",
|
||||
Self::Integrity(_) => "Integrity",
|
||||
Self::ReadBackDiffers(_) => "ReadBackDiffers",
|
||||
}
|
||||
}
|
||||
|
|
@ -40,7 +44,7 @@ impl Trouble {
|
|||
match self {
|
||||
Self::Misconfigured(_) | Self::InvalidInput(_) | Self::TooLarge { .. } => 2,
|
||||
Self::Unavailable(_) => 3,
|
||||
Self::Rejected(_) | Self::Unencodable(_) | Self::Internal(_) => 4,
|
||||
Self::Rejected(_) | Self::Unencodable(_) | Self::Internal(_) | Self::Integrity(_) => 4,
|
||||
Self::ReadBackDiffers(_) => 5,
|
||||
}
|
||||
}
|
||||
|
|
@ -60,6 +64,7 @@ impl fmt::Display for Trouble {
|
|||
| Self::Rejected(detail)
|
||||
| Self::Unencodable(detail)
|
||||
| Self::Internal(detail)
|
||||
| Self::Integrity(detail)
|
||||
| Self::ReadBackDiffers(detail) => f.write_str(detail),
|
||||
Self::TooLarge { bytes, limit } => {
|
||||
write!(f, "the request encodes to {bytes} bytes and one send carries at most {limit}")
|
||||
|
|
|
|||
|
|
@ -2468,3 +2468,256 @@ fn the_whole_hold_check_is_bounded_at_ten_seconds_and_fails_open() {
|
|||
assert!(ran.out.ends_with("read back 3 equal\n"), "{}", ran.out);
|
||||
assert!((9.5..15.0).contains(&took), "{took} s");
|
||||
}
|
||||
|
||||
|
||||
// ---- huginn import: a moved campaign replayed between two live daemons ----
|
||||
|
||||
const TAKER: &str = "heimdall";
|
||||
const ABROAD: &str = "other-campaign";
|
||||
|
||||
fn moving_to(mind: &str) -> PipelineDocument {
|
||||
PipelineDocument::Resolution(PipelineResolution {
|
||||
subject: PipelineRef { kind: PipelineKind::Campaign, id: Short(format!("{CAMPAIGN}:campaign:self")) },
|
||||
sequence: 1,
|
||||
outcome: ResolutionOutcome::Moved { to: slug(mind) },
|
||||
rationale: "So.".into(),
|
||||
resolved_on: Date("2026-11-08".into()),
|
||||
})
|
||||
}
|
||||
|
||||
fn answering_ruling(label: &str, campaign: &str, question: Option<String>) -> PipelineDocument {
|
||||
let PipelineDocument::Ruling(mut inner) = ruling_of(label, "A ruling") else { unreachable!() };
|
||||
inner.campaign = slug(campaign);
|
||||
inner.choice = question.as_ref().map(|_| "a".into());
|
||||
inner.answers = question.map(Short);
|
||||
PipelineDocument::Ruling(inner)
|
||||
}
|
||||
|
||||
/// A spec of the campaign citing the given rulings.
|
||||
fn spec_citing(cited: &[&str]) -> PipelineDocument {
|
||||
let mut value = spec("1");
|
||||
value["value"]["rulings"] = json!(cited);
|
||||
serde_json::from_value(value).unwrap()
|
||||
}
|
||||
|
||||
/// The giver's whole history: the campaign, two questions, a ruling answering one (two documents
|
||||
/// in one receipt), another campaign's ruling and a spec citing both rulings.
|
||||
fn giver_history(moved_to: Option<&str>) -> (TempDir, TestDaemon) {
|
||||
let q1 = format!("{CAMPAIGN}:question:Q1");
|
||||
let mut seed = vec![
|
||||
vec![campaign(), stewardship()],
|
||||
vec![question_of(CAMPAIGN, "Q1", "One"), question_of(CAMPAIGN, "Q2", "Two")],
|
||||
vec![answering_ruling("R1", CAMPAIGN, Some(q1))],
|
||||
vec![answering_ruling("X9", ABROAD, None)],
|
||||
vec![spec_citing(&[&format!("{CAMPAIGN}:ruling:R1"), &format!("{ABROAD}:ruling:X9")])],
|
||||
];
|
||||
seed.extend(moved_to.map(|to| vec![moving_to(to)]));
|
||||
mind(seed)
|
||||
}
|
||||
|
||||
fn taker_mind(holding_abroad: bool) -> (TempDir, TestDaemon) {
|
||||
let mut seed = vec![vec![stewardship_of(TAKER)]];
|
||||
if holding_abroad {
|
||||
seed.push(vec![answering_ruling("X9", ABROAD, None)]);
|
||||
}
|
||||
mind_as(TAKER, seed)
|
||||
}
|
||||
|
||||
fn plan_from(giver: &mut TestDaemon) -> huginn_mind::ImportPlan {
|
||||
let mut selection = Selection {
|
||||
fields: Some(vec![cultnet_rs::FieldPredicate { index: "root".into(), op: "any_of".into(), values: Some(vec![CAMPAIGN.into()]), number: None }]),
|
||||
projection: "document".into(),
|
||||
limit: Some(200),
|
||||
..Selection::default()
|
||||
};
|
||||
let mut views = Vec::new();
|
||||
loop {
|
||||
let request = HuginnMindRequest::Query { instance: slug(INSTANCE), selection: selection.clone(), semantic: None };
|
||||
let HuginnMindResponse::Query(page) = answered(giver.handle(request, chrono::Utc::now())) else { panic!("a query answers a page") };
|
||||
let PipelinePageItems::Documents(found) = page.items else { panic!("documents") };
|
||||
views.extend(found);
|
||||
match page.next {
|
||||
Some(next) => selection.cursor = Some(next),
|
||||
None => return huginn_mind::plan(&slug(CAMPAIGN), &slug(INSTANCE), &slug(TAKER), views).unwrap(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// What an interrupted import left in the receiver: the first `whole` batches, then the last `part` documents of the next.
|
||||
fn half_imported(taker: &mut TestDaemon, plan: &huginn_mind::ImportPlan, whole: usize, part: usize) {
|
||||
for (index, batch) in plan.batches.iter().enumerate().take(whole + usize::from(part > 0)) {
|
||||
let skip = if index < whole { 0 } else { batch.documents.len() - part };
|
||||
let documents = batch.documents.iter().skip(skip).map(|(_, document)| document.clone()).collect();
|
||||
let mut provenance = provenance();
|
||||
provenance.imported_from = Some(batch.source.clone());
|
||||
let batch = huginn_mind::PipelineAdmissionBatch { instance: slug(TAKER), provenance, documents };
|
||||
let outcome = answered(taker.handle(HuginnMindRequest::Admit(batch), chrono::Utc::now()));
|
||||
assert!(matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{outcome:?}");
|
||||
}
|
||||
}
|
||||
|
||||
fn importing(taker: &Server, from: &str, from_endpoint: SocketAddr) -> Run {
|
||||
let endpoint = format!("rudp://{from_endpoint}");
|
||||
run(
|
||||
&[("EUREKA_INSTANCE", TAKER), ("HUGINN_ENDPOINT", &format!("rudp://{}", taker.addr))],
|
||||
&["import", CAMPAIGN, "--from", from, "--from-endpoint", &endpoint, "--faculty", "SelfFaculty", "--agent", "mind-self-run", "--session", "import-1"],
|
||||
None,
|
||||
)
|
||||
}
|
||||
|
||||
fn shown_by(server: &Server, instance: &str, id: &str, raw: bool) -> Run {
|
||||
let env = [("EUREKA_INSTANCE", instance), ("HUGINN_ENDPOINT", &format!("rudp://{}", server.addr))];
|
||||
if raw { run(&env, &["view", id, "--raw"], None) } else { run(&env, &["view", id], None) }
|
||||
}
|
||||
|
||||
fn campaign_id() -> String {
|
||||
format!("{CAMPAIGN}:campaign:self")
|
||||
}
|
||||
|
||||
fn imported_lines(out: &str) -> Vec<&str> {
|
||||
out.lines().filter(|line| line.starts_with("imported ordinal")).collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_moves_a_campaign_between_two_live_daemons() {
|
||||
let (root, daemon) = giver_history(Some(TAKER));
|
||||
let giver = serve(root, daemon);
|
||||
let (root, daemon) = taker_mind(false);
|
||||
let taker = serve(root, daemon);
|
||||
|
||||
let ran = importing(&taker, INSTANCE, giver.addr);
|
||||
assert_eq!(ran.code, 0, "{}{}", ran.out, ran.err);
|
||||
let lines: Vec<&str> = ran.out.lines().collect();
|
||||
assert!(lines[0].starts_with(&format!("plan {CAMPAIGN} from {INSTANCE}: 4 receipts ")) && lines[0].contains(", 0 held, moved by "), "{}", lines[0]);
|
||||
assert_eq!(imported_lines(&ran.out).len(), 4);
|
||||
let last = lines.last().unwrap();
|
||||
assert!(last.starts_with(&format!("imported {CAMPAIGN} from {INSTANCE}: 4 receipts ")) && last.ends_with(", verified"), "{last}");
|
||||
|
||||
for id in [campaign_id(), format!("{CAMPAIGN}:question:Q1"), format!("{CAMPAIGN}:question:Q2"), format!("{CAMPAIGN}:ruling:R1"), format!("{CAMPAIGN}:cut_spec:cut-1.r1")] {
|
||||
let (there, here) = (shown_by(&taker, TAKER, &id, true), shown_by(&giver, INSTANCE, &id, true));
|
||||
assert_eq!((there.code, here.code), (0, 0), "{id}");
|
||||
assert_eq!(there.out, here.out, "{id} is the giver's document");
|
||||
let shown = shown_by(&taker, TAKER, &id, false).out;
|
||||
assert!(shown.starts_with(&format!("{id} ")) && shown.contains(&format!("imported from {INSTANCE} ordinal")), "{shown}");
|
||||
}
|
||||
assert!(shown_by(&taker, TAKER, &campaign_id(), false).out.starts_with(&format!("{} InForce", campaign_id())), "the receiver keeps the campaign");
|
||||
let moved = shown_by(&giver, INSTANCE, &campaign_id(), false).out;
|
||||
assert!(moved.starts_with(&format!("{} Resolved", campaign_id())) && moved.contains("Moved"), "{moved}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_resumes_after_a_stop() {
|
||||
let (root, mut giver_daemon) = giver_history(Some(TAKER));
|
||||
let plan = plan_from(&mut giver_daemon);
|
||||
let giver = serve(root, giver_daemon);
|
||||
let (root, mut daemon) = taker_mind(false);
|
||||
half_imported(&mut daemon, &plan, 2, 0);
|
||||
let taker = serve(root, daemon);
|
||||
|
||||
let ran = importing(&taker, INSTANCE, giver.addr);
|
||||
assert_eq!(ran.code, 0, "{}{}", ran.out, ran.err);
|
||||
assert!(ran.out.lines().next().unwrap().contains(", 2 held, moved by "), "{}", ran.out);
|
||||
assert_eq!(imported_lines(&ran.out).len(), plan.batches.len() - 2, "the held receipts are not sent again: {}", ran.out);
|
||||
assert!(imported_lines(&ran.out)[0].contains(&format!("ordinal {} ", plan.batches[2].source.ordinal)), "resumes at the first unheld: {}", ran.out);
|
||||
assert!(ran.out.trim_end().ends_with(", verified"));
|
||||
|
||||
let again = importing(&taker, INSTANCE, giver.addr);
|
||||
assert_eq!(again.code, 0, "{}{}", again.out, again.err);
|
||||
assert!(imported_lines(&again.out).is_empty() && again.out.lines().next().unwrap().contains(&format!(", {} held,", plan.batches.len())), "{}", again.out);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_refuses_a_receipt_the_receiver_holds_only_part_of() {
|
||||
let (root, mut giver_daemon) = giver_history(Some(TAKER));
|
||||
let plan = plan_from(&mut giver_daemon);
|
||||
let giver = serve(root, giver_daemon);
|
||||
let two = plan.batches.iter().position(|batch| batch.documents.len() > 1).expect("a receipt of two documents");
|
||||
let (root, mut daemon) = taker_mind(false);
|
||||
half_imported(&mut daemon, &plan, two, 1);
|
||||
let taker = serve(root, daemon);
|
||||
|
||||
let ran = importing(&taker, INSTANCE, giver.addr);
|
||||
assert_eq!(ran.code, 4, "{}{}", ran.out, ran.err);
|
||||
assert!(ran.err.starts_with("Integrity: source ordinal ") && ran.err.contains("is held in part (1 of 2"), "{}", ran.err);
|
||||
assert!(imported_lines(&ran.out).is_empty(), "nothing is sent over a receipt held in part: {}", ran.out);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_declares_only_unheld_foreign_referents() {
|
||||
for holding in [false, true] {
|
||||
let (root, daemon) = giver_history(Some(TAKER));
|
||||
let giver = serve(root, daemon);
|
||||
let (root, daemon) = taker_mind(holding);
|
||||
let taker = serve(root, daemon);
|
||||
let ran = importing(&taker, INSTANCE, giver.addr);
|
||||
assert_eq!(ran.code, 0, "holding {holding}: {}{}", ran.out, ran.err);
|
||||
let spec_line = imported_lines(&ran.out).into_iter().find(|line| line.ends_with("(4/4)")).expect("the spec's receipt is last");
|
||||
let declared = if holding { " 0 external " } else { " 1 external " };
|
||||
assert!(spec_line.contains(declared), "holding {holding}: {spec_line}");
|
||||
assert!(ran.out.trim_end().ends_with(", verified"));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_refuses_when_not_moved_to_her() {
|
||||
for (moved, said) in [(None, "Refused NotMoved"), (Some("kiko"), "Refused MovedElsewhere")] {
|
||||
let (root, daemon) = giver_history(moved);
|
||||
let giver = serve(root, daemon);
|
||||
let (root, daemon) = taker_mind(false);
|
||||
let taker = serve(root, daemon);
|
||||
let ran = importing(&taker, INSTANCE, giver.addr);
|
||||
assert_eq!(ran.code, 1, "{moved:?}: {}{}", ran.out, ran.err);
|
||||
assert!(ran.out.starts_with(said), "{}", ran.out);
|
||||
assert_eq!(shown_by(&taker, TAKER, &campaign_id(), false).code, 1, "nothing was written");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_refuses_her_own_records_under_the_root() {
|
||||
let (root, daemon) = giver_history(Some(TAKER));
|
||||
let giver = serve(root, daemon);
|
||||
let (root, mut daemon) = taker_mind(false);
|
||||
let own = answered(daemon.handle(HuginnMindRequest::Admit(batch(TAKER, vec![question_of(CAMPAIGN, "Mine", "Hers")])), chrono::Utc::now()));
|
||||
assert!(matches!(own, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })));
|
||||
let taker = serve(root, daemon);
|
||||
|
||||
let ran = importing(&taker, INSTANCE, giver.addr);
|
||||
assert_eq!(ran.code, 1, "{}{}", ran.out, ran.err);
|
||||
assert!(ran.out.starts_with("Refused the receiver already holds records under the root") && ran.out.contains(&format!("{CAMPAIGN}:question:Mine")), "{}", ran.out);
|
||||
assert_eq!(shown_by(&taker, TAKER, &campaign_id(), false).code, 1, "nothing was written");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_refuses_a_history_with_a_receipt_over_the_batch_limit() {
|
||||
// 33 rulings each answering a question: 33 documents admitted, 66 written in one receipt.
|
||||
let questions: Vec<PipelineDocument> = (0..33).map(|n| question_of(CAMPAIGN, &format!("Q{n}"), "A fork")).collect();
|
||||
let rulings: Vec<PipelineDocument> = (0..33).map(|n| answering_ruling(&format!("R{n}"), CAMPAIGN, Some(format!("{CAMPAIGN}:question:Q{n}")))).collect();
|
||||
let (root, daemon) = mind(vec![vec![campaign(), stewardship()], questions, rulings, vec![moving_to(TAKER)]]);
|
||||
let giver = serve(root, daemon);
|
||||
let (root, daemon) = taker_mind(false);
|
||||
let taker = serve(root, daemon);
|
||||
|
||||
let ran = importing(&taker, INSTANCE, giver.addr);
|
||||
assert_eq!(ran.code, 4, "{}{}", ran.out, ran.err);
|
||||
assert!(ran.err.starts_with("Integrity: BatchTooLarge { ordinal: ") && ran.err.contains("documents: 66 }"), "{}", ran.err);
|
||||
assert_eq!(shown_by(&taker, TAKER, &campaign_id(), false).code, 1, "refused before the first write");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_wrong_endpoint_is_foreign_instance() {
|
||||
let (root, daemon) = giver_history(Some(TAKER));
|
||||
let _giver = serve(root, daemon);
|
||||
let (root, daemon) = taker_mind(false);
|
||||
let taker = serve(root, daemon);
|
||||
|
||||
// The giver's name at the receiver's own endpoint: her daemon refuses a read declared for another mind.
|
||||
let ran = importing(&taker, INSTANCE, taker.addr);
|
||||
assert_eq!(ran.code, 1, "{}{}", ran.out, ran.err);
|
||||
assert!(ran.out.starts_with("Refused ") && ran.out.contains("ForeignInstance"), "{}", ran.out);
|
||||
assert_eq!(shown_by(&taker, TAKER, &campaign_id(), false).code, 1, "nothing was written");
|
||||
|
||||
// Her own name as the source is a wrong call before any send.
|
||||
let own = importing(&taker, TAKER, taker.addr);
|
||||
assert_eq!(own.code, 2, "{}{}", own.out, own.err);
|
||||
assert!(own.err.starts_with("InvalidInput: --from heimdall is this process's own instance"), "{}", own.err);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -120,10 +120,15 @@ pub fn answered(handled: Handled) -> HuginnMindResponse {
|
|||
/// A daemon over a temporary mind whose identity is admitted, and whatever
|
||||
/// else `seed` admits.
|
||||
pub fn mind(seed: Vec<Vec<PipelineDocument>>) -> (TempDir, TestDaemon) {
|
||||
mind_as(INSTANCE, seed)
|
||||
}
|
||||
|
||||
/// `mind` for another instance.
|
||||
pub fn mind_as(instance: &str, seed: Vec<Vec<PipelineDocument>>) -> (TempDir, TestDaemon) {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let mut daemon = Daemon::new(Mind::open(root.path(), &slug(INSTANCE)).unwrap(), Inert);
|
||||
for documents in std::iter::once(vec![identity(INSTANCE)]).chain(seed) {
|
||||
let outcome = answered(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), Utc::now()));
|
||||
let mut daemon = Daemon::new(Mind::open(root.path(), &slug(instance)).unwrap(), Inert);
|
||||
for documents in std::iter::once(vec![identity(instance)]).chain(seed) {
|
||||
let outcome = answered(daemon.handle(HuginnMindRequest::Admit(batch(instance, documents)), Utc::now()));
|
||||
assert!(matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{outcome:?}");
|
||||
}
|
||||
(root, daemon)
|
||||
|
|
@ -220,8 +225,12 @@ pub fn campaign() -> PipelineDocument {
|
|||
}
|
||||
|
||||
pub fn stewardship() -> PipelineDocument {
|
||||
stewardship_of(INSTANCE)
|
||||
}
|
||||
|
||||
pub fn stewardship_of(instance: &str) -> PipelineDocument {
|
||||
PipelineDocument::Stewardship(PipelineStewardship {
|
||||
instance: slug(INSTANCE),
|
||||
instance: slug(instance),
|
||||
repo: OrgRepo(REPO.into()),
|
||||
sequence: 1,
|
||||
assigned_on: Date("2026-09-29".into()),
|
||||
|
|
|
|||
266
crates/huginn-mind/src/import.rs
Normal file
266
crates/huginn-mind/src/import.rs
Normal file
|
|
@ -0,0 +1,266 @@
|
|||
//! 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<String, PipelineRef> = BTreeMap::new();
|
||||
for view in &group {
|
||||
for (_, target) in citations(&view.document) {
|
||||
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.to_string()).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 })
|
||||
}
|
||||
|
||||
#[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");
|
||||
}
|
||||
}
|
||||
|
|
@ -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:?}");
|
||||
|
|
|
|||
|
|
@ -108,3 +108,43 @@ fn a_tell_names_each_brief_the_plan_and_how_she_resolves_it() {
|
|||
assert!(text.ends_with("Plan: 5h at 40%."), "{text}");
|
||||
assert!(!render(&[], "").contains("brief"), "no brief, no plan, nothing");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn imported_briefs_are_not_her_briefs() {
|
||||
// A campaign moved to her: the giver's Persona admitted this brief, and her Self run imports it.
|
||||
let mind = TestMind::without_campaigns();
|
||||
let persona = run_ref("persona");
|
||||
let source = huginn_mind::ImportSource {
|
||||
instance: "eureka".into(),
|
||||
receipt_id: "mind-commit-giver".into(),
|
||||
ordinal: 7,
|
||||
committed_at: "2026-11-08T16:00:00Z".into(),
|
||||
faculty: huginn_mind::Faculty::SelfFaculty,
|
||||
agent: PERSONA.into(),
|
||||
session: "giver-persona-session".into(),
|
||||
tool: "huginn".into(),
|
||||
external: vec![persona.clone()],
|
||||
};
|
||||
let batch = huginn_mind::PipelineAdmissionBatch {
|
||||
instance: crate::testkit::INSTANCE.into(),
|
||||
provenance: huginn_mind::PipelineProvenance {
|
||||
faculty: huginn_mind::Faculty::SelfFaculty,
|
||||
agent: SELF_RUN.into(),
|
||||
session: "import-1".into(),
|
||||
tool: "huginn".into(),
|
||||
imported_from: Some(source),
|
||||
},
|
||||
documents: vec![crate::testkit::campaign_named(crate::testkit::CAMPAIGN, "Eureka body"), brief_doc("moved", OWNER, &persona)],
|
||||
};
|
||||
let huginn_mind::HuginnMindResponse::Admit(outcome) = mind.answer(huginn_mind::HuginnMindRequest::Admit(batch)) else { panic!("an admit answers an admit") };
|
||||
assert!(matches!(outcome, huginn_mind::PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
|
||||
|
||||
let in_force_ids: Vec<String> = headers(&mind, &in_force(of_kind(K::FollowUp))).unwrap().into_iter().map(|header| header.id.id.0).collect();
|
||||
assert_eq!(in_force_ids, vec![brief_ref("moved").id.0.clone()], "the imported follow-up stands in her mind");
|
||||
assert!(derive(&mind, &HashSet::new()).unwrap().is_empty(), "the giver's Persona's brief is not her Persona's brief");
|
||||
|
||||
// The same document admitted by her own Persona turn is a brief.
|
||||
mind.committed(vec![run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]);
|
||||
admitted_by(&mind, PERSONA, vec![brief_doc("hers", OWNER, &persona)]);
|
||||
assert_eq!(ids(derive(&mind, &HashSet::new()).unwrap()), vec![brief_ref("hers").id.0]);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue