huginn admit: hold a batch once when its rulings or questions have in-force neighbours
cut huginn-admit-hold r2. admit decodes, then asks near::neighbours (the one neighbour query, for admit now and recall dispatch next) for the nearest in-force rulings and questions to each of the batch's first three titled rulings or questions, minus the batch's own ids, the question a ruling answers and the subject of a Superseded resolution. A neighbour left holds the batch: exit 6, the neighbours and a token (16 hex of sha256 over the decoded batch and the agent) printed, nothing sent. The same command with --seen <token> admits it; any other token is InvalidInput. A trouble during the check prints a note on stderr and the admission proceeds. PipelineDocumentSummary::of is public so the neighbours print as header lines. sha2 joins eureka-state (already in the lock). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
6c9581cb18
commit
373d96691f
7 changed files with 615 additions and 5 deletions
1
Cargo.lock
generated
1
Cargo.lock
generated
|
|
@ -648,6 +648,7 @@ dependencies = [
|
|||
"serde",
|
||||
"serde_json",
|
||||
"serde_path_to_error",
|
||||
"sha2",
|
||||
"tempfile",
|
||||
"tokio",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
98
crates/eureka-state/src/bin/huginn/admit_hold.rs
Normal file
98
crates/eureka-state/src/bin/huginn/admit_hold.rs
Normal file
|
|
@ -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 <token>`
|
||||
//! 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<HuginnMindResponse, Trouble>,
|
||||
documents: &[PipelineDocument],
|
||||
token: &str,
|
||||
) -> Result<Option<String>, 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<String, Trouble> {
|
||||
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<Vec<String>, 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)
|
||||
}
|
||||
|
|
@ -13,10 +13,12 @@
|
|||
//! 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.
|
||||
//! 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\": <kind>, \"value\": {...}}, or - for stdin; one document object alone is a batch of one. \
|
||||
`huginn schema` lists the kinds and `huginn schema <kind>` 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 <n> 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 <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).";
|
||||
|
||||
#[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<String>,
|
||||
/// The documents file, or - for stdin.
|
||||
file: String,
|
||||
},
|
||||
|
|
@ -226,7 +231,7 @@ fn run(settings: &Settings, command: Command) -> Result<u8, Trouble> {
|
|||
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<u8, Trouble> {
|
||||
fn admit(
|
||||
settings: &Settings,
|
||||
faculty: Faculty,
|
||||
agent: String,
|
||||
session: String,
|
||||
seen: Option<String>,
|
||||
file: &str,
|
||||
) -> Result<u8, Trouble> {
|
||||
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| {
|
||||
|
|
|
|||
165
crates/eureka-state/src/bin/huginn/near.rs
Normal file
165
crates/eureka-state/src/bin/huginn/near.rs
Normal file
|
|
@ -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<HuginnMindResponse, Trouble>,
|
||||
text: &str,
|
||||
schemas: &[&str],
|
||||
excluded: &[String],
|
||||
k: usize,
|
||||
) -> Result<Vec<(f32, PipelineDocumentView)>, 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<Vec<f32>>) -> 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::<Vec<_>>(), [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));
|
||||
}
|
||||
}
|
||||
|
|
@ -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<PipelineDocumentView>) -> 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<PipelineDocumentView>, scores: Option<Vec<f32>>) -> 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<PipelineDocumentView>) -> (Server, Arc<Mutex<Vec<HuginnMindRequest>>>) {
|
||||
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<HuginnMindRequest>>) -> 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<Vec<HuginnMindRequest>>) -> 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<String> = 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::<Vec<_>>().join(" ");
|
||||
assert!(text.contains("--seen <SEEN>") && text.contains("held once: exit 6"), "{text}");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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(),
|
||||
|
|
|
|||
Loading…
Reference in a new issue