GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 2 Projects Releases Packages Wiki Activity Actions

eureka-state: typed TooLarge, typed InvalidInput, and the pins Soul found missing

Browse source 
- The client refuses a request whose encoded size exceeds what one RUDP send
  carries (65,507 less the 36-byte header and the 6-byte channel id), before
  connecting, as ClientError::TooLarge { bytes, limit }. It was reported as
  a retryable Unavailable ("Message too long"). The MCP tools surface it as
  isError with {"error":"TooLarge","bytes","limit"}.
- Arguments that fail to deserialize are still a tool error, now with the
  same typed body as every other one: {"error":"InvalidInput","detail"}.
- A non-UTF-8 EUREKA_INSTANCE or HUGINN_ENDPOINT is reported as not valid
  UTF-8 and not as unset.
- Tests: the schema walk asserts no $id or $schema below any tool's root; the
  15 s timeout is measured against a silent daemon; the semantic part of a
  query is observed on the wire; Internal's and TooLarge's bodies and the
  InvalidInput rewrite are pinned; attribution is refused through the MCP
  admit; oversize is TooLarge at and one past the limit.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 02:51:50 +02:00
commit 890a00b0de
4 changed files with 348 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

31
crates/eureka-state/src/lib.rs
View file

@ -23,7 +23,7 @@ use std::time::{Duration, Instant};
use anyhow::{Context, Result, anyhow, bail};
use cultnet_rs::{
CULTNET_OPERATION_CONNECTION_ID, CultMesh, CultMeshRudpSocketOptions, CultNetMessage,
CultNetRudpSocketTransportConnection, fetch_content,
CultNetRudpSocketTransportConnection, CultNetWireContract, encode_cultnet_message_to_vec, fetch_content,
};
use huginn_mind::envelope::{OperationFailure, decode_response, encode_request};
use huginn_mind::epiphany_pipeline::Slug;
@ -33,15 +33,25 @@ use huginn_mind::{HuginnMindRequest, HuginnMindResponse, MAX_DEFERRED_BODY_BYTES
const MESSAGE_ID: &str = "eureka-state-call";
const POLL: Duration = Duration::from_millis(2);
/// The largest encoded message one send carries. The client's RUDP session
/// does not fragment (`max_fragment_bytes` is unset), so a message is one UDP
/// datagram: at most 65,507 bytes over IPv4 (65,535 less the 8-byte UDP and
/// 20-byte IP headers), less RUDP's 36-byte fixed header (`RUDP_FIXED_HEADER_BYTES`
/// in `cultnet-rs`, private there) and the 6-byte `schema` channel id. A larger
/// message makes the operating system refuse the send ("Message too long").
pub const MAX_REQUEST_BYTES: usize = 65_507 - 36 - "schema".len();
/// Why a call produced no answer. `Unavailable` is the transport's failure and
/// the only one worth retrying: the daemon could not be reached, went quiet
/// past the call's deadline, or answered out of protocol. `Rejected` is the
/// daemon refusing the envelope itself, which a retry of the same call cannot
/// change.
/// daemon refusing the envelope itself, and `TooLarge` is the client refusing
/// a request that no single send can carry, before sending it; a retry of the
/// same call cannot change either.
#[derive(Debug)]
pub enum ClientError {
Unavailable { endpoint: SocketAddr, detail: String },
Rejected { endpoint: SocketAddr, code: String, detail: String },
TooLarge { bytes: usize, limit: usize },
}
impl fmt::Display for ClientError {
@ -53,6 +63,9 @@ impl fmt::Display for ClientError {
Self::Rejected { endpoint, code, detail } => {
write!(f, "the Huginn daemon at rudp://{endpoint} rejected the request: {code}: {detail}")
}
Self::TooLarge { bytes, limit } => {
write!(f, "the request encodes to {bytes} bytes and one send carries at most {limit}")
}
}
}
}
@ -96,8 +109,15 @@ impl HuginnClient {
let deadline = Instant::now() + self.timeout;
let unavailable =
|error: anyhow::Error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") };
let message = encode_request(MESSAGE_ID, &request, None).map_err(unavailable)?;
let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)
.map_err(unavailable)?
.len();
if bytes > MAX_REQUEST_BYTES {
return Err(ClientError::TooLarge { bytes, limit: MAX_REQUEST_BYTES });
}
let mut session = self.connect(deadline).map_err(unavailable)?;
let outcome = self.exchange(&mut session, &request, deadline);
let outcome = self.exchange(&mut session, &request, &message, deadline);
// Best effort: the daemon drops a session it never hears from, only later.
let _ = session.disconnect(Vec::new());
outcome.map_err(|failure| match failure {
@ -114,9 +134,10 @@ impl HuginnClient {
&self,
session: &mut CultNetRudpSocketTransportConnection,
request: &HuginnMindRequest,
message: &CultNetMessage,
deadline: Instant,
) -> Result<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());
};

