diff --git a/crates/eureka-state/src/lib.rs b/crates/eureka-state/src/lib.rs index 94dbfec..1ff2bb7 100644 --- a/crates/eureka-state/src/lib.rs +++ b/crates/eureka-state/src/lib.rs @@ -23,7 +23,7 @@ use std::time::{Duration, Instant}; use anyhow::{Context, Result, anyhow, bail}; use cultnet_rs::{ CULTNET_OPERATION_CONNECTION_ID, CultMesh, CultMeshRudpSocketOptions, CultNetMessage, - CultNetRudpSocketTransportConnection, fetch_content, + CultNetRudpSocketTransportConnection, CultNetWireContract, encode_cultnet_message_to_vec, fetch_content, }; use huginn_mind::envelope::{OperationFailure, decode_response, encode_request}; use huginn_mind::epiphany_pipeline::Slug; @@ -33,15 +33,25 @@ use huginn_mind::{HuginnMindRequest, HuginnMindResponse, MAX_DEFERRED_BODY_BYTES const MESSAGE_ID: &str = "eureka-state-call"; const POLL: Duration = Duration::from_millis(2); +/// The largest encoded message one send carries. The client's RUDP session +/// does not fragment (`max_fragment_bytes` is unset), so a message is one UDP +/// datagram: at most 65,507 bytes over IPv4 (65,535 less the 8-byte UDP and +/// 20-byte IP headers), less RUDP's 36-byte fixed header (`RUDP_FIXED_HEADER_BYTES` +/// in `cultnet-rs`, private there) and the 6-byte `schema` channel id. A larger +/// message makes the operating system refuse the send ("Message too long"). +pub const MAX_REQUEST_BYTES: usize = 65_507 - 36 - "schema".len(); + /// Why a call produced no answer. `Unavailable` is the transport's failure and /// the only one worth retrying: the daemon could not be reached, went quiet /// past the call's deadline, or answered out of protocol. `Rejected` is the -/// daemon refusing the envelope itself, which a retry of the same call cannot -/// change. +/// daemon refusing the envelope itself, and `TooLarge` is the client refusing +/// a request that no single send can carry, before sending it; a retry of the +/// same call cannot change either. #[derive(Debug)] pub enum ClientError { Unavailable { endpoint: SocketAddr, detail: String }, Rejected { endpoint: SocketAddr, code: String, detail: String }, + TooLarge { bytes: usize, limit: usize }, } impl fmt::Display for ClientError { @@ -53,6 +63,9 @@ impl fmt::Display for ClientError { Self::Rejected { endpoint, code, detail } => { write!(f, "the Huginn daemon at rudp://{endpoint} rejected the request: {code}: {detail}") } + Self::TooLarge { bytes, limit } => { + write!(f, "the request encodes to {bytes} bytes and one send carries at most {limit}") + } } } } @@ -96,8 +109,15 @@ impl HuginnClient { let deadline = Instant::now() + self.timeout; let unavailable = |error: anyhow::Error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") }; + let message = encode_request(MESSAGE_ID, &request, None).map_err(unavailable)?; + let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0) + .map_err(unavailable)? + .len(); + if bytes > MAX_REQUEST_BYTES { + return Err(ClientError::TooLarge { bytes, limit: MAX_REQUEST_BYTES }); + } let mut session = self.connect(deadline).map_err(unavailable)?; - let outcome = self.exchange(&mut session, &request, deadline); + let outcome = self.exchange(&mut session, &request, &message, deadline); // Best effort: the daemon drops a session it never hears from, only later. let _ = session.disconnect(Vec::new()); outcome.map_err(|failure| match failure { @@ -114,9 +134,10 @@ impl HuginnClient { &self, session: &mut CultNetRudpSocketTransportConnection, request: &HuginnMindRequest, + message: &CultNetMessage, deadline: Instant, ) -> Result { - let reply = self.ask(session, &encode_request(MESSAGE_ID, request, None)?, deadline)?; + let reply = self.ask(session, message, deadline)?; let CultNetMessage::OperationResponse { operation, .. } = &reply else { return Err(anyhow!("the daemon answered with something other than an operation response").into()); }; diff --git a/crates/eureka-state/src/main.rs b/crates/eureka-state/src/main.rs index e6d529f..d5a8bc5 100644 --- a/crates/eureka-state/src/main.rs +++ b/crates/eureka-state/src/main.rs @@ -11,8 +11,10 @@ //! startup vanishes from Claude Code without a word. `whoami` reports it and //! every other tool refuses with it. A refusal from the mind is an answer and //! comes back as a normal result; a daemon that cannot be reached or that -//! rejects the envelope is a result with `isError: true`, typed, so a caller -//! can tell a down daemon from a permanent mismatch. +//! rejects the envelope, or a request too large for one send, is a result with +//! `isError: true`, typed, so a caller can tell a down daemon from a permanent +//! mismatch. Arguments that do not fit a tool's input are typed the same way +//! (`InvalidInput`), as a tool error and not a protocol one. use std::net::SocketAddr; use std::sync::Arc; @@ -25,9 +27,11 @@ use huginn_mind::{ Faculty, HuginnMindRequest, HuginnMindResponse, PipelineAdmissionBatch, PipelineProvenance, SemanticQuery, }; use rmcp::handler::server::router::tool::ToolRouter; +use rmcp::handler::server::tool::ToolCallContext; use rmcp::handler::server::wrapper::Parameters; -use rmcp::model::{CallToolResult, ContentBlock, ServerCapabilities, ServerInfo}; -use rmcp::{ServerHandler, ServiceExt, tool, tool_handler, tool_router}; +use rmcp::model::{CallToolRequestParams, CallToolResult, ContentBlock, ServerCapabilities, ServerInfo}; +use rmcp::service::RequestContext; +use rmcp::{ErrorData, RoleServer, ServerHandler, ServiceExt, tool, tool_handler, tool_router}; use schemars::{JsonSchema, Schema, SchemaGenerator}; use serde::Deserialize; use serde_json::{Value, json}; @@ -101,6 +105,9 @@ impl Trouble { Self::Client(ClientError::Rejected { endpoint, code, detail }) => { json!({ "error": "Rejected", "endpoint": format!("rudp://{endpoint}"), "code": code, "detail": detail }) } + Self::Client(ClientError::TooLarge { bytes, limit }) => { + json!({ "error": "TooLarge", "bytes": bytes, "limit": limit }) + } Self::Internal(detail) => json!({ "error": "Internal", "detail": detail }), } } @@ -116,18 +123,28 @@ struct Settings { impl Settings { fn from_env() -> Self { - let instance = std::env::var("EUREKA_INSTANCE").ok(); - let endpoint = std::env::var("HUGINN_ENDPOINT").ok(); - let client = client(instance.as_deref(), endpoint.as_deref()); - Self { instance, endpoint, client } + let instance = variable("EUREKA_INSTANCE"); + let endpoint = variable("HUGINN_ENDPOINT"); + let shown = |value: &Result, String>| value.as_ref().ok().cloned().flatten(); + let (shown_instance, shown_endpoint) = (shown(&instance), shown(&endpoint)); + Self { instance: shown_instance, endpoint: shown_endpoint, client: client(instance, endpoint) } } } -fn client(instance: Option<&str>, endpoint: Option<&str>) -> Result { - let instance = instance.ok_or("EUREKA_INSTANCE is not set")?; - let slug = Slug(instance.to_string()); +/// An environment variable: its value, `None` when unset, or why it cannot be read. +fn variable(name: &str) -> Result, String> { + match std::env::var(name) { + Ok(value) => Ok(Some(value)), + Err(std::env::VarError::NotPresent) => Ok(None), + Err(std::env::VarError::NotUnicode(_)) => Err(format!("{name} is not valid UTF-8")), + } +} + +fn client(instance: Result, String>, endpoint: Result, String>) -> Result { + let instance = instance?.ok_or("EUREKA_INSTANCE is not set")?; + let slug = Slug(instance); slug.validate_slug().map_err(|refusal| format!("EUREKA_INSTANCE is not a valid instance name: {refusal:?}"))?; - let endpoint = endpoint.ok_or("HUGINN_ENDPOINT is not set")?; + let endpoint = endpoint?.ok_or("HUGINN_ENDPOINT is not set")?; let address = endpoint .strip_prefix("rudp://") .and_then(|address| address.parse::().ok()) @@ -138,7 +155,6 @@ fn client(instance: Option<&str>, endpoint: Option<&str>) -> Result, - #[expect(dead_code, reason = "the tool_handler macro reads this router field")] tool_router: ToolRouter, } @@ -244,8 +260,32 @@ impl EurekaState { } } +/// Every tool error is a typed JSON body. The ones this server writes already +/// are; the one rmcp writes, for arguments that do not deserialize into the +/// tool's input, is plain text, so it is put into the same shape. +fn typed_error(result: CallToolResult) -> CallToolResult { + if result.is_error != Some(true) { + return result; + } + let Some(text) = result.content.first().and_then(ContentBlock::as_text).map(|text| text.text.as_str()) else { + return result; + }; + if serde_json::from_str::(text).is_ok_and(|body| body.is_object()) { + return result; + } + reply(json!({ "error": "InvalidInput", "detail": text }), true) +} + #[tool_handler] impl ServerHandler for EurekaState { + async fn call_tool( + &self, + request: CallToolRequestParams, + context: RequestContext, + ) -> Result { + Ok(typed_error(self.tool_router.call(ToolCallContext::new(self, request, context)).await?)) + } + fn get_info(&self) -> ServerInfo { ServerInfo::new(ServerCapabilities::builder().enable_tools().build()).with_instructions( "Huginn's typed mind for this instance: whoami, admit, view, query. Refusals are answers; isError means the daemon could not be reached or rejected the request.", @@ -259,3 +299,46 @@ async fn main() -> anyhow::Result<()> { server.waiting().await?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + fn body_of(result: &CallToolResult) -> (bool, String) { + (result.is_error == Some(true), result.content[0].as_text().unwrap().text.clone()) + } + + /// Reaching `Internal` for real takes a panic in a blocking task, so its + /// body is pinned where it is built. + #[test] + fn every_trouble_has_its_typed_body() { + assert_eq!(Trouble::Internal("task panicked".into()).body(), json!({ "error": "Internal", "detail": "task panicked" })); + assert_eq!( + Trouble::Client(ClientError::TooLarge { bytes: 70_000, limit: 65_000 }).body(), + json!({ "error": "TooLarge", "bytes": 70_000, "limit": 65_000 }) + ); + assert_eq!(Trouble::Misconfigured("bad".into()).body(), json!({ "error": "Misconfigured", "detail": "bad" })); + } + + /// Only the text rmcp writes is rewritten: our own typed bodies and every + /// success pass through untouched. + #[test] + fn typed_error_rewrites_only_rmcp_plain_text() { + let plain = CallToolResult::error(vec![ContentBlock::text("failed to deserialize parameters: missing field `faculty`")]); + let (is_error, text) = body_of(&typed_error(plain)); + assert!(is_error); + assert_eq!( + serde_json::from_str::(&text).unwrap(), + json!({ "error": "InvalidInput", "detail": "failed to deserialize parameters: missing field `faculty`" }) + ); + + let typed = reply(json!({ "error": "Unavailable", "detail": "down" }), true); + assert_eq!(body_of(&typed_error(typed.clone())), body_of(&typed)); + + let success = CallToolResult::success(vec![ContentBlock::text("not json, and fine")]); + assert_eq!(body_of(&typed_error(success)), (false, "not json, and fine".to_string())); + + let quoted = CallToolResult::error(vec![ContentBlock::text("[1, 2]")]); + assert_eq!(serde_json::from_str::(&body_of(&typed_error(quoted)).1).unwrap()["error"], json!("InvalidInput")); + } +} diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index 56ab80c..38cfe14 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -11,9 +11,9 @@ use cultnet_rs::{ CultNetMessage, CultNetWireContract, Selection, answer_content_chunk_request, encode_cultnet_message_to_vec, pack_content, }; -use eureka_state::{ClientError, HuginnClient}; +use eureka_state::{ClientError, HuginnClient, MAX_REQUEST_BYTES}; use huginn_daemon::serve::MAX_RESPONSE_BYTES; -use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response}; +use huginn_mind::envelope::{OperationFailure, encode_failure, encode_request, encode_response}; use huginn_mind::epiphany_pipeline::{ AuthorityMap, CodeLocation, CutDelete, CutVerification, FileChange, Line, NegativeCheck, OrgRepo, PipelineCutSpec, PipelineDocument, PipelineKind, PipelineRef, Short, StructuralDelta, Title, @@ -128,14 +128,14 @@ fn client(addr: SocketAddr) -> HuginnClient { fn unavailable(error: ClientError) -> (SocketAddr, String) { match error { ClientError::Unavailable { endpoint, detail } => (endpoint, detail), - rejected @ ClientError::Rejected { .. } => panic!("expected Unavailable, got {rejected}"), + other @ (ClientError::Rejected { .. } | ClientError::TooLarge { .. }) => panic!("expected Unavailable, got {other}"), } } fn rejected(error: ClientError) -> (SocketAddr, String, String) { match error { ClientError::Rejected { endpoint, code, detail } => (endpoint, code, detail), - unavailable @ ClientError::Unavailable { .. } => panic!("expected Rejected, got {unavailable}"), + other @ (ClientError::Unavailable { .. } | ClientError::TooLarge { .. }) => panic!("expected Rejected, got {other}"), } } @@ -380,3 +380,83 @@ fn a_body_over_the_deferral_cap_is_refused() { let (_, detail) = unavailable(client(server.addr).call(HuginnMindRequest::Whoami).unwrap_err()); assert!(detail.contains("exceeds"), "{detail}"); } + +/// An admission whose attribution is `agent_pad` and `session_pad` characters +/// long: the two knobs that size a request without changing what it asks. +fn padded_admit(agent_pad: usize, session_pad: usize) -> HuginnMindRequest { + let mut request = batch(INSTANCE, vec![stewardship()]); + request.provenance.agent = Short("a".repeat(agent_pad)); + request.provenance.session = Short("s".repeat(session_pad)); + HuginnMindRequest::Admit(request) +} + +/// What one send has to carry for `request`, measured the way the client does. +fn encoded_bytes(request: &HuginnMindRequest) -> usize { + let message = encode_request("eureka-state-call", request, None).unwrap(); + encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap().len() +} + +/// A request that encodes to exactly `target` bytes: the session padding is +/// searched for each agent padding until the base64 steps land on it. +fn request_of_size(target: usize) -> HuginnMindRequest { + for agent_pad in 1..16 { + let (mut low, mut high) = (0, target); + while low < high { + let middle = (low + high) / 2; + if encoded_bytes(&padded_admit(agent_pad, middle)) < target { + low = middle + 1; + } else { + high = middle; + } + } + let request = padded_admit(agent_pad, low); + if encoded_bytes(&request) == target { + return request; + } + } + panic!("no padding encodes to {target} bytes"); +} + +fn admitted_answer() -> CultNetMessage { + encode_response( + "eureka-state-call", + "admit", + &HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { receipt_id: "r".into() }), + "scripted", + ) + .unwrap() +} + +/// A request larger than one send carries is refused by the client before it +/// connects to anything, as its own error and not as a transport failure the +/// caller would retry; one byte under the limit is sent, and answered. +#[test] +fn a_request_over_what_one_send_carries_is_too_large_and_one_at_the_limit_is_sent() { + let closed = UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap(); + let over = request_of_size(MAX_REQUEST_BYTES + 1); + let started = Instant::now(); + match client(closed).call(over) { + Err(ClientError::TooLarge { bytes, limit }) => assert_eq!((bytes, limit), (MAX_REQUEST_BYTES + 1, MAX_REQUEST_BYTES)), + other => panic!("expected TooLarge, got {other:?}"), + } + assert!(started.elapsed() < Duration::from_secs(1), "refused only after trying the transport"); + + let server = answering(admitted_answer()); + for size in [MAX_REQUEST_BYTES - 1, MAX_REQUEST_BYTES] { + let answer = client(server.addr).call(request_of_size(size)); + assert!( + matches!(answer, Ok(HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { .. }))), + "a request of {size} bytes: {answer:?}" + ); + } +} + +/// The transport's own refusal of an oversize send ("Message too long") is +/// what an unchecked request used to report, as an `Unavailable` daemon a +/// caller would retry forever. +#[test] +fn an_oversize_admission_is_never_reported_as_an_unavailable_daemon() { + let server = answering(admitted_answer()); + let error = client(server.addr).call(padded_admit(6, 70_000)).unwrap_err(); + assert!(matches!(error, ClientError::TooLarge { .. }), "{error}"); +} diff --git a/crates/eureka-state/tests/mcp.rs b/crates/eureka-state/tests/mcp.rs index ab5a924..f6def4c 100644 --- a/crates/eureka-state/tests/mcp.rs +++ b/crates/eureka-state/tests/mcp.rs @@ -10,7 +10,7 @@ use std::sync::mpsc::{Receiver, channel}; use std::time::Duration; use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response}; -use huginn_mind::HuginnMindResponse; +use huginn_mind::{HuginnMindRequest, HuginnMindResponse}; use huginn_mind::epiphany_pipeline::{PipelineKind, PipelineRef, Short}; use serde_json::{Value, json}; @@ -38,11 +38,17 @@ impl Drop for Mcp { impl Mcp { /// The binary with exactly the environment given, initialised. fn start(env: &[(&str, &str)]) -> Self { + Self::spawn(|command| { + command.envs(env.iter().copied()); + }) + } + + /// The binary with no configuration but what `configure` gives its command. + fn spawn(configure: impl FnOnce(&mut Command)) -> Self { let mut command = Command::new(env!("CARGO_BIN_EXE_eureka-state")); + command.env_remove("EUREKA_INSTANCE").env_remove("HUGINN_ENDPOINT"); + configure(&mut command); command - .env_remove("EUREKA_INSTANCE") - .env_remove("HUGINN_ENDPOINT") - .envs(env.iter().copied()) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::null()); @@ -130,6 +136,18 @@ fn refs(schema: &Value, found: &mut Vec) { } } +/// Every key named `key` anywhere below `schema`'s root. +fn keys_below_root(schema: &Value, key: &str) -> usize { + fn count(value: &Value, key: &str) -> usize { + match value { + Value::Object(map) => map.iter().map(|(name, inner)| usize::from(name == key) + count(inner, key)).sum(), + Value::Array(items) => items.iter().map(|item| count(item, key)).sum(), + _ => 0, + } + } + schema.as_object().unwrap().values().map(|inner| count(inner, key)).sum() +} + /// Four tools, each input schema a self-contained document: every `$ref` is /// internal and resolves, and the selection's definitions are bundled in. #[test] @@ -147,7 +165,12 @@ fn tools_list_offers_four_tools_with_self_contained_schemas() { let name = reference.strip_prefix("#/$defs/").unwrap_or_else(|| panic!("{}: remote $ref {reference}", tool["name"])); assert!(schema["$defs"].get(name).is_some(), "{}: {reference} does not resolve", tool["name"]); } + // The vendored selection schema brings its own `$id` and `$schema`, + // which would make its `$ref`s resolve against a remote document. assert!(schema.get("$id").is_none(), "{}: a remote identity", tool["name"]); + for key in ["$id", "$schema"] { + assert_eq!(keys_below_root(schema, key), 0, "{}: a nested {key}", tool["name"]); + } } let query = tools.iter().find(|tool| tool["name"] == "query").unwrap(); for bundled in ["fieldPredicate", "citation", "incoming", "recordRef"] { @@ -363,11 +386,125 @@ fn malformed_input_is_refused_before_any_call() { let response = mcp.rpc("tools/call", json!({ "name": "admit", "arguments": { "faculty": "Hands" } })); assert!(response.get("error").is_none(), "{response}"); assert_eq!(response["result"]["isError"], json!(true)); - assert!(response["result"]["content"][0]["text"].as_str().unwrap().contains("missing field"), "{response}"); + let body: Value = serde_json::from_str(response["result"]["content"][0]["text"].as_str().unwrap()).expect("a JSON body"); + assert_eq!(body["error"], json!("InvalidInput"), "{body}"); + assert!(body["detail"].as_str().unwrap().contains("missing field"), "{body}"); let (error, _) = mcp.call("whoami", json!({})); assert!(!error); } +/// A request too large for one send is the client's typed refusal, an error +/// result that names the size and the limit and does not call the daemon +/// unavailable. +#[test] +fn an_oversize_admit_is_a_typed_too_large_error() { + let mut mcp = Mcp::at(INSTANCE, closed_port()); + let mut arguments = faculty_args(vec![]); + arguments["session"] = json!("s".repeat(90_000)); + let (error, body) = mcp.call("admit", arguments); + assert!(error); + assert_eq!(body["error"], json!("TooLarge"), "{body}"); + assert_eq!(body["limit"], json!(eureka_state::MAX_REQUEST_BYTES)); + assert!(body["bytes"].as_u64().unwrap() > eureka_state::MAX_REQUEST_BYTES as u64, "{body}"); +} + +/// Who admitted a batch is validated by the mind, whichever door the batch +/// came through: a blank, whitespace, multi-line or oversized session or agent +/// is a typed refusal, and nothing is committed. +#[test] +fn admit_refuses_an_unusable_attribution() { + let (root, daemon) = mind(vec![]); + let server = serve(root, daemon); + let mut mcp = Mcp::at(INSTANCE, server.addr); + for (field, value) in [ + ("session", "".to_string()), + ("session", " ".to_string()), + ("session", "x".repeat(5_000)), + ("session", "line\nbreak".to_string()), + ("agent", "".to_string()), + ("agent", "two\nlines".to_string()), + ("agent", "y".repeat(201)), + ] { + let mut arguments = faculty_args(vec![campaign_json()]); + arguments[field] = json!(value); + let (error, refused) = mcp.call("admit", arguments); + assert!(!error, "{field}={value:?}"); + let refusal = &refused["Refused"]["Document"]; + let named = json!(format!("provenance.{field}")); + assert!( + refusal["InvalidFormat"]["field"] == named || refusal["FieldBound"]["field"] == named, + "{field}={value:?}: {refused}" + ); + } + let (_, whoami) = mcp.call("whoami", json!({})); + assert_eq!(whoami["status"]["receipts"], json!(1), "only the seed is committed: {whoami}"); + let (error, admitted) = mcp.call("admit", faculty_args(vec![campaign_json()])); + assert!(!error); + assert!(admitted.get("Committed").is_some(), "{admitted}"); +} + +/// The tool forwards the caller's semantic query, and only that, to the mind. +#[test] +fn query_sends_the_semantic_request_on_the_wire() { + let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let sink = std::sync::Arc::clone(&seen); + let server = scripted(move |message| { + let (_, request) = huginn_mind::envelope::decode_request(message).expect("a mind request"); + sink.lock().unwrap().push(request); + encode_response("eureka-state-call", "query", &HuginnMindResponse::View(None), "scripted").unwrap() + }); + let mut mcp = Mcp::at(INSTANCE, server.addr); + mcp.call("query", json!({ "selection": { "projection": "document" }, "semantic": { "text": "drift", "top_k": 7 } })); + mcp.call("query", json!({ "selection": {} })); + let seen = seen.lock().unwrap(); + let [HuginnMindRequest::Query { instance: first, semantic: Some(asked), .. }, HuginnMindRequest::Query { semantic: None, .. }] = + &seen[..] + else { + panic!("two queries expected, the first with its semantic part: {seen:?}"); + }; + assert_eq!(first.0, INSTANCE); + assert_eq!((asked.text.0.as_str(), asked.top_k), ("drift", 7)); +} + +/// A daemon that never answers is given up on after the server's own call +/// timeout, 15 seconds: not earlier, and not much later. +#[test] +fn a_silent_daemon_is_given_up_on_after_fifteen_seconds() { + let silent = UdpSocket::bind("127.0.0.1:0").unwrap(); + let mut mcp = Mcp::at(INSTANCE, silent.local_addr().unwrap()); + let started = std::time::Instant::now(); + let (error, body) = mcp.call("view", json!({ "id": absent() })); + let elapsed = started.elapsed(); + assert!(error); + assert_eq!(body["error"], json!("Unavailable"), "{body}"); + assert!(body["detail"].as_str().unwrap().contains("15s"), "{body}"); + assert!(elapsed >= Duration::from_secs(15), "gave up after {elapsed:?}"); + assert!(elapsed < Duration::from_secs(20), "gave up after {elapsed:?}"); +} + +/// An instance name that is not UTF-8 is reported as that, not as unset. +#[test] +fn a_non_utf8_instance_is_reported_as_such() { + let mut mcp = Mcp::spawn(|command| { + command.env("EUREKA_INSTANCE", non_utf8()).env("HUGINN_ENDPOINT", "rudp://127.0.0.1:1"); + }); + let (error, whoami) = mcp.call("whoami", json!({})); + assert!(!error); + let detail = whoami["error"]["detail"].as_str().unwrap(); + assert!(detail.contains("EUREKA_INSTANCE is not valid UTF-8"), "{whoami}"); + assert_eq!(whoami["instance"], Value::Null); +} + +#[cfg(unix)] +fn non_utf8() -> std::ffi::OsString { + std::os::unix::ffi::OsStringExt::from_vec(vec![b'e', 0xff]) +} + +#[cfg(windows)] +fn non_utf8() -> std::ffi::OsString { + std::os::windows::ffi::OsStringExt::from_wide(&[0x65, 0xd800]) +} + /// A well-formed id of a document no mind here holds. fn absent() -> Value { serde_json::to_value(PipelineRef { kind: PipelineKind::CutSpec, id: Short("eureka-state:cut_spec:cut-99.r1".into()) })