From 8041d82fe1d65fbdbb05d9080be217c5f2e56108 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 00:36:18 +0200 Subject: [PATCH 01/12] eureka-state tests: the in-process daemon fixtures move to tests/common The daemon half of the client's fixtures (an in-process daemon over a temporary mind on a loopback port, and the scripted server) becomes tests/common/mod.rs, so the MCP binary's tests (Cut 13b) drive the same daemon instead of a second copy of it. No behaviour change; client.rs shrinks by the moved lines. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/tests/client.rs | 153 +-------------------- crates/eureka-state/tests/common/mod.rs | 174 ++++++++++++++++++++++++ 2 files changed, 177 insertions(+), 150 deletions(-) create mode 100644 crates/eureka-state/tests/common/mod.rs diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index 07f458a..1ef33e8 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -28,29 +28,12 @@ use huginn_mind::{ }; use tempfile::TempDir; -const INSTANCE: &str = "eureka"; +mod common; +use common::*; + const OTHER: &str = "thought-cage"; -const CAMPAIGN: &str = "eureka-state"; -const REPO: &str = "GameCult/Epiphany"; const TIMEOUT: Duration = Duration::from_secs(20); -fn slug(value: &str) -> Slug { - Slug(value.into()) -} - -fn sha() -> Sha { - Sha("5f98228d".into()) -} - -fn identity(name: &str) -> PipelineDocument { - PipelineDocument::Instance(PipelineInstance { - instance: slug(name), - display_name: Short(format!("{name} mind")), - created_at: Date("2026-09-29".into()), - host: Short(name.into()), - }) -} - fn stewardship() -> PipelineDocument { PipelineDocument::Stewardship(PipelineStewardship { instance: slug(INSTANCE), @@ -61,29 +44,6 @@ fn stewardship() -> PipelineDocument { }) } -fn campaign() -> PipelineDocument { - PipelineDocument::Campaign(PipelineCampaign { - slug: slug(CAMPAIGN), - title: Title("Eureka pipeline state".into()), - repos: vec![OrgRepo(REPO.into())], - working_branch: Short("hands/cut13a".into()), - target_doc: DocRef { path: Short("notes/target.md".into()), start_line: 1, end_line: 9, commit: sha() }, - }) -} - -fn batch(instance: &str, documents: Vec) -> PipelineAdmissionBatch { - PipelineAdmissionBatch { - instance: slug(instance), - provenance: PipelineProvenance { - faculty: Faculty::Hands, - agent: Short("claude".into()), - session: Short("session-1".into()), - tool: Short("admit".into()), - }, - documents, - } -} - /// A `Line` and a `Short` at the leaf's own bound. fn line(n: usize) -> Line { Line(format!("{n:0>1000}")) @@ -151,108 +111,6 @@ fn cut_ref(cut: &str) -> PipelineRef { PipelineRef { kind: PipelineKind::CutSpec, id: Short(format!("{CAMPAIGN}:cut_spec:cut-{cut}.r1")) } } -/// No index: the client's subject is the transport, not the projection. -struct Inert; - -impl IndexSink for Inert { - fn committed(&mut self, _mind: &Mind, _writes: &[PipelineRef]) -> anyhow::Result<()> { - Ok(()) - } - fn status(&self) -> IndexStatus { - IndexStatus::Current - } - fn search(&mut self, _text: &str, _top_k: u32) -> Result { - Err("this index cannot search".into()) - } - fn search_deadline(&self) -> Duration { - Duration::ZERO - } - fn searched(&mut self) -> Vec<(SearchTicket, Result)> { - Vec::new() - } - fn abandon(&mut self, _ticket: SearchTicket) {} -} - -type TestDaemon = Daemon; - -fn answered(handled: Handled) -> HuginnMindResponse { - match handled { - Handled::Answered(response) => response, - Handled::Searching(_) => panic!("the inert index cannot search"), - } -} - -/// A daemon over a temporary mind whose identity is admitted, and whatever -/// else `seed` admits. -fn mind(seed: Vec>) -> (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())); - assert!(matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{outcome:?}"); - } - (root, daemon) -} - -/// A server on a loopback port that stops when dropped. -struct Server { - addr: SocketAddr, - stopping: Arc, - thread: Option>, - /// The sessions a scripted server has seen end: what the client's - /// disconnect looks like from the daemon's side. - ended: Arc, - _root: Option, -} - -impl Drop for Server { - fn drop(&mut self) { - self.stopping.store(true, Ordering::Relaxed); - if let Some(thread) = self.thread.take() { - let _ = thread.join(); - } - } -} - -fn serve(root: TempDir, mut daemon: TestDaemon) -> Server { - let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap(); - let addr = hub.local_addr().unwrap(); - let stopping = Arc::new(AtomicBool::new(false)); - let loop_stopping = Arc::clone(&stopping); - let registry = schema_registry().unwrap(); - let thread = std::thread::spawn(move || { - run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()).unwrap(); - }); - Server { addr, stopping, thread: Some(thread), ended: Arc::default(), _root: Some(root) } -} - -/// A server that answers each schema message with whatever `script` says. -fn scripted(mut script: impl FnMut(&CultNetMessage) -> CultNetMessage + Send + 'static) -> Server { - let mut hub = bind("127.0.0.1:0".parse().unwrap(), "scripted").unwrap(); - let addr = hub.local_addr().unwrap(); - let stopping = Arc::new(AtomicBool::new(false)); - let loop_stopping = Arc::clone(&stopping); - let ended = Arc::new(AtomicUsize::new(0)); - let loop_ended = Arc::clone(&ended); - let thread = std::thread::spawn(move || { - while !loop_stopping.load(Ordering::Relaxed) { - hub.poll_resends().unwrap(); - while let Some(event) = hub.receive_event_once().unwrap() { - if matches!(event, CultNetRudpServerEvent::Disconnected { .. }) { - loop_ended.fetch_add(1, Ordering::Relaxed); - } - let CultNetRudpServerEvent::Frame { session, frame } = event else { continue }; - let message = - decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0).unwrap(); - hub.send_schema_message(&session, &script(&message)).unwrap(); - } - std::thread::sleep(Duration::from_millis(2)); - } - }); - Server { addr, stopping, thread: Some(thread), ended, _root: None } -} - -/// A scripted server that answers any operation with a deferral of `body` and /// serves its chunks faithfully. fn deferring(body: Vec) -> Server { deferring_in_chunks(body, 256 * 1024, Duration::ZERO) @@ -298,11 +156,6 @@ fn rejected(error: ClientError) -> (SocketAddr, String, String) { } } -/// A scripted server whose every answer is `reply`. -fn answering(reply: CultNetMessage) -> Server { - scripted(move |_| reply.clone()) -} - fn nothing_found(message_id: &str, operation: &str) -> CultNetMessage { encode_response(message_id, operation, &HuginnMindResponse::View(None), "scripted").unwrap() } diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs new file mode 100644 index 0000000..5729e68 --- /dev/null +++ b/crates/eureka-state/tests/common/mod.rs @@ -0,0 +1,174 @@ +//! The daemon half of the fixtures: an in-process Huginn daemon over a +//! temporary mind, on a loopback port, shared by the client and the server +//! tests. +#![allow(dead_code)] + +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::thread::JoinHandle; +use std::time::Duration; + +use chrono::Utc; +use cultnet_rs::{CultNetMessage, CultNetRudpServerEvent, CultNetWireContract, decode_cultnet_message_from_slice}; +use huginn_daemon::serve::{bind, run, schema_registry}; +use huginn_daemon::{Daemon, Handled, Hits, IndexSink, SearchTicket, ServeOptions}; +use huginn_mind::epiphany_pipeline::{ + Date, DocRef, OrgRepo, PipelineCampaign, PipelineDocument, PipelineInstance, Sha, Short, Slug, Title, +}; +use huginn_mind::{ + Faculty, HuginnMindRequest, HuginnMindResponse, IndexStatus, Mind, MindStore, OwnedRedbMessagePackBackingStore, + PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineProvenance, +}; +use tempfile::TempDir; + +pub const INSTANCE: &str = "eureka"; + +pub fn slug(value: &str) -> Slug { + Slug(value.into()) +} + +pub fn identity(name: &str) -> PipelineDocument { + PipelineDocument::Instance(PipelineInstance { + instance: slug(name), + display_name: Short(format!("{name} mind")), + created_at: Date("2026-09-29".into()), + host: Short(name.into()), + }) +} + +pub fn batch(instance: &str, documents: Vec) -> PipelineAdmissionBatch { + PipelineAdmissionBatch { + instance: slug(instance), + provenance: PipelineProvenance { + faculty: Faculty::Hands, + agent: Short("claude".into()), + session: Short("session-1".into()), + tool: Short("admit".into()), + }, + documents, + } +} + +/// No index: the client's subject is the transport, not the projection. +pub struct Inert; + +impl IndexSink for Inert { + fn committed(&mut self, _mind: &Mind, _writes: &[PipelineRef]) -> anyhow::Result<()> { + Ok(()) + } + fn status(&self) -> IndexStatus { + IndexStatus::Current + } + fn search(&mut self, _text: &str, _top_k: u32) -> Result { + Err("this index cannot search".into()) + } + fn search_deadline(&self) -> Duration { + Duration::ZERO + } + fn searched(&mut self) -> Vec<(SearchTicket, Result)> { + Vec::new() + } + fn abandon(&mut self, _ticket: SearchTicket) {} +} + +pub type TestDaemon = Daemon; + +pub fn answered(handled: Handled) -> HuginnMindResponse { + match handled { + Handled::Answered(response) => response, + Handled::Searching(_) => panic!("the inert index cannot search"), + } +} + +/// A daemon over a temporary mind whose identity is admitted, and whatever +/// else `seed` admits. +pub fn mind(seed: Vec>) -> (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())); + assert!(matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{outcome:?}"); + } + (root, daemon) +} + +/// A server on a loopback port that stops when dropped. +pub struct Server { + pub addr: SocketAddr, + pub stopping: Arc, + pub thread: Option>, + /// The sessions a scripted server has seen end: what the client's + /// disconnect looks like from the daemon's side. + pub ended: Arc, + pub _root: Option, +} + +impl Drop for Server { + fn drop(&mut self) { + self.stopping.store(true, Ordering::Relaxed); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + +pub fn serve(root: TempDir, mut daemon: TestDaemon) -> Server { + let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap(); + let addr = hub.local_addr().unwrap(); + let stopping = Arc::new(AtomicBool::new(false)); + let loop_stopping = Arc::clone(&stopping); + let registry = schema_registry().unwrap(); + let thread = std::thread::spawn(move || { + run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()).unwrap(); + }); + Server { addr, stopping, thread: Some(thread), ended: Arc::default(), _root: Some(root) } +} + +/// A server that answers each schema message with whatever `script` says. +pub fn scripted(mut script: impl FnMut(&CultNetMessage) -> CultNetMessage + Send + 'static) -> Server { + let mut hub = bind("127.0.0.1:0".parse().unwrap(), "scripted").unwrap(); + let addr = hub.local_addr().unwrap(); + let stopping = Arc::new(AtomicBool::new(false)); + let loop_stopping = Arc::clone(&stopping); + let ended = Arc::new(AtomicUsize::new(0)); + let loop_ended = Arc::clone(&ended); + let thread = std::thread::spawn(move || { + while !loop_stopping.load(Ordering::Relaxed) { + hub.poll_resends().unwrap(); + while let Some(event) = hub.receive_event_once().unwrap() { + if matches!(event, CultNetRudpServerEvent::Disconnected { .. }) { + loop_ended.fetch_add(1, Ordering::Relaxed); + } + let CultNetRudpServerEvent::Frame { session, frame } = event else { continue }; + let message = + decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0).unwrap(); + hub.send_schema_message(&session, &script(&message)).unwrap(); + } + std::thread::sleep(Duration::from_millis(2)); + } + }); + Server { addr, stopping, thread: Some(thread), ended, _root: None } +} + +/// A scripted server whose every answer is `reply`. +pub fn answering(reply: CultNetMessage) -> Server { + scripted(move |_| reply.clone()) +} + +pub const CAMPAIGN: &str = "eureka-state"; +pub const REPO: &str = "GameCult/Epiphany"; + +pub fn sha() -> Sha { + Sha("5f98228d".into()) +} + +pub fn campaign() -> PipelineDocument { + PipelineDocument::Campaign(PipelineCampaign { + slug: slug(CAMPAIGN), + title: Title("Eureka pipeline state".into()), + repos: vec![OrgRepo(REPO.into())], + working_branch: Short("hands/cut13a".into()), + target_doc: DocRef { path: Short("notes/target.md".into()), start_line: 1, end_line: 9, commit: sha() }, + }) +} From e0f07c3f53cbf9496bf12b3235832a642a27c3ef Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 00:36:33 +0200 Subject: [PATCH 02/12] eureka-state: the MCP server, four tools 1:1 with the wire The eureka-state binary serves whoami, admit, view and query over stdio (rmcp 2.2.0), each one HuginnClient call. EUREKA_INSTANCE and HUGINN_ENDPOINT are process configuration: the server fills in the instance and provenance.tool, so input can name neither. Bad configuration never stops startup; whoami reports it and every other tool refuses with the typed body. A mind's refusal is a normal result; Unavailable, Rejected and Misconfigured are results with isError. The query input schema bundles CultLib's selection schema into $defs (vendored; a test pins it to the cultnet-rs revision). Co-Authored-By: Claude Sonnet 5.5 --- Cargo.lock | 254 ++++++++++++ crates/eureka-state/Cargo.toml | 15 +- .../schemas/cultnet.selection.schema.json | 100 +++++ crates/eureka-state/src/main.rs | 260 +++++++++++++ crates/eureka-state/tests/mcp.rs | 366 ++++++++++++++++++ 5 files changed, 992 insertions(+), 3 deletions(-) create mode 100644 crates/eureka-state/schemas/cultnet.selection.schema.json create mode 100644 crates/eureka-state/src/main.rs create mode 100644 crates/eureka-state/tests/mcp.rs diff --git a/Cargo.lock b/Cargo.lock index f278655..71de095 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -52,6 +52,17 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -258,6 +269,40 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "darling" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d" +dependencies = [ + "darling_core", + "darling_macro", +] + +[[package]] +name = "darling_core" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9865a50f7c335f53564bb694ef660825eb8610e0a53d3e11bf1b0d3df31e03b0" +dependencies = [ + "ident_case", + "proc-macro2", + "quote", + "strsim", + "syn 2.0.119", +] + +[[package]] +name = "darling_macro" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" +dependencies = [ + "darling_core", + "quote", + "syn 2.0.119", +] + [[package]] name = "der" version = "0.7.10" @@ -341,8 +386,13 @@ dependencies = [ "cultnet-rs", "huginn-daemon", "huginn-mind", + "rmcp", "rmp-serde", + "schemars", + "serde", + "serde_json", "tempfile", + "tokio", ] [[package]] @@ -373,12 +423,71 @@ dependencies = [ "winapi", ] +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", + "futures-sink", +] + [[package]] name = "futures-core" version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + [[package]] name = "futures-task" version = "0.3.34" @@ -391,8 +500,13 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -536,6 +650,12 @@ dependencies = [ "cc", ] +[[package]] +name = "ident_case" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" + [[package]] name = "inout" version = "0.1.4" @@ -607,6 +727,12 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" +[[package]] +name = "pastey" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ee67f1008b1ba2321834326597b8e186293b049a023cdef258527550b9935b4" + [[package]] name = "percent-encoding" version = "2.3.2" @@ -747,6 +873,40 @@ dependencies = [ "syn 3.0.5", ] +[[package]] +name = "rmcp" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "14db48ee17a9ba61810ab1a9c1beb7d06d8136ae39ac25a1137f10d357af01af" +dependencies = [ + "async-trait", + "chrono", + "futures", + "pastey", + "pin-project-lite", + "rmcp-macros", + "schemars", + "serde", + "serde_json", + "thiserror", + "tokio", + "tokio-util", + "tracing", +] + +[[package]] +name = "rmcp-macros" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "783d787bf21813b285f13019adc49e11af501c658890c1e519f31f937c68b7e3" +dependencies = [ + "darling", + "proc-macro2", + "quote", + "serde_json", + "syn 2.0.119", +] + [[package]] name = "rmp" version = "0.8.15" @@ -811,6 +971,7 @@ version = "1.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a" dependencies = [ + "chrono", "dyn-clone", "ref-cast", "schemars_derive", @@ -972,6 +1133,12 @@ dependencies = [ "der", ] +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + [[package]] name = "subtle" version = "2.6.1" @@ -1013,6 +1180,93 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "thiserror" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "bytes", + "pin-project-lite", + "tokio-macros", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "tokio-util" +version = "0.7.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + [[package]] name = "typenum" version = "1.20.1" diff --git a/crates/eureka-state/Cargo.toml b/crates/eureka-state/Cargo.toml index 0de0909..be76e34 100644 --- a/crates/eureka-state/Cargo.toml +++ b/crates/eureka-state/Cargo.toml @@ -5,18 +5,27 @@ edition.workspace = true license.workspace = true publish.workspace = true -# The client core: one call over CultNet to a Huginn daemon. It owns the -# transport and nothing of the mind. The operation envelope, the wire +# The client core (`lib.rs`): one call over CultNet to a Huginn daemon. It owns +# the transport and nothing of the mind. The operation envelope, the wire # vocabulary and the deferred-body bound are `huginn-mind`'s, so the client and # the daemon cannot spell them differently; the content fetch and its # verification are `cultnet-rs`'s. +# +# The MCP server (`main.rs`, the `eureka-state` binary) is the client's one +# consumer: four tools over stdio, one per wire operation, holding no state. [dependencies] anyhow = "1" cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultnet-rs" } huginn-mind = { path = "../huginn-mind" } +rmcp = { version = "2.2.0", default-features = false, features = ["server", "macros", "transport-io"] } rmp-serde = "1" +schemars = "1" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { version = "1", features = ["rt", "macros", "io-std"] } -# Tests run a real daemon in-process against a temporary mind. +# Tests run a real daemon in-process against a temporary mind, and the real +# binary against it. [dev-dependencies] chrono = "0.4.44" huginn-daemon = { path = "../huginn-daemon" } diff --git a/crates/eureka-state/schemas/cultnet.selection.schema.json b/crates/eureka-state/schemas/cultnet.selection.schema.json new file mode 100644 index 0000000..7076428 --- /dev/null +++ b/crates/eureka-state/schemas/cultnet.selection.schema.json @@ -0,0 +1,100 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/GameCult/CultLib/contracts/cultnet/cultnet.selection.schema.json", + "title": "CultNet Typed Selection", + "type": "object", + "additionalProperties": false, + "properties": { + "schemas": { + "type": "array", + "items": { "type": "string", "minLength": 1 }, + "minItems": 1, + "description": "Absent means every schema. Present and empty is refused at the door (selection_invalid), never treated as \"every schema\": omit the field instead." + }, + "keys": { + "type": "array", + "items": { "type": "string", "minLength": 1 }, + "minItems": 1, + "description": "Absent means every key. Present and empty is refused at the door (selection_invalid), never treated as \"every key\": omit the field instead." + }, + "fields": { + "type": "array", + "items": { "$ref": "#/$defs/fieldPredicate" } + }, + "cites": { "$ref": "#/$defs/citation" }, + "cited": { "$ref": "#/$defs/incoming" }, + "projection": { + "type": "string", + "enum": ["header", "document"], + "default": "header" + }, + "descending": { "type": "boolean", "default": false }, + "limit": { + "type": "integer", + "minimum": 1, + "maximum": 200, + "description": "Clamped to 1..=200 by the answering server, not refused outside that range: an absent value or one outside the bound is treated as the nearer bound (absent = 200)." + }, + "cursor": { "type": "string" } + }, + "$defs": { + "recordRef": { + "type": "object", + "required": ["schemaId", "recordKey"], + "additionalProperties": false, + "properties": { + "schemaId": { "type": "string", "minLength": 1 }, + "recordKey": { "type": "string", "minLength": 1 } + } + }, + "fieldPredicate": { + "type": "object", + "required": ["index", "op"], + "additionalProperties": false, + "properties": { + "index": { "type": "string", "minLength": 1 }, + "op": { "type": "string", "enum": ["any_of", "lt", "le", "ge", "gt"] }, + "values": { + "type": "array", + "items": { "type": "string" }, + "minItems": 1 + }, + "number": { + "type": "string", + "description": "Q-J: a canonical decimal string, never a JSON number. Matches ^-?(0|[1-9][0-9]*)(\\.[0-9]*[1-9])?$ and is not \"-0\"; any other spelling is refused typed at the door (selection_invalid), never normalised.", + "pattern": "^(?!-0$)-?(0|[1-9][0-9]*)(\\.[0-9]*[1-9])?$" + } + }, + "oneOf": [ + { + "required": ["values"], + "not": { "required": ["number"] }, + "properties": { "op": { "const": "any_of" } } + }, + { + "required": ["number"], + "not": { "required": ["values"] }, + "properties": { "op": { "enum": ["lt", "le", "ge", "gt"] } } + } + ] + }, + "citation": { + "type": "object", + "required": ["target"], + "additionalProperties": false, + "properties": { + "target": { "$ref": "#/$defs/recordRef" }, + "role": { "type": "string", "minLength": 1 } + } + }, + "incoming": { + "type": "object", + "required": ["role", "exists"], + "additionalProperties": false, + "properties": { + "role": { "type": "string", "minLength": 1 }, + "exists": { "type": "boolean" } + } + } + } +} diff --git a/crates/eureka-state/src/main.rs b/crates/eureka-state/src/main.rs new file mode 100644 index 0000000..a7f4a58 --- /dev/null +++ b/crates/eureka-state/src/main.rs @@ -0,0 +1,260 @@ +//! `eureka-state`: Huginn's mind as four MCP tools over stdio. +//! +//! One tool per wire operation (`whoami`, `admit`, `view`, `query`), each a +//! `HuginnClient` call and nothing else. The declared identity is this +//! process's configuration, not a per-call argument: `EUREKA_INSTANCE` names +//! the mind and `HUGINN_ENDPOINT` (`rudp://host:port`) the daemon that holds +//! it. The server fills in `instance` and `provenance.tool`, so a caller can +//! name neither. +//! +//! Bad configuration does not stop the server: an MCP server that dies at +//! startup vanishes from Claude Code without a word. `whoami` reports it and +//! every other tool refuses with it. A refusal from the mind is an answer and +//! comes back as a normal result; a daemon that cannot be reached or that +//! rejects the envelope is a result with `isError: true`, typed, so a caller +//! can tell a down daemon from a permanent mismatch. + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use cultnet_rs::Selection; +use eureka_state::{ClientError, HuginnClient}; +use huginn_mind::epiphany_pipeline::{PipelineDocument, PipelineRef, Short, Slug}; +use huginn_mind::{ + Faculty, HuginnMindRequest, HuginnMindResponse, PipelineAdmissionBatch, PipelineProvenance, SemanticQuery, +}; +use rmcp::handler::server::router::tool::ToolRouter; +use rmcp::handler::server::wrapper::Parameters; +use rmcp::model::{CallToolResult, ContentBlock, ServerCapabilities, ServerInfo}; +use rmcp::{ServerHandler, ServiceExt, tool, tool_handler, tool_router}; +use schemars::{JsonSchema, Schema, SchemaGenerator}; +use serde::Deserialize; +use serde_json::{Value, json}; + +/// The `provenance.tool` every admission from this server carries. +const TOOL: &str = "eureka-state"; +/// How long one call may take, connection to last byte. +const TIMEOUT: Duration = Duration::from_secs(15); + +/// CultLib's published shape of a selection, vendored because the selection +/// type carries no schema of its own. `selection_schema_matches_cultlib` pins +/// it to the CultLib revision `cultnet-rs` is built from. +const SELECTION_SCHEMA: &str = include_str!("../schemas/cultnet.selection.schema.json"); + +/// The selection's schema, self-contained: its own `$defs` move into the +/// tool's, so every `$ref` in the input schema is `#/$defs/...`. +fn selection_schema(generator: &mut SchemaGenerator) -> Schema { + let Ok(Value::Object(mut selection)) = serde_json::from_str(SELECTION_SCHEMA) else { + unreachable!("the vendored selection schema is a JSON object"); + }; + selection.remove("$schema"); + selection.remove("$id"); + if let Some(Value::Object(defs)) = selection.remove("$defs") { + generator.definitions_mut().extend(defs); + } + Schema::try_from(Value::Object(selection)).expect("an object is a schema") +} + +#[derive(Deserialize, JsonSchema)] +struct AdmitInput { + /// The faculty the batch is admitted as. Attribution, not authority. + faculty: Faculty, + /// The admitting agent's name. + agent: Short, + /// The admitting session, recorded on the receipt so that several sessions + /// sharing one mind stay tellable apart. + session: Short, + /// The documents to admit as one batch. The mind admits them whole or refuses them. + documents: Vec, +} + +#[derive(Deserialize, JsonSchema)] +struct ViewInput { + /// The document's id, as a query or an admission returned it. + id: PipelineRef, +} + +#[derive(Deserialize, JsonSchema)] +struct QueryInput { + /// A CultNet typed selection over the mind's documents. + #[schemars(schema_with = "selection_schema")] + selection: Selection, + /// Optional: rank by nearness to this text instead of ordering by key. + semantic: Option, +} + +/// Why a call produced no answer. Each is a typed body with `isError: true`. +enum Trouble { + Misconfigured(String), + Client(ClientError), + Internal(String), +} + +impl Trouble { + fn body(&self) -> Value { + match self { + Self::Misconfigured(detail) => json!({ "error": "Misconfigured", "detail": detail }), + Self::Client(ClientError::Unavailable { endpoint, detail }) => { + json!({ "error": "Unavailable", "endpoint": format!("rudp://{endpoint}"), "detail": detail }) + } + Self::Client(ClientError::Rejected { endpoint, code, detail }) => { + json!({ "error": "Rejected", "endpoint": format!("rudp://{endpoint}"), "code": code, "detail": detail }) + } + Self::Internal(detail) => json!({ "error": "Internal", "detail": detail }), + } + } +} + +/// The two environment values as given, and the client they make when both are +/// valid. +struct Settings { + instance: Option, + endpoint: Option, + client: Result, +} + +impl Settings { + fn from_env() -> Self { + let instance = std::env::var("EUREKA_INSTANCE").ok(); + let endpoint = std::env::var("HUGINN_ENDPOINT").ok(); + let client = client(instance.as_deref(), endpoint.as_deref()); + Self { instance, endpoint, client } + } +} + +fn client(instance: Option<&str>, endpoint: Option<&str>) -> Result { + let instance = instance.ok_or("EUREKA_INSTANCE is not set")?; + let slug = Slug(instance.to_string()); + 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::().ok()) + .ok_or_else(|| format!("HUGINN_ENDPOINT is not rudp://:: {endpoint}"))?; + Ok(HuginnClient::new(address, slug, TIMEOUT)) +} + +#[derive(Clone)] +struct EurekaState { + settings: Arc, + tool_router: ToolRouter, +} + +impl EurekaState { + fn new(settings: Settings) -> Self { + Self { settings: Arc::new(settings), tool_router: Self::tool_router() } + } + + /// One wire call for the configured instance. The request is built from + /// that instance, never from anything the caller sent. + async fn call(&self, request: impl FnOnce(Slug) -> HuginnMindRequest) -> Result { + let client = self.settings.client.as_ref().map_err(|detail| Trouble::Misconfigured(detail.clone()))?.clone(); + let request = request(client.instance().clone()); + tokio::task::spawn_blocking(move || client.call(request)) + .await + .map_err(|error| Trouble::Internal(error.to_string()))? + .map_err(Trouble::Client) + } + + async fn answer(&self, request: impl FnOnce(Slug) -> HuginnMindRequest) -> CallToolResult { + match self.call(request).await { + Ok(response) => reply(payload(response), false), + Err(trouble) => reply(trouble.body(), true), + } + } +} + +/// The answer as the tool's output: the operation's own payload, or the typed +/// refusal, tagged, when the mind refused the request. +fn payload(response: HuginnMindResponse) -> Value { + let value = match response { + HuginnMindResponse::Whoami(status) => serde_json::to_value(status), + HuginnMindResponse::Admit(outcome) => serde_json::to_value(outcome), + HuginnMindResponse::View(view) => serde_json::to_value(view), + HuginnMindResponse::Query(page) => serde_json::to_value(page), + other => serde_json::to_value(other), + }; + value.expect("wire types serialize") +} + +fn reply(body: Value, is_error: bool) -> CallToolResult { + let content = vec![ContentBlock::text(body.to_string())]; + if is_error { CallToolResult::error(content) } else { CallToolResult::success(content) } +} + +#[tool_router] +impl EurekaState { + #[tool( + description = "Who this server acts for and whether the Huginn daemon answers: the configured instance and endpoint, reachable, and the mind's status when it does. Reports bad configuration instead of failing." + )] + async fn whoami(&self) -> CallToolResult { + let mut out = json!({ + "instance": self.settings.instance, + "endpoint": self.settings.endpoint, + "reachable": false, + "status": null, + }); + match self.call(|_| HuginnMindRequest::Whoami).await { + Ok(HuginnMindResponse::Whoami(status)) => { + out["reachable"] = json!(true); + out["status"] = serde_json::to_value(status).expect("wire types serialize"); + } + Ok(other) => out["error"] = json!({ "error": "Internal", "detail": format!("unexpected answer: {other:?}") }), + Err(trouble) => out["error"] = trouble.body(), + } + reply(out, false) + } + + #[tool( + description = "Admit one batch of pipeline documents into the mind, whole or not at all. Returns the outcome: committed with a receipt, already admitted, conflict, or a typed refusal. The instance is this server's configuration; do not pass one." + )] + async fn admit(&self, Parameters(input): Parameters) -> CallToolResult { + self.answer(|instance| { + HuginnMindRequest::Admit(PipelineAdmissionBatch { + instance, + provenance: PipelineProvenance { + faculty: input.faculty, + agent: input.agent, + session: input.session, + tool: Short(TOOL.into()), + }, + documents: input.documents, + }) + }) + .await + } + + #[tool(description = "Read one document by id, with its admission facts and status. null when the mind has no such document.")] + async fn view(&self, Parameters(input): Parameters) -> CallToolResult { + self.answer(|instance| HuginnMindRequest::View { instance, id: input.id }).await + } + + #[tool( + description = "Query the mind with a CultNet typed selection, optionally ranked by semantic nearness. Returns one page; pass its cursor back in the selection for the next." + )] + async fn query(&self, Parameters(input): Parameters) -> CallToolResult { + self.answer(|instance| HuginnMindRequest::Query { + instance, + selection: input.selection, + semantic: input.semantic, + }) + .await + } +} + +#[tool_handler] +impl ServerHandler for EurekaState { + fn get_info(&self) -> ServerInfo { + ServerInfo::new(ServerCapabilities::builder().enable_tools().build()).with_instructions( + "Huginn's typed mind for this instance: whoami, admit, view, query. Refusals are answers; isError means the daemon could not be reached or rejected the request.", + ) + } +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> anyhow::Result<()> { + let server = EurekaState::new(Settings::from_env()).serve(rmcp::transport::stdio()).await?; + server.waiting().await?; + Ok(()) +} diff --git a/crates/eureka-state/tests/mcp.rs b/crates/eureka-state/tests/mcp.rs new file mode 100644 index 0000000..adbfcda --- /dev/null +++ b/crates/eureka-state/tests/mcp.rs @@ -0,0 +1,366 @@ +//! The `eureka-state` binary over stdio, as Claude Code drives it: real +//! processes, real JSON-RPC, a real daemon in-process behind it (or a scripted +//! server where the daemon cannot be made to misbehave). + +use std::collections::BTreeSet; +use std::io::{BufRead, BufReader, Write}; +use std::net::UdpSocket; +use std::process::{Child, ChildStdin, Command, Stdio}; +use std::sync::mpsc::{Receiver, channel}; +use std::time::Duration; + +use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response}; +use huginn_mind::HuginnMindResponse; +use huginn_mind::epiphany_pipeline::{PipelineKind, PipelineRef, Short}; +use serde_json::{Value, json}; + +mod common; +use common::*; + +const WAIT: Duration = Duration::from_secs(60); + +struct Mcp { + child: Child, + stdin: ChildStdin, + lines: Receiver, + next: u64, +} + +impl Drop for Mcp { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +impl Mcp { + /// The binary with exactly the environment given, initialised. + fn start(env: &[(&str, &str)]) -> Self { + let mut command = Command::new(env!("CARGO_BIN_EXE_eureka-state")); + command + .env_remove("EUREKA_INSTANCE") + .env_remove("HUGINN_ENDPOINT") + .envs(env.iter().copied()) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()); + let mut child = command.spawn().expect("the binary starts"); + let stdin = child.stdin.take().unwrap(); + let stdout = child.stdout.take().unwrap(); + let (sender, lines) = channel(); + std::thread::spawn(move || { + for line in BufReader::new(stdout).lines().map_while(Result::ok) { + let _ = sender.send(line); + } + }); + let mut mcp = Self { child, stdin, lines, next: 1 }; + let init = mcp.rpc( + "initialize", + json!({ "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": { "name": "test", "version": "0" } }), + ); + assert!(init.get("result").is_some(), "{init}"); + mcp.send(json!({ "jsonrpc": "2.0", "method": "notifications/initialized" })); + mcp + } + + /// Configured for `instance` against a daemon at `addr`. + fn at(instance: &str, addr: std::net::SocketAddr) -> Self { + Self::start(&[("EUREKA_INSTANCE", instance), ("HUGINN_ENDPOINT", &format!("rudp://{addr}"))]) + } + + fn send(&mut self, message: Value) { + writeln!(self.stdin, "{message}").unwrap(); + self.stdin.flush().unwrap(); + } + + /// One request and its response, whole (`result` or `error`). + fn rpc(&mut self, method: &str, params: Value) -> Value { + let id = self.next; + self.next += 1; + self.send(json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params })); + loop { + let line = self.lines.recv_timeout(WAIT).expect("the server answers"); + let message: Value = serde_json::from_str(&line).expect("stdout carries only JSON-RPC"); + if message["id"] == json!(id) { + return message; + } + } + } + + /// A tool call's result: whether it is an error, and its JSON body. + fn call(&mut self, tool: &str, arguments: Value) -> (bool, Value) { + let response = self.rpc("tools/call", json!({ "name": tool, "arguments": arguments })); + let result = &response["result"]; + assert!(result.is_object(), "a tool result, not a protocol error: {response}"); + let text = result["content"][0]["text"].as_str().expect("text content"); + (result["isError"] == json!(true), serde_json::from_str(text).expect("a JSON body")) + } +} + +/// A loopback UDP port nothing listens on. +fn closed_port() -> std::net::SocketAddr { + UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap() +} + +fn faculty_args(documents: Vec) -> Value { + json!({ "faculty": "Hands", "agent": "claude", "session": "session-7", "documents": documents }) +} + +fn campaign_json() -> Value { + serde_json::to_value(campaign()).unwrap() +} + +/// Every `$ref` in a schema, and the `$defs` it can resolve against. +fn refs(schema: &Value, found: &mut Vec) { + match schema { + Value::Object(map) => { + for (key, value) in map { + if key == "$ref" { + found.push(value.as_str().unwrap().to_string()); + } else { + refs(value, found); + } + } + } + Value::Array(items) => items.iter().for_each(|item| refs(item, found)), + _ => {} + } +} + +/// Four tools, each input schema a self-contained document: every `$ref` is +/// internal and resolves, and the selection's definitions are bundled in. +#[test] +fn tools_list_offers_four_tools_with_self_contained_schemas() { + let mut mcp = Mcp::at(INSTANCE, closed_port()); + let listed = mcp.rpc("tools/list", json!({})); + let tools = listed["result"]["tools"].as_array().expect("tools"); + let names: BTreeSet<&str> = tools.iter().map(|tool| tool["name"].as_str().unwrap()).collect(); + assert_eq!(names, BTreeSet::from(["whoami", "admit", "view", "query"])); + for tool in tools { + let schema = &tool["inputSchema"]; + let mut found = Vec::new(); + refs(schema, &mut found); + for reference in &found { + let name = reference.strip_prefix("#/$defs/").unwrap_or_else(|| panic!("{}: remote $ref {reference}", tool["name"])); + assert!(schema["$defs"].get(name).is_some(), "{}: {reference} does not resolve", tool["name"]); + } + assert!(schema.get("$id").is_none(), "{}: a remote identity", tool["name"]); + } + let query = tools.iter().find(|tool| tool["name"] == "query").unwrap(); + for bundled in ["fieldPredicate", "citation", "incoming", "recordRef"] { + assert!(query["inputSchema"]["$defs"].get(bundled).is_some(), "{bundled} is not bundled"); + } + assert_eq!(query["inputSchema"]["properties"]["selection"]["properties"]["limit"]["maximum"], json!(200)); +} + +/// The vendored selection schema is CultLib's, at the revision `cultnet-rs` +/// is built from. +#[test] +fn selection_schema_matches_cultlib() { + let cargo = std::env::var("CARGO").unwrap_or_else(|_| "cargo".into()); + let metadata = Command::new(cargo) + .args(["metadata", "--format-version", "1", "--offline", "--locked"]) + .current_dir(env!("CARGO_MANIFEST_DIR")) + .output() + .unwrap(); + assert!(metadata.status.success(), "{}", String::from_utf8_lossy(&metadata.stderr)); + let metadata: Value = serde_json::from_slice(&metadata.stdout).unwrap(); + let manifest = metadata["packages"] + .as_array() + .unwrap() + .iter() + .find(|package| package["name"] == "cultnet-rs") + .expect("cultnet-rs is a dependency")["manifest_path"] + .as_str() + .unwrap(); + let published = std::path::Path::new(manifest).join("../../../contracts/cultnet/cultnet.selection.schema.json"); + let theirs: Value = serde_json::from_str(&std::fs::read_to_string(published).unwrap()).unwrap(); + let ours: Value = serde_json::from_str( + &std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/schemas/cultnet.selection.schema.json")).unwrap(), + ) + .unwrap(); + assert_eq!(ours, theirs); +} + +/// Every tool against a live daemon, and what each returns. +#[test] +fn every_tool_round_trips_against_a_live_daemon() { + let (root, daemon) = mind(vec![]); + let server = serve(root, daemon); + let mut mcp = Mcp::at(INSTANCE, server.addr); + + let (error, whoami) = mcp.call("whoami", json!({})); + assert!(!error); + assert_eq!(whoami["instance"], json!(INSTANCE)); + assert_eq!(whoami["endpoint"], json!(format!("rudp://{}", server.addr))); + assert_eq!(whoami["reachable"], json!(true)); + assert_eq!(whoami["status"]["instance"], json!(INSTANCE)); + assert_eq!(whoami["status"]["documents"], json!(1)); + assert!(whoami.get("error").is_none(), "{whoami}"); + + let (error, admitted) = mcp.call("admit", faculty_args(vec![campaign_json()])); + assert!(!error); + let writes = admitted["Committed"]["writes"].as_array().expect("committed").clone(); + assert_eq!(writes.len(), 1, "{admitted}"); + + let (error, viewed) = mcp.call("view", json!({ "id": writes[0] })); + assert!(!error); + assert_eq!(viewed["id"], writes[0]); + assert_eq!(viewed["admission"]["provenance"]["session"], json!("session-7")); + + let (error, nothing) = mcp.call("view", json!({ "id": absent() })); + assert!(!error); + assert_eq!(nothing, Value::Null); + + let (error, page) = mcp.call("query", json!({ "selection": { "projection": "document" } })); + assert!(!error); + assert_eq!(page["matched"], json!(2), "{page}"); + assert_eq!(page["items"]["Documents"].as_array().unwrap().len(), 2); + + let (_, counted) = mcp.call("whoami", json!({})); + assert_eq!(counted["status"]["receipts"], json!(2)); +} + +/// Declared identity is configuration, not a per-call argument. Input that +/// names another instance, or another tool, changes neither: the batch is +/// admitted for the configured instance, and the receipt says `eureka-state`. +#[test] +fn admit_ignores_any_instance_in_input() { + let (root, daemon) = mind(vec![]); + let server = serve(root, daemon); + let mut mcp = Mcp::at(INSTANCE, server.addr); + + let mut arguments = faculty_args(vec![campaign_json()]); + arguments["instance"] = json!("thought-cage"); + arguments["tool"] = json!("someone-else"); + arguments["provenance"] = json!({ "faculty": "Operator", "agent": "x", "session": "y", "tool": "someone-else" }); + let (error, admitted) = mcp.call("admit", arguments); + assert!(!error); + let writes = admitted["Committed"]["writes"].as_array().unwrap_or_else(|| panic!("not committed: {admitted}")); + + let (_, viewed) = mcp.call("view", json!({ "id": writes[0] })); + let provenance = &viewed["admission"]["provenance"]; + assert_eq!(provenance["tool"], json!("eureka-state")); + assert_eq!(provenance["faculty"], json!("Hands")); + assert_eq!(provenance["agent"], json!("claude")); + assert_eq!(provenance["session"], json!("session-7")); +} + +/// A mind's refusal is an answer: a successful result carrying the typed +/// refusal. Here the configured instance is not the daemon's. +#[test] +fn a_refusal_is_a_successful_result() { + let (root, daemon) = mind(vec![]); + let server = serve(root, daemon); + let mut mcp = Mcp::at("thought-cage", server.addr); + + let (error, viewed) = mcp.call("view", json!({ "id": absent() })); + assert!(!error); + assert!(viewed.get("Refused").is_some(), "{viewed}"); + let (error, queried) = mcp.call("query", json!({ "selection": {} })); + assert!(!error); + assert!(queried.get("Refused").is_some(), "{queried}"); + let (error, admitted) = mcp.call("admit", faculty_args(vec![campaign_json()])); + assert!(!error); + assert!(admitted.get("Refused").is_some(), "{admitted}"); +} + +/// A daemon that cannot be reached is `isError` with the endpoint, and +/// `whoami` reports it as `reachable: false` without a protocol error. +#[test] +fn a_closed_port_is_reported_and_is_not_a_protocol_error() { + let addr = closed_port(); + let mut mcp = Mcp::at(INSTANCE, addr); + + let whoami = mcp.rpc("tools/call", json!({ "name": "whoami", "arguments": {} })); + assert!(whoami.get("error").is_none(), "{whoami}"); + assert_ne!(whoami["result"]["isError"], json!(true)); + let body: Value = serde_json::from_str(whoami["result"]["content"][0]["text"].as_str().unwrap()).unwrap(); + assert_eq!(body["reachable"], json!(false)); + assert_eq!(body["status"], Value::Null); + assert_eq!(body["error"]["error"], json!("Unavailable")); + + let (error, view) = mcp.call("view", json!({ "id": absent() })); + assert!(error); + assert_eq!(view["error"], json!("Unavailable")); + assert_eq!(view["endpoint"], json!(format!("rudp://{addr}"))); +} + +/// A daemon rejecting the envelope is a permanent mismatch, not a down +/// daemon: `isError` with its own code, distinct from `Unavailable`. +#[test] +fn a_rejected_request_is_a_typed_error_distinct_from_unavailable() { + let server = answering(encode_failure( + "eureka-state-call", + "view", + &OperationFailure { code: "wrong-service".into(), message: "not huginn.mind".into() }, + "scripted", + )); + let mut mcp = Mcp::at(INSTANCE, server.addr); + let (error, body) = mcp.call("view", json!({ "id": absent() })); + assert!(error); + assert_eq!(body["error"], json!("Rejected")); + assert_eq!(body["code"], json!("wrong-service")); + assert_eq!(body["detail"], json!("not huginn.mind")); + assert_eq!(body["endpoint"], json!(format!("rudp://{}", server.addr))); +} + +/// A `whoami` answered with another operation's payload is not a mind's +/// status: it is reported, not called reachable. +#[test] +fn whoami_does_not_take_another_answer_for_a_status() { + let server = answering(encode_response("eureka-state-call", "whoami", &HuginnMindResponse::View(None), "scripted").unwrap()); + let mut mcp = Mcp::at(INSTANCE, server.addr); + let (error, body) = mcp.call("whoami", json!({})); + assert!(!error); + assert_eq!(body["reachable"], json!(false)); + assert_eq!(body["status"], Value::Null); + assert!(body["error"]["detail"].as_str().unwrap().contains("unexpected answer"), "{body}"); +} + +/// Bad configuration does not stop the server. `whoami` says what is wrong and +/// every other tool refuses with the same typed body. +#[test] +fn bad_configuration_is_served_and_reported() { + let cases: [(&[(&str, &str)], &str); 5] = [ + (&[("HUGINN_ENDPOINT", "rudp://127.0.0.1:1")], "EUREKA_INSTANCE is not set"), + (&[("EUREKA_INSTANCE", "Not A Slug"), ("HUGINN_ENDPOINT", "rudp://127.0.0.1:1")], "EUREKA_INSTANCE is not a valid"), + (&[("EUREKA_INSTANCE", "eureka")], "HUGINN_ENDPOINT is not set"), + (&[("EUREKA_INSTANCE", "eureka"), ("HUGINN_ENDPOINT", "127.0.0.1:1")], "HUGINN_ENDPOINT is not rudp://"), + (&[("EUREKA_INSTANCE", "eureka"), ("HUGINN_ENDPOINT", "rudp://not-an-address")], "HUGINN_ENDPOINT is not rudp://"), + ]; + for (env, expected) in cases { + let mut mcp = Mcp::start(env); + let (error, whoami) = mcp.call("whoami", json!({})); + assert!(!error, "{expected}"); + assert_eq!(whoami["reachable"], json!(false), "{expected}"); + assert_eq!(whoami["error"]["error"], json!("Misconfigured"), "{expected}"); + assert!(whoami["error"]["detail"].as_str().unwrap().contains(expected), "{expected}: {whoami}"); + for (tool, arguments) in [ + ("view", json!({ "id": absent() })), + ("query", json!({ "selection": {} })), + ("admit", faculty_args(vec![])), + ] { + let (error, refusal) = mcp.call(tool, arguments); + assert!(error, "{tool}: {expected}"); + assert_eq!(refusal["error"], json!("Misconfigured"), "{tool}"); + } + } +} + +/// Input that does not fit the tool's schema is `invalid_params`, a protocol +/// error, and the server keeps serving. +#[test] +fn malformed_input_is_invalid_params() { + let mut mcp = Mcp::at(INSTANCE, closed_port()); + let response = mcp.rpc("tools/call", json!({ "name": "admit", "arguments": { "faculty": "Hands" } })); + assert_eq!(response["error"]["code"], json!(-32602), "{response}"); + let (error, _) = mcp.call("whoami", json!({})); + assert!(!error); +} + +/// A well-formed id of a document no mind here holds. +fn absent() -> Value { + serde_json::to_value(PipelineRef { kind: PipelineKind::CutSpec, id: Short("eureka-state:cut_spec:cut-99.r1".into()) }) + .unwrap() +} From 03eb670db33fa1a1516bdaa05c6bbd1a1c7dd3d7 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 00:39:03 +0200 Subject: [PATCH 03/12] eureka-state: drop the imports the fixture move left unused, and expect the router field Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/src/main.rs | 1 + crates/eureka-state/tests/client.rs | 23 ++++++++--------------- 2 files changed, 9 insertions(+), 15 deletions(-) diff --git a/crates/eureka-state/src/main.rs b/crates/eureka-state/src/main.rs index a7f4a58..e6d529f 100644 --- a/crates/eureka-state/src/main.rs +++ b/crates/eureka-state/src/main.rs @@ -138,6 +138,7 @@ fn client(instance: Option<&str>, endpoint: Option<&str>) -> Result, + #[expect(dead_code, reason = "the tool_handler macro reads this router field")] tool_router: ToolRouter, } diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index 1ef33e8..bb63d4e 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -3,30 +3,23 @@ //! made to misbehave. use std::net::{SocketAddr, UdpSocket}; -use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::thread::JoinHandle; +use std::sync::atomic::Ordering; use std::time::{Duration, Instant}; use chrono::Utc; use cultnet_rs::{ - CultNetMessage, CultNetRudpServerEvent, CultNetWireContract, Selection, answer_content_chunk_request, - decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, pack_content, + CultNetMessage, CultNetWireContract, Selection, answer_content_chunk_request, encode_cultnet_message_to_vec, + pack_content, }; use eureka_state::{ClientError, HuginnClient}; -use huginn_daemon::serve::{MAX_RESPONSE_BYTES, bind, run, schema_registry}; -use huginn_daemon::{Daemon, Handled, Hits, IndexSink, SearchTicket, ServeOptions}; +use huginn_daemon::serve::MAX_RESPONSE_BYTES; use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response}; use huginn_mind::epiphany_pipeline::{ - AuthorityMap, CodeLocation, CutDelete, CutVerification, Date, DocRef, FileChange, Line, NegativeCheck, OrgRepo, - PipelineCampaign, PipelineCutSpec, PipelineDocument, PipelineInstance, PipelineKind, PipelineRef, - PipelineStewardship, Sha, Short, Slug, StructuralDelta, Title, VerificationTest, + AuthorityMap, CodeLocation, CutDelete, CutVerification, Date, FileChange, Line, NegativeCheck, OrgRepo, + PipelineCutSpec, PipelineDocument, PipelineKind, PipelineRef, PipelineStewardship, Short, StructuralDelta, Title, + VerificationTest, }; -use huginn_mind::{ - DeferredAnswer, Faculty, HuginnMindRequest, HuginnMindResponse, IndexStatus, Mind, MindRefusal, MindStore, - OwnedRedbMessagePackBackingStore, PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineProvenance, -}; -use tempfile::TempDir; +use huginn_mind::{DeferredAnswer, HuginnMindRequest, HuginnMindResponse, MindRefusal, PipelineAdmissionOutcome}; mod common; use common::*; From 295b602b2e86b15113aca0cb8b14cff413d46f25 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 00:42:07 +0200 Subject: [PATCH 04/12] eureka-state tests: common names PipelineRef 8041d82 did not build on its own: the moved Inert index names PipelineRef and the import stayed behind in client.rs. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/tests/common/mod.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs index 5729e68..e8fc821 100644 --- a/crates/eureka-state/tests/common/mod.rs +++ b/crates/eureka-state/tests/common/mod.rs @@ -14,7 +14,8 @@ use cultnet_rs::{CultNetMessage, CultNetRudpServerEvent, CultNetWireContract, de use huginn_daemon::serve::{bind, run, schema_registry}; use huginn_daemon::{Daemon, Handled, Hits, IndexSink, SearchTicket, ServeOptions}; use huginn_mind::epiphany_pipeline::{ - Date, DocRef, OrgRepo, PipelineCampaign, PipelineDocument, PipelineInstance, Sha, Short, Slug, Title, + Date, DocRef, OrgRepo, PipelineCampaign, PipelineDocument, PipelineInstance, PipelineRef, Sha, Short, Slug, + Title, }; use huginn_mind::{ Faculty, HuginnMindRequest, HuginnMindResponse, IndexStatus, Mind, MindStore, OwnedRedbMessagePackBackingStore, From 63ac5038c96ee591bdec002d89a598d8fba1a006 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 01:01:22 +0200 Subject: [PATCH 05/12] eureka-state tests: seed the stewardship, pin the schema path, malformed input is a tool error Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/tests/client.rs | 14 ++--------- crates/eureka-state/tests/common/mod.rs | 12 +++++++++- crates/eureka-state/tests/mcp.rs | 32 +++++++++++++++---------- 3 files changed, 32 insertions(+), 26 deletions(-) diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index bb63d4e..56ab80c 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -15,8 +15,8 @@ use eureka_state::{ClientError, HuginnClient}; use huginn_daemon::serve::MAX_RESPONSE_BYTES; use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response}; use huginn_mind::epiphany_pipeline::{ - AuthorityMap, CodeLocation, CutDelete, CutVerification, Date, FileChange, Line, NegativeCheck, OrgRepo, - PipelineCutSpec, PipelineDocument, PipelineKind, PipelineRef, PipelineStewardship, Short, StructuralDelta, Title, + AuthorityMap, CodeLocation, CutDelete, CutVerification, FileChange, Line, NegativeCheck, OrgRepo, + PipelineCutSpec, PipelineDocument, PipelineKind, PipelineRef, Short, StructuralDelta, Title, VerificationTest, }; use huginn_mind::{DeferredAnswer, HuginnMindRequest, HuginnMindResponse, MindRefusal, PipelineAdmissionOutcome}; @@ -27,16 +27,6 @@ use common::*; const OTHER: &str = "thought-cage"; const TIMEOUT: Duration = Duration::from_secs(20); -fn stewardship() -> PipelineDocument { - PipelineDocument::Stewardship(PipelineStewardship { - instance: slug(INSTANCE), - repo: OrgRepo(REPO.into()), - sequence: 1, - assigned_on: Date("2026-09-29".into()), - note: "assigned".into(), - }) -} - /// A `Line` and a `Short` at the leaf's own bound. fn line(n: usize) -> Line { Line(format!("{n:0>1000}")) diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs index e8fc821..27a0705 100644 --- a/crates/eureka-state/tests/common/mod.rs +++ b/crates/eureka-state/tests/common/mod.rs @@ -14,7 +14,7 @@ use cultnet_rs::{CultNetMessage, CultNetRudpServerEvent, CultNetWireContract, de use huginn_daemon::serve::{bind, run, schema_registry}; use huginn_daemon::{Daemon, Handled, Hits, IndexSink, SearchTicket, ServeOptions}; use huginn_mind::epiphany_pipeline::{ - Date, DocRef, OrgRepo, PipelineCampaign, PipelineDocument, PipelineInstance, PipelineRef, Sha, Short, Slug, + Date, DocRef, OrgRepo, PipelineCampaign, PipelineDocument, PipelineInstance, PipelineRef, PipelineStewardship, Sha, Short, Slug, Title, }; use huginn_mind::{ @@ -173,3 +173,13 @@ pub fn campaign() -> PipelineDocument { target_doc: DocRef { path: Short("notes/target.md".into()), start_line: 1, end_line: 9, commit: sha() }, }) } + +pub fn stewardship() -> PipelineDocument { + PipelineDocument::Stewardship(PipelineStewardship { + instance: slug(INSTANCE), + repo: OrgRepo(REPO.into()), + sequence: 1, + assigned_on: Date("2026-09-29".into()), + note: "assigned".into(), + }) +} diff --git a/crates/eureka-state/tests/mcp.rs b/crates/eureka-state/tests/mcp.rs index adbfcda..2d9b3ad 100644 --- a/crates/eureka-state/tests/mcp.rs +++ b/crates/eureka-state/tests/mcp.rs @@ -173,7 +173,8 @@ fn selection_schema_matches_cultlib() { .expect("cultnet-rs is a dependency")["manifest_path"] .as_str() .unwrap(); - let published = std::path::Path::new(manifest).join("../../../contracts/cultnet/cultnet.selection.schema.json"); + let crate_dir = std::path::Path::new(manifest).parent().unwrap(); + let published = crate_dir.join("../../contracts/cultnet/cultnet.selection.schema.json"); let theirs: Value = serde_json::from_str(&std::fs::read_to_string(published).unwrap()).unwrap(); let ours: Value = serde_json::from_str( &std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/schemas/cultnet.selection.schema.json")).unwrap(), @@ -185,7 +186,7 @@ fn selection_schema_matches_cultlib() { /// Every tool against a live daemon, and what each returns. #[test] fn every_tool_round_trips_against_a_live_daemon() { - let (root, daemon) = mind(vec![]); + let (root, daemon) = mind(vec![vec![stewardship()]]); let server = serve(root, daemon); let mut mcp = Mcp::at(INSTANCE, server.addr); @@ -195,7 +196,7 @@ fn every_tool_round_trips_against_a_live_daemon() { assert_eq!(whoami["endpoint"], json!(format!("rudp://{}", server.addr))); assert_eq!(whoami["reachable"], json!(true)); assert_eq!(whoami["status"]["instance"], json!(INSTANCE)); - assert_eq!(whoami["status"]["documents"], json!(1)); + assert_eq!(whoami["status"]["documents"], json!(2)); assert!(whoami.get("error").is_none(), "{whoami}"); let (error, admitted) = mcp.call("admit", faculty_args(vec![campaign_json()])); @@ -214,11 +215,11 @@ fn every_tool_round_trips_against_a_live_daemon() { let (error, page) = mcp.call("query", json!({ "selection": { "projection": "document" } })); assert!(!error); - assert_eq!(page["matched"], json!(2), "{page}"); - assert_eq!(page["items"]["Documents"].as_array().unwrap().len(), 2); + assert_eq!(page["matched"], json!(3), "{page}"); + assert_eq!(page["items"]["Documents"].as_array().unwrap().len(), 3); let (_, counted) = mcp.call("whoami", json!({})); - assert_eq!(counted["status"]["receipts"], json!(2)); + assert_eq!(counted["status"]["receipts"], json!(3)); } /// Declared identity is configuration, not a per-call argument. Input that @@ -226,7 +227,7 @@ fn every_tool_round_trips_against_a_live_daemon() { /// admitted for the configured instance, and the receipt says `eureka-state`. #[test] fn admit_ignores_any_instance_in_input() { - let (root, daemon) = mind(vec![]); + let (root, daemon) = mind(vec![vec![stewardship()]]); let server = serve(root, daemon); let mut mcp = Mcp::at(INSTANCE, server.addr); @@ -250,7 +251,7 @@ fn admit_ignores_any_instance_in_input() { /// refusal. Here the configured instance is not the daemon's. #[test] fn a_refusal_is_a_successful_result() { - let (root, daemon) = mind(vec![]); + let (root, daemon) = mind(vec![vec![stewardship()]]); let server = serve(root, daemon); let mut mcp = Mcp::at("thought-cage", server.addr); @@ -348,13 +349,18 @@ fn bad_configuration_is_served_and_reported() { } } -/// Input that does not fit the tool's schema is `invalid_params`, a protocol -/// error, and the server keeps serving. +/// Input that does not fit the tool's schema is refused before any call is +/// made. rmcp 2.2 reports it as a tool result with `isError`, not as a +/// JSON-RPC `invalid_params`, so the model sees what to correct. The server +/// keeps serving. #[test] -fn malformed_input_is_invalid_params() { - let mut mcp = Mcp::at(INSTANCE, closed_port()); +fn malformed_input_is_refused_before_any_call() { + let server = answering(encode_response("eureka-state-call", "admit", &HuginnMindResponse::View(None), "scripted").unwrap()); + let mut mcp = Mcp::at(INSTANCE, server.addr); let response = mcp.rpc("tools/call", json!({ "name": "admit", "arguments": { "faculty": "Hands" } })); - assert_eq!(response["error"]["code"], json!(-32602), "{response}"); + assert!(response.get("error").is_none(), "{response}"); + assert_eq!(response["result"]["isError"], json!(true)); + assert!(response["result"]["content"][0]["text"].as_str().unwrap().contains("missing field"), "{response}"); let (error, _) = mcp.call("whoami", json!({})); assert!(!error); } From 120e0a6e5a0fc2ca14756789c302c0f727084aef Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 01:41:14 +0200 Subject: [PATCH 06/12] eureka-state tests: initialize must advertise the tools capability cargo-mutants on the Cut 13b diff: replacing get_info with Default::default() survived, and a client that does not see a tools capability does not list tools. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/tests/mcp.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/crates/eureka-state/tests/mcp.rs b/crates/eureka-state/tests/mcp.rs index 2d9b3ad..ab5a924 100644 --- a/crates/eureka-state/tests/mcp.rs +++ b/crates/eureka-state/tests/mcp.rs @@ -24,6 +24,8 @@ struct Mcp { stdin: ChildStdin, lines: Receiver, next: u64, + /// The server's answer to `initialize`. + init: Value, } impl Drop for Mcp { @@ -53,12 +55,13 @@ impl Mcp { let _ = sender.send(line); } }); - let mut mcp = Self { child, stdin, lines, next: 1 }; + let mut mcp = Self { child, stdin, lines, next: 1, init: Value::Null }; let init = mcp.rpc( "initialize", json!({ "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": { "name": "test", "version": "0" } }), ); assert!(init.get("result").is_some(), "{init}"); + mcp.init = init["result"].clone(); mcp.send(json!({ "jsonrpc": "2.0", "method": "notifications/initialized" })); mcp } @@ -370,3 +373,10 @@ fn absent() -> Value { serde_json::to_value(PipelineRef { kind: PipelineKind::CutSpec, id: Short("eureka-state:cut_spec:cut-99.r1".into()) }) .unwrap() } + +/// A client only lists tools from a server that says it has them. +#[test] +fn initialize_advertises_tools() { + let mcp = Mcp::at(INSTANCE, closed_port()); + assert!(mcp.init["capabilities"]["tools"].is_object(), "{}", mcp.init); +} From 1d3f917d9edac5a93595ead910f0982c8361e767 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 02:51:15 +0200 Subject: [PATCH 07/12] huginn-mind: admission validates who admitted the batch A batch's agent and session were committed as given: "", " ", a 5,000 character string and a string with a line break all became the receipt's admitting session. The mind now refuses, at A1 before any document is read, an agent or session that is longer than Short's 200 bytes (FieldBound), blank after trim, or spread over more than one line (InvalidFormat), so every writer gets the check. The refusals are the leaf's own, since the fields are the leaf's Short type; the published wire schema does not change. Short exposes no validator, so the bound is a constant pinned to the maxLength its schema publishes. Co-Authored-By: Claude Sonnet 5.5 --- crates/huginn-mind/src/admission.rs | 87 ++++++++++++++++++++++++++++- 1 file changed, 86 insertions(+), 1 deletion(-) diff --git a/crates/huginn-mind/src/admission.rs b/crates/huginn-mind/src/admission.rs index 4b85cdd..0288ad0 100644 --- a/crates/huginn-mind/src/admission.rs +++ b/crates/huginn-mind/src/admission.rs @@ -5,7 +5,8 @@ //! hand-off and import, and Cut 13's tools. The steps run in order and the //! first failing step is the outcome: //! -//! - A1 the batch names this mind; A2 one to sixty-four envelopes; +//! - A1 the batch names this mind, and attributes itself to a named agent and +//! session (`refuse_provenance`); A2 one to sixty-four envelopes; //! - A3 every envelope passes the leaf's bounds, formats and key //! recomputation; A4 identities are unique in the batch; A5 every document //! carrying an instance names this mind; @@ -107,6 +108,7 @@ impl Mind { ) -> Result { // A1 self.require_instance(instance)?; + refuse_provenance(&provenance)?; // A2 if envelopes.is_empty() || envelopes.len() > BATCH_MAX { return Err(MindRefusal::BatchSize { actual: envelopes.len() as u32 }); @@ -171,6 +173,30 @@ impl Mind { } /// A3 through the leaf, then A5. +/// The bound `epiphany_pipeline::Short` declares, in UTF-8 bytes. The leaf +/// exposes no validator for it, so `short_bound_is_the_leafs_own` pins this to +/// the `maxLength` its schema publishes. +const SHORT_MAX: usize = 200; + +/// Who admitted a batch is the receipt's only record of it, so the mind checks +/// it here and every writer, whatever its transport, gets the check: the agent +/// and the session are within `Short`'s bound, hold something besides +/// whitespace, and sit on one line. Refused as the leaf's own refusals, since +/// the fields are the leaf's own type. +fn refuse_provenance(provenance: &PipelineProvenance) -> Result<(), MindRefusal> { + for (field, value) in [("provenance.agent", &provenance.agent), ("provenance.session", &provenance.session)] { + let text = value.0.as_str(); + if text.len() > SHORT_MAX { + return Err(PipelineRefusal::FieldBound { field: field.into(), limit: SHORT_MAX as u32, actual: text.len() as u32 }.into()); + } + let line_break = |c: char| matches!(c, '\n' | '\r' | '\u{0B}' | '\u{0C}' | '\u{85}' | '\u{2028}' | '\u{2029}'); + if text.trim().is_empty() || text.contains(line_break) { + return Err(PipelineRefusal::InvalidFormat { field: field.into(), value: text.into() }.into()); + } + } + Ok(()) +} + fn stage(envelope: CultCacheEnvelope, mind: &Slug) -> Result { validate_pipeline_write_envelope(&envelope)?; let document = PipelineDocument::decode(&envelope)?; @@ -693,6 +719,65 @@ mod tests { assert_eq!(mind.genesis_receipt_id().unwrap(), Some(first)); } + fn admit_attributed(agent: &str, session: &str) -> (Mind, PipelineAdmissionOutcome) { + let mut mind = opened(MemoryStore::new(), INSTANCE); + let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: s(agent), session: s(session), tool: s("admit") }; + let batch = PipelineAdmissionBatch { instance: slug(INSTANCE), provenance, documents: vec![instance(INSTANCE), stewardship(INSTANCE, REPO)] }; + let outcome = mind.admit(batch, now()); + (mind, outcome) + } + + /// Who admitted a batch is validated by the mind, so no writer can commit + /// a blank, multi-line or oversized attribution, and a refused batch leaves + /// nothing behind. + #[test] + fn provenance_names_an_agent_and_a_session_on_one_line_within_short() { + let over = "x".repeat(SHORT_MAX + 1); + let exact = "y".repeat(SHORT_MAX); + for (field, agent, session) in [ + ("provenance.agent", "", "session-1"), + ("provenance.agent", " \t ", "session-1"), + ("provenance.session", "claude", ""), + ("provenance.session", "claude", " "), + ("provenance.session", "claude", "\u{2003}"), + ("provenance.session", "claude", "line\nbreak"), + ("provenance.session", "claude", "line\rbreak"), + ("provenance.session", "claude", "a\u{2028}b"), + ("provenance.agent", "two\nlines", "session-1"), + ] { + let (mind, outcome) = admit_attributed(agent, session); + let refused = refusal(outcome); + assert!( + matches!(&refused, MindRefusal::Document(PipelineRefusal::InvalidFormat { field: named, .. }) if named == field), + "{agent:?}/{session:?}: {refused:?}" + ); + assert!(mind.is_empty(), "{agent:?}/{session:?}: a refused batch wrote"); + } + let (mind, outcome) = admit_attributed(&over, "session-1"); + assert_eq!( + refusal(outcome), + MindRefusal::Document(PipelineRefusal::FieldBound { field: "provenance.agent".into(), limit: 200, actual: 201 }) + ); + assert!(mind.is_empty()); + let (_, outcome) = admit_attributed("claude", &over); + assert_eq!( + refusal(outcome), + MindRefusal::Document(PipelineRefusal::FieldBound { field: "provenance.session".into(), limit: 200, actual: 201 }) + ); + // The bound is inclusive and a name with inner spaces is a name. + let (mind, outcome) = admit_attributed(&exact, "session 7"); + committed(outcome); + let receipts = mind.receipts().unwrap(); + assert_eq!(receipts[0].provenance.agent, s(&exact)); + assert_eq!(receipts[0].provenance.session, s("session 7")); + } + + #[test] + fn short_bound_is_the_leafs_own() { + let schema = serde_json::to_value(schemars::schema_for!(Short)).unwrap(); + assert_eq!(schema["maxLength"], serde_json::json!(SHORT_MAX)); + } + #[test] fn admission_refuses_a_foreign_instance_whatever_the_transport() { let store = MemoryStore::new(); From 890a00b0de2cbc42bc5182a6fc06046497ab5034 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 02:51:50 +0200 Subject: [PATCH 08/12] eureka-state: typed TooLarge, typed InvalidInput, and the pins Soul found missing - The client refuses a request whose encoded size exceeds what one RUDP send carries (65,507 less the 36-byte header and the 6-byte channel id), before connecting, as ClientError::TooLarge { bytes, limit }. It was reported as a retryable Unavailable ("Message too long"). The MCP tools surface it as isError with {"error":"TooLarge","bytes","limit"}. - Arguments that fail to deserialize are still a tool error, now with the same typed body as every other one: {"error":"InvalidInput","detail"}. - A non-UTF-8 EUREKA_INSTANCE or HUGINN_ENDPOINT is reported as not valid UTF-8 and not as unset. - Tests: the schema walk asserts no $id or $schema below any tool's root; the 15 s timeout is measured against a silent daemon; the semantic part of a query is observed on the wire; Internal's and TooLarge's bodies and the InvalidInput rewrite are pinned; attribution is refused through the MCP admit; oversize is TooLarge at and one past the limit. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/src/lib.rs | 31 +++++- crates/eureka-state/src/main.rs | 109 ++++++++++++++++++--- crates/eureka-state/tests/client.rs | 88 ++++++++++++++++- crates/eureka-state/tests/mcp.rs | 147 +++++++++++++++++++++++++++- 4 files changed, 348 insertions(+), 27 deletions(-) diff --git a/crates/eureka-state/src/lib.rs b/crates/eureka-state/src/lib.rs index 94dbfec..1ff2bb7 100644 --- a/crates/eureka-state/src/lib.rs +++ b/crates/eureka-state/src/lib.rs @@ -23,7 +23,7 @@ use std::time::{Duration, Instant}; use anyhow::{Context, Result, anyhow, bail}; use cultnet_rs::{ CULTNET_OPERATION_CONNECTION_ID, CultMesh, CultMeshRudpSocketOptions, CultNetMessage, - CultNetRudpSocketTransportConnection, fetch_content, + CultNetRudpSocketTransportConnection, CultNetWireContract, encode_cultnet_message_to_vec, fetch_content, }; use huginn_mind::envelope::{OperationFailure, decode_response, encode_request}; use huginn_mind::epiphany_pipeline::Slug; @@ -33,15 +33,25 @@ use huginn_mind::{HuginnMindRequest, HuginnMindResponse, MAX_DEFERRED_BODY_BYTES const MESSAGE_ID: &str = "eureka-state-call"; const POLL: Duration = Duration::from_millis(2); +/// The largest encoded message one send carries. The client's RUDP session +/// does not fragment (`max_fragment_bytes` is unset), so a message is one UDP +/// datagram: at most 65,507 bytes over IPv4 (65,535 less the 8-byte UDP and +/// 20-byte IP headers), less RUDP's 36-byte fixed header (`RUDP_FIXED_HEADER_BYTES` +/// in `cultnet-rs`, private there) and the 6-byte `schema` channel id. A larger +/// message makes the operating system refuse the send ("Message too long"). +pub const MAX_REQUEST_BYTES: usize = 65_507 - 36 - "schema".len(); + /// Why a call produced no answer. `Unavailable` is the transport's failure and /// the only one worth retrying: the daemon could not be reached, went quiet /// past the call's deadline, or answered out of protocol. `Rejected` is the -/// daemon refusing the envelope itself, which a retry of the same call cannot -/// change. +/// daemon refusing the envelope itself, and `TooLarge` is the client refusing +/// a request that no single send can carry, before sending it; a retry of the +/// same call cannot change either. #[derive(Debug)] pub enum ClientError { Unavailable { endpoint: SocketAddr, detail: String }, Rejected { endpoint: SocketAddr, code: String, detail: String }, + TooLarge { bytes: usize, limit: usize }, } impl fmt::Display for ClientError { @@ -53,6 +63,9 @@ impl fmt::Display for ClientError { Self::Rejected { endpoint, code, detail } => { write!(f, "the Huginn daemon at rudp://{endpoint} rejected the request: {code}: {detail}") } + Self::TooLarge { bytes, limit } => { + write!(f, "the request encodes to {bytes} bytes and one send carries at most {limit}") + } } } } @@ -96,8 +109,15 @@ impl HuginnClient { let deadline = Instant::now() + self.timeout; let unavailable = |error: anyhow::Error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") }; + let message = encode_request(MESSAGE_ID, &request, None).map_err(unavailable)?; + let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0) + .map_err(unavailable)? + .len(); + if bytes > MAX_REQUEST_BYTES { + return Err(ClientError::TooLarge { bytes, limit: MAX_REQUEST_BYTES }); + } let mut session = self.connect(deadline).map_err(unavailable)?; - let outcome = self.exchange(&mut session, &request, deadline); + let outcome = self.exchange(&mut session, &request, &message, deadline); // Best effort: the daemon drops a session it never hears from, only later. let _ = session.disconnect(Vec::new()); outcome.map_err(|failure| match failure { @@ -114,9 +134,10 @@ impl HuginnClient { &self, session: &mut CultNetRudpSocketTransportConnection, request: &HuginnMindRequest, + message: &CultNetMessage, deadline: Instant, ) -> Result { - let reply = self.ask(session, &encode_request(MESSAGE_ID, request, None)?, deadline)?; + let reply = self.ask(session, message, deadline)?; let CultNetMessage::OperationResponse { operation, .. } = &reply else { return Err(anyhow!("the daemon answered with something other than an operation response").into()); }; diff --git a/crates/eureka-state/src/main.rs b/crates/eureka-state/src/main.rs index e6d529f..d5a8bc5 100644 --- a/crates/eureka-state/src/main.rs +++ b/crates/eureka-state/src/main.rs @@ -11,8 +11,10 @@ //! startup vanishes from Claude Code without a word. `whoami` reports it and //! every other tool refuses with it. A refusal from the mind is an answer and //! comes back as a normal result; a daemon that cannot be reached or that -//! rejects the envelope is a result with `isError: true`, typed, so a caller -//! can tell a down daemon from a permanent mismatch. +//! rejects the envelope, or a request too large for one send, is a result with +//! `isError: true`, typed, so a caller can tell a down daemon from a permanent +//! mismatch. Arguments that do not fit a tool's input are typed the same way +//! (`InvalidInput`), as a tool error and not a protocol one. use std::net::SocketAddr; use std::sync::Arc; @@ -25,9 +27,11 @@ use huginn_mind::{ Faculty, HuginnMindRequest, HuginnMindResponse, PipelineAdmissionBatch, PipelineProvenance, SemanticQuery, }; use rmcp::handler::server::router::tool::ToolRouter; +use rmcp::handler::server::tool::ToolCallContext; use rmcp::handler::server::wrapper::Parameters; -use rmcp::model::{CallToolResult, ContentBlock, ServerCapabilities, ServerInfo}; -use rmcp::{ServerHandler, ServiceExt, tool, tool_handler, tool_router}; +use rmcp::model::{CallToolRequestParams, CallToolResult, ContentBlock, ServerCapabilities, ServerInfo}; +use rmcp::service::RequestContext; +use rmcp::{ErrorData, RoleServer, ServerHandler, ServiceExt, tool, tool_handler, tool_router}; use schemars::{JsonSchema, Schema, SchemaGenerator}; use serde::Deserialize; use serde_json::{Value, json}; @@ -101,6 +105,9 @@ impl Trouble { Self::Client(ClientError::Rejected { endpoint, code, detail }) => { json!({ "error": "Rejected", "endpoint": format!("rudp://{endpoint}"), "code": code, "detail": detail }) } + Self::Client(ClientError::TooLarge { bytes, limit }) => { + json!({ "error": "TooLarge", "bytes": bytes, "limit": limit }) + } Self::Internal(detail) => json!({ "error": "Internal", "detail": detail }), } } @@ -116,18 +123,28 @@ struct Settings { impl Settings { fn from_env() -> Self { - let instance = std::env::var("EUREKA_INSTANCE").ok(); - let endpoint = std::env::var("HUGINN_ENDPOINT").ok(); - let client = client(instance.as_deref(), endpoint.as_deref()); - Self { instance, endpoint, client } + let instance = variable("EUREKA_INSTANCE"); + let endpoint = variable("HUGINN_ENDPOINT"); + let shown = |value: &Result, String>| value.as_ref().ok().cloned().flatten(); + let (shown_instance, shown_endpoint) = (shown(&instance), shown(&endpoint)); + Self { instance: shown_instance, endpoint: shown_endpoint, client: client(instance, endpoint) } } } -fn client(instance: Option<&str>, endpoint: Option<&str>) -> Result { - let instance = instance.ok_or("EUREKA_INSTANCE is not set")?; - let slug = Slug(instance.to_string()); +/// An environment variable: its value, `None` when unset, or why it cannot be read. +fn variable(name: &str) -> Result, String> { + match std::env::var(name) { + Ok(value) => Ok(Some(value)), + Err(std::env::VarError::NotPresent) => Ok(None), + Err(std::env::VarError::NotUnicode(_)) => Err(format!("{name} is not valid UTF-8")), + } +} + +fn client(instance: Result, String>, endpoint: Result, String>) -> Result { + let instance = instance?.ok_or("EUREKA_INSTANCE is not set")?; + 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 endpoint = endpoint?.ok_or("HUGINN_ENDPOINT is not set")?; let address = endpoint .strip_prefix("rudp://") .and_then(|address| address.parse::().ok()) @@ -138,7 +155,6 @@ fn client(instance: Option<&str>, endpoint: Option<&str>) -> Result, - #[expect(dead_code, reason = "the tool_handler macro reads this router field")] tool_router: ToolRouter, } @@ -244,8 +260,32 @@ impl EurekaState { } } +/// Every tool error is a typed JSON body. The ones this server writes already +/// are; the one rmcp writes, for arguments that do not deserialize into the +/// tool's input, is plain text, so it is put into the same shape. +fn typed_error(result: CallToolResult) -> CallToolResult { + if result.is_error != Some(true) { + return result; + } + let Some(text) = result.content.first().and_then(ContentBlock::as_text).map(|text| text.text.as_str()) else { + return result; + }; + if serde_json::from_str::(text).is_ok_and(|body| body.is_object()) { + return result; + } + reply(json!({ "error": "InvalidInput", "detail": text }), true) +} + #[tool_handler] impl ServerHandler for EurekaState { + async fn call_tool( + &self, + request: CallToolRequestParams, + context: RequestContext, + ) -> Result { + Ok(typed_error(self.tool_router.call(ToolCallContext::new(self, request, context)).await?)) + } + fn get_info(&self) -> ServerInfo { ServerInfo::new(ServerCapabilities::builder().enable_tools().build()).with_instructions( "Huginn's typed mind for this instance: whoami, admit, view, query. Refusals are answers; isError means the daemon could not be reached or rejected the request.", @@ -259,3 +299,46 @@ async fn main() -> anyhow::Result<()> { server.waiting().await?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + fn body_of(result: &CallToolResult) -> (bool, String) { + (result.is_error == Some(true), result.content[0].as_text().unwrap().text.clone()) + } + + /// Reaching `Internal` for real takes a panic in a blocking task, so its + /// body is pinned where it is built. + #[test] + fn every_trouble_has_its_typed_body() { + assert_eq!(Trouble::Internal("task panicked".into()).body(), json!({ "error": "Internal", "detail": "task panicked" })); + assert_eq!( + Trouble::Client(ClientError::TooLarge { bytes: 70_000, limit: 65_000 }).body(), + json!({ "error": "TooLarge", "bytes": 70_000, "limit": 65_000 }) + ); + assert_eq!(Trouble::Misconfigured("bad".into()).body(), json!({ "error": "Misconfigured", "detail": "bad" })); + } + + /// Only the text rmcp writes is rewritten: our own typed bodies and every + /// success pass through untouched. + #[test] + fn typed_error_rewrites_only_rmcp_plain_text() { + let plain = CallToolResult::error(vec![ContentBlock::text("failed to deserialize parameters: missing field `faculty`")]); + let (is_error, text) = body_of(&typed_error(plain)); + assert!(is_error); + assert_eq!( + serde_json::from_str::(&text).unwrap(), + json!({ "error": "InvalidInput", "detail": "failed to deserialize parameters: missing field `faculty`" }) + ); + + let typed = reply(json!({ "error": "Unavailable", "detail": "down" }), true); + assert_eq!(body_of(&typed_error(typed.clone())), body_of(&typed)); + + let success = CallToolResult::success(vec![ContentBlock::text("not json, and fine")]); + assert_eq!(body_of(&typed_error(success)), (false, "not json, and fine".to_string())); + + let quoted = CallToolResult::error(vec![ContentBlock::text("[1, 2]")]); + assert_eq!(serde_json::from_str::(&body_of(&typed_error(quoted)).1).unwrap()["error"], json!("InvalidInput")); + } +} diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index 56ab80c..38cfe14 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -11,9 +11,9 @@ use cultnet_rs::{ CultNetMessage, CultNetWireContract, Selection, answer_content_chunk_request, encode_cultnet_message_to_vec, pack_content, }; -use eureka_state::{ClientError, HuginnClient}; +use eureka_state::{ClientError, HuginnClient, MAX_REQUEST_BYTES}; use huginn_daemon::serve::MAX_RESPONSE_BYTES; -use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response}; +use huginn_mind::envelope::{OperationFailure, encode_failure, encode_request, encode_response}; use huginn_mind::epiphany_pipeline::{ AuthorityMap, CodeLocation, CutDelete, CutVerification, FileChange, Line, NegativeCheck, OrgRepo, PipelineCutSpec, PipelineDocument, PipelineKind, PipelineRef, Short, StructuralDelta, Title, @@ -128,14 +128,14 @@ fn client(addr: SocketAddr) -> HuginnClient { fn unavailable(error: ClientError) -> (SocketAddr, String) { match error { ClientError::Unavailable { endpoint, detail } => (endpoint, detail), - rejected @ ClientError::Rejected { .. } => panic!("expected Unavailable, got {rejected}"), + other @ (ClientError::Rejected { .. } | ClientError::TooLarge { .. }) => panic!("expected Unavailable, got {other}"), } } fn rejected(error: ClientError) -> (SocketAddr, String, String) { match error { ClientError::Rejected { endpoint, code, detail } => (endpoint, code, detail), - unavailable @ ClientError::Unavailable { .. } => panic!("expected Rejected, got {unavailable}"), + other @ (ClientError::Unavailable { .. } | ClientError::TooLarge { .. }) => panic!("expected Rejected, got {other}"), } } @@ -380,3 +380,83 @@ fn a_body_over_the_deferral_cap_is_refused() { let (_, detail) = unavailable(client(server.addr).call(HuginnMindRequest::Whoami).unwrap_err()); assert!(detail.contains("exceeds"), "{detail}"); } + +/// An admission whose attribution is `agent_pad` and `session_pad` characters +/// long: the two knobs that size a request without changing what it asks. +fn padded_admit(agent_pad: usize, session_pad: usize) -> HuginnMindRequest { + let mut request = batch(INSTANCE, vec![stewardship()]); + request.provenance.agent = Short("a".repeat(agent_pad)); + request.provenance.session = Short("s".repeat(session_pad)); + HuginnMindRequest::Admit(request) +} + +/// What one send has to carry for `request`, measured the way the client does. +fn encoded_bytes(request: &HuginnMindRequest) -> usize { + let message = encode_request("eureka-state-call", request, None).unwrap(); + encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap().len() +} + +/// A request that encodes to exactly `target` bytes: the session padding is +/// searched for each agent padding until the base64 steps land on it. +fn request_of_size(target: usize) -> HuginnMindRequest { + for agent_pad in 1..16 { + let (mut low, mut high) = (0, target); + while low < high { + let middle = (low + high) / 2; + if encoded_bytes(&padded_admit(agent_pad, middle)) < target { + low = middle + 1; + } else { + high = middle; + } + } + let request = padded_admit(agent_pad, low); + if encoded_bytes(&request) == target { + return request; + } + } + panic!("no padding encodes to {target} bytes"); +} + +fn admitted_answer() -> CultNetMessage { + encode_response( + "eureka-state-call", + "admit", + &HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { receipt_id: "r".into() }), + "scripted", + ) + .unwrap() +} + +/// A request larger than one send carries is refused by the client before it +/// connects to anything, as its own error and not as a transport failure the +/// caller would retry; one byte under the limit is sent, and answered. +#[test] +fn a_request_over_what_one_send_carries_is_too_large_and_one_at_the_limit_is_sent() { + let closed = UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap(); + let over = request_of_size(MAX_REQUEST_BYTES + 1); + let started = Instant::now(); + match client(closed).call(over) { + Err(ClientError::TooLarge { bytes, limit }) => assert_eq!((bytes, limit), (MAX_REQUEST_BYTES + 1, MAX_REQUEST_BYTES)), + other => panic!("expected TooLarge, got {other:?}"), + } + assert!(started.elapsed() < Duration::from_secs(1), "refused only after trying the transport"); + + let server = answering(admitted_answer()); + for size in [MAX_REQUEST_BYTES - 1, MAX_REQUEST_BYTES] { + let answer = client(server.addr).call(request_of_size(size)); + assert!( + matches!(answer, Ok(HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { .. }))), + "a request of {size} bytes: {answer:?}" + ); + } +} + +/// The transport's own refusal of an oversize send ("Message too long") is +/// what an unchecked request used to report, as an `Unavailable` daemon a +/// caller would retry forever. +#[test] +fn an_oversize_admission_is_never_reported_as_an_unavailable_daemon() { + let server = answering(admitted_answer()); + let error = client(server.addr).call(padded_admit(6, 70_000)).unwrap_err(); + assert!(matches!(error, ClientError::TooLarge { .. }), "{error}"); +} diff --git a/crates/eureka-state/tests/mcp.rs b/crates/eureka-state/tests/mcp.rs index ab5a924..f6def4c 100644 --- a/crates/eureka-state/tests/mcp.rs +++ b/crates/eureka-state/tests/mcp.rs @@ -10,7 +10,7 @@ use std::sync::mpsc::{Receiver, channel}; use std::time::Duration; use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response}; -use huginn_mind::HuginnMindResponse; +use huginn_mind::{HuginnMindRequest, HuginnMindResponse}; use huginn_mind::epiphany_pipeline::{PipelineKind, PipelineRef, Short}; use serde_json::{Value, json}; @@ -38,11 +38,17 @@ impl Drop for Mcp { impl Mcp { /// The binary with exactly the environment given, initialised. fn start(env: &[(&str, &str)]) -> Self { + Self::spawn(|command| { + command.envs(env.iter().copied()); + }) + } + + /// The binary with no configuration but what `configure` gives its command. + fn spawn(configure: impl FnOnce(&mut Command)) -> Self { let mut command = Command::new(env!("CARGO_BIN_EXE_eureka-state")); + command.env_remove("EUREKA_INSTANCE").env_remove("HUGINN_ENDPOINT"); + configure(&mut command); command - .env_remove("EUREKA_INSTANCE") - .env_remove("HUGINN_ENDPOINT") - .envs(env.iter().copied()) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::null()); @@ -130,6 +136,18 @@ fn refs(schema: &Value, found: &mut Vec) { } } +/// Every key named `key` anywhere below `schema`'s root. +fn keys_below_root(schema: &Value, key: &str) -> usize { + fn count(value: &Value, key: &str) -> usize { + match value { + Value::Object(map) => map.iter().map(|(name, inner)| usize::from(name == key) + count(inner, key)).sum(), + Value::Array(items) => items.iter().map(|item| count(item, key)).sum(), + _ => 0, + } + } + schema.as_object().unwrap().values().map(|inner| count(inner, key)).sum() +} + /// Four tools, each input schema a self-contained document: every `$ref` is /// internal and resolves, and the selection's definitions are bundled in. #[test] @@ -147,7 +165,12 @@ fn tools_list_offers_four_tools_with_self_contained_schemas() { let name = reference.strip_prefix("#/$defs/").unwrap_or_else(|| panic!("{}: remote $ref {reference}", tool["name"])); assert!(schema["$defs"].get(name).is_some(), "{}: {reference} does not resolve", tool["name"]); } + // The vendored selection schema brings its own `$id` and `$schema`, + // which would make its `$ref`s resolve against a remote document. assert!(schema.get("$id").is_none(), "{}: a remote identity", tool["name"]); + for key in ["$id", "$schema"] { + assert_eq!(keys_below_root(schema, key), 0, "{}: a nested {key}", tool["name"]); + } } let query = tools.iter().find(|tool| tool["name"] == "query").unwrap(); for bundled in ["fieldPredicate", "citation", "incoming", "recordRef"] { @@ -363,11 +386,125 @@ fn malformed_input_is_refused_before_any_call() { let response = mcp.rpc("tools/call", json!({ "name": "admit", "arguments": { "faculty": "Hands" } })); assert!(response.get("error").is_none(), "{response}"); assert_eq!(response["result"]["isError"], json!(true)); - assert!(response["result"]["content"][0]["text"].as_str().unwrap().contains("missing field"), "{response}"); + let body: Value = serde_json::from_str(response["result"]["content"][0]["text"].as_str().unwrap()).expect("a JSON body"); + assert_eq!(body["error"], json!("InvalidInput"), "{body}"); + assert!(body["detail"].as_str().unwrap().contains("missing field"), "{body}"); let (error, _) = mcp.call("whoami", json!({})); assert!(!error); } +/// A request too large for one send is the client's typed refusal, an error +/// result that names the size and the limit and does not call the daemon +/// unavailable. +#[test] +fn an_oversize_admit_is_a_typed_too_large_error() { + let mut mcp = Mcp::at(INSTANCE, closed_port()); + let mut arguments = faculty_args(vec![]); + arguments["session"] = json!("s".repeat(90_000)); + let (error, body) = mcp.call("admit", arguments); + assert!(error); + assert_eq!(body["error"], json!("TooLarge"), "{body}"); + assert_eq!(body["limit"], json!(eureka_state::MAX_REQUEST_BYTES)); + assert!(body["bytes"].as_u64().unwrap() > eureka_state::MAX_REQUEST_BYTES as u64, "{body}"); +} + +/// Who admitted a batch is validated by the mind, whichever door the batch +/// came through: a blank, whitespace, multi-line or oversized session or agent +/// is a typed refusal, and nothing is committed. +#[test] +fn admit_refuses_an_unusable_attribution() { + let (root, daemon) = mind(vec![]); + let server = serve(root, daemon); + let mut mcp = Mcp::at(INSTANCE, server.addr); + for (field, value) in [ + ("session", "".to_string()), + ("session", " ".to_string()), + ("session", "x".repeat(5_000)), + ("session", "line\nbreak".to_string()), + ("agent", "".to_string()), + ("agent", "two\nlines".to_string()), + ("agent", "y".repeat(201)), + ] { + let mut arguments = faculty_args(vec![campaign_json()]); + arguments[field] = json!(value); + let (error, refused) = mcp.call("admit", arguments); + assert!(!error, "{field}={value:?}"); + let refusal = &refused["Refused"]["Document"]; + let named = json!(format!("provenance.{field}")); + assert!( + refusal["InvalidFormat"]["field"] == named || refusal["FieldBound"]["field"] == named, + "{field}={value:?}: {refused}" + ); + } + let (_, whoami) = mcp.call("whoami", json!({})); + assert_eq!(whoami["status"]["receipts"], json!(1), "only the seed is committed: {whoami}"); + let (error, admitted) = mcp.call("admit", faculty_args(vec![campaign_json()])); + assert!(!error); + assert!(admitted.get("Committed").is_some(), "{admitted}"); +} + +/// The tool forwards the caller's semantic query, and only that, to the mind. +#[test] +fn query_sends_the_semantic_request_on_the_wire() { + let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let sink = std::sync::Arc::clone(&seen); + let server = scripted(move |message| { + let (_, request) = huginn_mind::envelope::decode_request(message).expect("a mind request"); + sink.lock().unwrap().push(request); + encode_response("eureka-state-call", "query", &HuginnMindResponse::View(None), "scripted").unwrap() + }); + let mut mcp = Mcp::at(INSTANCE, server.addr); + mcp.call("query", json!({ "selection": { "projection": "document" }, "semantic": { "text": "drift", "top_k": 7 } })); + mcp.call("query", json!({ "selection": {} })); + let seen = seen.lock().unwrap(); + let [HuginnMindRequest::Query { instance: first, semantic: Some(asked), .. }, HuginnMindRequest::Query { semantic: None, .. }] = + &seen[..] + else { + panic!("two queries expected, the first with its semantic part: {seen:?}"); + }; + assert_eq!(first.0, INSTANCE); + assert_eq!((asked.text.0.as_str(), asked.top_k), ("drift", 7)); +} + +/// A daemon that never answers is given up on after the server's own call +/// timeout, 15 seconds: not earlier, and not much later. +#[test] +fn a_silent_daemon_is_given_up_on_after_fifteen_seconds() { + let silent = UdpSocket::bind("127.0.0.1:0").unwrap(); + let mut mcp = Mcp::at(INSTANCE, silent.local_addr().unwrap()); + let started = std::time::Instant::now(); + let (error, body) = mcp.call("view", json!({ "id": absent() })); + let elapsed = started.elapsed(); + assert!(error); + assert_eq!(body["error"], json!("Unavailable"), "{body}"); + assert!(body["detail"].as_str().unwrap().contains("15s"), "{body}"); + assert!(elapsed >= Duration::from_secs(15), "gave up after {elapsed:?}"); + assert!(elapsed < Duration::from_secs(20), "gave up after {elapsed:?}"); +} + +/// An instance name that is not UTF-8 is reported as that, not as unset. +#[test] +fn a_non_utf8_instance_is_reported_as_such() { + let mut mcp = Mcp::spawn(|command| { + command.env("EUREKA_INSTANCE", non_utf8()).env("HUGINN_ENDPOINT", "rudp://127.0.0.1:1"); + }); + let (error, whoami) = mcp.call("whoami", json!({})); + assert!(!error); + let detail = whoami["error"]["detail"].as_str().unwrap(); + assert!(detail.contains("EUREKA_INSTANCE is not valid UTF-8"), "{whoami}"); + assert_eq!(whoami["instance"], Value::Null); +} + +#[cfg(unix)] +fn non_utf8() -> std::ffi::OsString { + std::os::unix::ffi::OsStringExt::from_vec(vec![b'e', 0xff]) +} + +#[cfg(windows)] +fn non_utf8() -> std::ffi::OsString { + std::os::windows::ffi::OsStringExt::from_wide(&[0x65, 0xd800]) +} + /// A well-formed id of a document no mind here holds. fn absent() -> Value { serde_json::to_value(PipelineRef { kind: PipelineKind::CutSpec, id: Short("eureka-state:cut_spec:cut-99.r1".into()) }) From db40a6647bfd1a691b6c20ee7d3eaba6541dbe48 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 02:54:54 +0200 Subject: [PATCH 09/12] eureka-state tests: size the oversize requests the way base64 allows Request sizes step by four (the payload rides as base64), so a request of exactly the limit is not reachable through the request. The client tests use the largest request that fits and the smallest that does not; the raw-socket test tops a message up with a source runtime id to reach the limit and one byte past it, which pins the limit against the transport's own refusal. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/tests/client.rs | 125 +++++++++++++++++++--------- 1 file changed, 85 insertions(+), 40 deletions(-) diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index 38cfe14..ff7d50c 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -8,8 +8,8 @@ use std::time::{Duration, Instant}; use chrono::Utc; use cultnet_rs::{ - CultNetMessage, CultNetWireContract, Selection, answer_content_chunk_request, encode_cultnet_message_to_vec, - pack_content, + CULTNET_OPERATION_CONNECTION_ID, CultMesh, CultMeshRudpSocketOptions, CultNetMessage, CultNetWireContract, Selection, + answer_content_chunk_request, encode_cultnet_message_to_vec, pack_content, }; use eureka_state::{ClientError, HuginnClient, MAX_REQUEST_BYTES}; use huginn_daemon::serve::MAX_RESPONSE_BYTES; @@ -381,40 +381,60 @@ fn a_body_over_the_deferral_cap_is_refused() { assert!(detail.contains("exceeds"), "{detail}"); } -/// An admission whose attribution is `agent_pad` and `session_pad` characters -/// long: the two knobs that size a request without changing what it asks. -fn padded_admit(agent_pad: usize, session_pad: usize) -> HuginnMindRequest { +/// An admission whose session is `session_pad` characters long: the knob that +/// sizes a request without changing what it asks. +fn padded_admit(session_pad: usize) -> HuginnMindRequest { let mut request = batch(INSTANCE, vec![stewardship()]); - request.provenance.agent = Short("a".repeat(agent_pad)); request.provenance.session = Short("s".repeat(session_pad)); HuginnMindRequest::Admit(request) } -/// What one send has to carry for `request`, measured the way the client does. -fn encoded_bytes(request: &HuginnMindRequest) -> usize { - let message = encode_request("eureka-state-call", request, None).unwrap(); - encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap().len() +/// What one send has to carry for `message`, measured the way the client does. +fn encoded_bytes(message: &CultNetMessage) -> usize { + encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0).unwrap().len() } -/// A request that encodes to exactly `target` bytes: the session padding is -/// searched for each agent padding until the base64 steps land on it. -fn request_of_size(target: usize) -> HuginnMindRequest { - for agent_pad in 1..16 { - let (mut low, mut high) = (0, target); - while low < high { - let middle = (low + high) / 2; - if encoded_bytes(&padded_admit(agent_pad, middle)) < target { - low = middle + 1; - } else { - high = middle; - } - } - let request = padded_admit(agent_pad, low); - if encoded_bytes(&request) == target { - return request; +fn message_for(request: &HuginnMindRequest, source: Option) -> CultNetMessage { + encode_request("eureka-state-call", request, source).unwrap() +} + +/// The smallest session padding whose request encodes to at least `target` +/// bytes. The payload rides as base64, so request sizes step by four and only +/// every fourth size is reachable through the request alone. +fn pad_for(target: usize) -> usize { + let (mut low, mut high) = (0, target); + while low < high { + let middle = (low + high) / 2; + if encoded_bytes(&message_for(&padded_admit(middle), None)) < target { + low = middle + 1; + } else { + high = middle; } } - panic!("no padding encodes to {target} bytes"); + low +} + +/// The largest request one send carries, and the smallest it does not. +fn largest_request_that_fits() -> HuginnMindRequest { + padded_admit(pad_for(MAX_REQUEST_BYTES + 1) - 1) +} + +fn smallest_request_over() -> HuginnMindRequest { + padded_admit(pad_for(MAX_REQUEST_BYTES + 1)) +} + +/// A message that encodes to exactly `target` bytes: the largest request that +/// fits under it, topped up with a source runtime id, which rides outside the +/// payload and moves the size a byte at a time. +fn message_of_size(target: usize) -> CultNetMessage { + let request = padded_admit(pad_for(target + 1) - 1); + for pad in 0..16 { + let message = message_for(&request, (pad > 0).then(|| "r".repeat(pad))); + if encoded_bytes(&message) == target { + return message; + } + } + panic!("no message encodes to {target} bytes"); } fn admitted_answer() -> CultNetMessage { @@ -429,26 +449,27 @@ fn admitted_answer() -> CultNetMessage { /// A request larger than one send carries is refused by the client before it /// connects to anything, as its own error and not as a transport failure the -/// caller would retry; one byte under the limit is sent, and answered. +/// caller would retry; the largest request that fits is sent, and answered. #[test] -fn a_request_over_what_one_send_carries_is_too_large_and_one_at_the_limit_is_sent() { +fn a_request_over_what_one_send_carries_is_too_large_and_one_that_fits_is_sent() { let closed = UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap(); - let over = request_of_size(MAX_REQUEST_BYTES + 1); + let over = smallest_request_over(); let started = Instant::now(); - match client(closed).call(over) { - Err(ClientError::TooLarge { bytes, limit }) => assert_eq!((bytes, limit), (MAX_REQUEST_BYTES + 1, MAX_REQUEST_BYTES)), + match client(closed).call(over.clone()) { + Err(ClientError::TooLarge { bytes, limit }) => { + assert_eq!(bytes, encoded_bytes(&message_for(&over, None))); + assert_eq!(limit, MAX_REQUEST_BYTES); + assert!(bytes > limit); + } other => panic!("expected TooLarge, got {other:?}"), } assert!(started.elapsed() < Duration::from_secs(1), "refused only after trying the transport"); + let fits = largest_request_that_fits(); + assert!(MAX_REQUEST_BYTES - encoded_bytes(&message_for(&fits, None)) < 4, "not the largest that fits"); let server = answering(admitted_answer()); - for size in [MAX_REQUEST_BYTES - 1, MAX_REQUEST_BYTES] { - let answer = client(server.addr).call(request_of_size(size)); - assert!( - matches!(answer, Ok(HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { .. }))), - "a request of {size} bytes: {answer:?}" - ); - } + let answer = client(server.addr).call(fits); + assert!(matches!(answer, Ok(HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { .. }))), "{answer:?}"); } /// The transport's own refusal of an oversize send ("Message too long") is @@ -457,6 +478,30 @@ fn a_request_over_what_one_send_carries_is_too_large_and_one_at_the_limit_is_sen #[test] fn an_oversize_admission_is_never_reported_as_an_unavailable_daemon() { let server = answering(admitted_answer()); - let error = client(server.addr).call(padded_admit(6, 70_000)).unwrap_err(); + let error = client(server.addr).call(padded_admit(70_000)).unwrap_err(); assert!(matches!(error, ClientError::TooLarge { .. }), "{error}"); } + +/// The limit is the transport's own: a message of `MAX_REQUEST_BYTES` goes out +/// on an RUDP session, and one byte more is refused by the socket. +#[test] +fn the_limit_is_the_largest_message_the_transport_sends() { + let server = answering(admitted_answer()); + let mut session = CultMesh::create_rudp_client_for_endpoint( + "limit-probe".to_string(), + CULTNET_OPERATION_CONNECTION_ID, + &format!("rudp://{}", server.addr), + CultMeshRudpSocketOptions::default(), + ) + .unwrap(); + session.connect(Vec::new()).unwrap(); + let until = Instant::now() + Duration::from_secs(5); + while !session.connected() { + session.poll_resends().unwrap(); + let _ = session.receive_once().unwrap(); + assert!(Instant::now() < until, "the server never accepted"); + std::thread::sleep(Duration::from_millis(2)); + } + session.send_schema_message(&message_of_size(MAX_REQUEST_BYTES)).expect("the limit is carried"); + assert!(session.send_schema_message(&message_of_size(MAX_REQUEST_BYTES + 1)).is_err(), "one byte more is sent"); +} From c2b2952749e4cfafcefc304dee7fa72262358b89 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 03:04:04 +0200 Subject: [PATCH 10/12] eureka-state tests: reach the limit through the source runtime id, seed the stewardship The source runtime id adds its field name when present, so exact sizes are searched over both paddings. The attribution test admits a campaign, which needs its repo stewarded first. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/tests/client.rs | 19 +++++++++++-------- crates/eureka-state/tests/mcp.rs | 4 ++-- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index ff7d50c..6a054ea 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -423,15 +423,18 @@ fn smallest_request_over() -> HuginnMindRequest { padded_admit(pad_for(MAX_REQUEST_BYTES + 1)) } -/// A message that encodes to exactly `target` bytes: the largest request that -/// fits under it, topped up with a source runtime id, which rides outside the -/// payload and moves the size a byte at a time. +/// A message that encodes to exactly `target` bytes: a request a little under +/// it, topped up with a source runtime id, which rides outside the payload and +/// moves the size a byte at a time once the field is there at all. fn message_of_size(target: usize) -> CultNetMessage { - let request = padded_admit(pad_for(target + 1) - 1); - for pad in 0..16 { - let message = message_for(&request, (pad > 0).then(|| "r".repeat(pad))); - if encoded_bytes(&message) == target { - return message; + let fitting = pad_for(target + 1) - 1; + for padding in (fitting.saturating_sub(80)..=fitting).rev() { + let request = padded_admit(padding); + for source in 1..80 { + let message = message_for(&request, Some("r".repeat(source))); + if encoded_bytes(&message) == target { + return message; + } } } panic!("no message encodes to {target} bytes"); diff --git a/crates/eureka-state/tests/mcp.rs b/crates/eureka-state/tests/mcp.rs index f6def4c..3a51eb0 100644 --- a/crates/eureka-state/tests/mcp.rs +++ b/crates/eureka-state/tests/mcp.rs @@ -413,7 +413,7 @@ fn an_oversize_admit_is_a_typed_too_large_error() { /// is a typed refusal, and nothing is committed. #[test] fn admit_refuses_an_unusable_attribution() { - let (root, daemon) = mind(vec![]); + let (root, daemon) = mind(vec![vec![stewardship()]]); let server = serve(root, daemon); let mut mcp = Mcp::at(INSTANCE, server.addr); for (field, value) in [ @@ -437,7 +437,7 @@ fn admit_refuses_an_unusable_attribution() { ); } let (_, whoami) = mcp.call("whoami", json!({})); - assert_eq!(whoami["status"]["receipts"], json!(1), "only the seed is committed: {whoami}"); + assert_eq!(whoami["status"]["receipts"], json!(2), "only the seed is committed: {whoami}"); let (error, admitted) = mcp.call("admit", faculty_args(vec![campaign_json()])); assert!(!error); assert!(admitted.get("Committed").is_some(), "{admitted}"); From 3d9965983f4dc1695eebd2f8e23b725a3f767bf0 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 04:55:04 +0200 Subject: [PATCH 11/12] huginn-mind: provenance follows the leaf's Title rule and covers the tool refuse_provenance validated only agent and session, and only for blank and a handful of line breaks. It now covers agent, session and tool with the leaf's Title character rule (an alphanumeric character, no control or line-break character, no bidi control) inside Short's byte bound. The leaf keeps that rule private, so a differential test holds the copy to a campaign Title. Restores the stage doc comment that drifted above SHORT_MAX. Multibyte bound test kills a chars-count mutant. Co-Authored-By: Claude Sonnet 5.5 --- crates/huginn-mind/src/admission.rs | 136 ++++++++++++++++++++-------- 1 file changed, 98 insertions(+), 38 deletions(-) diff --git a/crates/huginn-mind/src/admission.rs b/crates/huginn-mind/src/admission.rs index 0288ad0..6443e94 100644 --- a/crates/huginn-mind/src/admission.rs +++ b/crates/huginn-mind/src/admission.rs @@ -172,31 +172,40 @@ impl Mind { } } -/// A3 through the leaf, then A5. /// The bound `epiphany_pipeline::Short` declares, in UTF-8 bytes. The leaf /// exposes no validator for it, so `short_bound_is_the_leafs_own` pins this to /// the `maxLength` its schema publishes. const SHORT_MAX: usize = 200; /// Who admitted a batch is the receipt's only record of it, so the mind checks -/// it here and every writer, whatever its transport, gets the check: the agent -/// and the session are within `Short`'s bound, hold something besides -/// whitespace, and sit on one line. Refused as the leaf's own refusals, since -/// the fields are the leaf's own type. +/// it here and every writer, whatever its transport, gets the check: the agent, +/// the session and the tool follow the leaf's `Title` rule (one alphanumeric +/// character, no control or line-break character, no bidi control) within +/// `Short`'s bound. The leaf keeps that rule private, so +/// `provenance_text_is_the_leafs_title_rule` pins this to it. Refused as the +/// leaf's own refusals, since the fields are the leaf's own type. fn refuse_provenance(provenance: &PipelineProvenance) -> Result<(), MindRefusal> { - for (field, value) in [("provenance.agent", &provenance.agent), ("provenance.session", &provenance.session)] { + let bidi = |c: char| { + matches!(c, '\u{061C}' | '\u{200E}' | '\u{200F}' | '\u{202A}'..='\u{202E}' | '\u{2066}'..='\u{2069}') + }; + let control = |c: char| c.is_control() || matches!(c, '\u{2028}' | '\u{2029}'); + for (field, value) in [ + ("provenance.agent", &provenance.agent), + ("provenance.session", &provenance.session), + ("provenance.tool", &provenance.tool), + ] { let text = value.0.as_str(); + if !text.chars().any(char::is_alphanumeric) || text.contains(control) || text.contains(bidi) { + return Err(PipelineRefusal::InvalidFormat { field: field.into(), value: text.into() }.into()); + } if text.len() > SHORT_MAX { return Err(PipelineRefusal::FieldBound { field: field.into(), limit: SHORT_MAX as u32, actual: text.len() as u32 }.into()); } - let line_break = |c: char| matches!(c, '\n' | '\r' | '\u{0B}' | '\u{0C}' | '\u{85}' | '\u{2028}' | '\u{2029}'); - if text.trim().is_empty() || text.contains(line_break) { - return Err(PipelineRefusal::InvalidFormat { field: field.into(), value: text.into() }.into()); - } } Ok(()) } +/// A3 through the leaf, then A5. fn stage(envelope: CultCacheEnvelope, mind: &Slug) -> Result { validate_pipeline_write_envelope(&envelope)?; let document = PipelineDocument::decode(&envelope)?; @@ -720,8 +729,12 @@ mod tests { } fn admit_attributed(agent: &str, session: &str) -> (Mind, PipelineAdmissionOutcome) { + admit_signed(agent, session, "admit") + } + + fn admit_signed(agent: &str, session: &str, tool: &str) -> (Mind, PipelineAdmissionOutcome) { let mut mind = opened(MemoryStore::new(), INSTANCE); - let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: s(agent), session: s(session), tool: s("admit") }; + let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: s(agent), session: s(session), tool: s(tool) }; let batch = PipelineAdmissionBatch { instance: slug(INSTANCE), provenance, documents: vec![instance(INSTANCE), stewardship(INSTANCE, REPO)] }; let outcome = mind.admit(batch, now()); (mind, outcome) @@ -731,45 +744,92 @@ mod tests { /// a blank, multi-line or oversized attribution, and a refused batch leaves /// nothing behind. #[test] - fn provenance_names_an_agent_and_a_session_on_one_line_within_short() { + fn provenance_names_an_agent_a_session_and_a_tool_on_one_clean_line_within_short() { let over = "x".repeat(SHORT_MAX + 1); let exact = "y".repeat(SHORT_MAX); - for (field, agent, session) in [ - ("provenance.agent", "", "session-1"), - ("provenance.agent", " \t ", "session-1"), - ("provenance.session", "claude", ""), - ("provenance.session", "claude", " "), - ("provenance.session", "claude", "\u{2003}"), - ("provenance.session", "claude", "line\nbreak"), - ("provenance.session", "claude", "line\rbreak"), - ("provenance.session", "claude", "a\u{2028}b"), - ("provenance.agent", "two\nlines", "session-1"), + for bad in [ + "", + " \t ", + "\u{2003}", + "\u{200B}", + "line\nbreak", + "line\rbreak", + "a\u{0B}b", + "a\u{0C}b", + "a\u{85}b", + "a\u{2028}b", + "a\u{2029}b", + "a\0b", + "a\u{1B}[31mb", + "a\u{202E}b", + "a\u{2066}b", + "a\u{200F}b", ] { - let (mind, outcome) = admit_attributed(agent, session); - let refused = refusal(outcome); - assert!( - matches!(&refused, MindRefusal::Document(PipelineRefusal::InvalidFormat { field: named, .. }) if named == field), - "{agent:?}/{session:?}: {refused:?}" - ); - assert!(mind.is_empty(), "{agent:?}/{session:?}: a refused batch wrote"); + for field in ["provenance.agent", "provenance.session", "provenance.tool"] { + let (mind, outcome) = match field { + "provenance.agent" => admit_signed(bad, "session-1", "admit"), + "provenance.session" => admit_signed("claude", bad, "admit"), + _ => admit_signed("claude", "session-1", bad), + }; + let refused = refusal(outcome); + assert!( + matches!(&refused, MindRefusal::Document(PipelineRefusal::InvalidFormat { field: named, .. }) if named == field), + "{field}={bad:?}: {refused:?}" + ); + assert!(mind.is_empty(), "{field}={bad:?}: a refused batch wrote"); + } } - let (mind, outcome) = admit_attributed(&over, "session-1"); + for (field, outcome) in [ + ("provenance.agent", admit_signed(&over, "session-1", "admit")), + ("provenance.session", admit_signed("claude", &over, "admit")), + ("provenance.tool", admit_signed("claude", "session-1", &over)), + ] { + let (mind, outcome) = outcome; + assert_eq!( + refusal(outcome), + MindRefusal::Document(PipelineRefusal::FieldBound { field: field.into(), limit: 200, actual: 201 }) + ); + assert!(mind.is_empty()); + } + // The bound is in bytes, not characters: 101 two-byte characters are 202 bytes. + let (_, outcome) = admit_attributed(&"é".repeat(101), "session-1"); assert_eq!( refusal(outcome), - MindRefusal::Document(PipelineRefusal::FieldBound { field: "provenance.agent".into(), limit: 200, actual: 201 }) - ); - assert!(mind.is_empty()); - let (_, outcome) = admit_attributed("claude", &over); - assert_eq!( - refusal(outcome), - MindRefusal::Document(PipelineRefusal::FieldBound { field: "provenance.session".into(), limit: 200, actual: 201 }) + MindRefusal::Document(PipelineRefusal::FieldBound { field: "provenance.agent".into(), limit: 200, actual: 202 }) ); + let (_, outcome) = admit_signed("claude", "session-1", &"é".repeat(101)); + assert!(matches!(refusal(outcome), MindRefusal::Document(PipelineRefusal::FieldBound { actual: 202, .. }))); + let (_, outcome) = admit_attributed(&"é".repeat(100), "session-1"); + committed(outcome); // The bound is inclusive and a name with inner spaces is a name. - let (mind, outcome) = admit_attributed(&exact, "session 7"); + let (mind, outcome) = admit_signed(&exact, "session 7", "admit tool"); committed(outcome); let receipts = mind.receipts().unwrap(); assert_eq!(receipts[0].provenance.agent, s(&exact)); assert_eq!(receipts[0].provenance.session, s("session 7")); + assert_eq!(receipts[0].provenance.tool, s("admit tool")); + } + + /// The leaf's `Title` rule is private, so this holds the provenance rule to + /// it: the two accept and refuse the same text. + #[test] + fn provenance_text_is_the_leafs_title_rule() { + let leaf_accepts = |text: &str| { + let D::Campaign(mut campaign) = campaign(&[REPO]) else { unreachable!() }; + campaign.title = epiphany_pipeline::Title(text.into()); + D::Campaign(campaign).validate().is_ok() + }; + let mine_accepts = |text: &str| { + let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: s(text), session: s("s"), tool: s("t") }; + refuse_provenance(&provenance).is_ok() + }; + for text in [ + "claude", "session 7", "", " ", "\t", "\u{200B}", "\u{2003}", "!!!", "a\nb", "a\u{0B}b", "a\u{85}b", "a\u{2028}b", + "a\0b", "a\u{7F}b", "a\u{202E}b", "a\u{061C}b", "a\u{2069}b", "a\u{200D}b", "é", "日本", "a\u{FEFF}b", + &"x".repeat(200), &"x".repeat(201), &"é".repeat(100), &"é".repeat(101), + ] { + assert_eq!(mine_accepts(text), leaf_accepts(text), "{text:?}"); + } } #[test] From cd3dfc0268210736f059cb7501173ec2a36ad7ef Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 04:55:05 +0200 Subject: [PATCH 12/12] eureka-state: an encode failure is not retryable, and a non-UTF-8 endpoint is pinned A request that fails to encode is a defect in this client, not a transport failure, so it is ClientError::Unencodable, typed Unencodable at the tool, instead of the retryable Unavailable. A test reports a non-UTF-8 HUGINN_ENDPOINT as such. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/src/lib.rs | 12 ++++++++---- crates/eureka-state/src/main.rs | 5 +++++ crates/eureka-state/tests/client.rs | 4 ++-- crates/eureka-state/tests/mcp.rs | 13 +++++++++++++ 4 files changed, 28 insertions(+), 6 deletions(-) diff --git a/crates/eureka-state/src/lib.rs b/crates/eureka-state/src/lib.rs index 1ff2bb7..b6208e8 100644 --- a/crates/eureka-state/src/lib.rs +++ b/crates/eureka-state/src/lib.rs @@ -45,13 +45,15 @@ pub const MAX_REQUEST_BYTES: usize = 65_507 - 36 - "schema".len(); /// the only one worth retrying: the daemon could not be reached, went quiet /// past the call's deadline, or answered out of protocol. `Rejected` is the /// daemon refusing the envelope itself, and `TooLarge` is the client refusing -/// a request that no single send can carry, before sending it; a retry of the -/// same call cannot change either. +/// a request that no single send can carry, before sending it; `Unencodable` is +/// the request failing to encode at all, a defect in this client. A retry of the +/// same call cannot change any of those three. #[derive(Debug)] pub enum ClientError { Unavailable { endpoint: SocketAddr, detail: String }, Rejected { endpoint: SocketAddr, code: String, detail: String }, TooLarge { bytes: usize, limit: usize }, + Unencodable { detail: String }, } impl fmt::Display for ClientError { @@ -66,6 +68,7 @@ impl fmt::Display for ClientError { Self::TooLarge { bytes, limit } => { write!(f, "the request encodes to {bytes} bytes and one send carries at most {limit}") } + Self::Unencodable { detail } => write!(f, "the request could not be encoded: {detail}"), } } } @@ -109,9 +112,10 @@ impl HuginnClient { let deadline = Instant::now() + self.timeout; let unavailable = |error: anyhow::Error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") }; - let message = encode_request(MESSAGE_ID, &request, None).map_err(unavailable)?; + let unencodable = |error: anyhow::Error| ClientError::Unencodable { detail: format!("{error:#}") }; + let message = encode_request(MESSAGE_ID, &request, None).map_err(unencodable)?; let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0) - .map_err(unavailable)? + .map_err(unencodable)? .len(); if bytes > MAX_REQUEST_BYTES { return Err(ClientError::TooLarge { bytes, limit: MAX_REQUEST_BYTES }); diff --git a/crates/eureka-state/src/main.rs b/crates/eureka-state/src/main.rs index d5a8bc5..d368a23 100644 --- a/crates/eureka-state/src/main.rs +++ b/crates/eureka-state/src/main.rs @@ -108,6 +108,7 @@ impl Trouble { Self::Client(ClientError::TooLarge { bytes, limit }) => { json!({ "error": "TooLarge", "bytes": bytes, "limit": limit }) } + Self::Client(ClientError::Unencodable { detail }) => json!({ "error": "Unencodable", "detail": detail }), Self::Internal(detail) => json!({ "error": "Internal", "detail": detail }), } } @@ -317,6 +318,10 @@ mod tests { Trouble::Client(ClientError::TooLarge { bytes: 70_000, limit: 65_000 }).body(), json!({ "error": "TooLarge", "bytes": 70_000, "limit": 65_000 }) ); + assert_eq!( + Trouble::Client(ClientError::Unencodable { detail: "no".into() }).body(), + json!({ "error": "Unencodable", "detail": "no" }) + ); assert_eq!(Trouble::Misconfigured("bad".into()).body(), json!({ "error": "Misconfigured", "detail": "bad" })); } diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index 6a054ea..3f34512 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -128,14 +128,14 @@ fn client(addr: SocketAddr) -> HuginnClient { fn unavailable(error: ClientError) -> (SocketAddr, String) { match error { ClientError::Unavailable { endpoint, detail } => (endpoint, detail), - other @ (ClientError::Rejected { .. } | ClientError::TooLarge { .. }) => panic!("expected Unavailable, got {other}"), + other @ (ClientError::Rejected { .. } | ClientError::TooLarge { .. } | ClientError::Unencodable { .. }) => panic!("expected Unavailable, got {other}"), } } fn rejected(error: ClientError) -> (SocketAddr, String, String) { match error { ClientError::Rejected { endpoint, code, detail } => (endpoint, code, detail), - other @ (ClientError::Unavailable { .. } | ClientError::TooLarge { .. }) => panic!("expected Rejected, got {other}"), + other @ (ClientError::Unavailable { .. } | ClientError::TooLarge { .. } | ClientError::Unencodable { .. }) => panic!("expected Rejected, got {other}"), } } diff --git a/crates/eureka-state/tests/mcp.rs b/crates/eureka-state/tests/mcp.rs index 3a51eb0..27ee9fd 100644 --- a/crates/eureka-state/tests/mcp.rs +++ b/crates/eureka-state/tests/mcp.rs @@ -495,6 +495,19 @@ fn a_non_utf8_instance_is_reported_as_such() { assert_eq!(whoami["instance"], Value::Null); } +/// Likewise an endpoint that is not UTF-8: reported as that, not as unset. +#[test] +fn a_non_utf8_endpoint_is_reported_as_such() { + let mut mcp = Mcp::spawn(|command| { + command.env("EUREKA_INSTANCE", INSTANCE).env("HUGINN_ENDPOINT", non_utf8()); + }); + let (error, whoami) = mcp.call("whoami", json!({})); + assert!(!error); + let detail = whoami["error"]["detail"].as_str().unwrap(); + assert!(detail.contains("HUGINN_ENDPOINT is not valid UTF-8"), "{whoami}"); + assert_eq!(whoami["endpoint"], Value::Null); +} + #[cfg(unix)] fn non_utf8() -> std::ffi::OsString { std::os::unix::ffi::OsStringExt::from_vec(vec![b'e', 0xff])