109
crates/eureka-state/src/main.rs
View file

@ -11,8 +11,10 @@
//! startup vanishes from Claude Code without a word. `whoami` reports it and
//! every other tool refuses with it. A refusal from the mind is an answer and
//! comes back as a normal result; a daemon that cannot be reached or that
//! rejects the envelope is a result with `isError: true`, typed, so a caller
//! can tell a down daemon from a permanent mismatch.
//! rejects the envelope, or a request too large for one send, is a result with
//! `isError: true`, typed, so a caller can tell a down daemon from a permanent
//! mismatch. Arguments that do not fit a tool's input are typed the same way
//! (`InvalidInput`), as a tool error and not a protocol one.
use std::net::SocketAddr;
use std::sync::Arc;
@ -25,9 +27,11 @@ use huginn_mind::{
Faculty, HuginnMindRequest, HuginnMindResponse, PipelineAdmissionBatch, PipelineProvenance, SemanticQuery,
};
use rmcp::handler::server::router::tool::ToolRouter;
use rmcp::handler::server::tool::ToolCallContext;
use rmcp::handler::server::wrapper::Parameters;
use rmcp::model::{CallToolResult, ContentBlock, ServerCapabilities, ServerInfo};
use rmcp::{ServerHandler, ServiceExt, tool, tool_handler, tool_router};
use rmcp::model::{CallToolRequestParams, CallToolResult, ContentBlock, ServerCapabilities, ServerInfo};
use rmcp::service::RequestContext;
use rmcp::{ErrorData, RoleServer, ServerHandler, ServiceExt, tool, tool_handler, tool_router};
use schemars::{JsonSchema, Schema, SchemaGenerator};
use serde::Deserialize;
use serde_json::{Value, json};
@ -101,6 +105,9 @@ impl Trouble {
Self::Client(ClientError::Rejected { endpoint, code, detail }) => {
json!({ "error": "Rejected", "endpoint": format!("rudp://{endpoint}"), "code": code, "detail": detail })
}
Self::Client(ClientError::TooLarge { bytes, limit }) => {
json!({ "error": "TooLarge", "bytes": bytes, "limit": limit })
}
Self::Internal(detail) => json!({ "error": "Internal", "detail": detail }),
}
}
@ -116,18 +123,28 @@ struct Settings {
impl Settings {
fn from_env() -> Self {
let instance = std::env::var("EUREKA_INSTANCE").ok();
let endpoint = std::env::var("HUGINN_ENDPOINT").ok();
let client = client(instance.as_deref(), endpoint.as_deref());
Self { instance, endpoint, client }
let instance = variable("EUREKA_INSTANCE");
let endpoint = variable("HUGINN_ENDPOINT");
let shown = |value: &Result<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) }
}
}
fn client(instance: Option<&str>, endpoint: Option<&str>) -> Result<HuginnClient, String> {
let instance = instance.ok_or("EUREKA_INSTANCE is not set")?;
let slug = Slug(instance.to_string());
/// An environment variable: its value, `None` when unset, or why it cannot be read.
fn variable(name: &str) -> Result<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 endpoint = endpoint?.ok_or("HUGINN_ENDPOINT is not set")?;
let address = endpoint
.strip_prefix("rudp://")
.and_then(|address| address.parse::<SocketAddr>().ok())
@ -138,7 +155,6 @@ fn client(instance: Option<&str>, endpoint: Option<&str>) -> Result<HuginnClient
#[derive(Clone)]
struct EurekaState {
settings: Arc<Settings>,
#[expect(dead_code, reason = "the tool_handler macro reads this router field")]
tool_router: ToolRouter<Self>,
}
@ -244,8 +260,32 @@ impl EurekaState {
}
}
/// Every tool error is a typed JSON body. The ones this server writes already
/// are; the one rmcp writes, for arguments that do not deserialize into the
/// tool's input, is plain text, so it is put into the same shape.
fn typed_error(result: CallToolResult) -> CallToolResult {
if result.is_error != Some(true) {
return result;
}
let Some(text) = result.content.first().and_then(ContentBlock::as_text).map(|text| text.text.as_str()) else {
return result;
};
if serde_json::from_str::<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.",
@ -259,3 +299,46 @@ async fn main() -> anyhow::Result<()> {
server.waiting().await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn body_of(result: &CallToolResult) -> (bool, String) {
(result.is_error == Some(true), result.content[0].as_text().unwrap().text.clone())
}
/// Reaching `Internal` for real takes a panic in a blocking task, so its
/// body is pinned where it is built.
#[test]
fn every_trouble_has_its_typed_body() {
assert_eq!(Trouble::Internal("task panicked".into()).body(), json!({ "error": "Internal", "detail": "task panicked" }));
assert_eq!(
Trouble::Client(ClientError::TooLarge { bytes: 70_000, limit: 65_000 }).body(),
json!({ "error": "TooLarge", "bytes": 70_000, "limit": 65_000 })
);
assert_eq!(Trouble::Misconfigured("bad".into()).body(), json!({ "error": "Misconfigured", "detail": "bad" }));
}
/// Only the text rmcp writes is rewritten: our own typed bodies and every
/// success pass through untouched.
#[test]
fn typed_error_rewrites_only_rmcp_plain_text() {
let plain = CallToolResult::error(vec![ContentBlock::text("failed to deserialize parameters: missing field `faculty`")]);
let (is_error, text) = body_of(&typed_error(plain));
assert!(is_error);
assert_eq!(
serde_json::from_str::<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"));
}
}

88
crates/eureka-state/tests/client.rs
View file

@ -11,9 +11,9 @@ use cultnet_rs::{
CultNetMessage, CultNetWireContract, Selection, answer_content_chunk_request, encode_cultnet_message_to_vec,
pack_content,
};
use eureka_state::{ClientError, HuginnClient};
use eureka_state::{ClientError, HuginnClient, MAX_REQUEST_BYTES};
use huginn_daemon::serve::MAX_RESPONSE_BYTES;
use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response};
use huginn_mind::envelope::{OperationFailure, encode_failure, encode_request, encode_response};
use huginn_mind::epiphany_pipeline::{
AuthorityMap, CodeLocation, CutDelete, CutVerification, FileChange, Line, NegativeCheck, OrgRepo,
PipelineCutSpec, PipelineDocument, PipelineKind, PipelineRef, Short, StructuralDelta, Title,
@ -128,14 +128,14 @@ fn client(addr: SocketAddr) -> HuginnClient {
fn unavailable(error: ClientError) -> (SocketAddr, String) {
match error {
ClientError::Unavailable { endpoint, detail } => (endpoint, detail),
rejected @ ClientError::Rejected { .. } => panic!("expected Unavailable, got {rejected}"),
other @ (ClientError::Rejected { .. } | ClientError::TooLarge { .. }) => panic!("expected Unavailable, got {other}"),
}
}
fn rejected(error: ClientError) -> (SocketAddr, String, String) {
match error {
ClientError::Rejected { endpoint, code, detail } => (endpoint, code, detail),
unavailable @ ClientError::Unavailable { .. } => panic!("expected Rejected, got {unavailable}"),
other @ (ClientError::Unavailable { .. } | ClientError::TooLarge { .. }) => panic!("expected Rejected, got {other}"),
}
}
@ -380,3 +380,83 @@ fn a_body_over_the_deferral_cap_is_refused() {
let (_, detail) = unavailable(client(server.addr).call(HuginnMindRequest::Whoami).unwrap_err());
assert!(detail.contains("exceeds"), "{detail}");
}
/// An admission whose attribution is `agent_pad` and `session_pad` characters
/// long: the two knobs that size a request without changing what it asks.
fn padded_admit(agent_pad: usize, session_pad: usize) -> HuginnMindRequest {
let mut request = batch(INSTANCE, vec![stewardship()]);
request.provenance.agent = Short("a".repeat(agent_pad));
request.provenance.session = Short("s".repeat(session_pad));
HuginnMindRequest::Admit(request)
}
/// What one send has to carry for `request`, measured the way the client does.
fn encoded_bytes(request: &HuginnMindRequest) -> usize {
let message = encode_request("eureka-state-call", request, None).unwrap();
encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap().len()
}
/// A request that encodes to exactly `target` bytes: the session padding is
/// searched for each agent padding until the base64 steps land on it.
fn request_of_size(target: usize) -> HuginnMindRequest {
for agent_pad in 1..16 {
let (mut low, mut high) = (0, target);
while low < high {
let middle = (low + high) / 2;
if encoded_bytes(&padded_admit(agent_pad, middle)) < target {
low = middle + 1;
} else {
high = middle;
}
}
let request = padded_admit(agent_pad, low);
if encoded_bytes(&request) == target {
return request;
}
}
panic!("no padding encodes to {target} bytes");
}
fn admitted_answer() -> CultNetMessage {
encode_response(
"eureka-state-call",
"admit",
&HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { receipt_id: "r".into() }),
"scripted",
)
.unwrap()
}
/// A request larger than one send carries is refused by the client before it
/// connects to anything, as its own error and not as a transport failure the
/// caller would retry; one byte under the limit is sent, and answered.
#[test]
fn a_request_over_what_one_send_carries_is_too_large_and_one_at_the_limit_is_sent() {
let closed = UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap();
let over = request_of_size(MAX_REQUEST_BYTES + 1);
let started = Instant::now();
match client(closed).call(over) {
Err(ClientError::TooLarge { bytes, limit }) => assert_eq!((bytes, limit), (MAX_REQUEST_BYTES + 1, MAX_REQUEST_BYTES)),
other => panic!("expected TooLarge, got {other:?}"),
}
assert!(started.elapsed() < Duration::from_secs(1), "refused only after trying the transport");
let server = answering(admitted_answer());
for size in [MAX_REQUEST_BYTES - 1, MAX_REQUEST_BYTES] {
let answer = client(server.addr).call(request_of_size(size));
assert!(
matches!(answer, Ok(HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { .. }))),
"a request of {size} bytes: {answer:?}"
);
}
}
/// The transport's own refusal of an oversize send ("Message too long") is
/// what an unchecked request used to report, as an `Unavailable` daemon a
/// caller would retry forever.
#[test]
fn an_oversize_admission_is_never_reported_as_an_unavailable_daemon() {
let server = answering(admitted_answer());
let error = client(server.addr).call(padded_admit(6, 70_000)).unwrap_err();
assert!(matches!(error, ClientError::TooLarge { .. }), "{error}");
}

147
crates/eureka-state/tests/mcp.rs
View file

@ -10,7 +10,7 @@ use std::sync::mpsc::{Receiver, channel};
use std::time::Duration;
use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response};
use huginn_mind::HuginnMindResponse;
use huginn_mind::{HuginnMindRequest, HuginnMindResponse};
use huginn_mind::epiphany_pipeline::{PipelineKind, PipelineRef, Short};
use serde_json::{Value, json};
@ -38,11 +38,17 @@ impl Drop for Mcp {
impl Mcp {
/// The binary with exactly the environment given, initialised.
fn start(env: &[(&str, &str)]) -> Self {
Self::spawn(|command| {
command.envs(env.iter().copied());
})
}
/// The binary with no configuration but what `configure` gives its command.
fn spawn(configure: impl FnOnce(&mut Command)) -> Self {
let mut command = Command::new(env!("CARGO_BIN_EXE_eureka-state"));
command.env_remove("EUREKA_INSTANCE").env_remove("HUGINN_ENDPOINT");
configure(&mut command);
command
.env_remove("EUREKA_INSTANCE")
.env_remove("HUGINN_ENDPOINT")
.envs(env.iter().copied())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null());
@ -130,6 +136,18 @@ fn refs(schema: &Value, found: &mut Vec<String>) {
}
}
/// Every key named `key` anywhere below `schema`'s root.
fn keys_below_root(schema: &Value, key: &str) -> usize {
fn count(value: &Value, key: &str) -> usize {
match value {
Value::Object(map) => map.iter().map(|(name, inner)| usize::from(name == key) + count(inner, key)).sum(),
Value::Array(items) => items.iter().map(|item| count(item, key)).sum(),
_ => 0,
}
}
schema.as_object().unwrap().values().map(|inner| count(inner, key)).sum()
}
/// Four tools, each input schema a self-contained document: every `$ref` is
/// internal and resolves, and the selection's definitions are bundled in.
#[test]
@ -147,7 +165,12 @@ fn tools_list_offers_four_tools_with_self_contained_schemas() {
let name = reference.strip_prefix("#/$defs/").unwrap_or_else(|| panic!("{}: remote $ref {reference}", tool["name"]));
assert!(schema["$defs"].get(name).is_some(), "{}: {reference} does not resolve", tool["name"]);
}
// The vendored selection schema brings its own `$id` and `$schema`,
// which would make its `$ref`s resolve against a remote document.
assert!(schema.get("$id").is_none(), "{}: a remote identity", tool["name"]);
for key in ["$id", "$schema"] {
assert_eq!(keys_below_root(schema, key), 0, "{}: a nested {key}", tool["name"]);
}
}
let query = tools.iter().find(|tool| tool["name"] == "query").unwrap();
for bundled in ["fieldPredicate", "citation", "incoming", "recordRef"] {
@ -363,11 +386,125 @@ fn malformed_input_is_refused_before_any_call() {
let response = mcp.rpc("tools/call", json!({ "name": "admit", "arguments": { "faculty": "Hands" } }));
assert!(response.get("error").is_none(), "{response}");
assert_eq!(response["result"]["isError"], json!(true));
assert!(response["result"]["content"][0]["text"].as_str().unwrap().contains("missing field"), "{response}");
let body: Value = serde_json::from_str(response["result"]["content"][0]["text"].as_str().unwrap()).expect("a JSON body");
assert_eq!(body["error"], json!("InvalidInput"), "{body}");
assert!(body["detail"].as_str().unwrap().contains("missing field"), "{body}");
let (error, _) = mcp.call("whoami", json!({}));
assert!(!error);
}
/// A request too large for one send is the client's typed refusal, an error
/// result that names the size and the limit and does not call the daemon
/// unavailable.
#[test]
fn an_oversize_admit_is_a_typed_too_large_error() {
let mut mcp = Mcp::at(INSTANCE, closed_port());
let mut arguments = faculty_args(vec![]);
arguments["session"] = json!("s".repeat(90_000));
let (error, body) = mcp.call("admit", arguments);
assert!(error);
assert_eq!(body["error"], json!("TooLarge"), "{body}");
assert_eq!(body["limit"], json!(eureka_state::MAX_REQUEST_BYTES));
assert!(body["bytes"].as_u64().unwrap() > eureka_state::MAX_REQUEST_BYTES as u64, "{body}");
}
/// Who admitted a batch is validated by the mind, whichever door the batch
/// came through: a blank, whitespace, multi-line or oversized session or agent
/// is a typed refusal, and nothing is committed.
#[test]
fn admit_refuses_an_unusable_attribution() {
let (root, daemon) = mind(vec![]);
let server = serve(root, daemon);
let mut mcp = Mcp::at(INSTANCE, server.addr);
for (field, value) in [
("session", "".to_string()),
("session", " ".to_string()),
("session", "x".repeat(5_000)),
("session", "line\nbreak".to_string()),
("agent", "".to_string()),
("agent", "two\nlines".to_string()),
("agent", "y".repeat(201)),
] {
let mut arguments = faculty_args(vec![campaign_json()]);
arguments[field] = json!(value);
let (error, refused) = mcp.call("admit", arguments);
assert!(!error, "{field}={value:?}");
let refusal = &refused["Refused"]["Document"];
let named = json!(format!("provenance.{field}"));
assert!(
refusal["InvalidFormat"]["field"] == named || refusal["FieldBound"]["field"] == named,
"{field}={value:?}: {refused}"
);
}
let (_, whoami) = mcp.call("whoami", json!({}));
assert_eq!(whoami["status"]["receipts"], json!(1), "only the seed is committed: {whoami}");
let (error, admitted) = mcp.call("admit", faculty_args(vec![campaign_json()]));
assert!(!error);
assert!(admitted.get("Committed").is_some(), "{admitted}");
}
/// The tool forwards the caller's semantic query, and only that, to the mind.
#[test]
fn query_sends_the_semantic_request_on_the_wire() {
let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let sink = std::sync::Arc::clone(&seen);
let server = scripted(move |message| {
let (_, request) = huginn_mind::envelope::decode_request(message).expect("a mind request");
sink.lock().unwrap().push(request);
encode_response("eureka-state-call", "query", &HuginnMindResponse::View(None), "scripted").unwrap()
});
let mut mcp = Mcp::at(INSTANCE, server.addr);
mcp.call("query", json!({ "selection": { "projection": "document" }, "semantic": { "text": "drift", "top_k": 7 } }));
mcp.call("query", json!({ "selection": {} }));
let seen = seen.lock().unwrap();
let [HuginnMindRequest::Query { instance: first, semantic: Some(asked), .. }, HuginnMindRequest::Query { semantic: None, .. }] =
&seen[..]
else {
panic!("two queries expected, the first with its semantic part: {seen:?}");
};
assert_eq!(first.0, INSTANCE);
assert_eq!((asked.text.0.as_str(), asked.top_k), ("drift", 7));
}
/// A daemon that never answers is given up on after the server's own call
/// timeout, 15 seconds: not earlier, and not much later.
#[test]
fn a_silent_daemon_is_given_up_on_after_fifteen_seconds() {
let silent = UdpSocket::bind("127.0.0.1:0").unwrap();
let mut mcp = Mcp::at(INSTANCE, silent.local_addr().unwrap());
let started = std::time::Instant::now();
let (error, body) = mcp.call("view", json!({ "id": absent() }));
let elapsed = started.elapsed();
assert!(error);
assert_eq!(body["error"], json!("Unavailable"), "{body}");
assert!(body["detail"].as_str().unwrap().contains("15s"), "{body}");
assert!(elapsed >= Duration::from_secs(15), "gave up after {elapsed:?}");
assert!(elapsed < Duration::from_secs(20), "gave up after {elapsed:?}");
}
/// An instance name that is not UTF-8 is reported as that, not as unset.
#[test]
fn a_non_utf8_instance_is_reported_as_such() {
let mut mcp = Mcp::spawn(|command| {
command.env("EUREKA_INSTANCE", non_utf8()).env("HUGINN_ENDPOINT", "rudp://127.0.0.1:1");
});
let (error, whoami) = mcp.call("whoami", json!({}));
assert!(!error);
let detail = whoami["error"]["detail"].as_str().unwrap();
assert!(detail.contains("EUREKA_INSTANCE is not valid UTF-8"), "{whoami}");
assert_eq!(whoami["instance"], Value::Null);
}
#[cfg(unix)]
fn non_utf8() -> std::ffi::OsString {
std::os::unix::ffi::OsStringExt::from_vec(vec![b'e', 0xff])
}
#[cfg(windows)]
fn non_utf8() -> std::ffi::OsString {
std::os::windows::ffi::OsStringExt::from_wide(&[0x65, 0xd800])
}
/// A well-formed id of a document no mind here holds.
fn absent() -> Value {
serde_json::to_value(PipelineRef { kind: PipelineKind::CutSpec, id: Short("eureka-state:cut_spec:cut-99.r1".into()) })