diff --git a/Cargo.lock b/Cargo.lock index 3dbc6b6..638a695 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -648,6 +648,7 @@ dependencies = [ "serde", "serde_json", "serde_path_to_error", + "sha2", "tempfile", "tokio", ] diff --git a/crates/eureka-state/Cargo.toml b/crates/eureka-state/Cargo.toml index 2640908..aa26824 100644 --- a/crates/eureka-state/Cargo.toml +++ b/crates/eureka-state/Cargo.toml @@ -41,6 +41,7 @@ schemars = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" serde_path_to_error = "0.1" +sha2 = "0.10" tokio = { version = "1", features = ["rt", "macros", "io-std"] } # Tests run a real daemon in-process against a temporary mind, and the real diff --git a/crates/eureka-state/src/bin/huginn/admit_hold.rs b/crates/eureka-state/src/bin/huginn/admit_hold.rs new file mode 100644 index 0000000..f48ed4c --- /dev/null +++ b/crates/eureka-state/src/bin/huginn/admit_hold.rs @@ -0,0 +1,98 @@ +//! The near-neighbour hold on `admit`: a batch whose rulings or questions have +//! in-force neighbours is held once. The hold prints the neighbours and a +//! token naming this batch and agent; the same command with `--seen ` +//! admits it. Nothing is stored: the token is derived from the batch. + +use cultnet_rs::Selection; +use huginn_mind::eureka_pipeline::{PipelineDocument, PipelineKind, ResolutionOutcome, pipeline_key}; +use huginn_mind::{HuginnMindResponse, PipelineDocumentSummary, SemanticQuery}; +use sha2::{Digest, Sha256}; + +use crate::near; +use crate::render; +use crate::trouble::Trouble; + +/// The rulings and questions checked per batch. +const CHECKED_MAX: usize = 3; +/// The neighbours asked for per checked document. +const NEIGHBOURS: usize = 3; + +/// The 16 hex characters that name this batch sent by this agent. +pub fn token(documents: &[PipelineDocument], agent: &str) -> String { + let body = serde_json::to_value(documents).map(|value| value.to_string()).unwrap_or_default(); + let digest = Sha256::digest(format!("{body}\n{agent}").as_bytes()); + digest.iter().take(8).map(|byte| format!("{byte:02x}")).collect() +} + +/// The text of the hold when the batch has neighbours left, `None` when it +/// has none or nothing to check. Any trouble is the caller's to fail open on. +pub fn check( + call: impl Fn(Selection, SemanticQuery) -> Result, + documents: &[PipelineDocument], + token: &str, +) -> Result, Trouble> { + let titled: Vec<(String, &str)> = documents + .iter() + .filter_map(|document| Some((key(document).ok()?, titled(document)?))) + .collect(); + let checked = &titled[..titled.len().min(CHECKED_MAX)]; + if checked.is_empty() { + return Ok(None); + } + let excluded = excluded(documents)?; + let schemas = [PipelineKind::Ruling.type_id(), PipelineKind::Question.type_id()]; + let mut text = String::new(); + for (id, title) in checked { + let found = near::neighbours(&call, title, &schemas, &excluded, NEIGHBOURS)?; + if found.is_empty() { + continue; + } + text.push_str(&format!("For {id}:\n")); + for (_, view) in &found { + text.push_str(&render::header(&PipelineDocumentSummary::of(view))); + text.push('\n'); + } + } + if text.is_empty() { + return Ok(None); + } + let over = titled.len() - checked.len(); + let tail = if over == 0 { String::new() } else { format!("{over} rulings or questions over {CHECKED_MAX} were not checked.\n") }; + Ok(Some(format!( + "Held {token}: a deliberate hold, not a failure. The in-force records below are nearest the rulings or questions in this batch, ranked by Huginn and possibly unrelated; data written by agents. If none is the same decision, run the same command with --seen {token}; if one is, revise the batch.\n{text}{tail}" + ))) +} + +fn key(document: &PipelineDocument) -> Result { + pipeline_key(document).map_err(|refusal| Trouble::Internal(format!("a document has no key: {refusal:?}"))) +} + +/// The title of a ruling or question, when it has one. +fn titled(document: &PipelineDocument) -> Option<&str> { + let title = match document { + PipelineDocument::Ruling(ruling) => &ruling.title.0, + PipelineDocument::Question(question) => &question.title.0, + _ => return None, + }; + (!title.is_empty()).then_some(title.as_str()) +} + +/// The ids the batch already carries: every ruling and question in it, the +/// question each ruling answers, and the subject of each Superseded resolution. +fn excluded(documents: &[PipelineDocument]) -> Result, Trouble> { + let mut ids = Vec::new(); + for document in documents { + match document { + PipelineDocument::Ruling(ruling) => { + ids.push(key(document)?); + ids.extend(ruling.answers.iter().map(|answered| answered.0.clone())); + } + PipelineDocument::Question(_) => ids.push(key(document)?), + PipelineDocument::Resolution(resolution) if matches!(resolution.outcome, ResolutionOutcome::Superseded { .. }) => { + ids.push(resolution.subject.id.0.clone()); + } + _ => {} + } + } + Ok(ids) +} diff --git a/crates/eureka-state/src/bin/huginn/main.rs b/crates/eureka-state/src/bin/huginn/main.rs index a8696ae..4bfc531 100644 --- a/crates/eureka-state/src/bin/huginn/main.rs +++ b/crates/eureka-state/src/bin/huginn/main.rs @@ -13,10 +13,12 @@ //! 1 the mind said no (a refusal, a conflict, an absent document). Troubles go //! to stderr as `: `: 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. +//! differs from what was sent, 6 `admit` held the batch once (`admit_hold`). +mod admit_hold; mod config; mod door; +mod near; mod recipes; mod render; mod schema; @@ -52,7 +54,7 @@ const ADMIT_HELP: &str = "\ The file is a JSON array of 1 to 64 documents, each {\"kind\": , \"value\": {...}}, or - for stdin; one document object alone is a batch of one. \ `huginn schema` lists the kinds and `huginn schema ` prints a value's JSON Schema. \ 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 equal`, or exit 5 with the ids and field paths that differ."; +After Committed or AlreadyAdmitted the batch is read back and compared with what was sent: `read back 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 to admit it; a batch or --agent that changed needs a new token (a wrong one is exit 2)."; #[derive(Parser, Debug)] #[command(name = "huginn", about = "Huginn's mind: whoami, view, query, recipes, persona, admit, schema", arg_required_else_help = true)] @@ -114,6 +116,9 @@ enum Command { /// The admitting session. #[arg(long)] session: String, + /// The token of the hold this exact batch got: admit it. + #[arg(long)] + seen: Option, /// The documents file, or - for stdin. file: String, }, @@ -226,7 +231,7 @@ fn run(settings: &Settings, command: Command) -> Result { let ask = recipe.ask()?; query(settings, door::selection_value(&ask.selection)?, ask.semantic, ask.page.cursor, ask.page.documents) } - Command::Admit { faculty, agent, session, file } => admit(settings, faculty, agent, session, &file), + Command::Admit { faculty, agent, session, seen, file } => admit(settings, faculty, agent, session, seen, &file), 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) @@ -414,11 +419,38 @@ fn show(page: &PipelineSelectionPage) { say(format_args!("matched {} as_of {} next {}", page.matched, page.as_of, page.next.as_deref().unwrap_or("-"))); } -fn admit(settings: &Settings, faculty: Faculty, agent: String, session: String, file: &str) -> Result { +fn admit( + settings: &Settings, + faculty: Faculty, + agent: String, + session: String, + seen: Option, + file: &str, +) -> Result { let documents = door::documents(&door::file_or_stdin(file)?)?; if documents.is_empty() || documents.len() > BATCH_MAX { return Err(Trouble::invalid(format!("a batch is 1 to {BATCH_MAX} documents, this one has {}", documents.len()))); } + let token = admit_hold::token(&documents, &agent); + match seen { + Some(seen) if seen != token => { + return Err(Trouble::invalid("--seen is not the token of this batch and agent; run without it to get one")); + } + Some(_) => {} + None => { + let call = |selection, semantic| { + settings.ask(|instance| HuginnMindRequest::Query { instance: instance.clone(), selection, semantic: Some(semantic) }) + }; + match admit_hold::check(call, &documents, &token) { + Ok(Some(held)) => { + say(held.trim_end()); + return Ok(6); + } + Ok(None) => {} + Err(trouble) => eprintln!("near-neighbour check did not run ({}); this admit was not held", trouble.code()), + } + } + } let sent = documents.clone(); let provenance = PipelineProvenance { faculty, agent: Short(agent), session: Short(session), tool: Short(TOOL.to_string()) }; let request = |instance: &Slug| { diff --git a/crates/eureka-state/src/bin/huginn/near.rs b/crates/eureka-state/src/bin/huginn/near.rs new file mode 100644 index 0000000..0fd525b --- /dev/null +++ b/crates/eureka-state/src/bin/huginn/near.rs @@ -0,0 +1,165 @@ +//! The one neighbour query of the CLI: the in-force documents of some schemas +//! nearest a text, minus the ones the caller already holds. `admit_hold` asks +//! it before an admission; recall dispatch asks it next. It holds no client +//! and no policy beyond the rule here: the call is the caller's, so each +//! caller brings its own deadline. + +use cultnet_rs::{PROJECTION_DOCUMENT, Selection}; +use huginn_mind::eureka_pipeline::Line; +use huginn_mind::{HuginnMindResponse, PipelineDocumentView, PipelinePageItems, PipelineSelectionPage, SemanticQuery}; +use serde_json::json; + +use crate::trouble::Trouble; + +/// The query text is cut here, at a character boundary. +const TEXT_MAX_BYTES: usize = 1000; + +/// The in-force documents of `schemas` nearest `text`, as (score, document) +/// pairs: `excluded` ids dropped, at most `k` kept, nearest first as the mind +/// ranked them. A score the answer does not carry is 0. The mind is asked for +/// `k` plus the excluded count, so exclusions cannot leave fewer than `k`. +pub fn neighbours( + call: impl FnOnce(Selection, SemanticQuery) -> Result, + text: &str, + schemas: &[&str], + excluded: &[String], + k: usize, +) -> Result, Trouble> { + let selection = crate::door::selection_value(&json!({ + "schemas": schemas, + "projection": PROJECTION_DOCUMENT, + "fields": [{ "index": "in_force", "op": "any_of", "values": ["true"] }], + }))?; + let top_k = u32::try_from(k + excluded.len()).unwrap_or(u32::MAX); + let semantic = SemanticQuery { text: Line(cut(text, TEXT_MAX_BYTES).to_string()), top_k }; + let PipelineSelectionPage { items, scores, .. } = match call(selection, semantic)? { + HuginnMindResponse::Query(page) => page, + HuginnMindResponse::Refused(refusal) => return Err(Trouble::Internal(format!("the mind refused the query: {refusal:?}"))), + _ => return Err(Trouble::Internal("the mind answered a query with something else".to_string())), + }; + let PipelinePageItems::Documents(views) = items else { + return Err(Trouble::Internal("a document query answered headers".to_string())); + }; + let scores = scores.unwrap_or_default(); + Ok(views + .into_iter() + .enumerate() + .filter(|(_, view)| !excluded.contains(&view.id.id.0)) + .map(|(index, view)| (scores.get(index).copied().unwrap_or(0.0), view)) + .take(k) + .collect()) +} + +/// `text` cut to at most `max` bytes, backing off to a character boundary. +fn cut(text: &str, max: usize) -> &str { + &text[..text.floor_char_boundary(max)] +} + +#[cfg(test)] +mod tests { + use std::cell::RefCell; + + use huginn_mind::eureka_pipeline::{Date, PipelineDocument, PipelineInstance, PipelineRef, Short, Slug, pipeline_key}; + use huginn_mind::{AdmissionFacts, Faculty, PipelineProvenance, PipelineStatus}; + + use super::*; + + fn view(name: &str) -> PipelineDocumentView { + let document = PipelineDocument::Instance(PipelineInstance { + instance: Slug(name.into()), + display_name: Short(name.into()), + created_at: Date("2026-10-10".into()), + host: Short(name.into()), + }); + PipelineDocumentView { + id: PipelineRef { kind: document.kind(), id: Short(pipeline_key(&document).unwrap()) }, + document, + admission: AdmissionFacts { + receipt_id: "r".into(), + admitted_at: "2026-10-10T00:00:00Z".into(), + provenance: PipelineProvenance { + faculty: Faculty::Hands, + agent: Short("a".into()), + session: Short("s".into()), + tool: Short("huginn".into()), + }, + ordinal: 1, + }, + status: PipelineStatus::InForce, + } + } + + fn page(names: &[&str], scores: Option>) -> HuginnMindResponse { + HuginnMindResponse::Query(PipelineSelectionPage { + matched: 0, + as_of: 1, + next: None, + items: PipelinePageItems::Documents(names.iter().map(|name| view(name)).collect()), + edges: None, + scores, + }) + } + + #[test] + fn the_mind_is_asked_for_k_plus_the_excluded_and_the_cut_keeps_k_nearest_first() { + let asked = RefCell::new(None); + let found = neighbours( + |selection, semantic| { + *asked.borrow_mut() = Some((selection, semantic)); + Ok(page(&["a", "b", "c", "d", "e"], Some(vec![0.9, 0.8, 0.7, 0.6, 0.5]))) + }, + "words", + &["x:y", "z:w"], + &[view("b").id.id.0, view("d").id.id.0], + 2, + ) + .unwrap(); + let (selection, semantic) = asked.into_inner().unwrap(); + assert_eq!((semantic.text.0.as_str(), semantic.top_k), ("words", 4)); + assert_eq!(selection.schemas, Some(vec!["x:y".to_string(), "z:w".to_string()])); + assert_eq!(selection.projection, "document"); + let names: Vec<(f32, String)> = found.iter().map(|(score, view)| (*score, view.id.id.0.clone())).collect(); + assert_eq!(names, [(0.9, view("a").id.id.0), (0.7, view("c").id.id.0)]); + } + + #[test] + fn a_refusal_or_another_answer_is_a_trouble_that_says_which() { + let refused = neighbours(|_, _| Ok(HuginnMindResponse::Refused(huginn_mind::MindRefusal::MissingIdentity)), "t", &["s"], &[], 1); + assert!(matches!(&refused, Err(Trouble::Internal(detail)) if detail.contains("refused the query") && detail.contains("MissingIdentity")), "{refused:?}"); + let other = neighbours(|_, _| Ok(HuginnMindResponse::Persona(None)), "t", &["s"], &[], 1); + assert!(matches!(&other, Err(Trouble::Internal(detail)) if detail.contains("something else")), "{other:?}"); + } + + #[test] + fn a_missing_score_is_zero() { + let found = neighbours(|_, _| Ok(page(&["a", "b"], Some(vec![0.5]))), "t", &["s"], &[], 5).unwrap(); + assert_eq!(found.iter().map(|(score, _)| *score).collect::>(), [0.5, 0.0]); + let found = neighbours(|_, _| Ok(page(&["a"], None)), "t", &["s"], &[], 5).unwrap(); + assert_eq!(found[0].0, 0.0); + } + + #[test] + fn the_text_is_cut_to_a_thousand_bytes_at_a_character_boundary() { + let sent = RefCell::new(String::new()); + let ask = |text: &str| { + neighbours( + |_, semantic| { + *sent.borrow_mut() = semantic.text.0; + Ok(page(&[], None)) + }, + text, + &["s"], + &[], + 1, + ) + .unwrap(); + sent.borrow().clone() + }; + assert_eq!(ask(&"a".repeat(1000)).len(), 1000); + assert_eq!(ask(&"a".repeat(1001)).len(), 1000); + assert_eq!(ask("short"), "short"); + // 999 bytes then a 3-byte character straddling the limit: it is dropped whole. + let straddling = format!("{}€{}", "a".repeat(999), "b"); + assert_eq!(ask(&straddling), "a".repeat(999)); + } +} diff --git a/crates/eureka-state/tests/cli.rs b/crates/eureka-state/tests/cli.rs index ac2c737..b9aaa14 100644 --- a/crates/eureka-state/tests/cli.rs +++ b/crates/eureka-state/tests/cli.rs @@ -496,6 +496,7 @@ fn a_refused_read_back_is_the_minds_no_with_the_receipt_printed() { writes: vec![], }), ), + HuginnMindRequest::Query { semantic: Some(_), .. } => answer("query", &documents_page(vec![], None)), _ => answer("query", &HuginnMindResponse::Refused(MindRefusal::MissingIdentity)), } }); @@ -538,6 +539,8 @@ fn committing(stored: Vec) -> Server { writes: vec![], }), ), + // The near-neighbour check finds no neighbour. + HuginnMindRequest::Query { semantic: Some(_), .. } => answer("query", &documents_page(vec![], None)), HuginnMindRequest::Query { selection, .. } => { assert_eq!(selection.projection, "document"); assert_eq!(selection.limit, Some(u32::try_from(selection.keys.as_ref().unwrap().len()).unwrap())); @@ -1465,3 +1468,312 @@ fn a_persona_that_is_not_one_object_is_a_usage_error_and_sends_nothing() { assert!(ran.err.starts_with("InvalidInput: "), "{}", ran.err); assert!(seen.lock().unwrap().is_empty()); } + +fn ruling_of(label: &str, title: &str) -> PipelineDocument { + let PipelineDocument::Ruling(mut ruling) = ruling(None) else { unreachable!() }; + (ruling.label, ruling.title) = (label.into(), title.into()); + PipelineDocument::Ruling(ruling) +} + +fn question_of(campaign: &str, label: &str, title: &str) -> PipelineDocument { + PipelineDocument::Question(PipelineQuestion { + campaign: slug(campaign), + label: label.into(), + title: title.into(), + question: "Which?".into(), + options: ["a", "b"].iter().map(|name| QuestionOption { label: (*name).into(), text: "an option".into() }).collect(), + recommended: "a".into(), + depends: vec![], + raised_in: None, + asked_on: Date("2026-10-10".into()), + }) +} + +fn for_lines(page: &str) -> Vec<&str> { + page.lines().filter(|line| line.starts_with("For ")).collect() +} + +fn committed() -> HuginnMindResponse { + HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { + receipt_id: "mind-commit-x".into(), + committed_at: "2026-10-10T00:00:00Z".into(), + writes: vec![], + }) +} + +fn documents_page(views: Vec, scores: Option>) -> HuginnMindResponse { + HuginnMindResponse::Query(PipelineSelectionPage { + matched: 0, + as_of: 2, + next: None, + items: PipelinePageItems::Documents(views), + edges: None, + scores, + }) +} + +/// A scripted daemon for the near-neighbour hold: a semantic query is answered +/// with `neighbours` (scores 0.9, 0.8, ...), an admission commits what it was +/// sent and the read-back returns it. Every request is recorded. +fn holding(neighbours: Vec) -> (Server, Arc>>) { + let seen = Arc::new(Mutex::new(Vec::new())); + let sink = Arc::clone(&seen); + let stored = Mutex::new(Vec::new()); + let server = scripted(move |message| { + let (_, request) = decode_request(message).expect("a mind request"); + sink.lock().unwrap().push(request.clone()); + match request { + HuginnMindRequest::Admit(batch) => { + *stored.lock().unwrap() = batch.documents.iter().map(view_of).collect(); + answer("admit", &committed()) + } + HuginnMindRequest::Query { semantic: Some(_), .. } => { + let scores = (0..neighbours.len()).map(|n| 0.9 - n as f32 / 10.0).collect(); + answer("query", &documents_page(neighbours.clone(), Some(scores))) + } + HuginnMindRequest::Query { .. } => answer("query", &documents_page(stored.lock().unwrap().clone(), None)), + other => panic!("unexpected {other:?}"), + } + }); + (server, seen) +} + +fn semantic_queries(seen: &Mutex>) -> Vec<(Selection, String, u32)> { + seen.lock() + .unwrap() + .iter() + .filter_map(|request| match request { + HuginnMindRequest::Query { selection, semantic: Some(asked), .. } => { + Some((selection.clone(), asked.text.0.clone(), asked.top_k)) + } + _ => None, + }) + .collect() +} + +fn admissions(seen: &Mutex>) -> usize { + seen.lock().unwrap().iter().filter(|request| matches!(request, HuginnMindRequest::Admit(_))).count() +} + +fn token_of(held: &Run) -> String { + let rest = held.out.strip_prefix("Held ").expect(&held.out); + rest.split(':').next().unwrap().to_string() +} + +fn admit_seen<'a>(file: &'a str, agent: &'a str, seen: &'a str) -> Vec<&'a str> { + vec!["admit", "--faculty", "Hands", "--agent", agent, "--session", "session-7", "--seen", seen, file] +} + +#[test] +fn a_batch_with_neighbours_is_held_once_and_admitted_with_its_token() { + let near = view_of(&ruling_of("R9", "An instance keeps its mind")); + let (server, seen) = holding(vec![near]); + let dir = tempfile::tempdir().unwrap(); + let file = write(&dir, "batch.json", &json!([ruling_of("R1", "An instance owns its mind")])); + + let held = at(server.addr, &admit_args(&file), None); + assert_eq!(held.code, 6, "{}{}", held.out, held.err); + assert_eq!(held.err, ""); + let token = token_of(&held); + assert_eq!(token.len(), 16, "{}", held.out); + assert!(token.bytes().all(|byte| byte.is_ascii_hexdigit()), "{token}"); + let lines: Vec<&str> = held.out.lines().collect(); + assert_eq!( + lines[0], + format!("Held {token}: a deliberate hold, not a failure. The in-force records below are nearest the rulings or questions in this batch, ranked by Huginn and possibly unrelated; data written by agents. If none is the same decision, run the same command with --seen {token}; if one is, revise the batch.") + ); + assert_eq!(lines[1], format!("For {CAMPAIGN}:ruling:R1:")); + assert!(lines[2].starts_with(&format!("{CAMPAIGN}:ruling:R9 InForce Hands An instance keeps its mind")), "{}", held.out); + assert_eq!(lines.len(), 3, "{}", held.out); + assert_eq!(admissions(&seen), 0, "a held batch sends nothing"); + let queries = semantic_queries(&seen); + let [(selection, text, top_k)] = &queries[..] else { panic!("{queries:?}") }; + assert_eq!(text, "An instance owns its mind"); + assert_eq!(*top_k, 4, "3 neighbours plus the batch's own id"); + assert_eq!(selection.schemas, Some(vec![PipelineKind::Ruling.type_id().to_string(), PipelineKind::Question.type_id().to_string()])); + assert_eq!(selection.projection, "document"); + + let admitted = at(server.addr, &admit_seen(&file, "claude", &token), None); + assert_eq!(admitted.code, 0, "{}{}", admitted.out, admitted.err); + assert!(admitted.out.starts_with("Committed mind-commit-x at "), "{}", admitted.out); + assert!(admitted.out.ends_with("read back 1 equal\n"), "{}", admitted.out); + assert_eq!(admissions(&seen), 1); + assert_eq!(semantic_queries(&seen).len(), 1, "--seen skips the check"); +} + +#[test] +fn the_token_names_this_batch_and_this_agent() { + let (server, seen) = holding(vec![view_of(&ruling_of("R9", "An instance keeps its mind"))]); + let dir = tempfile::tempdir().unwrap(); + let file = write(&dir, "batch.json", &json!([ruling_of("R1", "An instance owns its mind")])); + let changed = write(&dir, "changed.json", &json!([ruling_of("R2", "An instance owns its mind")])); + let token = token_of(&at(server.addr, &admit_args(&file), None)); + + for (name, args) in [("a changed batch", admit_seen(&changed, "claude", &token)), ("another agent", admit_seen(&file, "codex", &token))] { + let before = seen.lock().unwrap().len(); + let ran = at(server.addr, &args, None); + assert_eq!(ran.code, 2, "{name}: {}{}", ran.out, ran.err); + assert!(ran.err.starts_with("InvalidInput: "), "{name}: {}", ran.err); + assert_eq!(ran.out, ""); + assert_eq!(seen.lock().unwrap().len(), before, "{name}: nothing is sent"); + } + let wrong = at(server.addr, &admit_seen(&file, "claude", "0000000000000000"), None); + assert_eq!(wrong.code, 2, "{}", wrong.out); + + // Without a token the changed batch and the other agent are held again, under tokens of their own. + let again = at(server.addr, &admit_args(&changed), None); + assert_eq!(again.code, 6); + assert_ne!(token_of(&again), token); + let other = at(server.addr, &["admit", "--faculty", "Hands", "--agent", "codex", "--session", "session-7", &file], None); + assert_eq!(other.code, 6); + assert_ne!(token_of(&other), token); + assert_ne!(token_of(&other), token_of(&again)); +} + +#[test] +fn the_batch_and_what_it_closes_are_not_neighbours_of_itself() { + let mut answer = ruling_of("R1", "An instance owns its mind"); + let PipelineDocument::Ruling(inner) = &mut answer else { unreachable!() }; + (inner.answers, inner.choice) = (Some(Short("other-campaign:question:Q9".into())), Some("a".into())); + let answered = question_of("other-campaign", "Q9", "Which owner?"); + let superseding = PipelineDocument::Resolution(PipelineResolution { + subject: PipelineRef { kind: PipelineKind::Ruling, id: Short(format!("{CAMPAIGN}:ruling:S1")) }, + sequence: 1, + outcome: ResolutionOutcome::Superseded { + by: vec![PipelineRef { kind: PipelineKind::Ruling, id: Short(format!("{CAMPAIGN}:ruling:R1")) }], + }, + rationale: "So.".into(), + resolved_on: Date("2026-10-10".into()), + }); + // A resolution that does not supersede its subject leaves the subject a neighbour. + let withdrawing = PipelineDocument::Resolution(PipelineResolution { + subject: PipelineRef { kind: PipelineKind::Ruling, id: Short(format!("{CAMPAIGN}:ruling:W1")) }, + sequence: 1, + outcome: ResolutionOutcome::Withdrawn { reason: "moot".into() }, + rationale: "So.".into(), + resolved_on: Date("2026-10-10".into()), + }); + let question = question_of(CAMPAIGN, "Q1", "Which store?"); + let near = vec![ + view_of(&ruling_of("W1", "The withdrawn one")), + view_of(&answer), + view_of(&answered), + view_of(&ruling_of("S1", "The superseded one")), + view_of(&question), + view_of(&ruling_of("N1", "A fresh neighbour")), + ]; + let (server, seen) = holding(near); + let dir = tempfile::tempdir().unwrap(); + let file = write(&dir, "batch.json", &json!([answer, question, superseding, withdrawing])); + + let held = at(server.addr, &admit_args(&file), None); + assert_eq!(held.code, 6, "{}{}", held.out, held.err); + assert_eq!(for_lines(&held.out), [format!("For {CAMPAIGN}:ruling:R1:"), format!("For {CAMPAIGN}:question:Q1:")]); + assert_eq!(held.out.matches(&format!("{CAMPAIGN}:ruling:N1 ")).count(), 2, "the fresh neighbour under each: {}", held.out); + assert_eq!(held.out.matches(&format!("{CAMPAIGN}:ruling:W1 ")).count(), 2, "a withdrawn subject is still a neighbour: {}", held.out); + for gone in [ + format!("{CAMPAIGN}:ruling:R1 "), + format!("{CAMPAIGN}:question:Q1 "), + "other-campaign:question:Q9 ".to_string(), + format!("{CAMPAIGN}:ruling:S1 "), + ] { + assert!(!held.out.contains(&gone), "{gone} is not its own neighbour: {}", held.out); + } + // top_k asks for 3 plus the four ids the batch excludes: R1, Q1, the answered question and S1. + let queries = semantic_queries(&seen); + assert_eq!(queries.len(), 2); + assert!(queries.iter().all(|(_, _, top_k)| *top_k == 7), "{queries:?}"); + + // The mind answering only excluded records leaves nothing to hold. + let (alone, seen) = holding(vec![view_of(&answer), view_of(&question)]); + let admitted = at(alone.addr, &admit_args(&file), None); + assert_eq!(admitted.code, 0, "{}{}", admitted.out, admitted.err); + assert_eq!(admissions(&seen), 1); +} + +#[test] +fn only_the_first_three_rulings_and_questions_are_checked_and_the_rest_counted() { + let (server, seen) = holding(vec![view_of(&ruling_of("R9", "A neighbour"))]); + let dir = tempfile::tempdir().unwrap(); + let batch = json!([ + research_json(), + ruling_of("R1", "First"), + question_of(CAMPAIGN, "Q2", "Second"), + research_json(), + ruling_of("R3", "Third"), + ruling_of("R4", "Fourth"), + ruling_of("R5", "Fifth"), + ]); + let file = write(&dir, "batch.json", &batch); + let held = at(server.addr, &admit_args(&file), None); + assert_eq!(held.code, 6, "{}{}", held.out, held.err); + let texts: Vec = semantic_queries(&seen).into_iter().map(|(_, text, _)| text).collect(); + assert_eq!(texts, ["First", "Second", "Third"]); + assert_eq!(held.out.lines().last(), Some("2 rulings or questions over 3 were not checked."), "{}", held.out); + assert_eq!(for_lines(&held.out).len(), 3); +} + +#[test] +fn a_batch_without_a_ruling_or_question_asks_nothing_and_a_wrong_token_is_still_refused() { + let (server, seen) = holding(vec![view_of(&ruling_of("R9", "A neighbour"))]); + let dir = tempfile::tempdir().unwrap(); + let file = write(&dir, "batch.json", &json!([campaign_json(), research_json()])); + let ran = at(server.addr, &admit_args(&file), None); + assert_eq!(ran.code, 0, "{}{}", ran.out, ran.err); + assert!(semantic_queries(&seen).is_empty()); + assert_eq!(ran.err, ""); + let wrong = at(server.addr, &admit_seen(&file, "claude", "0000000000000000"), None); + assert_eq!(wrong.code, 2, "{}", wrong.out); +} + +#[test] +fn a_check_that_cannot_run_admits_and_says_so() { + let dir = tempfile::tempdir().unwrap(); + let file = write(&dir, "batch.json", &json!([ruling_of("R1", "An instance owns its mind")])); + + // The mind rejects the query; the admission itself goes through. + let stored = Mutex::new(Vec::new()); + let refusing = scripted(move |message| { + let (_, request) = decode_request(message).expect("a mind request"); + match request { + HuginnMindRequest::Query { semantic: Some(_), .. } => encode_failure( + "eureka-state-call", + "query", + &OperationFailure { code: "index-behind".into(), message: "the index is behind".into() }, + "scripted", + ), + HuginnMindRequest::Admit(batch) => { + *stored.lock().unwrap() = batch.documents.iter().map(view_of).collect(); + answer("admit", &committed()) + } + HuginnMindRequest::Query { .. } => answer("query", &documents_page(stored.lock().unwrap().clone(), None)), + other => panic!("unexpected {other:?}"), + } + }); + let ran = at(refusing.addr, &admit_args(&file), None); + assert_eq!(ran.code, 0, "{}{}", ran.out, ran.err); + assert_eq!(ran.err, "near-neighbour check did not run (Rejected); this admit was not held\n"); + assert!(ran.out.ends_with("read back 1 equal\n"), "{}", ran.out); + + // A garbled answer (a headers page where documents were asked) is the same. + let (server, _) = counting(HuginnMindResponse::Query(empty_page()), "query"); + let ran = at(server.addr, &admit_args(&file), None); + assert_eq!(ran.err.lines().next(), Some("near-neighbour check did not run (Internal); this admit was not held"), "{}", ran.err); + assert_eq!(ran.code, 3, "the admission got an answer of another operation: {}{}", ran.out, ran.err); + + // No daemon: the note, then the admission fails on its own. + let down = at(closed_port(), &admit_args(&file), None); + assert_eq!(down.code, 3, "{}{}", down.out, down.err); + assert!( + down.err.starts_with("near-neighbour check did not run (Unavailable); this admit was not held\nUnavailable: "), + "{}", + down.err + ); +} + +#[test] +fn admit_help_documents_the_hold() { + let help = run(&[], &["admit", "--help"], None); + let text = help.out.split_whitespace().collect::>().join(" "); + assert!(text.contains("--seen ") && text.contains("held once: exit 6"), "{text}"); +} diff --git a/crates/huginn-mind/src/query.rs b/crates/huginn-mind/src/query.rs index f6540fd..b20629a 100644 --- a/crates/huginn-mind/src/query.rs +++ b/crates/huginn-mind/src/query.rs @@ -298,7 +298,8 @@ impl PipelineFacts { } impl PipelineDocumentSummary { - fn of(view: &PipelineDocumentView) -> Self { + /// The header of a view: what a `headers` projection answers for it. + pub fn of(view: &PipelineDocumentView) -> Self { Self { id: view.id.clone(), admission: view.admission.clone(),