Merge Cut 13b: the eureka-state MCP server
Four tools 1:1 with the wire (whoami, admit, view, query); the server fills instance and provenance.tool; config never kills startup; typed TooLarge, Rejected, Unencodable and InvalidInput errors. The mind now validates admission provenance (agent, session, tool) by the leaf's Title rule. Two Soul passes; registered by the operator 2026-09-30. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
commit
0c4e4107c2
9 changed files with 1751 additions and 188 deletions
254
Cargo.lock
generated
254
Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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" }
|
||||
|
|
|
|||
100
crates/eureka-state/schemas/cultnet.selection.schema.json
Normal file
100
crates/eureka-state/schemas/cultnet.selection.schema.json
Normal file
|
|
@ -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" }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<HuginnMindResponse, Failure> {
|
||||
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());
|
||||
};
|
||||
|
|
|
|||
349
crates/eureka-state/src/main.rs
Normal file
349
crates/eureka-state/src/main.rs
Normal file
|
|
@ -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<PipelineDocument>,
|
||||
}
|
||||
|
||||
#[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<SemanticQuery>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
endpoint: Option<String>,
|
||||
client: Result<HuginnClient, String>,
|
||||
}
|
||||
|
||||
impl Settings {
|
||||
fn from_env() -> Self {
|
||||
let instance = variable("EUREKA_INSTANCE");
|
||||
let endpoint = variable("HUGINN_ENDPOINT");
|
||||
let shown = |value: &Result<Option<String>, 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<Option<String>, 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<Option<String>, String>, endpoint: Result<Option<String>, String>) -> Result<HuginnClient, String> {
|
||||
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::<SocketAddr>().ok())
|
||||
.ok_or_else(|| format!("HUGINN_ENDPOINT is not rudp://<ip>:<port>: {endpoint}"))?;
|
||||
Ok(HuginnClient::new(address, slug, TIMEOUT))
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct EurekaState {
|
||||
settings: Arc<Settings>,
|
||||
tool_router: ToolRouter<Self>,
|
||||
}
|
||||
|
||||
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<HuginnMindResponse, Trouble> {
|
||||
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<AdmitInput>) -> 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<ViewInput>) -> 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<QueryInput>) -> 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::<Value>(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<RoleServer>,
|
||||
) -> Result<CallToolResult, ErrorData> {
|
||||
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::<Value>(&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::<Value>(&body_of(&typed_error(quoted)).1).unwrap()["error"], json!("InvalidInput"));
|
||||
}
|
||||
}
|
||||
|
|
@ -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<PipelineDocument>) -> 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<S: MindStore> IndexSink<S> for Inert {
|
||||
fn committed(&mut self, _mind: &Mind<S>, _writes: &[PipelineRef]) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn status(&self) -> IndexStatus {
|
||||
IndexStatus::Current
|
||||
}
|
||||
fn search(&mut self, _text: &str, _top_k: u32) -> Result<SearchTicket, String> {
|
||||
Err("this index cannot search".into())
|
||||
}
|
||||
fn search_deadline(&self) -> Duration {
|
||||
Duration::ZERO
|
||||
}
|
||||
fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
|
||||
Vec::new()
|
||||
}
|
||||
fn abandon(&mut self, _ticket: SearchTicket) {}
|
||||
}
|
||||
|
||||
type TestDaemon = Daemon<OwnedRedbMessagePackBackingStore, Inert>;
|
||||
|
||||
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<Vec<PipelineDocument>>) -> (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<AtomicBool>,
|
||||
thread: Option<JoinHandle<()>>,
|
||||
/// The sessions a scripted server has seen end: what the client's
|
||||
/// disconnect looks like from the daemon's side.
|
||||
ended: Arc<AtomicUsize>,
|
||||
_root: Option<TempDir>,
|
||||
}
|
||||
|
||||
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<u8>) -> 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<String>) -> 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");
|
||||
}
|
||||
|
|
|
|||
185
crates/eureka-state/tests/common/mod.rs
Normal file
185
crates/eureka-state/tests/common/mod.rs
Normal file
|
|
@ -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<PipelineDocument>) -> 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<S: MindStore> IndexSink<S> for Inert {
|
||||
fn committed(&mut self, _mind: &Mind<S>, _writes: &[PipelineRef]) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn status(&self) -> IndexStatus {
|
||||
IndexStatus::Current
|
||||
}
|
||||
fn search(&mut self, _text: &str, _top_k: u32) -> Result<SearchTicket, String> {
|
||||
Err("this index cannot search".into())
|
||||
}
|
||||
fn search_deadline(&self) -> Duration {
|
||||
Duration::ZERO
|
||||
}
|
||||
fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
|
||||
Vec::new()
|
||||
}
|
||||
fn abandon(&mut self, _ticket: SearchTicket) {}
|
||||
}
|
||||
|
||||
pub type TestDaemon = Daemon<OwnedRedbMessagePackBackingStore, Inert>;
|
||||
|
||||
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<Vec<PipelineDocument>>) -> (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<AtomicBool>,
|
||||
pub thread: Option<JoinHandle<()>>,
|
||||
/// The sessions a scripted server has seen end: what the client's
|
||||
/// disconnect looks like from the daemon's side.
|
||||
pub ended: Arc<AtomicUsize>,
|
||||
pub _root: Option<TempDir>,
|
||||
}
|
||||
|
||||
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(),
|
||||
})
|
||||
}
|
||||
532
crates/eureka-state/tests/mcp.rs
Normal file
532
crates/eureka-state/tests/mcp.rs
Normal file
|
|
@ -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<String>,
|
||||
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>) -> 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<String>) {
|
||||
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);
|
||||
}
|
||||
|
|
@ -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<S: MindStore> Mind<S> {
|
|||
) -> Result<PipelineAdmissionOutcome, MindRefusal> {
|
||||
// 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<S: MindStore> Mind<S> {
|
|||
}
|
||||
}
|
||||
|
||||
/// 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<Staged, MindRefusal> {
|
||||
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<MemoryStore>, PipelineAdmissionOutcome) {
|
||||
admit_signed(agent, session, "admit")
|
||||
}
|
||||
|
||||
fn admit_signed(agent: &str, session: &str, tool: &str) -> (Mind<MemoryStore>, 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();
|
||||
|
|
|
|||
Loading…
Reference in a new issue