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/lib.rs b/crates/eureka-state/src/lib.rs index 94dbfec..b6208e8 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,27 @@ 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; `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 { @@ -53,6 +65,10 @@ 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}") + } + Self::Unencodable { detail } => write!(f, "the request could not be encoded: {detail}"), } } } @@ -96,8 +112,16 @@ impl HuginnClient { let deadline = Instant::now() + self.timeout; let unavailable = |error: anyhow::Error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") }; + 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(unencodable)? + .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 +138,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 new file mode 100644 index 0000000..d368a23 --- /dev/null +++ b/crates/eureka-state/src/main.rs @@ -0,0 +1,349 @@ +//! `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, 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; +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::tool::ToolCallContext; +use rmcp::handler::server::wrapper::Parameters; +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}; + +/// 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::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 }), + } + } +} + +/// 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 = 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) } + } +} + +/// 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 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 + } +} + +/// 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.", + ) + } +} + +#[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(()) +} + +#[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::Client(ClientError::Unencodable { detail: "no".into() }).body(), + json!({ "error": "Unencodable", "detail": "no" }) + ); + 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 07f458a..3f34512 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -3,87 +3,30 @@ //! 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, + 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}; -use huginn_daemon::serve::{MAX_RESPONSE_BYTES, bind, run, schema_registry}; -use huginn_daemon::{Daemon, Handled, Hits, IndexSink, SearchTicket, ServeOptions}; -use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response}; +use eureka_state::{ClientError, HuginnClient, MAX_REQUEST_BYTES}; +use huginn_daemon::serve::MAX_RESPONSE_BYTES; +use huginn_mind::envelope::{OperationFailure, encode_failure, encode_request, 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, FileChange, Line, NegativeCheck, OrgRepo, + PipelineCutSpec, PipelineDocument, PipelineKind, PipelineRef, 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::*; -const INSTANCE: &str = "eureka"; 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), - repo: OrgRepo(REPO.into()), - sequence: 1, - assigned_on: Date("2026-09-29".into()), - note: "assigned".into(), - }) -} - -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 +94,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) @@ -287,22 +128,17 @@ 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 { .. } | ClientError::Unencodable { .. }) => 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 { .. } | ClientError::Unencodable { .. }) => panic!("expected Rejected, got {other}"), } } -/// 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() } @@ -544,3 +380,131 @@ 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 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.session = Short("s".repeat(session_pad)); + HuginnMindRequest::Admit(request) +} + +/// 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() +} + +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; + } + } + 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: 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 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"); +} + +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; the largest request that fits is sent, and answered. +#[test] +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 = smallest_request_over(); + let started = Instant::now(); + 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()); + 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 +/// 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(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"); +} diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs new file mode 100644 index 0000000..27a0705 --- /dev/null +++ b/crates/eureka-state/tests/common/mod.rs @@ -0,0 +1,185 @@ +//! 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, PipelineRef, PipelineStewardship, 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() }, + }) +} + +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 new file mode 100644 index 0000000..27ee9fd --- /dev/null +++ b/crates/eureka-state/tests/mcp.rs @@ -0,0 +1,532 @@ +//! 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::{HuginnMindRequest, 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, + /// The server's answer to `initialize`. + init: Value, +} + +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 { + 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 + .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, 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 + } + + /// 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)), + _ => {} + } +} + +/// 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] +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"]); + } + // 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"] { + 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 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(), + ) + .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![vec![stewardship()]]); + 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!(2)); + 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!(3), "{page}"); + assert_eq!(page["items"]["Documents"].as_array().unwrap().len(), 3); + + let (_, counted) = mcp.call("whoami", json!({})); + assert_eq!(counted["status"]["receipts"], json!(3)); +} + +/// 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![vec![stewardship()]]); + 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![vec![stewardship()]]); + 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 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_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!(response.get("error").is_none(), "{response}"); + assert_eq!(response["result"]["isError"], json!(true)); + 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![vec![stewardship()]]); + 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!(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}"); +} + +/// 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); +} + +/// 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]) +} + +#[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()) }) + .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); +} diff --git a/crates/huginn-mind/src/admission.rs b/crates/huginn-mind/src/admission.rs index 4b85cdd..6443e94 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 }); @@ -170,6 +172,39 @@ impl Mind { } } +/// 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, +/// 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> { + 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()); + } + } + Ok(()) +} + /// A3 through the leaf, then A5. fn stage(envelope: CultCacheEnvelope, mind: &Slug) -> Result { validate_pipeline_write_envelope(&envelope)?; @@ -693,6 +728,116 @@ mod tests { assert_eq!(mind.genesis_receipt_id().unwrap(), Some(first)); } + 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(tool) }; + 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_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 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", + ] { + 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"); + } + } + 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: 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_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] + 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();