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() +}