diff --git a/Cargo.toml b/Cargo.toml index 837f5f41c..ebb739d05 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -385,7 +385,7 @@ tokio = { version = "1", features = ["test-util"] } proptest = "1" [features] -default = ["tokenjuice-treesitter", "inference", "voice", "web3", "media", "documents", "meet", "skills", "flows", "mcp", "crash-reporting", "http-server", "desktop-automation", "channels", "tui"] +default = ["tokenjuice-treesitter", "inference", "voice", "web3", "media", "documents", "meet", "skills", "flows", "mcp", "crash-reporting", "http-server", "desktop-automation", "channels", "tui", "medulla-local"] # HTTP + Socket.IO server transport (#5048): the `/rpc` JSON-RPC endpoint and # its auth middleware/CORS layer (`core::jsonrpc`, `core::auth`), the `/v1` # OpenAI-compatible router (`inference::http`), the ad-hoc static-dir file @@ -596,6 +596,26 @@ desktop-automation = ["dep:uiautomation"] # build-fact "tui feature disabled at compile time" error from the untouched # `"tui" | "chat"` CLI arm (mirrors the `mcp` stub pattern). tui = ["dep:ratatui", "dep:crossterm", "dep:unicode-width"] +# Local Medulla brain (plan Flavor A, §3.1–§3.2): the `medulla_local` domain — +# a supervised `medulla-serve` Node child speaking the serve NDJSON protocol, +# host-side inference + tools port callbacks, the `medulla_local` RPC namespace, +# and the subconscious-replacement draft (`subconscious.engine = "medulla"`, +# §5.2). Default-ON — the desktop app ships it — and therefore FORWARDED to the +# Tauri shell's `openhuman_core` features (default-features=false there). Slim / +# headless builds opt out via `--no-default-features --features ""`. The module is `#[cfg(feature = "medulla-local")]` at its +# `pub mod` declaration (a registration-site gate, like `flows`); the +# `subconscious.engine` config field stays compiled in ALL builds (inert serde), +# and the subconscious tick's medulla branch is `#[cfg]`-gated so the default +# (`local`) path is byte-identical whether or not this feature is on. +# Sheds ZERO exclusive dependencies — tokio net/process, reqwest, serde are all +# load-bearing for other domains; the value is compile-time surface. +# The serve transport is unix-only (unix domain sockets): on non-unix targets +# (Windows desktop) the feature still compiles via a supervisor stub that +# reports a typed unsupported-platform error, so forwarding the default-ON +# gate to the Tauri shell never breaks Windows packaging. +medulla-local = [] + # Channels domain: `openhuman::channels` (external-messaging providers — Telegram, # Discord, Slack, Signal, WhatsApp, iMessage, IRC, … — plus the channel runtime, # controllers, host, proactive messaging and inbound dispatch) together with the diff --git a/app/src-tauri/Cargo.toml b/app/src-tauri/Cargo.toml index 349a35567..716eb3e9b 100644 --- a/app/src-tauri/Cargo.toml +++ b/app/src-tauri/Cargo.toml @@ -178,6 +178,7 @@ openhuman_core = { path = "../..", package = "openhuman", default-features = fal # (#5048). Enforced by the HTTP_SERVER_COMPILED_IN compile assert in lib.rs. "http-server", "desktop-automation", + "medulla-local", ] } tinyjuice = { version = "0.2.1", default-features = false } diff --git a/docs/TEST-COVERAGE-MATRIX.md b/docs/TEST-COVERAGE-MATRIX.md index e7c1a6f3a..1153bc166 100644 --- a/docs/TEST-COVERAGE-MATRIX.md +++ b/docs/TEST-COVERAGE-MATRIX.md @@ -516,6 +516,7 @@ End-to-end coverage of the agent harness via the web-chat RPC surface against an | ------ | --------------------------------------------------------------- | ----- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | 11.3.1 | Hosted-only orchestration (client = trigger + effects + render) | RI+VU | `tests/orchestration_hosted_client.rs`, `app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsx`, `app/src/components/orchestration/__tests__/AgentChatPanel.test.tsx` | ✅ | Local wake-graph brain retired (frontend_agent/graph/master_agent/reasoning_agent deleted). Client forwards events to the hosted brain (`POST /orchestration/v1/events`), uploads world-diffs, syncs hosted sessions/messages/steering into the render cache, and executes `send_dm`/`evict` socket effects; cloud-unreachable banner on outage. | | 11.3.2 | Direct paid Medulla orchestration with local OpenHuman tools | RU | `src/openhuman/orchestration/medulla.rs`, `src/openhuman/orchestration/schemas.rs` | ✅ | `openhuman.orchestration_run` checks the active paid plan, starts a hosted Medulla cycle, executes requested contact/session/send tools locally, and continues pending/tool-use events to a final result. Mocked HTTP tests cover direct and tool-loop success plus plan, pending, backend-error, unknown-tool, and tool-failure paths. | +| 11.3.3 | Local medulla-serve supervision (`medulla_local` RPC, Flavor A draft) | RU+RI | `src/openhuman/medulla_local/server_tests.rs`, `tests/medulla_local_e2e.rs` | ✅ | Behind the default-ON `medulla-local` feature. Unit tests drive the supervisor's handshake seam against a mock connector: restart-and-retry fires only on mid-request transport death of an idempotent op (exactly one respawn); the non-idempotent `instruct` is never replayed — a transport break surfaces the typed `MaybeApplied` error with no duplicate submission; an overall per-request deadline (`subconscious.medulla_local.request_deadline_secs`, default 300s) bounds a request even while the child keeps streaming frames or while a serve→host port callback (`inference.invoke`/`tools.invoke`) hangs on a stalled provider/tool — tripping as the typed transport error (`MaybeApplied` for `instruct`, restart-and-retry-once for `status`); a retry that also breaks mid-request resets the replacement connection, so the next request re-establishes cleanly instead of reusing a poisoned transport; child liveness is probed before the cached connection is trusted, so a child that dies between requests reports `running=false` and is respawned on the next request instead of misreporting `MaybeApplied`; `ok=false` serve rejections fail fast with the typed `RequestError` without killing the child; and the global cache is keyed on an order-canonical config fingerprint (map-field insertion order cannot invalidate it, a changed config rebuilds and bypasses the start-failure backoff). The transport is unix-only; non-unix targets compile a stub whose entry points report a typed unsupported-platform error (self-tested on those targets). The RI suite boots the real JSON-RPC router and asserts `medulla_local.status` returns a well-formed idle snapshot (running=false, actionable message) and `instruct` fails cleanly when no serve entry is configured — hermetic, no Node child or network. | --- diff --git a/src/core/all.rs b/src/core/all.rs index 0312dd93c..d764cc724 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -496,6 +496,16 @@ fn build_registered_controllers() -> Vec { DomainGroup::Platform, crate::openhuman::javascript::all_javascript_registered_controllers(), ); + // Local Medulla brain (plan Flavor A, §3.1–§3.2): status/instruct against a + // supervised `medulla-serve` child. Registration-site gate, like `flows` — + // with the feature off the `medulla_local.*` methods are simply absent + // (unknown-method), not a runtime error. + #[cfg(feature = "medulla-local")] + push( + &mut controllers, + DomainGroup::Agent, + crate::openhuman::medulla_local::all_medulla_local_registered_controllers(), + ); // Discovered SKILL.md skills and their bundled resources push( &mut controllers, @@ -931,6 +941,7 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "inference" => Some("Connect to configured text, vision, and embedding inference runtimes."), "migrate" => Some("Data migration utilities."), "javascript" => Some("First-class JavaScript runtime bridge for listing and dispatching tools."), + "medulla_local" => Some("Supervised local medulla-serve brain: status of the child and instruct enqueue (Flavor A draft)."), "monitor" => Some("Start, inspect, read, and stop bounded background command monitors."), "screen_intelligence" => Some("Screen capture, permissions, and accessibility automation."), "security" => Some("Security policy and autonomy guardrail metadata."), diff --git a/src/openhuman/config/schema/mod.rs b/src/openhuman/config/schema/mod.rs index f603397dd..6c5092e8b 100644 --- a/src/openhuman/config/schema/mod.rs +++ b/src/openhuman/config/schema/mod.rs @@ -9,6 +9,8 @@ pub use cloud_providers::{ generate_provider_id, is_slug_reserved, migrate_legacy_fields, AuthStyle, CloudProviderCreds, CloudProviderType, }; +pub mod subconscious; +pub use subconscious::{MedullaLocalConfig, SubconsciousConfig, SubconsciousEngine}; mod accessibility; mod agent; mod autocomplete; diff --git a/src/openhuman/config/schema/subconscious.rs b/src/openhuman/config/schema/subconscious.rs new file mode 100644 index 000000000..952c9625f --- /dev/null +++ b/src/openhuman/config/schema/subconscious.rs @@ -0,0 +1,270 @@ +//! Subconscious engine selection (plan §5.2 — the openhuman subconscious +//! replacement draft). +//! +//! `subconscious.engine` chooses which cognition drives the heartbeat tick's +//! observe/reflect/commit cycle: +//! +//! * `local` (default) — the existing local tinyagents graph. Unchanged. +//! * `medulla` — route each tick through a supervised local `medulla-serve` +//! child via `openhuman::medulla_local`. Draft; only wired when the crate is +//! built with the `medulla-local` feature. +//! +//! The default is `local`, so a config that omits the `[subconscious]` block — +//! every config today — behaves exactly as before. + +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +/// Which engine runs the subconscious reflect/commit cognition. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "lowercase")] +pub enum SubconsciousEngine { + /// The local tinyagents subconscious graph (unchanged default). + #[default] + Local, + /// Route ticks through a local `medulla-serve` child (draft). + Medulla, +} + +impl SubconsciousEngine { + /// Whether ticks should route through the local medulla brain. + pub fn is_medulla(self) -> bool { + matches!(self, Self::Medulla) + } +} + +/// Settings for the supervised local `medulla-serve` child. +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(default)] +pub struct MedullaLocalConfig { + /// Path to medulla-v1's built serve entry (`dist/serve/index.js`). Empty + /// falls back to the `OPENHUMAN_MEDULLA_SERVE_ENTRY` environment override; + /// with neither set the medulla engine reports its serve entry as + /// unconfigured (see [`Self::resolved_serve_entry`]). + #[serde(default)] + pub serve_entry: String, + /// Overall deadline, in seconds, for one serve request (from writing the + /// `req` to receiving its correlated `res`), regardless of interleaved + /// frame traffic. Distinct from the per-read idle timeout: a child that + /// keeps streaming frames without ever answering is bounded by this + /// ceiling. `0` falls back to the default + /// ([`DEFAULT_REQUEST_DEADLINE_SECS`], 300). + #[serde(default = "default_request_deadline_secs")] + pub request_deadline_secs: u64, +} + +impl Default for MedullaLocalConfig { + fn default() -> Self { + Self { + serve_entry: String::new(), + request_deadline_secs: DEFAULT_REQUEST_DEADLINE_SECS, + } + } +} + +/// Default overall per-request deadline for serve requests, in seconds. +pub const DEFAULT_REQUEST_DEADLINE_SECS: u64 = 300; + +/// Ceiling for [`MedullaLocalConfig::request_deadline_secs`]: 24 hours. Larger +/// values add no practical headroom and a near-`u64::MAX` duration would panic +/// in `Instant + Duration` arithmetic on the request path. +pub const MAX_REQUEST_DEADLINE_SECS: u64 = 24 * 60 * 60; + +fn default_request_deadline_secs() -> u64 { + DEFAULT_REQUEST_DEADLINE_SECS +} + +/// Environment override for the serve entry when `serve_entry` is left unset. +/// +/// There is no portable compiled-in default: medulla-v1's built `dist/serve` +/// lives outside this repo, so its location is deployment-specific. A developer +/// pointing at their umbrella checkout sets this env var (or the config field) +/// rather than relying on a machine-local path baked into the binary. +const SERVE_ENTRY_ENV: &str = "OPENHUMAN_MEDULLA_SERVE_ENTRY"; + +impl MedullaLocalConfig { + /// The resolved serve entry, or `None` when it is unconfigured. + /// + /// Precedence: the explicit `serve_entry` config value, then the + /// `OPENHUMAN_MEDULLA_SERVE_ENTRY` environment override. When neither is + /// set this returns `None` — the medulla engine then reports the serve + /// entry as unconfigured instead of pointing at a machine-local path. + pub fn resolved_serve_entry(&self) -> Option { + Self::resolve_entry(&self.serve_entry, std::env::var(SERVE_ENTRY_ENV).ok()) + } + + /// The effective overall per-request deadline. A configured `0` (an + /// explicit zero would disable the ceiling entirely — never wanted) falls + /// back to the default. + pub fn request_deadline(&self) -> std::time::Duration { + let secs = if self.request_deadline_secs == 0 { + tracing::warn!( + configured = self.request_deadline_secs, + effective_secs = DEFAULT_REQUEST_DEADLINE_SECS, + "medulla_local request_deadline_secs=0 would disable the request ceiling; using the default" + ); + DEFAULT_REQUEST_DEADLINE_SECS + } else if self.request_deadline_secs > MAX_REQUEST_DEADLINE_SECS { + tracing::warn!( + configured = self.request_deadline_secs, + effective_secs = MAX_REQUEST_DEADLINE_SECS, + "medulla_local request_deadline_secs exceeds the 24h ceiling; clamping" + ); + MAX_REQUEST_DEADLINE_SECS + } else { + self.request_deadline_secs + }; + std::time::Duration::from_secs(secs) + } + + /// Pure resolver shared by [`Self::resolved_serve_entry`], factored out so + /// the precedence rules are testable without mutating process env. + fn resolve_entry(configured: &str, env_override: Option) -> Option { + let trimmed = configured.trim(); + if !trimmed.is_empty() { + return Some(std::path::PathBuf::from(trimmed)); + } + let env = env_override?; + let env = env.trim(); + if env.is_empty() { + None + } else { + Some(std::path::PathBuf::from(env)) + } + } +} + +/// The `[subconscious]` config block. +#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)] +#[serde(default)] +pub struct SubconsciousConfig { + /// Which engine drives the subconscious tick. Default `local`. + #[serde(default)] + pub engine: SubconsciousEngine, + /// Local `medulla-serve` child settings (only used when `engine = medulla`). + #[serde(default)] + pub medulla_local: MedullaLocalConfig, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_engine_is_local() { + assert_eq!( + SubconsciousConfig::default().engine, + SubconsciousEngine::Local + ); + assert!(!SubconsciousConfig::default().engine.is_medulla()); + } + + #[test] + fn missing_block_deserializes_to_local() { + let config: SubconsciousConfig = serde_json::from_str("{}").unwrap(); + assert_eq!(config.engine, SubconsciousEngine::Local); + } + + #[test] + fn engine_serde_round_trip() { + assert_eq!( + serde_json::to_string(&SubconsciousEngine::Medulla).unwrap(), + r#""medulla""# + ); + assert_eq!( + serde_json::from_str::(r#""local""#).unwrap(), + SubconsciousEngine::Local + ); + } + + #[test] + fn explicit_serve_entry_wins_over_env() { + // An explicit config value is used verbatim, ignoring any env override. + assert_eq!( + MedullaLocalConfig::resolve_entry("/tmp/serve.js", Some("/env/serve.js".to_string())), + Some(std::path::PathBuf::from("/tmp/serve.js")) + ); + // Surrounding whitespace is trimmed. + assert_eq!( + MedullaLocalConfig::resolve_entry(" /tmp/serve.js ", None), + Some(std::path::PathBuf::from("/tmp/serve.js")) + ); + } + + #[test] + fn serve_entry_falls_back_to_env_override() { + assert_eq!( + MedullaLocalConfig::resolve_entry("", Some("/env/serve.js".to_string())), + Some(std::path::PathBuf::from("/env/serve.js")) + ); + assert_eq!( + MedullaLocalConfig::resolve_entry(" ", Some(" /env/serve.js ".to_string())), + Some(std::path::PathBuf::from("/env/serve.js")) + ); + } + + #[test] + fn serve_entry_unconfigured_resolves_to_none() { + // No machine-local default is baked in: unset config + unset (or blank) + // env resolves to None so the engine reports it unconfigured. + assert_eq!(MedullaLocalConfig::resolve_entry("", None), None); + assert_eq!( + MedullaLocalConfig::resolve_entry("", Some(" ".to_string())), + None + ); + } + + #[test] + fn request_deadline_defaults_and_zero_falls_back() { + // Both construction paths — `Default` and a config that omits the + // field — land on the same documented default. + assert_eq!( + MedullaLocalConfig::default().request_deadline_secs, + DEFAULT_REQUEST_DEADLINE_SECS + ); + let omitted: MedullaLocalConfig = serde_json::from_str("{}").unwrap(); + assert_eq!(omitted.request_deadline_secs, DEFAULT_REQUEST_DEADLINE_SECS); + + // A configured value is honoured… + let configured: MedullaLocalConfig = + serde_json::from_str(r#"{ "request_deadline_secs": 120 }"#).unwrap(); + assert_eq!( + configured.request_deadline(), + std::time::Duration::from_secs(120) + ); + + // …and an explicit zero (which would disable the ceiling) falls back + // to the default instead. + let zeroed: MedullaLocalConfig = + serde_json::from_str(r#"{ "request_deadline_secs": 0 }"#).unwrap(); + assert_eq!( + zeroed.request_deadline(), + std::time::Duration::from_secs(DEFAULT_REQUEST_DEADLINE_SECS) + ); + } + + #[test] + fn request_deadline_clamps_oversized_values_to_the_ceiling() { + // A near-u64::MAX duration would panic in `Instant + Duration` + // arithmetic on the request path; the accessor clamps instead. + let oversized: MedullaLocalConfig = serde_json::from_str(&format!( + r#"{{ "request_deadline_secs": {} }}"#, + u64::MAX - 1 + )) + .unwrap(); + assert_eq!( + oversized.request_deadline(), + std::time::Duration::from_secs(MAX_REQUEST_DEADLINE_SECS) + ); + // The boundary value itself is accepted un-clamped. + let at_max: MedullaLocalConfig = serde_json::from_str(&format!( + r#"{{ "request_deadline_secs": {} }}"#, + MAX_REQUEST_DEADLINE_SECS + )) + .unwrap(); + assert_eq!( + at_max.request_deadline(), + std::time::Duration::from_secs(MAX_REQUEST_DEADLINE_SECS) + ); + } +} diff --git a/src/openhuman/config/schema/types.rs b/src/openhuman/config/schema/types.rs index ce7acea93..833b6253d 100644 --- a/src/openhuman/config/schema/types.rs +++ b/src/openhuman/config/schema/types.rs @@ -230,6 +230,12 @@ pub struct Config { #[serde(default)] pub heartbeat: HeartbeatConfig, + /// Subconscious engine selection (local tinyagents graph vs. local + /// medulla-serve child). Default `local` — omitting this block preserves + /// the historical behavior exactly. + #[serde(default)] + pub subconscious: crate::openhuman::config::schema::SubconsciousConfig, + #[serde(default)] pub cron: CronConfig, @@ -764,6 +770,7 @@ impl Default for Config { model_routes: Vec::new(), embedding_routes: Vec::new(), heartbeat: HeartbeatConfig::default(), + subconscious: crate::openhuman::config::schema::SubconsciousConfig::default(), cron: CronConfig::default(), task_sources: TaskSourcesConfig::default(), channels_config: ChannelsConfig::default(), diff --git a/src/openhuman/medulla_local/host_ports.rs b/src/openhuman/medulla_local/host_ports.rs new file mode 100644 index 000000000..51c81edb0 --- /dev/null +++ b/src/openhuman/medulla_local/host_ports.rs @@ -0,0 +1,221 @@ +//! The concrete openhuman implementation of the serve [`HostPorts`] surface +//! (plan §3.2 — "the substance of the flavor"). +//! +//! * `inference` routes onto the existing per-role provider factory: the serve +//! tier (`orchestrator`/`reasoning`/`compress`) maps to an openhuman model +//! role via [`role_for_model_tier`]/[`provider_for_role`], the same routing +//! BYOK / local / managed-backend all already flow through. +//! * `tools` exposes a SMALL curated allowlist of **read-only** tools drawn +//! from the full local surface (`runtime_node::ops::build_runtime_tools`, +//! itself `tools::ops::all_tools_with_runtime`). Every result passes through +//! unchanged; write/execute tools are refused for this draft. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::{json, Value}; +use tracing::{debug, warn}; + +use super::ports::{HostPorts, PortError}; +use super::types::{InferenceCall, InferenceResult, ToolSpec, Usage}; +use crate::openhuman::config::Config; +use crate::openhuman::inference::provider::{ + create_chat_model_with_model_id, provider_for_role, role_for_model_tier, +}; +use crate::openhuman::tools::PermissionLevel; +use tinyagents::harness::message::Message; +use tinyagents::harness::model::ModelRequest; + +/// The curated read-only tool names serve may invoke this draft. The port also +/// enforces `PermissionLevel::ReadOnly` at call time, so an allowlisted name +/// that ever gained a write action is still refused. +const CURATED_READ_ONLY_TOOLS: &[&str] = &[ + "file_read", + "read_diff", + "list", + "glob", + "grep", + "web_fetch", +]; + +/// Adapter holding the config snapshot the port routing needs. +pub struct OpenhumanHostPorts { + config: Arc, +} + +impl OpenhumanHostPorts { + pub fn new(config: Arc) -> Self { + Self { config } + } + + /// Map a serve inference tier onto an openhuman model role. + /// + /// `orchestrator` → the agentic (tool-using) role, `reasoning` → the + /// reasoning role, `compress` → the summarization role; anything else + /// falls back through `role_for_model_tier`'s own default (chat). + fn role_for_tier(tier: &str) -> &'static str { + match tier { + "orchestrator" => "agentic", + "reasoning" => "reasoning", + "compress" => "summarization", + other => role_for_model_tier(other), + } + } + + /// The curated read-only tools drawn from the full local surface: the + /// intersection of [`CURATED_READ_ONLY_TOOLS`] with the built runtime tools, + /// filtered again on `PermissionLevel::ReadOnly` so an allowlisted name that + /// ever gained a write action is dropped. Both the `hello` advertisement + /// ([`Self::tool_specs`]) and the `tools.invoke` handler + /// ([`Self::invoke_tool`]) resolve tools through this one path, keeping the + /// advertised set and the invocable set identical. + fn curated_read_only_tools( + &self, + ) -> Result>, String> { + let tools = crate::openhuman::runtime_node::ops::build_runtime_tools(&self.config)?; + Ok(tools + .into_iter() + .filter(|tool| CURATED_READ_ONLY_TOOLS.contains(&tool.name())) + .filter(|tool| tool.permission_level() == PermissionLevel::ReadOnly) + .collect()) + } +} + +#[async_trait] +impl HostPorts for OpenhumanHostPorts { + fn tool_specs(&self) -> Vec { + let tools = match self.curated_read_only_tools() { + Ok(tools) => tools, + Err(error) => { + warn!( + "[medulla_local] building tool surface for hello advertisement failed: {error}" + ); + return Vec::new(); + } + }; + let specs: Vec = tools + .iter() + .map(|tool| ToolSpec { + name: tool.name().to_string(), + description: tool.description().to_string(), + parameters: tool.parameters_schema(), + }) + .collect(); + debug!( + count = specs.len(), + "[medulla_local] advertising curated read-only tools in hello" + ); + specs + } + + async fn invoke_inference(&self, call: InferenceCall) -> Result { + let role = Self::role_for_tier(&call.tier); + debug!( + tier = %call.tier, + role, + provider = %provider_for_role(role, &self.config), + messages = call.messages.len(), + "[medulla_local] inference callback routing" + ); + + let (model, model_id) = + create_chat_model_with_model_id(role, &self.config, 0.2).map_err(|error| { + PortError::internal(format!("model init for role `{role}`: {error}")) + })?; + + let messages: Vec = call + .messages + .iter() + .map(|message| match message.role.as_str() { + "system" => Message::system(message.content.clone()), + "assistant" => Message::assistant(message.content.clone()), + _ => Message::user(message.content.clone()), + }) + .collect(); + + let response = model + .invoke(&(), ModelRequest::new(messages).with_temperature(0.2)) + .await + .map_err(|error| PortError::internal(format!("inference invoke failed: {error}")))?; + + Ok(InferenceResult { + content: response.text(), + reasoning_content: None, + model: model_id, + tool_calls: Vec::new(), + usage: Usage::default(), + }) + } + + async fn invoke_tool(&self, name: &str, args: Value) -> Result { + if !CURATED_READ_ONLY_TOOLS.contains(&name) { + return Err(PortError::port_unavailable(format!( + "tool `{name}` is not on the curated read-only allowlist (draft)" + ))); + } + + // Resolve through the same curated (allowlisted + read-only) surface the + // `hello` advertisement is built from, so a tool the model was told about + // is exactly a tool this handler will run. The `curated_read_only_tools` + // filter already drops anything not read-only; the explicit re-check + // below is defence in depth against future drift. + let tools = self + .curated_read_only_tools() + .map_err(|error| PortError::internal(format!("building tool surface: {error}")))?; + let tool = tools + .into_iter() + .find(|tool| tool.name() == name) + .ok_or_else(|| { + PortError::port_unavailable(format!("tool `{name}` not present in local surface")) + })?; + + // Defence in depth: refuse anything that is not read-only even if the + // allowlist and the tool registry ever drift. + if tool.permission_level() != PermissionLevel::ReadOnly { + warn!( + name, + "[medulla_local] refusing non-read-only tool over medulla port" + ); + return Err(PortError::port_unavailable(format!( + "tool `{name}` is not read-only; refused for this draft" + ))); + } + + let result = tool + .execute(args) + .await + .map_err(|error| PortError::internal(format!("tool `{name}` failed: {error}")))?; + + // Pass the ToolResult through unchanged, adapting only the error-flag + // key to the medulla wire shape (`isError`, §5.2). + Ok(json!({ + "content": serde_json::to_value(&result.content).unwrap_or_else(|_| json!([])), + "isError": result.is_error, + })) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tier_maps_onto_expected_roles() { + assert_eq!(OpenhumanHostPorts::role_for_tier("orchestrator"), "agentic"); + assert_eq!(OpenhumanHostPorts::role_for_tier("reasoning"), "reasoning"); + assert_eq!( + OpenhumanHostPorts::role_for_tier("compress"), + "summarization" + ); + // Unknown tiers fall back through the factory's own default. + assert_eq!(OpenhumanHostPorts::role_for_tier("mystery"), "chat"); + } + + #[test] + fn curated_allowlist_is_read_only_by_name() { + // Sanity: the curated set never contains an obviously-mutating tool. + for forbidden in ["file_write", "edit", "apply_patch", "git_operations"] { + assert!(!CURATED_READ_ONLY_TOOLS.contains(&forbidden)); + } + } +} diff --git a/src/openhuman/medulla_local/mod.rs b/src/openhuman/medulla_local/mod.rs new file mode 100644 index 000000000..8239682b9 --- /dev/null +++ b/src/openhuman/medulla_local/mod.rs @@ -0,0 +1,48 @@ +//! Local Medulla brain — supervises a `medulla-serve` Node child and speaks the +//! medulla-serve NDJSON protocol, v1, as the host. +//! +//! This is Flavor A of the Medulla flavors plan (§3.1–§3.2): openhuman runs +//! medulla-v1's agent-harness facade in a supervised child and answers its port +//! callbacks against the local capability surface (inference routing, curated +//! read-only tools). It also backs the subconscious-replacement draft (§5.2): +//! with `subconscious.engine = "medulla"`, each heartbeat tick routes its +//! observe/reflect/commit cycle through one `instruct` instead of the local +//! tinyagents graph. The default (`local`) is untouched. +//! +//! Module shape (canonical `mod/types/…/ops/schemas`): +//! * [`protocol`] — wire frame envelopes (`ready`/`req`/`res`/`call`/`ret`/`event`). +//! * [`types`] — handshake / instruct / status / inference domain types. +//! * [`ports`] — the [`HostPorts`] seam serve's reverse-RPC drives. +//! * [`server`] — the supervisor: spawn, handshake, id-correlated NDJSON, +//! restart-and-retry-once, stderr drain (mirrors +//! `runtime_python_server/server.rs`). Unix-only — the +//! transport is a unix domain socket; non-unix targets get +//! a stub that reports the platform as unsupported. +//! * [`host_ports`] — the concrete openhuman [`HostPorts`] (inference + tools). +//! * [`ops`] — RPC handlers + the subconscious tick entrypoint. +//! * [`schemas`] — the `medulla_local` controller schemas. + +pub mod host_ports; +pub mod ops; +pub mod ports; +pub mod protocol; +pub mod schemas; +#[cfg(unix)] +pub mod server; +/// The serve transport is a unix domain socket, which `tokio::net` only +/// provides on unix targets. On other targets (Windows) the feature still +/// compiles: this stub keeps the same supervisor surface but every entry +/// point reports a typed unsupported-platform error. A portable transport +/// (e.g. stdio) can lift this later without touching the callers. +#[cfg(not(unix))] +#[path = "server_unsupported.rs"] +pub mod server; +pub mod types; + +pub use ports::{HostPorts, PortError}; +pub use schemas::{ + all_controller_schemas as all_medulla_local_controller_schemas, + all_registered_controllers as all_medulla_local_registered_controllers, +}; +pub use server::{ensure_started, MedullaSupervisor}; +pub use types::{InferenceCall, InferenceResult, InstructReceipt, MedullaLocalStatus}; diff --git a/src/openhuman/medulla_local/ops.rs b/src/openhuman/medulla_local/ops.rs new file mode 100644 index 000000000..b3b7e9583 --- /dev/null +++ b/src/openhuman/medulla_local/ops.rs @@ -0,0 +1,86 @@ +//! Business logic + RPC handlers for the `medulla_local` namespace. +//! +//! Thin surface over [`super::server`]: `status` reports the supervised serve +//! child's handshake state; `instruct` enqueues one instruction and returns the +//! synchronous receipt (§4.1). The subconscious tick path (§5.2) calls +//! [`instruct_tick`] directly rather than over RPC. + +use serde::Deserialize; +use serde_json::{json, Value}; +use tracing::warn; + +use super::server; +use super::types::{InstructReceipt, MedullaLocalStatus}; +use crate::openhuman::config::rpc as config_rpc; +use crate::openhuman::config::Config; +use crate::rpc::RpcOutcome; + +#[derive(Debug, Deserialize)] +pub(crate) struct InstructParams { + pub(crate) message: String, + #[serde(default)] + pub(crate) meta: Value, +} + +/// Enqueue one instruction against the supervised serve child. +pub async fn instruct_tick( + config: &Config, + message: &str, + meta: Value, +) -> anyhow::Result { + let supervisor = server::ensure_started(config).await?; + supervisor.instruct(message, meta).await +} + +/// Snapshot the supervised serve child's status without forcing a spawn on the +/// failure path (a failed `ensure_started` still yields a well-formed +/// unavailable status). +pub async fn status(config: &Config) -> MedullaLocalStatus { + match server::ensure_started(config).await { + Ok(supervisor) => supervisor.snapshot().await, + Err(error) => { + warn!("[medulla_local] medulla_local.status: serve unavailable: {error:#}"); + MedullaLocalStatus { + enabled: true, + running: false, + serve_version: None, + session_id: None, + ports: Vec::new(), + message: Some(error.to_string()), + } + } + } +} + +pub(crate) async fn status_handler() -> Result { + let config = config_rpc::load_config_with_timeout().await?; + let status = status(&config).await; + let payload = json!(status); + let log = vec![format!( + "medulla_local.status: running={} ports={}", + status.running, + status.ports.len() + )]; + RpcOutcome::new(payload, log).into_cli_compatible_json() +} + +pub(crate) async fn instruct_handler(params: InstructParams) -> Result { + let config = config_rpc::load_config_with_timeout().await?; + let meta = if params.meta.is_null() { + json!({ "origin": "rpc" }) + } else { + params.meta + }; + let receipt = instruct_tick(&config, ¶ms.message, meta) + .await + .map_err(|error| { + warn!("[medulla_local] medulla_local.instruct failed: {error:#}"); + format!("{error:#}") + })?; + let payload = json!(receipt); + let log = vec![format!( + "medulla_local.instruct: instruction_id={} cycle_id={}", + receipt.instruction_id, receipt.cycle_id + )]; + RpcOutcome::new(payload, log).into_cli_compatible_json() +} diff --git a/src/openhuman/medulla_local/ports.rs b/src/openhuman/medulla_local/ports.rs new file mode 100644 index 000000000..8cc26a371 --- /dev/null +++ b/src/openhuman/medulla_local/ports.rs @@ -0,0 +1,82 @@ +//! The host-side port surface a supervised serve child drives via reverse RPC. +//! +//! serve issues `call` frames for the ports it needs (§5); the supervisor +//! demultiplexes them and dispatches to a [`HostPorts`] implementation, then +//! writes the `ret`. This trait is the clean seam between the transport +//! (`server.rs`) and the openhuman capability surface (`host_ports.rs`): the +//! supervisor and its tests are generic over `Arc`. +//! +//! For this DRAFT milestone only two ports are answered — `inference` and +//! `tools` — matching plan §3.2's "the substance of the flavor". Every other +//! port serve might request is refused as `port_unavailable`, handled centrally +//! in `server.rs` so an implementer cannot forget to reject one. + +use async_trait::async_trait; +use serde_json::Value; + +use super::protocol::ServeError; +use super::types::{error_codes, InferenceCall, InferenceResult, ToolSpec}; + +/// A failure answering a port callback. Serialized into a `ret` error +/// envelope (§8) with one of the reserved [`error_codes`]. +#[derive(Debug, Clone)] +pub struct PortError { + pub code: &'static str, + pub message: String, +} + +impl PortError { + pub fn new(code: &'static str, message: impl Into) -> Self { + Self { + code, + message: message.into(), + } + } + + /// The host does not offer this port (or the tool/method is not on the + /// curated allowlist for this draft). + pub fn port_unavailable(message: impl Into) -> Self { + Self::new(error_codes::PORT_UNAVAILABLE, message) + } + + /// The port is offered but the specific method is not implemented. + pub fn unsupported_method(message: impl Into) -> Self { + Self::new(error_codes::UNSUPPORTED_METHOD, message) + } + + /// A host-side failure while servicing an otherwise-valid callback. + pub fn internal(message: impl Into) -> Self { + Self::new(error_codes::INTERNAL, message) + } + + pub fn to_serve_error(&self) -> ServeError { + ServeError::new(self.code, self.message.clone()) + } +} + +/// The subset of medulla-v1 ports this host answers. +/// +/// The wire ToolResult (`{content, isError}`, §5.2) is returned as a raw +/// [`Value`] so the concrete adapter can pass a tool's result through +/// **unchanged** (the draft's contract) without a lossy re-typing. +#[async_trait] +pub trait HostPorts: Send + Sync { + /// The host tool specs advertised in the `hello` handshake (§3), which serve + /// binds into the serve-side module registry so the model can see and emit + /// `tools.invoke` calls for them. + /// + /// This MUST correspond to the set [`Self::invoke_tool`] will actually + /// answer — an advertised tool the port later refuses is a phantom the model + /// wastes a turn on, and a tool the port answers but never advertises can + /// never be invoked at all. Both derive from the same curated allowlist in + /// the concrete adapter to keep them in lock-step. + fn tool_specs(&self) -> Vec; + + /// `inference.invoke` (§5.1) — route the call's tier onto the host's + /// per-role model routing and return an `InferenceResult`. + async fn invoke_inference(&self, call: InferenceCall) -> Result; + + /// `tools.invoke` (§5.2) — execute a host tool by name against the curated + /// read-only allowlist and return its wire `ToolResult` unchanged. + async fn invoke_tool(&self, name: &str, args: Value) -> Result; +} diff --git a/src/openhuman/medulla_local/protocol.rs b/src/openhuman/medulla_local/protocol.rs new file mode 100644 index 000000000..9b2848535 --- /dev/null +++ b/src/openhuman/medulla_local/protocol.rs @@ -0,0 +1,183 @@ +//! Wire-level frame envelopes for the `medulla-serve` NDJSON protocol. +//! +//! The contract follows the medulla-serve NDJSON protocol, v1 (plan §2.2). +//! `serve` is the Node child wrapping medulla-v1's agent-harness facade; +//! `host` is this Rust supervisor. Six frame kinds cross three flows: +//! +//! | `t` | direction | flow | +//! | ------- | ----------- | --------------------------------------- | +//! | `ready` | serve→host | handshake banner (unprompted, first) | +//! | `req` | host→serve | request | +//! | `res` | serve→host | response (correlated by `id`) | +//! | `call` | serve→host | port callback (reverse RPC into host) | +//! | `ret` | host→serve | port-callback return (correlated `id`) | +//! | `emit` | host→serve | streaming tap for an in-flight `call` | +//! | `event` | serve→host | unsolicited event stream (no `id`) | +//! +//! Frames are read as untyped [`serde_json::Value`] and dispatched on the `t` +//! discriminant (see [`FrameKind`]) so an unknown/extra frame is skipped and +//! logged rather than treated as fatal — mirroring +//! `runtime_python_server/server.rs`'s "unparseable response skipped". + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +/// Wire version the host speaks. The `ready` banner MUST carry the same value +/// or the host bails (§3, §7 handshake mismatch). +pub const PROTOCOL_VERSION: u32 = 1; + +/// The `t` discriminant on an inbound (serve→host) frame. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FrameKind { + Ready, + Res, + Call, + Event, + /// Any `t` value the host does not expect inbound (`req`/`ret`/`emit` are + /// host→serve). Skipped and logged. + Unknown, +} + +impl FrameKind { + /// Classify a decoded frame by its `t` field. + pub fn of(frame: &Value) -> Self { + match frame.get("t").and_then(Value::as_str) { + Some("ready") => Self::Ready, + Some("res") => Self::Res, + Some("call") => Self::Call, + Some("event") => Self::Event, + _ => Self::Unknown, + } + } +} + +/// The unprompted first line serve writes on connect (§3). `error` non-null ⇒ +/// startup failed and the host treats the child as unavailable. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ReadyLine { + #[serde(default)] + pub protocol: u32, + #[serde(default)] + pub serve: Option, + #[serde(default, rename = "sessionId")] + pub session_id: Option, + #[serde(default)] + pub capabilities: Vec, + #[serde(default)] + pub error: Option, +} + +/// The always-`{code,message}` error envelope shared by `res` and `ret` +/// (§8). Mirrors the `PythonServerError` shape. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ServeError { + pub code: String, + pub message: String, +} + +impl ServeError { + pub fn new(code: impl Into, message: impl Into) -> Self { + Self { + code: code.into(), + message: message.into(), + } + } +} + +/// A decoded `res` frame (serve→host response, correlated by `id`). +#[derive(Debug, Clone, Deserialize)] +pub struct ResFrame { + pub id: Option, + #[serde(default)] + pub ok: bool, + #[serde(default)] + pub result: Option, + #[serde(default)] + pub error: Option, +} + +/// A decoded `call` frame (serve→host reverse-RPC into a host port). +#[derive(Debug, Clone, Deserialize)] +pub struct CallFrame { + pub id: String, + pub port: String, + pub method: String, + #[serde(default)] + pub params: Value, +} + +/// A decoded `event` frame (serve→host unsolicited stream, §6). +#[derive(Debug, Clone, Deserialize)] +pub struct EventFrame { + #[serde(default)] + pub seq: u64, + #[serde(default)] + pub at: u64, + #[serde(default)] + pub event: Value, +} + +/// Build a host→serve `req` frame line body (§4). +pub fn req_frame(id: &str, op: &str, params: Value) -> Value { + json!({ "t": "req", "id": id, "op": op, "params": params }) +} + +/// Build a host→serve `ret` frame answering a port `call` with success (§5). +pub fn ret_ok(id: &str, result: Value) -> Value { + json!({ "t": "ret", "id": id, "ok": true, "result": result }) +} + +/// Build a host→serve `ret` frame answering a port `call` with failure (§5, §8). +pub fn ret_err(id: &str, error: &ServeError) -> Value { + json!({ "t": "ret", "id": id, "ok": false, "error": { "code": error.code, "message": error.message } }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ready_line_parses_session_and_capabilities() { + let ready: ReadyLine = serde_json::from_str( + r#"{"t":"ready","protocol":1,"serve":"3.12.0","sessionId":"agent","capabilities":["inference","tools"],"error":null}"#, + ) + .unwrap(); + assert_eq!(ready.protocol, PROTOCOL_VERSION); + assert_eq!(ready.session_id.as_deref(), Some("agent")); + assert_eq!(ready.capabilities, vec!["inference", "tools"]); + assert!(ready.error.is_none()); + } + + #[test] + fn frame_kind_classifies_by_t() { + assert_eq!( + FrameKind::of(&json!({"t":"call","id":"c1"})), + FrameKind::Call + ); + assert_eq!(FrameKind::of(&json!({"t":"event"})), FrameKind::Event); + assert_eq!(FrameKind::of(&json!({"t":"nope"})), FrameKind::Unknown); + assert_eq!(FrameKind::of(&json!({})), FrameKind::Unknown); + } + + #[test] + fn res_error_envelope_parses() { + let res: ResFrame = serde_json::from_str( + r#"{"t":"res","id":"7","ok":false,"error":{"code":"bad_request","message":"missing message"}}"#, + ) + .unwrap(); + assert!(!res.ok); + assert_eq!(res.id.as_deref(), Some("7")); + assert_eq!(res.error.unwrap().code, "bad_request"); + } + + #[test] + fn ret_frames_carry_correct_discriminant() { + let ok = ret_ok("c1", json!({"content": []})); + assert_eq!(ok["t"], "ret"); + assert_eq!(ok["ok"], true); + let err = ret_err("c2", &ServeError::new("port_unavailable", "no memory port")); + assert_eq!(err["t"], "ret"); + assert_eq!(err["ok"], false); + assert_eq!(err["error"]["code"], "port_unavailable"); + } +} diff --git a/src/openhuman/medulla_local/schemas.rs b/src/openhuman/medulla_local/schemas.rs new file mode 100644 index 000000000..df11343f5 --- /dev/null +++ b/src/openhuman/medulla_local/schemas.rs @@ -0,0 +1,132 @@ +//! Controller schemas for the `medulla_local` RPC namespace. +//! +//! Registered in `src/core/all.rs` under `DomainGroup::Agent` behind the +//! `medulla-local` feature. Two methods this draft: `status` and `instruct`. + +use serde_json::{Map, Value}; +use tracing::warn; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::openhuman::medulla_local::ops::{instruct_handler, status_handler, InstructParams}; + +pub fn all_controller_schemas() -> Vec { + vec![ + schemas("medulla_local_status"), + schemas("medulla_local_instruct"), + ] +} + +pub fn all_registered_controllers() -> Vec { + vec![ + RegisteredController { + schema: schemas("medulla_local_status"), + handler: handle_status, + }, + RegisteredController { + schema: schemas("medulla_local_instruct"), + handler: handle_instruct, + }, + ] +} + +pub fn schemas(function: &str) -> ControllerSchema { + match function { + "medulla_local_status" => ControllerSchema { + namespace: "medulla_local", + function: "status", + description: "Status of the supervised local medulla-serve child: whether it is connected, its serve version, session id, and negotiated port set.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "status", + ty: TypeSchema::Json, + comment: "MedullaLocalStatus: {enabled, running, serve_version?, session_id?, ports, message?}.", + required: true, + }], + }, + "medulla_local_instruct" => ControllerSchema { + namespace: "medulla_local", + function: "instruct", + description: "Enqueue one instruction against the local medulla-serve harness. Returns the synchronous receipt; the cycle runs async and is observed via the event stream.", + inputs: vec![ + FieldSchema { + name: "message", + ty: TypeSchema::String, + comment: "The instruction text for the harness cycle.", + required: true, + }, + FieldSchema { + name: "meta", + ty: TypeSchema::Option(Box::new(TypeSchema::Json)), + comment: "Optional instruction metadata (e.g. {origin: 'wake'}).", + required: false, + }, + ], + outputs: vec![ + FieldSchema { + name: "instructionId", + ty: TypeSchema::String, + comment: "Id of the enqueued instruction.", + required: true, + }, + FieldSchema { + name: "cycleId", + ty: TypeSchema::String, + comment: "Id of the cycle the instruction will run in.", + required: true, + }, + ], + }, + _ => ControllerSchema { + namespace: "medulla_local", + function: "unknown", + description: "Unknown medulla_local controller.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "error", + ty: TypeSchema::String, + comment: "Lookup error details.", + required: true, + }], + }, + } +} + +fn handle_status(_params: Map) -> ControllerFuture { + Box::pin(async move { status_handler().await }) +} + +fn handle_instruct(params: Map) -> ControllerFuture { + Box::pin(async move { + let params: InstructParams = + serde_json::from_value(Value::Object(params)).map_err(|error| { + warn!("[medulla_local] medulla_local.instruct rejected malformed params: {error}"); + error.to_string() + })?; + instruct_handler(params).await + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn catalog_lists_status_and_instruct() { + let schemas = all_controller_schemas(); + assert_eq!(schemas.len(), 2); + let names: Vec<&str> = schemas.iter().map(|schema| schema.function).collect(); + assert!(names.contains(&"status")); + assert!(names.contains(&"instruct")); + } + + #[test] + fn instruct_schema_requires_message() { + let schema = schemas("medulla_local_instruct"); + assert_eq!(schema.namespace, "medulla_local"); + assert!(schema + .inputs + .iter() + .any(|field| field.name == "message" && field.required)); + } +} diff --git a/src/openhuman/medulla_local/server.rs b/src/openhuman/medulla_local/server.rs new file mode 100644 index 000000000..89cf4acf7 --- /dev/null +++ b/src/openhuman/medulla_local/server.rs @@ -0,0 +1,1018 @@ +//! Supervisor for a local `medulla-serve` Node child. +//! +//! Mirrors `runtime_python_server/server.rs` conventions verbatim: a versioned +//! handshake, id-correlated NDJSON, per-request timeout, restart-and-retry-once +//! on transport failure, start-failure backoff, and a drained stderr pipe. The +//! transport is a unix domain socket (§1 of the serve protocol spec): serve +//! listens on `serve.sock` under the workspace state dir and the host connects. +//! +//! The wire is demultiplexed inline (single connection, single reader) exactly +//! like the Python server's response loop — extended to also service the +//! serve→host `call` port callbacks and fold the `event` stream, since medulla +//! runs a reverse-RPC plane the Python backend does not. Port dispatch goes +//! through the [`HostPorts`] seam so the transport stays testable without a +//! live Node process. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use anyhow::{bail, Context, Result}; +use async_trait::async_trait; +use serde::de::DeserializeOwned; +use serde_json::{json, Value}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines}; +use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::net::UnixStream; +use tokio::process::{Child, ChildStderr}; +use tokio::sync::Mutex; +use tracing::{debug, info, warn}; + +use super::host_ports::OpenhumanHostPorts; +use super::ports::HostPorts; +use super::protocol::{ + ret_err, ret_ok, CallFrame, EventFrame, FrameKind, ReadyLine, ResFrame, ServeError, + PROTOCOL_VERSION, +}; +use super::types::{ + error_codes, HarnessStatus, HelloParams, HelloResult, InferenceCall, InstructReceipt, + MedullaLocalStatus, +}; +use crate::openhuman::config::Config; + +/// Ceiling for the `ready` handshake (§7). Matches the Python server. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30); +/// Idle ceiling for a single frame read while awaiting a `res`. Resets on +/// every inbound frame, so it only catches a *silent* child; a child that +/// keeps streaming frames is bounded by the overall per-request deadline +/// carried on [`Connection`] (`subconscious.medulla_local.request_deadline_secs`). +const REQUEST_TIMEOUT: Duration = Duration::from_secs(60); +/// Backoff after a failed spawn before the host retries (§7). +const START_FAILURE_BACKOFF: Duration = Duration::from_secs(300); +/// Poll interval while waiting for the child to create its listening socket. +const SOCKET_CONNECT_POLL: Duration = Duration::from_millis(50); + +/// Why a supervised request failed — the axis the restart-and-retry policy +/// pivots on. Only [`RequestError::Transport`] (the established connection +/// broke mid-request: process death, closed socket, IO failure, read timeout) +/// is retryable, because respawning the child yields a fresh transport. The +/// other variants are deterministic protocol- or application-level failures: +/// killing a healthy child and replaying the request would hit the same +/// failure again and can lose session state or duplicate side effects, so +/// they fail fast. +#[derive(Debug, thiserror::Error)] +pub enum RequestError { + /// Spawning/connecting/handshaking a fresh child failed (including a + /// protocol-version mismatch in the handshake). The connect path already + /// has its own retry window, so this is terminal for the request. + #[error("{0:#}")] + Connect(anyhow::Error), + /// The established connection broke mid-request. Retryable once. + #[error("{0:#}")] + Transport(anyhow::Error), + /// serve answered `ok=false`: an application-level rejection over a + /// healthy connection (e.g. `bad_request`, `not_ready`). Not retryable. + #[error("medulla serve `{op}` failed: {code}: {message}")] + Serve { + op: String, + code: String, + message: String, + }, + /// The transport delivered a frame the host cannot use (an undecodable + /// result payload). Not retryable — a restart replays the same exchange. + #[error("{0:#}")] + Protocol(anyhow::Error), + /// The established connection broke mid-request on a **non-idempotent** + /// op: the request may or may not have reached serve before the break, so + /// replaying it could duplicate the side effect (e.g. enqueue the same + /// instruction twice). Never retried; the caller reconciles the true + /// outcome out of band (for `instruct`: `harness_status` shows the queue + /// and active instruction). + #[error( + "medulla serve `{op}` connection broke mid-request; the operation may or may not have \ + been applied — reconcile via `status` before re-issuing: {source:#}" + )] + MaybeApplied { + op: String, + #[source] + source: anyhow::Error, + }, +} + +impl RequestError { + /// Whether killing and respawning the child could plausibly change the + /// outcome of replaying this request. + pub fn is_retryable(&self) -> bool { + matches!(self, Self::Transport(_)) + } +} + +/// The typed overall-deadline failure: classified as +/// [`RequestError::Transport`] so it rides the existing policy axis — an +/// idempotent op is restarted-and-retried once, while a non-idempotent op +/// (e.g. `instruct`) surfaces as [`RequestError::MaybeApplied`] because the +/// request reached serve but its outcome was never observed. +fn deadline_exceeded(op: &str, deadline: Duration) -> RequestError { + RequestError::Transport(anyhow::anyhow!( + "medulla serve `{op}` exceeded the overall request deadline ({deadline:?}) while awaiting \ + its correlated response (interleaved frames or port callbacks kept the connection busy \ + but no `res` was seen)" + )) +} + +/// Whether replaying `op` after a mid-request transport break is safe. +/// +/// `instruct` is NOT idempotent: the wire op carries only `message`/`meta` — +/// there is no client-supplied instruction id the host could reuse to dedupe a +/// replay (`instructionId` is assigned serve-side and only returned in the +/// receipt, §4.1). If the first attempt reached serve before the connection +/// broke, a retry would enqueue the instruction twice. So a transport failure +/// on `instruct` fails fast as [`RequestError::MaybeApplied`] instead of +/// retrying; the caller reconciles via `status` (`HarnessStatus` exposes the +/// queue depth and active instruction). Read-only ops (`status`) and the +/// replayed handshake stay on the restart-and-retry-once path. +fn op_is_idempotent(op: &str) -> bool { + !matches!(op, "instruct") +} + +/// A live, handshaken connection to one serve child. +/// +/// Owns the split unix-socket halves, the next-id counter (§2), and the +/// [`HostPorts`] the reverse-RPC plane dispatches to. Dropping it kills the +/// child (the [`Child`] is spawned with `kill_on_drop`). +pub struct Connection { + writer: OwnedWriteHalf, + reader: Lines>, + next_id: u64, + ports: Arc, + ready: ReadyLine, + hello: HelloResult, + /// Kept alive for the connection's lifetime and probed for liveness + /// before the cached connection is trusted (see [`Self::child_has_exited`]); + /// `None` in tests that connect to a mock listener instead of spawning + /// Node. + child: Option, + last_event_seq: Option, + /// Overall wall-clock ceiling for one request (write → correlated `res`). + /// Unlike [`REQUEST_TIMEOUT`], interleaved `call`/`event` frames do NOT + /// reset it: a request either completes within this window or fails with + /// the typed transport-timeout error. + request_deadline: Duration, +} + +impl Connection { + /// Read the `ready` banner, negotiate `hello`, and return a live + /// connection. `child` is retained so the caller can tie the process + /// lifetime to the connection. `request_deadline` is the overall + /// per-request ceiling applied to every request on this connection + /// (including the `hello` negotiated here). + pub async fn establish( + stream: UnixStream, + ports: Arc, + hello: HelloParams, + child: Option, + request_deadline: Duration, + ) -> Result { + let (read_half, write_half) = stream.into_split(); + let mut reader = BufReader::new(read_half).lines(); + + let ready_line = match tokio::time::timeout(HANDSHAKE_TIMEOUT, reader.next_line()).await { + Ok(Ok(Some(line))) => line, + Ok(Ok(None)) => bail!("medulla serve closed before the ready handshake"), + Ok(Err(error)) => return Err(error).context("reading medulla serve ready line"), + Err(_) => bail!("medulla serve ready handshake timed out"), + }; + let ready: ReadyLine = serde_json::from_str(&ready_line) + .with_context(|| format!("parsing medulla serve ready line: {ready_line}"))?; + if let Some(error) = &ready.error { + bail!("medulla serve reported startup failure: {error}"); + } + if ready.protocol != PROTOCOL_VERSION { + bail!( + "medulla serve protocol mismatch: expected {PROTOCOL_VERSION}, got {}", + ready.protocol + ); + } + info!( + serve = ready.serve.as_deref().unwrap_or(""), + session = ready.session_id.as_deref().unwrap_or(""), + capabilities = ?ready.capabilities, + "[medulla_local] serve ready" + ); + + let mut conn = Self { + writer: write_half, + reader, + next_id: 0, + ports, + ready, + hello: HelloResult::default(), + child, + last_event_seq: None, + request_deadline, + }; + + let hello_value = serde_json::to_value(&hello).context("encoding medulla hello params")?; + let negotiated: HelloResult = conn + .request("hello", hello_value) + .await + .context("medulla serve hello handshake failed")?; + info!( + ports = ?negotiated.ports, + "[medulla_local] hello negotiated active port set" + ); + conn.hello = negotiated; + Ok(conn) + } + + /// Typed request (§4): write a `req`, then drive the read loop — + /// servicing interleaved `call` port callbacks and folding `event`s — + /// until the correlated `res` arrives. The error is classified (see + /// [`RequestError`]) so the supervisor can restrict restart-and-retry to + /// transport failures. + pub async fn request( + &mut self, + op: &str, + params: Value, + ) -> Result { + let value = self.request_raw(op, params).await?; + serde_json::from_value(value) + .with_context(|| format!("decoding medulla serve `{op}` result")) + .map_err(RequestError::Protocol) + } + + async fn request_raw(&mut self, op: &str, params: Value) -> Result { + let id = self.next_id.to_string(); + self.next_id += 1; + debug!(id = %id, op, "[medulla_local] sending req"); + self.write_frame(&super::protocol::req_frame(&id, op, params)) + .await + .map_err(RequestError::Transport)?; + + // One wall-clock deadline bounds the whole await-`res` loop. + // Interleaved `call`/`event` frames keep resetting the per-read idle + // timeout below, so without this ceiling a child that streams frames + // forever while never answering the correlated `res` would keep the + // request (and the supervisor's connection lock) pending indefinitely. + let deadline = Instant::now() + self.request_deadline; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(deadline_exceeded(op, self.request_deadline)); + } + let line = match self.next_line(REQUEST_TIMEOUT.min(remaining)).await { + Ok(line) => line, + // A read that fails once the deadline has elapsed is reported + // as the overall-deadline trip, not a per-read idle timeout — + // the shortened read window above is how the deadline fires. + Err(_) if Instant::now() >= deadline => { + return Err(deadline_exceeded(op, self.request_deadline)); + } + Err(error) => return Err(RequestError::Transport(error)), + }; + let frame: Value = match serde_json::from_str(&line) { + Ok(frame) => frame, + Err(error) => { + warn!( + "[medulla_local] unparseable frame skipped: {error}; line_len={}", + line.len() + ); + continue; + } + }; + match FrameKind::of(&frame) { + FrameKind::Res => { + let res: ResFrame = match serde_json::from_value(frame) { + Ok(res) => res, + Err(error) => { + warn!("[medulla_local] malformed res skipped: {error}"); + continue; + } + }; + if res.id.as_deref() != Some(id.as_str()) { + debug!(want = %id, got = ?res.id, "[medulla_local] skipping res for other id"); + continue; + } + if !res.ok { + // An application-level rejection over a healthy + // connection — surfaced typed so the supervisor + // fails fast instead of killing the child. + let (code, message) = + res.error.map(|e| (e.code, e.message)).unwrap_or_else(|| { + ( + "unknown_error".to_string(), + "unknown medulla serve error".to_string(), + ) + }); + return Err(RequestError::Serve { + op: op.to_string(), + code, + message, + }); + } + return Ok(res.result.unwrap_or(Value::Null)); + } + FrameKind::Call => { + // Port callbacks (`inference.invoke` / `tools.invoke`) are + // serviced under the SAME wall-clock deadline as the + // response wait: a hung provider or tool must not suspend + // the deadline and pin the request (and the supervisor's + // connection lock) past the configured ceiling. Expiry + // surfaces through the exact path a read-timeout expiry + // takes, so the instruct→MaybeApplied / status→retry + // policy split is preserved unchanged. + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() + || tokio::time::timeout(remaining, self.handle_call(frame)) + .await + .is_err() + { + return Err(deadline_exceeded(op, self.request_deadline)); + } + } + FrameKind::Event => self.fold_event(frame), + FrameKind::Ready | FrameKind::Unknown => { + debug!("[medulla_local] ignoring unexpected inbound frame while awaiting res"); + } + } + } + } + + /// Dispatch one serve→host port `call` to [`HostPorts`] and write the + /// `ret` (§5). Only `inference` and `tools` are answered this draft; every + /// other port is refused `port_unavailable` — centralised here so no + /// implementer can forget the refusal. + async fn handle_call(&mut self, frame: Value) { + let call: CallFrame = match serde_json::from_value(frame) { + Ok(call) => call, + Err(error) => { + warn!("[medulla_local] malformed call frame skipped: {error}"); + return; + } + }; + debug!(id = %call.id, port = %call.port, method = %call.method, "[medulla_local] port call"); + + let ret = match (call.port.as_str(), call.method.as_str()) { + ("inference", "invoke") => match serde_json::from_value::(call.params) { + Ok(inference_call) => match self.ports.invoke_inference(inference_call).await { + Ok(result) => ret_ok( + &call.id, + serde_json::to_value(result).unwrap_or(Value::Null), + ), + Err(port_error) => ret_err(&call.id, &port_error.to_serve_error()), + }, + Err(error) => ret_err( + &call.id, + &ServeError::new(error_codes::BAD_REQUEST, error.to_string()), + ), + }, + // Cancellation is a fresh call id naming the target (§5.1); the + // draft answers its own ret and lets the in-flight call settle. + ("inference", "cancel") => ret_ok(&call.id, json!({})), + ("tools", "invoke") => { + let name = call.params.get("name").and_then(Value::as_str); + let args = call + .params + .get("args") + .cloned() + .unwrap_or_else(|| json!({})); + match name { + Some(name) => match self.ports.invoke_tool(name, args).await { + Ok(result) => ret_ok(&call.id, result), + Err(port_error) => ret_err(&call.id, &port_error.to_serve_error()), + }, + None => ret_err( + &call.id, + &ServeError::new(error_codes::BAD_REQUEST, "tools.invoke missing `name`"), + ), + } + } + (port, method) => ret_err( + &call.id, + &ServeError::new( + error_codes::PORT_UNAVAILABLE, + format!("port `{port}.{method}` is not offered by this host (draft)"), + ), + ), + }; + if let Err(error) = self.write_frame(&ret).await { + warn!( + "[medulla_local] failed to write ret for call {}: {error}", + call.id + ); + } + } + + /// Fold one `event` frame (§6). Advisory-only for the draft: track the + /// high-water `seq` and log a gap (which in the full design would trigger a + /// `subscribe` replay + re-`status`). + fn fold_event(&mut self, frame: Value) { + let event: EventFrame = match serde_json::from_value(frame) { + Ok(event) => event, + Err(error) => { + warn!("[medulla_local] malformed event skipped: {error}"); + return; + } + }; + if let Some(prev) = self.last_event_seq { + if event.seq > prev + 1 { + warn!( + prev, + seq = event.seq, + "[medulla_local] event seq gap — a full host would resync via subscribe(replay)" + ); + } + } + self.last_event_seq = Some(event.seq); + debug!(seq = event.seq, "[medulla_local] folded event"); + } + + async fn write_frame(&mut self, frame: &Value) -> Result<()> { + let mut line = serde_json::to_string(frame).context("encoding medulla frame")?; + line.push('\n'); + self.writer + .write_all(line.as_bytes()) + .await + .context("writing medulla frame")?; + self.writer + .flush() + .await + .context("flushing medulla frame")?; + Ok(()) + } + + async fn next_line(&mut self, timeout: Duration) -> Result { + match tokio::time::timeout(timeout, self.reader.next_line()).await { + Ok(Ok(Some(line))) => Ok(line), + Ok(Ok(None)) => bail!("medulla serve closed the connection"), + Ok(Err(error)) => Err(error).context("reading medulla serve frame"), + Err(_) => bail!("medulla serve frame read timed out"), + } + } + + /// Whether the supervised child has exited (killed externally, crashed + /// between requests). `try_wait` reaps without blocking. A connection + /// established without a child (tests dialing a mock listener) has no + /// process to supervise and reports alive; so does an indeterminate probe + /// error — discarding a possibly-healthy child on a probe failure would + /// be worse than one optimistic status report, and the next request's + /// transport error corrects it anyway. + fn child_has_exited(&mut self) -> bool { + match self.child.as_mut() { + Some(child) => matches!(child.try_wait(), Ok(Some(_))), + None => false, + } + } + + /// The `ready`/`hello` state captured at handshake, for status reporting. + fn status(&self) -> MedullaLocalStatus { + MedullaLocalStatus { + enabled: true, + running: true, + serve_version: self.ready.serve.clone(), + session_id: self + .hello + .session_id + .clone() + .or_else(|| self.ready.session_id.clone()), + ports: self.hello.ports.clone(), + message: None, + } + } +} + +/// A source of live [`Connection`]s. Production spawns a Node child and connects +/// its socket; tests connect to a mock listener. Keeping this a trait is the +/// seam that makes the supervisor's restart-and-retry logic testable without a +/// real process. +#[async_trait] +pub trait Connector: Send + Sync { + async fn connect(&self, ports: Arc) -> Result; + /// Human-readable identity for logs. + fn describe(&self) -> String; +} + +/// Production connector: resolves Node via `NodeBootstrap`, spawns medulla-v1's +/// `dist/serve` entry pointed at a unix socket, drains stderr, and connects. +pub struct NodeServeConnector { + node_bootstrap: Arc, + /// `None` when neither `subconscious.medulla_local.serve_entry` nor the + /// `OPENHUMAN_MEDULLA_SERVE_ENTRY` env override is set; `connect` then bails + /// with an actionable message rather than probing a bogus path. + serve_entry: Option, + socket_path: PathBuf, + host_identity: String, + /// Overall per-request deadline handed to every [`Connection`] this + /// connector establishes (from `subconscious.medulla_local`). + request_deadline: Duration, +} + +impl NodeServeConnector { + pub fn new( + node_bootstrap: Arc, + serve_entry: Option, + socket_path: PathBuf, + host_identity: String, + request_deadline: Duration, + ) -> Self { + Self { + node_bootstrap, + serve_entry, + socket_path, + host_identity, + request_deadline, + } + } +} + +#[async_trait] +impl Connector for NodeServeConnector { + async fn connect(&self, ports: Arc) -> Result { + let serve_entry = self.serve_entry.as_ref().ok_or_else(|| { + anyhow::anyhow!( + "medulla serve entry not configured: set subconscious.medulla_local.serve_entry \ + or the OPENHUMAN_MEDULLA_SERVE_ENTRY env var to the medulla-serve entry point \ + (path supplied via config/env, e.g. a built `dist/serve/index.js`)" + ) + })?; + if !serve_entry.is_file() { + bail!( + "medulla serve entry not found: {} (build medulla-v1 `dist/serve` or set \ + subconscious.medulla_local.serve_entry / OPENHUMAN_MEDULLA_SERVE_ENTRY)", + serve_entry.display() + ); + } + let node = self + .node_bootstrap + .resolve() + .await + .context("resolving Node toolchain for medulla serve")?; + + if let Some(parent) = self.socket_path.parent() { + tokio::fs::create_dir_all(parent) + .await + .with_context(|| format!("creating medulla socket dir {}", parent.display()))?; + } + // serve `rm`s a stale path before listen; be defensive in case a prior + // child died without cleanup so `connect` does not attach to a dead + // socket inode. + let _ = tokio::fs::remove_file(&self.socket_path).await; + + info!( + node = %node.node_bin.display(), + entry = %serve_entry.display(), + socket = %self.socket_path.display(), + "[medulla_local] spawning serve child" + ); + let mut command = tokio::process::Command::new(&node.node_bin); + command + .arg(serve_entry) + .arg("--socket") + .arg(&self.socket_path) + .env("PATH", prepend_path(&node.bin_dir)) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::piped()) + .kill_on_drop(true); + let mut child = command.spawn().context("spawning medulla serve child")?; + if let Some(stderr) = child.stderr.take() { + drain_stderr(stderr); + } + + let stream = connect_unix_retry(&self.socket_path, HANDSHAKE_TIMEOUT).await?; + // Advertise the curated read-only tool surface so serve binds it into + // its module registry and the model can emit `tools.invoke` for these tools. + // The spec set comes from the same `HostPorts` the `tools` port callback + // dispatches to, so what is advertised is exactly what can be invoked. + let tools = ports.tool_specs(); + let hello = HelloParams { + protocol: PROTOCOL_VERSION, + host: self.host_identity.clone(), + ports: vec!["inference".to_string(), "tools".to_string()], + tools, + }; + Connection::establish(stream, ports, hello, Some(child), self.request_deadline).await + } + + fn describe(&self) -> String { + format!("node serve @ {}", self.socket_path.display()) + } +} + +fn prepend_path(bin_dir: &Path) -> String { + match std::env::var("PATH") { + Ok(existing) => format!("{}:{}", bin_dir.display(), existing), + Err(_) => bin_dir.display().to_string(), + } +} + +/// Connect to the child's listening socket, retrying until it appears or the +/// handshake deadline elapses. +async fn connect_unix_retry(path: &Path, deadline: Duration) -> Result { + let start = Instant::now(); + loop { + match UnixStream::connect(path).await { + Ok(stream) => return Ok(stream), + Err(error) => { + if start.elapsed() >= deadline { + return Err(error).with_context(|| { + format!( + "connecting medulla serve socket {} timed out", + path.display() + ) + }); + } + tokio::time::sleep(SOCKET_CONNECT_POLL).await; + } + } + } +} + +/// Drain the child's stderr so a chatty serve never blocks on a full pipe +/// (mirrors `drain_server_stderr`). +fn drain_stderr(stderr: ChildStderr) { + tokio::spawn(async move { + let mut reader = BufReader::new(stderr).lines(); + while let Ok(Some(line)) = reader.next_line().await { + debug!("[medulla_local] serve stderr: {line}"); + } + debug!("[medulla_local] serve stderr drain closed"); + }); +} + +/// Supervises one serve connection with restart-and-retry-once semantics. +pub struct MedullaSupervisor { + connector: Arc, + ports: Arc, + connection: Mutex>, +} + +impl std::fmt::Debug for MedullaSupervisor { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MedullaSupervisor") + .field("connector", &self.connector.describe()) + .finish_non_exhaustive() + } +} + +impl MedullaSupervisor { + pub fn new(connector: Arc, ports: Arc) -> Self { + Self { + connector, + ports, + connection: Mutex::new(None), + } + } + + /// Ensure a connection exists (lazy spawn + handshake). A cached + /// connection whose child has since died is discarded and respawned. + pub async fn ensure(&self) -> Result<()> { + let mut guard = self.connection.lock().await; + Self::prune_dead_child(&mut guard); + if guard.is_none() { + *guard = Some(self.connector.connect(self.ports.clone()).await?); + } + Ok(()) + } + + /// Drop the cached connection when its supervised child has exited + /// (killed externally, crashed between requests). Returns whether a stale + /// connection was discarded. Probing BEFORE trusting the cache keeps two + /// contracts honest: `snapshot` never advertises `running: true` over a + /// dead child, and a non-idempotent request is never written into a + /// known-dead transport — which would have surfaced a spurious + /// [`RequestError::MaybeApplied`] for an operation that provably never + /// reached serve. + fn prune_dead_child(guard: &mut Option) -> bool { + let dead = guard.as_mut().is_some_and(Connection::child_has_exited); + if dead { + warn!("[medulla_local] supervised serve child has exited; discarding the stale connection"); + *guard = None; + } + dead + } + + /// Request with restart-and-retry-once (§7) — restricted to transport + /// failures on **idempotent** ops: only when the established connection + /// broke mid-request (process death, closed socket, IO failure, timeout) + /// does the host reset, respawn via the connector (which replays + /// `hello`), and retry exactly once. A retry that ALSO breaks + /// mid-request resets the replacement connection as well, so the next + /// request starts from a clean establish instead of reusing a + /// possibly-poisoned transport. Application-level rejections + /// (`ok=false`), undecodable results, and connect/handshake failures are + /// deterministic — retrying them would kill a healthy child — so they + /// fail fast with the typed [`RequestError`]. Non-idempotent ops (see + /// [`op_is_idempotent`]) are excluded from the retry entirely: the first + /// attempt may have reached serve before the break, so the connection is + /// reset (it is broken regardless) but the request is NOT replayed — + /// the caller gets [`RequestError::MaybeApplied`] and reconciles out of + /// band. + pub async fn request(&self, op: &str, params: Value) -> Result { + match self.request_once(op, params.clone()).await { + Ok(value) => Ok(value), + Err(error) if error.is_retryable() && !op_is_idempotent(op) => { + warn!( + "[medulla_local] non-idempotent request `{op}` hit a transport failure; \ + NOT retrying (the op may or may not have been applied), resetting {}: {error}", + self.connector.describe() + ); + self.reset().await; + match error { + RequestError::Transport(source) => Err(RequestError::MaybeApplied { + op: op.to_string(), + source, + } + .into()), + other => Err(other.into()), + } + } + Err(error) if error.is_retryable() => { + warn!( + "[medulla_local] request `{op}` hit a transport failure; restarting {} before retry: {error}", + self.connector.describe() + ); + self.reset().await; + match self.request_once(op, params).await { + Ok(value) => Ok(value), + Err(retry_error) => { + // The replacement connection is not left cached when + // the retry ALSO breaks mid-request: a known-bad + // transport would cost the next caller a full deadline + // — and could misreport `MaybeApplied` for an + // `instruct` written into it — before self-correcting. + // Reset so the next request starts from a clean + // establish. + if retry_error.is_retryable() { + warn!( + "[medulla_local] retry of `{op}` hit another transport failure; \ + resetting {} so the next request re-establishes: {retry_error}", + self.connector.describe() + ); + self.reset().await; + } + Err(retry_error.into()) + } + } + } + Err(error) => { + debug!("[medulla_local] request `{op}` failed non-retryably: {error}"); + Err(error.into()) + } + } + } + + async fn request_once( + &self, + op: &str, + params: Value, + ) -> Result { + let mut guard = self.connection.lock().await; + // A cached connection over a dead child would fail mid-request — for + // a non-idempotent op that misreads as MaybeApplied even though the + // request never reached serve. Detect and respawn up front instead. + Self::prune_dead_child(&mut guard); + if guard.is_none() { + *guard = Some( + self.connector + .connect(self.ports.clone()) + .await + .map_err(RequestError::Connect)?, + ); + } + let connection = match guard.as_mut() { + Some(connection) => connection, + None => { + return Err(RequestError::Connect(anyhow::anyhow!( + "medulla connection missing after connect" + ))) + } + }; + connection.request(op, params).await + } + + async fn reset(&self) { + // Dropping the connection kills the child (`kill_on_drop`). + let mut guard = self.connection.lock().await; + *guard = None; + } + + /// Enqueue one instruction (§4.1). Returns the synchronous receipt. + pub async fn instruct(&self, message: &str, meta: Value) -> Result { + self.request("instruct", json!({ "message": message, "meta": meta })) + .await + } + + /// Snapshot of `HarnessStatus` (§4.4). + pub async fn harness_status(&self) -> Result { + self.request("status", json!({})).await + } + + /// Non-spawning status snapshot from the currently-cached connection. + /// + /// The cached handshake state alone is not proof of life: the child can + /// die between requests without any I/O touching the socket. Liveness is + /// verified first, so a dead child is reported as `running: false` (and + /// the cache transitions to the restartable empty state) instead of + /// advertising a healthy supervisor that no longer exists. + pub async fn snapshot(&self) -> MedullaLocalStatus { + let mut guard = self.connection.lock().await; + let child_exited = Self::prune_dead_child(&mut guard); + match guard.as_ref() { + Some(connection) => connection.status(), + None => MedullaLocalStatus { + enabled: true, + running: false, + serve_version: None, + session_id: None, + ports: Vec::new(), + message: Some(if child_exited { + "medulla serve child exited; it will be respawned on the next request" + .to_string() + } else { + "medulla serve not connected".to_string() + }), + }, + } + } +} + +/// Cached global supervisor + start-failure backoff, mirroring the Python +/// server's `ServerCache`. Both live variants carry the fingerprint of the +/// config they were built from, so a config change invalidates them (the +/// Python server does the same via its `backends()` comparison). +enum SupervisorCache { + Empty, + Ready { + supervisor: Arc, + config_fingerprint: u64, + }, + Failed { + message: String, + retry_after: Instant, + config_fingerprint: u64, + }, +} + +static SUPERVISOR: std::sync::OnceLock> = std::sync::OnceLock::new(); + +fn supervisor_slot() -> &'static Mutex { + SUPERVISOR.get_or_init(|| Mutex::new(SupervisorCache::Empty)) +} + +/// Fingerprint of the config snapshot a supervisor is built from. The whole +/// config is hashed because the cached [`OpenhumanHostPorts`] captures the +/// whole `Arc`: security policy, `action_dir`, trusted roots, and +/// model/provider routing all feed the port callbacks, so any change must +/// invalidate the cached child. +fn config_fingerprint(config: &Config) -> Result { + use std::hash::{Hash, Hasher}; + let encoded = + serde_json::to_value(config).context("encoding config for medulla cache fingerprint")?; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + // Hash the JSON tree with object keys visited in sorted order, NOT the + // serialized string: `Config` contains `HashMap`-backed fields whose + // serde emission order is unstable (per-instance hash seeds), so hashing + // the raw encoding would spuriously invalidate the cache — and restart + // the child — for a byte-identical config. + hash_canonical_json(&encoded, &mut hasher); + // The runtime-resolved path roots are `#[serde(skip)]` and therefore + // absent from the encoding, but both feed the supervisor — the socket + // path lives under `workspace_dir` and the tool sandbox resolves against + // `action_dir` — so fold them in explicitly. + config.workspace_dir.hash(&mut hasher); + config.action_dir.hash(&mut hasher); + Ok(hasher.finish()) +} + +/// Feed a JSON value into `hasher` in a canonical order: object entries are +/// visited sorted by key (recursively), arrays in element order, and every +/// node is tagged with a type discriminant so differently-shaped trees cannot +/// collide by concatenation. +fn hash_canonical_json(value: &Value, hasher: &mut H) { + use std::hash::Hash; + match value { + Value::Null => 0u8.hash(hasher), + Value::Bool(flag) => { + 1u8.hash(hasher); + flag.hash(hasher); + } + Value::Number(number) => { + 2u8.hash(hasher); + number.to_string().hash(hasher); + } + Value::String(text) => { + 3u8.hash(hasher); + text.hash(hasher); + } + Value::Array(items) => { + 4u8.hash(hasher); + items.len().hash(hasher); + for item in items { + hash_canonical_json(item, hasher); + } + } + Value::Object(entries) => { + 5u8.hash(hasher); + entries.len().hash(hasher); + let mut keys: Vec<&String> = entries.keys().collect(); + keys.sort(); + for key in keys { + key.hash(hasher); + hash_canonical_json(&entries[key.as_str()], hasher); + } + } + } +} + +/// Resolve (and lazily start) the process-global supervisor for `config`. +/// +/// The cache is keyed on [`config_fingerprint`]: a cached supervisor built +/// from a different config snapshot is dropped (killing the child once the +/// last in-flight handle releases its connection) and rebuilt, so callbacks +/// never keep answering under a stale security policy or routing table. A +/// config change also bypasses the start-failure backoff, since the new +/// config may be exactly what fixes the startup failure. +pub async fn ensure_started(config: &Config) -> Result> { + let fingerprint = config_fingerprint(config)?; + let mut guard = supervisor_slot().lock().await; + match &*guard { + SupervisorCache::Ready { + supervisor, + config_fingerprint, + } if *config_fingerprint == fingerprint => { + let existing = supervisor.clone(); + drop(guard); + existing.ensure().await?; + return Ok(existing); + } + SupervisorCache::Ready { .. } => { + info!("[medulla_local] config changed; rebuilding serve supervisor"); + *guard = SupervisorCache::Empty; + } + SupervisorCache::Failed { + message, + retry_after, + config_fingerprint, + } if *config_fingerprint == fingerprint && Instant::now() < *retry_after => { + bail!("medulla serve unavailable after previous startup failure: {message}"); + } + SupervisorCache::Failed { .. } | SupervisorCache::Empty => {} + } + + match build_supervisor(config).await { + Ok(supervisor) => { + *guard = SupervisorCache::Ready { + supervisor: supervisor.clone(), + config_fingerprint: fingerprint, + }; + Ok(supervisor) + } + Err(error) => { + let message = format!("{error:#}"); + warn!( + "[medulla_local] startup failed; backing off {:?}: {message}", + START_FAILURE_BACKOFF + ); + *guard = SupervisorCache::Failed { + message: message.clone(), + retry_after: Instant::now() + START_FAILURE_BACKOFF, + config_fingerprint: fingerprint, + }; + bail!("medulla serve unavailable: {message}"); + } + } +} + +async fn build_supervisor(config: &Config) -> Result> { + let node_bootstrap = Arc::new(crate::openhuman::runtime_node::NodeBootstrap::new( + config.node.clone(), + config.workspace_dir.clone(), + reqwest::Client::new(), + )); + let serve_entry = config.subconscious.medulla_local.resolved_serve_entry(); + let socket_path = medulla_socket_path(config); + let host_identity = format!("openhuman/{}", env!("CARGO_PKG_VERSION")); + let request_deadline = config.subconscious.medulla_local.request_deadline(); + let connector = Arc::new(NodeServeConnector::new( + node_bootstrap, + serve_entry, + socket_path, + host_identity, + request_deadline, + )); + let ports: Arc = Arc::new(OpenhumanHostPorts::new(Arc::new(config.clone()))); + let supervisor = Arc::new(MedullaSupervisor::new(connector, ports)); + supervisor.ensure().await?; + Ok(supervisor) +} + +/// `serve.sock` under the workspace state dir (§1 path precedence, second hop — +/// `$XDG_RUNTIME_DIR` is the daemon's concern and left to a later milestone). +pub fn medulla_socket_path(config: &Config) -> PathBuf { + config.workspace_dir.join("medulla").join("serve.sock") +} + +#[cfg(test)] +#[path = "server_tests.rs"] +mod tests; diff --git a/src/openhuman/medulla_local/server_tests.rs b/src/openhuman/medulla_local/server_tests.rs new file mode 100644 index 000000000..0958d2bf7 --- /dev/null +++ b/src/openhuman/medulla_local/server_tests.rs @@ -0,0 +1,1194 @@ +//! Supervisor tests against a mock NDJSON serve server on a unix socket. +//! +//! Mirrors the `runtime_python_server` mock-JSONL tests: a mock listener plays +//! `serve`, the supervisor plays `host`. Covers the handshake, an `instruct` +//! round trip, `inference` port-callback routing, and restart-on-death. + +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use async_trait::async_trait; +use serde_json::{json, Value}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{UnixListener, UnixStream}; + +use super::*; +use crate::openhuman::medulla_local::ports::{HostPorts, PortError}; +use crate::openhuman::medulla_local::types::{InferenceCall, InferenceResult, ToolSpec, Usage}; + +static SOCKET_COUNTER: AtomicU64 = AtomicU64::new(0); + +fn unique_socket_path(tag: &str) -> PathBuf { + let n = SOCKET_COUNTER.fetch_add(1, Ordering::SeqCst); + std::env::temp_dir().join(format!( + "oh-medulla-{}-{}-{}.sock", + tag, + std::process::id(), + n + )) +} + +/// Records what the host ports were asked to do, and answers with canned data. +#[derive(Default)] +struct RecordingState { + inference_tiers: Mutex>, + tool_names: Mutex>, +} + +struct RecordingPorts { + state: Arc, + /// Never resolve `invoke_inference` (after recording the dispatch): the + /// shape of a hung model provider, so a test can assert the overall + /// request deadline bounds port-callback servicing too. + stall_inference: bool, +} + +/// The curated read-only tool the recording ports advertise and answer. A real +/// [`OpenhumanHostPorts`] derives this list from the runtime tool surface; the +/// test uses one fixed spec so both the `hello` advertisement and the +/// `tools.invoke` dispatch can be asserted end to end. +fn recording_tool_spec() -> ToolSpec { + ToolSpec { + name: "file_read".to_string(), + description: "Read a file from the workspace".to_string(), + parameters: json!({ "type": "object", "properties": { "path": { "type": "string" } } }), + } +} + +#[async_trait] +impl HostPorts for RecordingPorts { + fn tool_specs(&self) -> Vec { + vec![recording_tool_spec()] + } + + async fn invoke_inference(&self, call: InferenceCall) -> Result { + self.state + .inference_tiers + .lock() + .unwrap() + .push(call.tier.clone()); + if self.stall_inference { + std::future::pending::<()>().await; + } + Ok(InferenceResult { + content: "canned-answer".to_string(), + reasoning_content: None, + model: "test-model".to_string(), + tool_calls: Vec::new(), + usage: Usage { + input_tokens: 5, + output_tokens: 2, + }, + }) + } + + async fn invoke_tool(&self, name: &str, _args: Value) -> Result { + self.state.tool_names.lock().unwrap().push(name.to_string()); + Ok(json!({ "content": [{ "type": "text", "text": "ok" }], "isError": false })) + } +} + +/// Behaviour knobs for one mock serve run. +#[derive(Clone, Default)] +struct MockOpts { + /// Issue an `inference.invoke` port call before answering the first + /// `instruct`, asserting the returned `ret` content. + issue_inference: bool, + /// Issue a `tools.invoke` port call before answering the first `instruct`, + /// asserting the returned `ret` content. + issue_tool_call: bool, + /// Drop the connection on the first `instruct` of the first connection. + /// `instruct` is non-idempotent, so the supervisor must fail fast with + /// `MaybeApplied` instead of restart-and-retry. + die_first_instruct: bool, + /// Drop the connection on the first `status` of the first connection, + /// forcing the supervisor to restart-and-retry (status is idempotent). + die_first_status: bool, + /// Counts `instruct` requests that actually reached the mock, so a test + /// can assert a transport break did not cause a duplicate submission. + instruct_count: Option>, + /// Answer every `instruct` with `ok=false` (`bad_request`): a healthy + /// connection issuing an application-level rejection, which must NOT + /// trigger restart-and-retry. + reject_instruct: bool, + /// Counts accepted connections, so a test can assert whether the + /// supervisor restarted (2) or failed fast on the live child (1). + connection_count: Option>, + /// Sink recording the tool names advertised in the `hello` request, so a + /// test can assert the host advertised its curated surface. + observed_hello_tools: Option>>>, + /// On `instruct`/`status`, stream `event` frames forever instead of ever + /// answering the correlated `res`. Each frame feeds the host's per-read + /// idle timeout, so only the overall per-request deadline can end the + /// wait — the regression shape for a wedged child that keeps emitting. + stream_events_instead_of_res: bool, + /// On `instruct`/`status`, issue an `inference.invoke` port call and then + /// go silent (never answer the correlated `res`). Paired with a host whose + /// inference port never resolves, this is the regression shape for a hung + /// provider/tool holding the connection during callback servicing — only + /// the overall per-request deadline can end the wait. + stall_call_before_res: bool, + /// Drop the connection on `status` for each of the first N accepted + /// connections (generalizes `die_first_status`), so the retry attempt can + /// be made to fail too. + die_status_connections: u64, +} + +/// Spawn a mock serve loop on `listener`, one accepted connection at a time. +fn spawn_mock_serve(listener: UnixListener, opts: MockOpts) { + tokio::spawn(async move { + let mut conn_index = 0u64; + loop { + let (stream, _) = match listener.accept().await { + Ok(pair) => pair, + Err(_) => break, + }; + if let Some(counter) = &opts.connection_count { + counter.fetch_add(1, Ordering::SeqCst); + } + serve_connection(stream, conn_index, &opts).await; + conn_index += 1; + } + }); +} + +async fn serve_connection(stream: UnixStream, conn_index: u64, opts: &MockOpts) { + let (read_half, mut write_half) = stream.into_split(); + let mut reader = BufReader::new(read_half).lines(); + + // Unprompted ready banner (§3). + write_line( + &mut write_half, + &json!({ + "t": "ready", "protocol": 1, "serve": "3.12.0-test", + "sessionId": "agent", "capabilities": ["inference", "tools"], "error": null + }), + ) + .await; + + while let Ok(Some(line)) = reader.next_line().await { + let frame: Value = match serde_json::from_str(&line) { + Ok(frame) => frame, + Err(_) => continue, + }; + // Host→serve frames we handle: req and ret. We only initiate ret reads + // inline (below), so here we only see `req`. + if frame.get("t").and_then(Value::as_str) != Some("req") { + continue; + } + let id = frame + .get("id") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + let op = frame.get("op").and_then(Value::as_str).unwrap_or(""); + if opts.stream_events_instead_of_res && matches!(op, "instruct" | "status") { + if op == "instruct" { + if let Some(counter) = &opts.instruct_count { + counter.fetch_add(1, Ordering::SeqCst); + } + } + // Never answer the correlated res; keep the connection chatty so + // the per-read idle timeout is fed on every iteration. Stop when + // the host gives up and drops the connection (write error). + let mut seq = 0u64; + loop { + seq += 1; + let mut line = json!({ + "t": "event", "seq": seq, "at": 0, + "event": { "type": "progress" } + }) + .to_string(); + line.push('\n'); + if write_half.write_all(line.as_bytes()).await.is_err() + || write_half.flush().await.is_err() + { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + if opts.stall_call_before_res && matches!(op, "instruct" | "status") { + if op == "instruct" { + if let Some(counter) = &opts.instruct_count { + counter.fetch_add(1, Ordering::SeqCst); + } + } + // Reverse-RPC into the host, then go silent: no further frames. + // The host's inference port never resolves either, so only its + // overall per-request deadline can end the callback servicing. + write_line( + &mut write_half, + &json!({ + "t": "call", "id": format!("stall-{conn_index}"), + "port": "inference", "method": "invoke", + "params": { + "tier": "orchestrator", "op": "orchestrate", "cycleId": "cyc:1", + "messages": [{ "role": "user", "content": "stall" }] + } + }), + ) + .await; + // Drain until the host gives up and drops the connection. + while let Ok(Some(_)) = reader.next_line().await {} + return; + } + match op { + "hello" => { + if let Some(sink) = &opts.observed_hello_tools { + let names: Vec = frame + .get("params") + .and_then(|params| params.get("tools")) + .and_then(Value::as_array) + .map(|tools| { + tools + .iter() + .filter_map(|tool| tool.get("name").and_then(Value::as_str)) + .map(str::to_string) + .collect() + }) + .unwrap_or_default(); + *sink.lock().unwrap() = names; + } + write_line( + &mut write_half, + &json!({ + "t": "res", "id": id, "ok": true, + "result": { "protocol": 1, "sessionId": "agent", "ports": ["inference", "tools"] } + }), + ) + .await; + } + "status" => { + if (opts.die_first_status && conn_index == 0) + || conn_index < opts.die_status_connections + { + // Drop the connection mid-request: status is idempotent, + // so the host must restart and retry. + return; + } + write_line( + &mut write_half, + &json!({ + "t": "res", "id": id, "ok": true, + "result": { "state": "running", "queued": 0 } + }), + ) + .await; + } + "instruct" => { + if let Some(counter) = &opts.instruct_count { + counter.fetch_add(1, Ordering::SeqCst); + } + if opts.die_first_instruct && conn_index == 0 { + // Drop the connection mid-request: instruct is + // non-idempotent, so the host must NOT replay it. + return; + } + if opts.reject_instruct { + // Application-level rejection over a healthy connection: + // the host must fail fast, not kill and respawn us. + write_line( + &mut write_half, + &json!({ + "t": "res", "id": id, "ok": false, + "error": { "code": "bad_request", "message": "instruct refused by mock" } + }), + ) + .await; + continue; + } + if opts.issue_inference { + // Reverse-RPC into the host inference port, then read its ret. + write_line( + &mut write_half, + &json!({ + "t": "call", "id": "c1", "port": "inference", "method": "invoke", + "params": { + "tier": "orchestrator", "op": "orchestrate", "cycleId": "cyc:1", + "messages": [{ "role": "user", "content": "reconcile" }] + } + }), + ) + .await; + let ret = read_frame(&mut reader) + .await + .expect("host must answer the call"); + assert_eq!(ret["t"], "ret"); + assert_eq!(ret["id"], "c1"); + assert_eq!(ret["ok"], true); + assert_eq!(ret["result"]["content"], "canned-answer"); + } + if opts.issue_tool_call { + // Reverse-RPC into the host tools port, then read its ret. + write_line( + &mut write_half, + &json!({ + "t": "call", "id": "c2", "port": "tools", "method": "invoke", + "params": { + "name": "file_read", + "args": { "path": "README.md" }, + "callId": "cyc:1:tool_call:0", "cycleId": "cyc:1" + } + }), + ) + .await; + let ret = read_frame(&mut reader) + .await + .expect("host must answer the tools call"); + assert_eq!(ret["t"], "ret"); + assert_eq!(ret["id"], "c2"); + assert_eq!(ret["ok"], true); + assert_eq!(ret["result"]["isError"], false); + assert_eq!(ret["result"]["content"][0]["text"], "ok"); + } + write_line( + &mut write_half, + &json!({ + "t": "res", "id": id, "ok": true, + "result": { "instructionId": "inst-agent-0", "cycleId": "cyc:agent:agent:0" } + }), + ) + .await; + } + _ => { + write_line( + &mut write_half, + &json!({ + "t": "res", "id": id, "ok": false, + "error": { "code": "unknown_op", "message": op } + }), + ) + .await; + } + } + } +} + +async fn write_line(writer: &mut tokio::net::unix::OwnedWriteHalf, frame: &Value) { + let mut line = serde_json::to_string(frame).unwrap(); + line.push('\n'); + writer.write_all(line.as_bytes()).await.unwrap(); + writer.flush().await.unwrap(); +} + +async fn read_frame( + reader: &mut tokio::io::Lines>, +) -> Option { + let line = reader.next_line().await.ok()??; + serde_json::from_str(&line).ok() +} + +/// A connector that dials the mock listener instead of spawning Node. +struct MockConnector { + path: PathBuf, + request_deadline: Duration, +} + +#[async_trait] +impl Connector for MockConnector { + async fn connect(&self, ports: Arc) -> anyhow::Result { + let stream = connect_unix_retry(&self.path, Duration::from_secs(5)).await?; + let hello = mock_hello(&ports); + Connection::establish(stream, ports, hello, None, self.request_deadline).await + } + + fn describe(&self) -> String { + "mock".to_string() + } +} + +fn mock_hello(ports: &Arc) -> super::HelloParams { + super::HelloParams { + protocol: super::PROTOCOL_VERSION, + host: "openhuman/test".to_string(), + ports: vec!["inference".to_string(), "tools".to_string()], + tools: ports.tool_specs(), + } +} + +/// A connector that dials the mock listener AND spawns a stand-in child +/// process (`sleep`) whose only job is to be supervised, so the child-liveness +/// probe can be exercised hermetically: the transport stays "healthy" (the +/// mock listener never closes) while the supervised process dies. +struct ChildSpawningConnector { + path: PathBuf, + /// Pids of every stand-in child spawned, in connect order, so the test + /// can kill one externally. + spawned_pids: Arc>>, +} + +#[async_trait] +impl Connector for ChildSpawningConnector { + async fn connect(&self, ports: Arc) -> anyhow::Result { + let child = tokio::process::Command::new("sleep") + .arg("600") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .kill_on_drop(true) + .spawn()?; + self.spawned_pids + .lock() + .unwrap() + .push(child.id().expect("freshly spawned child has a pid")); + let stream = connect_unix_retry(&self.path, Duration::from_secs(5)).await?; + let hello = mock_hello(&ports); + Connection::establish(stream, ports, hello, Some(child), Duration::from_secs(300)).await + } + + fn describe(&self) -> String { + "mock+child".to_string() + } +} + +fn build(path: PathBuf, state: Arc) -> MedullaSupervisor { + build_with_deadline(path, state, Duration::from_secs(300)) +} + +fn build_with_deadline( + path: PathBuf, + state: Arc, + request_deadline: Duration, +) -> MedullaSupervisor { + build_with_ports( + path, + Arc::new(RecordingPorts { + state, + stall_inference: false, + }), + request_deadline, + ) +} + +fn build_with_ports( + path: PathBuf, + ports: Arc, + request_deadline: Duration, +) -> MedullaSupervisor { + MedullaSupervisor::new( + Arc::new(MockConnector { + path, + request_deadline, + }), + ports, + ) +} + +#[tokio::test] +async fn handshake_negotiates_ready_and_hello() { + let path = unique_socket_path("handshake"); + let listener = UnixListener::bind(&path).unwrap(); + spawn_mock_serve(listener, MockOpts::default()); + + let supervisor = build(path.clone(), Arc::new(RecordingState::default())); + supervisor.ensure().await.expect("handshake should succeed"); + + let status = supervisor.snapshot().await; + assert!(status.running); + assert_eq!(status.serve_version.as_deref(), Some("3.12.0-test")); + assert_eq!(status.session_id.as_deref(), Some("agent")); + assert_eq!(status.ports, vec!["inference", "tools"]); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn instruct_round_trip_returns_receipt() { + let path = unique_socket_path("instruct"); + let listener = UnixListener::bind(&path).unwrap(); + spawn_mock_serve(listener, MockOpts::default()); + + let supervisor = build(path.clone(), Arc::new(RecordingState::default())); + let receipt = supervisor + .instruct("reconcile the world", json!({ "origin": "wake" })) + .await + .expect("instruct should return a receipt"); + assert_eq!(receipt.instruction_id, "inst-agent-0"); + assert_eq!(receipt.cycle_id, "cyc:agent:agent:0"); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn inference_callback_routes_to_host_ports() { + let path = unique_socket_path("inference"); + let listener = UnixListener::bind(&path).unwrap(); + spawn_mock_serve( + listener, + MockOpts { + issue_inference: true, + ..MockOpts::default() + }, + ); + + let state = Arc::new(RecordingState::default()); + let supervisor = build(path.clone(), state.clone()); + let receipt = supervisor + .instruct("reconcile", json!({})) + .await + .expect("instruct with an inference callback should complete"); + assert_eq!(receipt.instruction_id, "inst-agent-0"); + + // The serve inference call was dispatched to the host ports with its tier. + let tiers = state.inference_tiers.lock().unwrap().clone(); + assert_eq!(tiers, vec!["orchestrator"]); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn tools_are_advertised_and_invocable() { + let path = unique_socket_path("tools"); + let listener = UnixListener::bind(&path).unwrap(); + let observed = Arc::new(Mutex::new(Vec::new())); + spawn_mock_serve( + listener, + MockOpts { + issue_tool_call: true, + observed_hello_tools: Some(observed.clone()), + ..MockOpts::default() + }, + ); + + let state = Arc::new(RecordingState::default()); + let supervisor = build(path.clone(), state.clone()); + let receipt = supervisor + .instruct("reconcile", json!({})) + .await + .expect("instruct with a tools callback should complete"); + assert_eq!(receipt.instruction_id, "inst-agent-0"); + + // The host advertised its curated tool surface in the hello handshake, so + // serve could bind the tool and drive a `tools.invoke` for it. + assert_eq!(observed.lock().unwrap().clone(), vec!["file_read"]); + // And that invocation reached the host ports with the advertised name. + let names = state.tool_names.lock().unwrap().clone(); + assert_eq!(names, vec!["file_read"]); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn restart_on_death_retries_idempotent_status_once() { + let path = unique_socket_path("restart"); + let listener = UnixListener::bind(&path).unwrap(); + let connections = Arc::new(AtomicU64::new(0)); + spawn_mock_serve( + listener, + MockOpts { + die_first_status: true, + connection_count: Some(connections.clone()), + ..MockOpts::default() + }, + ); + + let supervisor = build(path.clone(), Arc::new(RecordingState::default())); + // First connection dies mid-status; status is idempotent, so the + // supervisor restarts and the second connection answers. + let status = supervisor + .harness_status() + .await + .expect("restart-and-retry-once should recover an idempotent op"); + assert_eq!(status.state, "running"); + assert_eq!( + connections.load(Ordering::SeqCst), + 2, + "a mid-request transport death must trigger exactly one respawn" + ); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn instruct_transport_failure_fails_fast_without_replay() { + let path = unique_socket_path("instruct-transport"); + let listener = UnixListener::bind(&path).unwrap(); + let connections = Arc::new(AtomicU64::new(0)); + let instructs = Arc::new(AtomicU64::new(0)); + spawn_mock_serve( + listener, + MockOpts { + die_first_instruct: true, + connection_count: Some(connections.clone()), + instruct_count: Some(instructs.clone()), + ..MockOpts::default() + }, + ); + + let supervisor = build(path.clone(), Arc::new(RecordingState::default())); + let error = supervisor + .instruct("reconcile", json!({})) + .await + .expect_err("a mid-instruct transport break must surface as an error"); + + // The error is the typed maybe-applied outcome, telling the caller the + // instruction may or may not have been enqueued… + let request_error = error + .downcast_ref::() + .expect("supervisor errors must stay downcastable to RequestError"); + assert!( + matches!(request_error, RequestError::MaybeApplied { op, .. } if op == "instruct"), + "expected MaybeApplied for the non-idempotent op, got: {request_error:?}" + ); + assert!(!request_error.is_retryable()); + + // …the instruct was submitted exactly once — no duplicate enqueue… + assert_eq!( + instructs.load(Ordering::SeqCst), + 1, + "a non-idempotent op must never be replayed after a transport break" + ); + // …and no respawn-driven retry connection was made for it. + assert_eq!( + connections.load(Ordering::SeqCst), + 1, + "failing fast must not respawn to replay the instruct" + ); + + // The broken connection was reset: a later idempotent request reconnects. + let status = supervisor + .harness_status() + .await + .expect("the supervisor must recover on the next request"); + assert_eq!(status.state, "running"); + assert_eq!(connections.load(Ordering::SeqCst), 2); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn serve_rejection_fails_fast_without_restart() { + let path = unique_socket_path("reject"); + let listener = UnixListener::bind(&path).unwrap(); + let connections = Arc::new(AtomicU64::new(0)); + spawn_mock_serve( + listener, + MockOpts { + reject_instruct: true, + connection_count: Some(connections.clone()), + ..MockOpts::default() + }, + ); + + let supervisor = build(path.clone(), Arc::new(RecordingState::default())); + let error = supervisor + .instruct("reconcile", json!({})) + .await + .expect_err("an ok=false serve rejection must surface as an error"); + + // The error is the typed, non-retryable serve rejection… + let request_error = error + .downcast_ref::() + .expect("supervisor errors must stay downcastable to RequestError"); + assert!( + matches!(request_error, RequestError::Serve { code, .. } if code == "bad_request"), + "expected a Serve error carrying the wire code, got: {request_error:?}" + ); + assert!(!request_error.is_retryable()); + + // …the healthy child was NOT killed and respawned… + assert_eq!( + connections.load(Ordering::SeqCst), + 1, + "an application-level rejection must not trigger a restart" + ); + + // …and the connection is still live for the next request. + let status = supervisor.snapshot().await; + assert!( + status.running, + "the connection must survive a serve rejection" + ); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn overall_deadline_bounds_instruct_despite_continuous_events() { + let path = unique_socket_path("deadline-instruct"); + let listener = UnixListener::bind(&path).unwrap(); + let connections = Arc::new(AtomicU64::new(0)); + let instructs = Arc::new(AtomicU64::new(0)); + spawn_mock_serve( + listener, + MockOpts { + stream_events_instead_of_res: true, + connection_count: Some(connections.clone()), + instruct_count: Some(instructs.clone()), + ..MockOpts::default() + }, + ); + + let supervisor = build_with_deadline( + path.clone(), + Arc::new(RecordingState::default()), + Duration::from_millis(300), + ); + let started = std::time::Instant::now(); + let error = supervisor + .instruct("reconcile", json!({})) + .await + .expect_err("a never-answered instruct must trip the overall deadline"); + + // The request ended promptly even though event frames kept feeding the + // per-read idle timeout — the overall deadline is what fired. + assert!( + started.elapsed() < Duration::from_secs(10), + "the overall deadline must bound the wait; waited {:?}", + started.elapsed() + ); + + // The deadline on a non-idempotent op keeps the fail-fast MaybeApplied + // contract: the instruct reached serve but its outcome was never observed. + let request_error = error + .downcast_ref::() + .expect("supervisor errors must stay downcastable to RequestError"); + assert!( + matches!(request_error, RequestError::MaybeApplied { op, .. } if op == "instruct"), + "expected MaybeApplied for the deadline on a non-idempotent op, got: {request_error:?}" + ); + assert!(format!("{request_error:#}").contains("deadline")); + assert_eq!( + instructs.load(Ordering::SeqCst), + 1, + "the deadline must not cause a silent replay of the instruct" + ); + assert_eq!( + connections.load(Ordering::SeqCst), + 1, + "failing fast on the deadline must not respawn to replay the instruct" + ); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn overall_deadline_bounds_idempotent_status_with_one_retry() { + let path = unique_socket_path("deadline-status"); + let listener = UnixListener::bind(&path).unwrap(); + let connections = Arc::new(AtomicU64::new(0)); + spawn_mock_serve( + listener, + MockOpts { + stream_events_instead_of_res: true, + connection_count: Some(connections.clone()), + ..MockOpts::default() + }, + ); + + let supervisor = build_with_deadline( + path.clone(), + Arc::new(RecordingState::default()), + Duration::from_millis(300), + ); + let started = std::time::Instant::now(); + let error = supervisor + .harness_status() + .await + .expect_err("a never-answered status must trip the deadline on both attempts"); + assert!( + started.elapsed() < Duration::from_secs(10), + "both attempts together must stay bounded; waited {:?}", + started.elapsed() + ); + + // The deadline is a transport-class failure, so the idempotent op kept + // its restart-and-retry-once semantics: exactly one respawn happened. + let request_error = error + .downcast_ref::() + .expect("supervisor errors must stay downcastable to RequestError"); + assert!( + matches!(request_error, RequestError::Transport(_)), + "expected the typed transport deadline error, got: {request_error:?}" + ); + assert!(format!("{request_error:#}").contains("deadline")); + assert_eq!( + connections.load(Ordering::SeqCst), + 2, + "an idempotent deadline trip must restart-and-retry exactly once" + ); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn overall_deadline_bounds_instruct_during_hung_port_callback() { + let path = unique_socket_path("deadline-callback-instruct"); + let listener = UnixListener::bind(&path).unwrap(); + let connections = Arc::new(AtomicU64::new(0)); + let instructs = Arc::new(AtomicU64::new(0)); + spawn_mock_serve( + listener, + MockOpts { + stall_call_before_res: true, + connection_count: Some(connections.clone()), + instruct_count: Some(instructs.clone()), + ..MockOpts::default() + }, + ); + + let state = Arc::new(RecordingState::default()); + let supervisor = build_with_ports( + path.clone(), + Arc::new(RecordingPorts { + state: state.clone(), + stall_inference: true, + }), + Duration::from_millis(300), + ); + let started = std::time::Instant::now(); + let error = supervisor + .instruct("reconcile", json!({})) + .await + .expect_err("a callback that never returns must trip the overall deadline"); + + // The hung port callback could not suspend the deadline: the request + // ended promptly instead of pinning the connection on the provider. + assert!( + started.elapsed() < Duration::from_secs(10), + "the overall deadline must bound callback servicing; waited {:?}", + started.elapsed() + ); + + // Deadline expiry rides the same path as a read-timeout expiry, so the + // non-idempotent op keeps its fail-fast MaybeApplied contract. + let request_error = error + .downcast_ref::() + .expect("supervisor errors must stay downcastable to RequestError"); + assert!( + matches!(request_error, RequestError::MaybeApplied { op, .. } if op == "instruct"), + "expected MaybeApplied for the deadline on a non-idempotent op, got: {request_error:?}" + ); + assert!(format!("{request_error:#}").contains("deadline")); + assert_eq!( + instructs.load(Ordering::SeqCst), + 1, + "the deadline must not cause a silent replay of the instruct" + ); + assert_eq!( + connections.load(Ordering::SeqCst), + 1, + "failing fast on the deadline must not respawn to replay the instruct" + ); + // The callback was dispatched into the host ports exactly once. + assert_eq!(state.inference_tiers.lock().unwrap().len(), 1); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn overall_deadline_bounds_status_during_hung_port_callback() { + let path = unique_socket_path("deadline-callback-status"); + let listener = UnixListener::bind(&path).unwrap(); + let connections = Arc::new(AtomicU64::new(0)); + spawn_mock_serve( + listener, + MockOpts { + stall_call_before_res: true, + connection_count: Some(connections.clone()), + ..MockOpts::default() + }, + ); + + let state = Arc::new(RecordingState::default()); + let supervisor = build_with_ports( + path.clone(), + Arc::new(RecordingPorts { + state, + stall_inference: true, + }), + Duration::from_millis(300), + ); + let started = std::time::Instant::now(); + let error = supervisor + .harness_status() + .await + .expect_err("a hung callback must trip the deadline on both attempts"); + assert!( + started.elapsed() < Duration::from_secs(10), + "both attempts together must stay bounded; waited {:?}", + started.elapsed() + ); + + // The deadline surfaces as the typed transport error, so the idempotent + // op kept restart-and-retry-once: exactly one respawn happened. + let request_error = error + .downcast_ref::() + .expect("supervisor errors must stay downcastable to RequestError"); + assert!( + matches!(request_error, RequestError::Transport(_)), + "expected the typed transport deadline error, got: {request_error:?}" + ); + assert!(format!("{request_error:#}").contains("deadline")); + assert_eq!( + connections.load(Ordering::SeqCst), + 2, + "an idempotent deadline trip must restart-and-retry exactly once" + ); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn retry_failure_resets_cache_so_next_request_reestablishes() { + let path = unique_socket_path("retry-failure-reset"); + let listener = UnixListener::bind(&path).unwrap(); + let connections = Arc::new(AtomicU64::new(0)); + let instructs = Arc::new(AtomicU64::new(0)); + spawn_mock_serve( + listener, + MockOpts { + // Both the first attempt AND its retry die mid-status; the third + // connection behaves. + die_status_connections: 2, + connection_count: Some(connections.clone()), + instruct_count: Some(instructs.clone()), + ..MockOpts::default() + }, + ); + + let supervisor = build(path.clone(), Arc::new(RecordingState::default())); + let error = supervisor + .harness_status() + .await + .expect_err("a retry that also dies mid-request must surface an error"); + let request_error = error + .downcast_ref::() + .expect("supervisor errors must stay downcastable to RequestError"); + assert!( + matches!(request_error, RequestError::Transport(_)), + "expected the transport error from the failed retry, got: {request_error:?}" + ); + assert_eq!( + connections.load(Ordering::SeqCst), + 2, + "restart-and-retry must have attempted exactly one respawn" + ); + + // The broken replacement connection was reset, not left cached: the next + // request — the non-idempotent instruct — starts from a clean establish + // and succeeds, instead of being written into the stale transport and + // misreading as MaybeApplied. + let receipt = supervisor + .instruct("reconcile", json!({})) + .await + .expect("the request after a failed retry must re-establish and succeed"); + assert_eq!(receipt.instruction_id, "inst-agent-0"); + assert_eq!( + instructs.load(Ordering::SeqCst), + 1, + "the instruct must be submitted exactly once, on the fresh connection" + ); + assert_eq!( + connections.load(Ordering::SeqCst), + 3, + "the request after a failed retry must open a fresh connection" + ); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn dead_child_reports_not_running_and_respawns_on_next_request() { + let path = unique_socket_path("dead-child"); + let listener = UnixListener::bind(&path).unwrap(); + let connections = Arc::new(AtomicU64::new(0)); + spawn_mock_serve( + listener, + MockOpts { + connection_count: Some(connections.clone()), + ..MockOpts::default() + }, + ); + + let spawned_pids = Arc::new(Mutex::new(Vec::new())); + let ports: Arc = Arc::new(RecordingPorts { + state: Arc::new(RecordingState::default()), + stall_inference: false, + }); + let supervisor = MedullaSupervisor::new( + Arc::new(ChildSpawningConnector { + path: path.clone(), + spawned_pids: spawned_pids.clone(), + }), + ports, + ); + + supervisor.ensure().await.expect("handshake should succeed"); + assert!( + supervisor.snapshot().await.running, + "a live supervised child must report running" + ); + + // Kill the supervised child externally, leaving the cached transport + // untouched — the exact shape of a child dying between requests. + let pid = spawned_pids.lock().unwrap()[0]; + let killed = std::process::Command::new("kill") + .args(["-9", &pid.to_string()]) + .status() + .expect("kill must run"); + assert!(killed.success(), "kill -9 must reach the stand-in child"); + + // Signal delivery is asynchronous: poll until the snapshot notices. + let mut status = supervisor.snapshot().await; + let poll_started = std::time::Instant::now(); + while status.running && poll_started.elapsed() < Duration::from_secs(5) { + tokio::time::sleep(Duration::from_millis(20)).await; + status = supervisor.snapshot().await; + } + assert!( + !status.running, + "a dead child must not be reported as running from the cached handshake" + ); + assert!( + status + .message + .as_deref() + .is_some_and(|message| message.contains("exited")), + "the status must say the child exited, got: {:?}", + status.message + ); + + // The cache moved to the restartable state: the next request respawns a + // fresh child instead of writing into the dead one — in particular the + // non-idempotent instruct succeeds rather than misreporting MaybeApplied. + let receipt = supervisor + .instruct("reconcile", json!({})) + .await + .expect("a request after child death must respawn and succeed"); + assert_eq!(receipt.instruction_id, "inst-agent-0"); + assert_eq!( + connections.load(Ordering::SeqCst), + 2, + "child death must lead to exactly one respawn on the next request" + ); + assert_eq!(spawned_pids.lock().unwrap().len(), 2); + assert!( + supervisor.snapshot().await.running, + "the respawned child must report running again" + ); + let _ = std::fs::remove_file(&path); +} + +#[tokio::test] +async fn ensure_started_backoff_is_keyed_on_config_fingerprint() { + // An explicit (nonexistent) serve entry wins over any env override, so + // the build fails deterministically before touching Node or the network. + let mut config_a = crate::openhuman::config::Config::default(); + config_a.subconscious.medulla_local.serve_entry = + "/nonexistent/medulla-local-test/serve-a.js".to_string(); + + let first = ensure_started(&config_a) + .await + .expect_err("a missing serve entry must fail startup"); + assert!( + format!("{first:#}").contains("serve entry not found"), + "unexpected first startup error: {first:#}" + ); + + // Same config within the backoff window: fail fast on the cached failure. + let cached = ensure_started(&config_a) + .await + .expect_err("the same config must stay in start-failure backoff"); + assert!( + format!("{cached:#}").contains("after previous startup failure"), + "unexpected backoff error: {cached:#}" + ); + + // A changed config bypasses the backoff and attempts a fresh build — the + // new snapshot may be exactly what fixes the failure. + let mut config_b = config_a.clone(); + config_b.subconscious.medulla_local.serve_entry = + "/nonexistent/medulla-local-test/serve-b.js".to_string(); + let rebuilt = ensure_started(&config_b) + .await + .expect_err("the changed config still points at a missing entry"); + let rebuilt_message = format!("{rebuilt:#}"); + assert!( + !rebuilt_message.contains("after previous startup failure"), + "a config change must bypass the stale backoff: {rebuilt_message}" + ); + assert!( + rebuilt_message.contains("serve-b.js"), + "the fresh build must run against the NEW config: {rebuilt_message}" + ); +} + +#[test] +fn config_fingerprint_tracks_relevant_config_changes() { + let base = crate::openhuman::config::Config::default(); + let base_fingerprint = config_fingerprint(&base).unwrap(); + assert_eq!( + base_fingerprint, + config_fingerprint(&base.clone()).unwrap(), + "the fingerprint must be deterministic for an identical config" + ); + + // The cached ports capture the whole config: a serve-entry change, a + // security-root change, and an action-dir change must each invalidate. + let mut serve_changed = base.clone(); + serve_changed.subconscious.medulla_local.serve_entry = "/elsewhere/serve.js".to_string(); + assert_ne!( + base_fingerprint, + config_fingerprint(&serve_changed).unwrap() + ); + + let mut action_dir_changed = base.clone(); + action_dir_changed.action_dir = std::path::PathBuf::from("/elsewhere/projects"); + assert_ne!( + base_fingerprint, + config_fingerprint(&action_dir_changed).unwrap() + ); +} + +#[test] +fn config_fingerprint_is_stable_across_map_insertion_orders() { + use crate::openhuman::config::schema::TeamModelConfig; + + // `Config` carries HashMap-backed fields whose serde emission order is + // unstable (per-instance hash seeds). Two configs holding the SAME team + // pins inserted in opposite orders must fingerprint identically — a + // mismatch would spuriously kill and respawn the supervised child. + let team = |model: &str| TeamModelConfig { + lead_model: Some(model.to_string()), + agent_model: None, + }; + let mut config_a = crate::openhuman::config::Config::default(); + for name in ["alpha", "beta", "gamma", "delta", "epsilon"] { + config_a.teams.insert(name.to_string(), team(name)); + } + let mut config_b = crate::openhuman::config::Config::default(); + for name in ["epsilon", "delta", "gamma", "beta", "alpha"] { + config_b.teams.insert(name.to_string(), team(name)); + } + + assert_eq!( + config_fingerprint(&config_a).unwrap(), + config_fingerprint(&config_b).unwrap(), + "identical configs must fingerprint identically regardless of map order" + ); + + // And a genuinely different map value still changes the fingerprint. + let mut config_c = config_a.clone(); + config_c.teams.insert("alpha".to_string(), team("changed")); + assert_ne!( + config_fingerprint(&config_a).unwrap(), + config_fingerprint(&config_c).unwrap() + ); +} + +#[test] +fn canonical_json_hash_ignores_object_key_order() { + use std::collections::hash_map::DefaultHasher; + use std::hash::Hasher; + + let hash = |value: &Value| { + let mut hasher = DefaultHasher::new(); + hash_canonical_json(value, &mut hasher); + hasher.finish() + }; + + // serde_json is built with `preserve_order` in this workspace, so these + // two literals genuinely serialize with different key orders. + let ab = json!({ "a": 1, "b": { "x": [1, 2], "y": null } }); + let ba = json!({ "b": { "y": null, "x": [1, 2] }, "a": 1 }); + assert_eq!(hash(&ab), hash(&ba), "key order must not affect the hash"); + + // Same keys, different value → different hash. + let changed = json!({ "a": 2, "b": { "x": [1, 2], "y": null } }); + assert_ne!(hash(&ab), hash(&changed)); + // Array order stays significant. + let reordered_array = json!({ "a": 1, "b": { "x": [2, 1], "y": null } }); + assert_ne!(hash(&ab), hash(&reordered_array)); +} diff --git a/src/openhuman/medulla_local/server_unsupported.rs b/src/openhuman/medulla_local/server_unsupported.rs new file mode 100644 index 000000000..2aeffa97e --- /dev/null +++ b/src/openhuman/medulla_local/server_unsupported.rs @@ -0,0 +1,87 @@ +//! Non-unix stub for the medulla-local supervisor surface. +//! +//! The local serve transport is a unix domain socket, which `tokio::net` +//! only provides on unix targets. On other targets (Windows) the +//! `medulla-local` feature still compiles: this stub exposes the same public +//! surface `ops.rs` drives, with every entry point reporting a typed +//! unsupported-platform error instead of breaking the build. A portable +//! transport (e.g. stdio) can replace this stub later without touching the +//! callers. + +use std::sync::Arc; + +use anyhow::Result; +use serde_json::Value; + +use super::types::{HarnessStatus, InstructReceipt, MedullaLocalStatus}; +use crate::openhuman::config::Config; + +/// Typed marker for "this build's target has no unix-socket transport", so +/// callers can distinguish a platform limitation from a runtime failure. +#[derive(Debug, Clone, Copy, thiserror::Error)] +#[error("medulla-local is unavailable on this platform: the local serve transport requires unix domain sockets")] +pub struct UnsupportedPlatformError; + +/// Stub supervisor: never connectable on this platform. [`ensure_started`] +/// always fails first, so these methods exist only to keep the call sites in +/// `ops.rs` compiling identically on every target. +pub struct MedullaSupervisor; + +impl MedullaSupervisor { + /// Always fails with [`UnsupportedPlatformError`]. + pub async fn instruct(&self, _message: &str, _meta: Value) -> Result { + Err(UnsupportedPlatformError.into()) + } + + /// Always fails with [`UnsupportedPlatformError`]. + pub async fn harness_status(&self) -> Result { + Err(UnsupportedPlatformError.into()) + } + + /// A well-formed not-running snapshot carrying the platform message. + pub async fn snapshot(&self) -> MedullaLocalStatus { + MedullaLocalStatus { + enabled: true, + running: false, + serve_version: None, + session_id: None, + ports: Vec::new(), + message: Some(UnsupportedPlatformError.to_string()), + } + } +} + +/// Always reports the platform as unsupported (typed, downcastable): `status` +/// folds it into a well-formed not-running snapshot and `instruct` fails +/// cleanly, exactly like an unconfigured serve entry does on unix. +pub async fn ensure_started(_config: &Config) -> Result> { + Err(UnsupportedPlatformError.into()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn ensure_started_reports_unsupported_platform() { + let error = ensure_started(&Config::default()) + .await + .expect_err("non-unix targets must not report a live supervisor"); + assert!( + error.downcast_ref::().is_some(), + "error must be the typed platform marker: {error:#}" + ); + } + + #[tokio::test] + async fn stub_supervisor_snapshot_is_well_formed_and_not_running() { + let snapshot = MedullaSupervisor.snapshot().await; + assert!(snapshot.enabled); + assert!(!snapshot.running); + assert!(snapshot + .message + .as_deref() + .unwrap_or_default() + .contains("unavailable on this platform")); + } +} diff --git a/src/openhuman/medulla_local/types.rs b/src/openhuman/medulla_local/types.rs new file mode 100644 index 000000000..f84f517a9 --- /dev/null +++ b/src/openhuman/medulla_local/types.rs @@ -0,0 +1,221 @@ +//! Domain types for the `medulla_local` flow: the `hello` handshake, the +//! `instruct`/`status` request results, and the `inference` port-callback +//! request/response shapes (§3, §4, §5.1 of the serve protocol spec). +//! +//! These mirror the medulla-v1 types named in the spec exactly enough to +//! decode/encode the wire; fields the draft does not consume are kept +//! `#[serde(default)]` / optional so a serve version that adds fields never +//! breaks the host. + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// Reserved error codes (§8). A port callback the host cannot answer returns +/// one of these in a `ret` error envelope. +pub mod error_codes { + pub const BAD_REQUEST: &str = "bad_request"; + pub const NOT_READY: &str = "not_ready"; + pub const UNKNOWN_OP: &str = "unknown_op"; + pub const UNKNOWN_PORT: &str = "unknown_port"; + pub const UNSUPPORTED_METHOD: &str = "unsupported_method"; + pub const PORT_UNAVAILABLE: &str = "port_unavailable"; + pub const TIMEOUT: &str = "timeout"; + pub const RETRYABLE: &str = "retryable"; + pub const INTERNAL: &str = "internal"; +} + +/// A host tool spec advertised in the `hello` request (`ToolSpec`, §3). The +/// serve side binds these into its module registry. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ToolSpec { + pub name: String, + pub description: String, + pub parameters: Value, +} + +/// The `hello` request params the host sends after receiving `ready` (§3). +/// `ports` is the set of port callbacks the host will answer; the active set +/// is the intersection with serve's advertised `capabilities`. +#[derive(Debug, Clone, Serialize)] +pub struct HelloParams { + pub protocol: u32, + pub host: String, + pub ports: Vec, + pub tools: Vec, +} + +/// The `hello` response (§3): the negotiated active port set. +#[derive(Debug, Clone, Default, Deserialize)] +pub struct HelloResult { + #[serde(default)] + pub protocol: u32, + #[serde(default, rename = "sessionId")] + pub session_id: Option, + #[serde(default)] + pub ports: Vec, +} + +/// Synchronous receipt for an `instruct` (§4.1). The cycle itself runs async +/// and is observed via `event` frames. +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +pub struct InstructReceipt { + #[serde(rename = "instructionId")] + pub instruction_id: String, + #[serde(rename = "cycleId")] + pub cycle_id: String, +} + +/// Token accounting shared by `status` and `inference` (§4.4, §5.1). +#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq, Eq)] +pub struct Usage { + #[serde(default, rename = "inputTokens")] + pub input_tokens: u64, + #[serde(default, rename = "outputTokens")] + pub output_tokens: u64, +} + +/// Snapshot of `HarnessStatus` (§4.4). Unmodelled sub-shapes ride as raw +/// `Value` so a richer serve status never fails to decode. +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +pub struct HarnessStatus { + #[serde(default)] + pub state: String, + #[serde(default)] + pub queued: u64, + #[serde(default, rename = "activeInstructionId")] + pub active_instruction_id: Option, + #[serde(default, rename = "activeCycleId")] + pub active_cycle_id: Option, + #[serde(default)] + pub tasks: Vec, + #[serde(default, rename = "runningDelegations")] + pub running_delegations: u64, + #[serde(default)] + pub usage: Value, + #[serde(default)] + pub escalations: Vec, +} + +/// A single chat message crossing the `inference` port (§5.1). +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct WireChatMessage { + pub role: String, + #[serde(default)] + pub content: String, +} + +/// The `inference.invoke` port-callback params (§5.1). `tier` is one of +/// `orchestrator` / `reasoning` / `compress`; the host maps it onto its +/// per-role model routing. +#[derive(Debug, Clone, Deserialize)] +pub struct InferenceCall { + #[serde(default)] + pub tier: String, + #[serde(default)] + pub op: String, + #[serde(default, rename = "cycleId")] + pub cycle_id: Option, + #[serde(default)] + pub messages: Vec, + #[serde(default)] + pub tools: Vec, + #[serde(default)] + pub meta: Value, +} + +/// A tool call the model requested, returned inside [`InferenceResult`] (§5.1). +#[derive(Debug, Clone, Default, Serialize)] +pub struct WireToolCall { + pub id: String, + pub name: String, + pub args: Value, +} + +/// The `InferenceResult` a host returns for an `inference.invoke` (§5.1). +#[derive(Debug, Clone, Default, Serialize)] +pub struct InferenceResult { + pub content: String, + #[serde(rename = "reasoningContent")] + pub reasoning_content: Option, + pub model: String, + #[serde(rename = "toolCalls")] + pub tool_calls: Vec, + pub usage: Usage, +} + +/// Status of the supervised serve child, surfaced over the `medulla_local` +/// RPC namespace. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct MedullaLocalStatus { + /// Whether the `medulla-local` engine is selected/available. + pub enabled: bool, + /// Whether a serve child is currently connected. + pub running: bool, + /// serve version string from the `ready` banner, if connected. + pub serve_version: Option, + /// Session id negotiated in the handshake, if connected. + pub session_id: Option, + /// Active port set negotiated in `hello`. + pub ports: Vec, + /// Last error, if the supervisor is in a failed/unavailable state. + pub message: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn instruct_receipt_round_trips_camel_case() { + let receipt: InstructReceipt = serde_json::from_str( + r#"{"instructionId":"inst-agent-0","cycleId":"cyc:agent:agent:0"}"#, + ) + .unwrap(); + assert_eq!(receipt.instruction_id, "inst-agent-0"); + assert_eq!(receipt.cycle_id, "cyc:agent:agent:0"); + } + + #[test] + fn inference_call_decodes_tier_and_messages() { + let call: InferenceCall = serde_json::from_value(json!({ + "tier": "orchestrator", + "op": "orchestrate", + "cycleId": "cyc:1", + "messages": [{"role": "user", "content": "hi"}], + "meta": {"priority": "root"} + })) + .unwrap(); + assert_eq!(call.tier, "orchestrator"); + assert_eq!(call.messages.len(), 1); + assert_eq!(call.messages[0].role, "user"); + } + + #[test] + fn inference_result_serializes_camel_case() { + let result = InferenceResult { + content: "done".into(), + reasoning_content: None, + model: "orchestrator-v1".into(), + tool_calls: vec![], + usage: Usage { + input_tokens: 9, + output_tokens: 3, + }, + }; + let value = serde_json::to_value(&result).unwrap(); + assert_eq!(value["content"], "done"); + assert_eq!(value["reasoningContent"], Value::Null); + assert_eq!(value["usage"]["inputTokens"], 9); + assert_eq!(value["toolCalls"], json!([])); + } + + #[test] + fn harness_status_tolerates_sparse_payload() { + let status: HarnessStatus = + serde_json::from_str(r#"{"state":"running","queued":1}"#).unwrap(); + assert_eq!(status.state, "running"); + assert_eq!(status.queued, 1); + assert!(status.tasks.is_empty()); + } +} diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index 3f2f4c02a..1e4c11734 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -78,6 +78,8 @@ pub mod mcp_registry; pub mod mcp_server; #[cfg(feature = "media")] pub mod media_generation; +#[cfg(feature = "medulla-local")] +pub mod medulla_local; #[cfg(feature = "meet")] pub mod meet; pub mod meet_agent; diff --git a/src/openhuman/runtime_node/ops.rs b/src/openhuman/runtime_node/ops.rs index 0bc43c19c..d82091fc0 100644 --- a/src/openhuman/runtime_node/ops.rs +++ b/src/openhuman/runtime_node/ops.rs @@ -75,7 +75,7 @@ fn command_class_for_tool( } } -fn build_runtime_tools(config: &Config) -> Result>, String> { +pub fn build_runtime_tools(config: &Config) -> Result>, String> { debug!( workspace = %config.workspace_dir.display(), "[runtime_node::ops] build_runtime_tools: start" diff --git a/src/openhuman/subconscious/instance.rs b/src/openhuman/subconscious/instance.rs index 19de2aa56..e3166da0e 100644 --- a/src/openhuman/subconscious/instance.rs +++ b/src/openhuman/subconscious/instance.rs @@ -252,6 +252,17 @@ impl SubconsciousInstance { /// The tick body given a loaded config. Split out so tests can drive it with /// a crafted `Config` without hitting disk (`Config::load_or_init`). async fn run_tick(&self, config: Config) -> Result { + // Subconscious-replacement draft (plan §5.2): when + // `subconscious.engine = "medulla"`, route observe→reflect→commit + // through one local `medulla-serve` instruct instead of the local + // tinyagents graph. The default (`local`) falls through to the + // unchanged body below, so default behaviour is byte-identical — the + // branch compiles out entirely when the `medulla-local` feature is off. + #[cfg(feature = "medulla-local")] + if config.subconscious.engine.is_medulla() { + return self.run_tick_medulla(config).await; + } + let prefix = self.log_prefix(); let started = std::time::Instant::now(); let tick_at = now_secs(); @@ -338,6 +349,105 @@ impl SubconsciousInstance { }) } + /// The medulla-engine tick body (plan §5.2 draft). Observes the world the + /// same way the local graph does, then — instead of the local reflect turn + /// — enqueues ONE `instruct` on the supervised `medulla-serve` child + /// summarising the tick context, and advances the baseline via the + /// profile's own `commit`. Quiet windows short-circuit exactly like the + /// local path (no instruct, still commit). The hosted cycle runs async and + /// is observed via the serve event stream; this method only awaits the + /// synchronous receipt. + #[cfg(feature = "medulla-local")] + async fn run_tick_medulla(&self, config: Config) -> Result { + use crate::openhuman::medulla_local::ops::instruct_tick; + use serde_json::json; + + let prefix = self.log_prefix(); + let started = std::time::Instant::now(); + let tick_at = now_secs(); + let my_generation = self.tick_generation.fetch_add(1, Ordering::SeqCst) + 1; + + let observation = self.profile.observe(&config).await; + debug!( + "{prefix} medulla tick observe has_changes={} external={}", + observation.has_changes, observation.has_external_content + ); + + if !observation.has_changes { + // Superseded-generation discard (mirrors the local graph path): + // a newer tick owns the baseline now, so committing this stale + // observation would clobber it. + if self.tick_generation.load(Ordering::SeqCst) != my_generation { + info!("{prefix} medulla quiet tick superseded by newer tick, discarding"); + let mut state = self.state.lock().await; + state.total_ticks += 1; + return Ok(self.quiet_result(tick_at, started)); + } + // Quiet window: no billed cycle, but still advance the baseline so + // the next tick observes only genuinely-new content (mirrors the + // local observe→commit quiet edge). + self.profile.commit(&config, &observation).await; + let mut state = self.state.lock().await; + state.total_ticks += 1; + state.consecutive_failures = 0; + state.last_tick_at = tick_at; + persist_last_tick_at(&self.workspace_dir, self.profile.id(), tick_at); + return Ok(self.quiet_result(tick_at, started)); + } + + let message = format!( + "Subconscious wake for world `{}`. Reconcile the following changes:\n\n{}", + self.profile.id(), + observation.rendered + ); + let meta = json!({ + "origin": "wake", + "world": self.profile.id(), + "tickId": tick_at as u64, + }); + + let response_chars = match instruct_tick(&config, &message, meta).await { + Ok(receipt) => { + info!( + "{prefix} medulla instruct enqueued instruction_id={} cycle_id={}", + receipt.instruction_id, receipt.cycle_id + ); + // Superseded-generation discard (mirrors the local graph + // path): the instruction is already enqueued serve-side, but a + // newer tick owns the baseline now, so this stale observation + // must not commit or advance last_tick_at. + if self.tick_generation.load(Ordering::SeqCst) != my_generation { + info!("{prefix} medulla tick superseded by newer tick, discarding"); + let mut state = self.state.lock().await; + state.total_ticks += 1; + return Ok(self.quiet_result(tick_at, started)); + } + // Baseline advances on a successful enqueue; the cycle itself is + // observed via events (a failed cycle does not re-observe here). + self.profile.commit(&config, &observation).await; + let mut state = self.state.lock().await; + state.total_ticks += 1; + state.consecutive_failures = 0; + state.last_tick_at = tick_at; + persist_last_tick_at(&self.workspace_dir, self.profile.id(), tick_at); + observation.rendered.chars().count() + } + Err(error) => { + warn!("{prefix} medulla instruct failed: {error:#}"); + let mut state = self.state.lock().await; + state.consecutive_failures += 1; + state.total_ticks += 1; + 0 + } + }; + + Ok(TickResult { + tick_at, + duration_ms: started.elapsed().as_millis() as u64, + response_chars, + }) + } + fn quiet_result(&self, tick_at: f64, started: std::time::Instant) -> TickResult { TickResult { tick_at, diff --git a/src/openhuman/subconscious/instance_tests.rs b/src/openhuman/subconscious/instance_tests.rs index 4c23a54df..16f1ab0eb 100644 --- a/src/openhuman/subconscious/instance_tests.rs +++ b/src/openhuman/subconscious/instance_tests.rs @@ -24,6 +24,10 @@ struct FakeProfile { /// newer tick superseding the in-flight one. supersede: Mutex>>, supersede_pending: AtomicUsize, + /// Like `supersede_pending`, but fires during `observe` — the medulla + /// engine has no reflect stage, so supersession is injected at its only + /// pre-commit await point. + supersede_in_observe: AtomicUsize, } impl FakeProfile { @@ -37,6 +41,7 @@ impl FakeProfile { commit_calls: AtomicUsize::new(0), supersede: Mutex::new(None), supersede_pending: AtomicUsize::new(0), + supersede_in_observe: AtomicUsize::new(0), } } @@ -69,6 +74,11 @@ impl SubconsciousProfile for FakeProfile { } async fn observe(&self, _config: &Config) -> Observation { self.observe_calls.fetch_add(1, Ordering::SeqCst); + if self.supersede_in_observe.swap(0, Ordering::SeqCst) > 0 { + if let Some(gen) = self.supersede.lock().unwrap().as_ref() { + gen.fetch_add(1, Ordering::SeqCst); + } + } let mut obs = self.observations.lock().unwrap(); if obs.len() > 1 { obs.remove(0) @@ -341,3 +351,113 @@ async fn superseded_tick_discards_result_and_skips_commit() { "not counted a failure" ); } + +// ── Subconscious-replacement draft (plan §5.2): default path is untouched ──── + +/// The subconscious engine defaults to `local` when the `[subconscious]` block +/// (or its `engine` field) is unset. This test pins that a default-config tick +/// runs the LOCAL tinyagents graph — proven by `reflect` executing — rather +/// than the medulla instruct path (which observes → instructs → commits and +/// never calls `reflect`). It is the "byte-identical when the flag is unset" +/// guarantee for the draft: whether or not the `medulla-local` feature is +/// compiled in, an unset flag routes exactly as before. +#[tokio::test] +async fn default_engine_runs_local_graph_not_medulla() { + let dir = tempfile::tempdir().unwrap(); + let profile = Arc::new(FakeProfile::new( + FakeProfile::changed(), + Ok(Reflection::Acted { response_chars: 7 }), + )); + let instance = build(profile.clone(), dir.path()); + + let config = test_config(dir.path()); + // Precondition: the flag is unset, so the engine is the default `local`. + assert!( + !config.subconscious.engine.is_medulla(), + "default config must select the local engine" + ); + + let result = instance.run_tick_for_test(config).await.unwrap(); + + // The local graph's reflect stage ran — the medulla path never calls it. + assert_eq!( + profile.reflect_calls.load(Ordering::SeqCst), + 1, + "default engine must drive the local reflect graph" + ); + assert_eq!(profile.commit_calls.load(Ordering::SeqCst), 1); + assert_eq!(result.response_chars, 7); +} + +/// A medulla-engine config for the hermetic quiet-path tests: the quiet edge +/// commits the baseline without ever touching the serve supervisor, so no +/// child process or socket is involved. +#[cfg(feature = "medulla-local")] +fn medulla_config(workspace: &std::path::Path) -> Config { + use crate::openhuman::config::schema::SubconsciousEngine; + let mut cfg = test_config(workspace); + cfg.subconscious.engine = SubconsciousEngine::Medulla; + cfg +} + +/// The medulla quiet edge mirrors the local one: no reflect, no instruct, +/// baseline committed and `last_tick_at` advanced. +#[cfg(feature = "medulla-local")] +#[tokio::test] +async fn medulla_quiet_tick_commits_and_advances() { + let dir = tempfile::tempdir().unwrap(); + let profile = Arc::new(FakeProfile::new(FakeProfile::quiet(), Ok(Reflection::Idle))); + let instance = build(profile.clone(), dir.path()); + + let result = instance + .run_tick_for_test(medulla_config(dir.path())) + .await + .unwrap(); + + assert_eq!(profile.observe_calls.load(Ordering::SeqCst), 1); + assert_eq!( + profile.reflect_calls.load(Ordering::SeqCst), + 0, + "the medulla engine never runs the local reflect graph" + ); + assert_eq!(profile.commit_calls.load(Ordering::SeqCst), 1, "committed"); + let status = instance.status().await; + assert!(status.last_tick_at.is_some()); + assert_eq!(status.total_ticks, 1); + assert_eq!(result.response_chars, 0); +} + +/// A superseded medulla tick must not commit its stale observation — the same +/// generation guard the local graph path applies before its commit. +#[cfg(feature = "medulla-local")] +#[tokio::test] +async fn medulla_superseded_tick_skips_commit() { + let dir = tempfile::tempdir().unwrap(); + let profile = Arc::new(FakeProfile::new(FakeProfile::quiet(), Ok(Reflection::Idle))); + let instance = build(profile.clone(), dir.path()); + // Wire the profile to bump the instance's generation during observe — the + // medulla path's only pre-commit await point. + *profile.supersede.lock().unwrap() = Some(instance.generation_handle()); + profile.supersede_in_observe.store(1, Ordering::SeqCst); + + let result = instance + .run_tick_for_test(medulla_config(dir.path())) + .await + .unwrap(); + + assert_eq!(profile.observe_calls.load(Ordering::SeqCst), 1); + assert_eq!( + profile.commit_calls.load(Ordering::SeqCst), + 0, + "a superseded medulla tick must not commit" + ); + assert_eq!(result.response_chars, 0); + let status = instance.status().await; + assert!(status.last_tick_at.is_none(), "no baseline advance"); + assert_eq!(status.total_ticks, 1, "the discarded tick is still counted"); + assert_eq!( + instance.snapshot_failures().await, + 0, + "not counted a failure" + ); +} diff --git a/tests/medulla_local_e2e.rs b/tests/medulla_local_e2e.rs new file mode 100644 index 000000000..a96f77b39 --- /dev/null +++ b/tests/medulla_local_e2e.rs @@ -0,0 +1,352 @@ +//! JSON-RPC E2E coverage for the `medulla_local` namespace (Flavor A draft). +//! +//! Boots the real Axum JSON-RPC router (same pattern as +//! `domain_modules_e2e.rs`) and exercises the `medulla_local` controllers +//! hermetically: no `medulla-serve` entry is configured, so `status` must +//! report a well-formed not-running snapshot without spawning Node or touching +//! the network, and `instruct` must fail cleanly with the actionable +//! "serve entry not configured" error. Run with: +//! `cargo test --test medulla_local_e2e`. +#![cfg(feature = "medulla-local")] + +use std::net::SocketAddr; +use std::path::Path; +use std::sync::{Mutex, OnceLock}; +use std::time::Duration; + +use axum::http::header::AUTHORIZATION; +use reqwest::StatusCode; +use serde_json::{json, Value}; +use tempfile::{tempdir, TempDir}; + +use openhuman_core::core::auth::{init_rpc_token, CORE_TOKEN_ENV_VAR}; +use openhuman_core::core::jsonrpc::build_core_http_router; + +const TEST_RPC_TOKEN: &str = "medulla-local-e2e-token"; + +/// The startup-failure needle the supervisor must surface with no serve entry +/// configured: the actionable configuration error on unix, and the typed +/// unsupported-platform error on targets without unix domain sockets (where +/// the serve transport is stubbed out). +#[cfg(unix)] +const STARTUP_UNAVAILABLE_NEEDLE: &str = "serve entry not configured"; +#[cfg(not(unix))] +const STARTUP_UNAVAILABLE_NEEDLE: &str = "unavailable on this platform"; + +static AUTH_INIT: OnceLock<()> = OnceLock::new(); +static ENV_LOCK: OnceLock> = OnceLock::new(); + +struct EnvVarGuard { + key: &'static str, + old: Option, +} + +impl EnvVarGuard { + fn set_to_path(key: &'static str, path: &Path) -> Self { + let old = std::env::var(key).ok(); + std::env::set_var(key, path.as_os_str()); + Self { key, old } + } + + fn set(key: &'static str, value: &str) -> Self { + let old = std::env::var(key).ok(); + std::env::set_var(key, value); + Self { key, old } + } + + fn unset(key: &'static str) -> Self { + let old = std::env::var(key).ok(); + std::env::remove_var(key); + Self { key, old } + } +} + +impl Drop for EnvVarGuard { + fn drop(&mut self) { + match &self.old { + Some(value) => std::env::set_var(self.key, value), + None => std::env::remove_var(self.key), + } + } +} + +/// Serializes tests in this binary: `HOME` and the serve-entry env override are +/// process-global, as is the medulla supervisor cache. +fn env_lock() -> std::sync::MutexGuard<'static, ()> { + let mutex = ENV_LOCK.get_or_init(|| Mutex::new(())); + match mutex.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + } +} + +fn ensure_rpc_auth() { + AUTH_INIT.get_or_init(|| { + // SAFETY: guarded by OnceLock and set once before the router for this + // test binary is used concurrently. + unsafe { std::env::set_var(CORE_TOKEN_ENV_VAR, TEST_RPC_TOKEN) }; + let token_dir = std::env::temp_dir().join("openhuman-medulla-local-e2e-auth"); + init_rpc_token(&token_dir).expect("init rpc auth token"); + }); +} + +async fn serve_rpc() -> ( + SocketAddr, + tokio::task::JoinHandle>, +) { + ensure_rpc_auth(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind rpc listener"); + let addr = listener.local_addr().expect("rpc listener addr"); + let router = build_core_http_router(false); + let join = tokio::spawn(async move { axum::serve(listener, router).await }); + (addr, join) +} + +fn write_min_config(openhuman_dir: &Path) { + std::fs::create_dir_all(openhuman_dir).expect("create .openhuman"); + let cfg = r#"api_url = "http://127.0.0.1:9" +default_model = "e2e-model" +default_temperature = 0.2 + +[secrets] +encrypt = false + +[local_ai] +enabled = false + +[memory] +provider = "none" +embedding_provider = "none" +embedding_model = "none" +embedding_dimensions = 0 + +[memory_tree] +embedding_strict = false +"#; + std::fs::write(openhuman_dir.join("config.toml"), cfg).expect("write config.toml"); + let _: openhuman_core::openhuman::config::Config = + toml::from_str(cfg).expect("test config must match schema"); +} + +struct TestHarness { + _tmp: TempDir, + _guards: Vec, + rpc_base: String, + join: tokio::task::JoinHandle>, +} + +async fn setup() -> TestHarness { + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + write_min_config(&openhuman_home); + + let guards = vec![ + EnvVarGuard::set_to_path("HOME", home), + EnvVarGuard::unset("OPENHUMAN_WORKSPACE"), + EnvVarGuard::unset("BACKEND_URL"), + EnvVarGuard::unset("VITE_BACKEND_URL"), + EnvVarGuard::unset("OPENHUMAN_API_URL"), + // The whole point of this suite: no serve entry anywhere, so the + // supervisor must fail its spawn attempt with the actionable + // configuration error instead of probing Node or the network. + EnvVarGuard::unset("OPENHUMAN_MEDULLA_SERVE_ENTRY"), + EnvVarGuard::set("OPENHUMAN_KEYRING_BACKEND", "file"), + EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_STRICT", "false"), + EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_ENDPOINT", ""), + EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_MODEL", ""), + ]; + + let (addr, join) = serve_rpc().await; + TestHarness { + _tmp: tmp, + _guards: guards, + rpc_base: format!("http://{addr}"), + join, + } +} + +async fn rpc(rpc_base: &str, id: i64, method: &str, params: Value) -> Value { + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(30)) + .build() + .expect("client"); + let url = format!("{}/rpc", rpc_base.trim_end_matches('/')); + let response = client + .post(&url) + .header(AUTHORIZATION, format!("Bearer {TEST_RPC_TOKEN}")) + .json(&json!({ + "jsonrpc": "2.0", + "id": id, + "method": method, + "params": params, + })) + .send() + .await + .unwrap_or_else(|err| panic!("POST {url} {method}: {err}")); + assert_eq!( + response.status(), + StatusCode::OK, + "HTTP transport should accept {method}" + ); + response + .json::() + .await + .unwrap_or_else(|err| panic!("json for {method}: {err}")) +} + +fn ok<'a>(value: &'a Value, context: &str) -> &'a Value { + if let Some(error) = value.get("error") { + panic!("{context}: unexpected JSON-RPC error: {error}"); + } + value + .get("result") + .unwrap_or_else(|| panic!("{context}: missing result: {value}")) +} + +fn err<'a>(value: &'a Value, context: &str) -> &'a Value { + value + .get("error") + .unwrap_or_else(|| panic!("{context}: expected JSON-RPC error, got: {value}")) +} + +fn err_message(value: &Value, context: &str) -> String { + err(value, context) + .get("message") + .and_then(Value::as_str) + .unwrap_or_else(|| panic!("{context}: error without string message: {value}")) + .to_string() +} + +/// Unwrap the CLI-compatible envelope: handlers that attach logs return +/// `{ "result": , "logs": [...] }`; otherwise the payload is bare. +fn payload<'a>(value: &'a Value, context: &str) -> &'a Value { + let result = ok(value, context); + result.get("result").unwrap_or(result) +} + +#[tokio::test] +async fn medulla_local_schema_catalog_exposes_status_and_instruct() { + let _lock = env_lock(); + let harness = setup().await; + + let url = format!("{}/schema", harness.rpc_base.trim_end_matches('/')); + let schema = reqwest::get(&url) + .await + .unwrap_or_else(|err| panic!("GET {url}: {err}")) + .json::() + .await + .expect("schema json"); + let methods: Vec = schema + .get("methods") + .and_then(Value::as_array) + .expect("schema methods array") + .iter() + .map(|method| { + method + .get("method") + .and_then(Value::as_str) + .expect("method name") + .to_string() + }) + .collect(); + + for method in [ + "openhuman.medulla_local_status", + "openhuman.medulla_local_instruct", + ] { + assert!( + methods.iter().any(|name| name == method), + "schema catalog must expose {method}; got {} methods", + methods.len() + ); + } + + harness.join.abort(); +} + +#[tokio::test] +async fn medulla_local_status_and_instruct_round_trip_without_serve_child() { + let _lock = env_lock(); + let harness = setup().await; + + // `status` with no serve entry configured: a well-formed idle snapshot, + // never a spawn attempt against Node or the network. The failed startup is + // folded into `message` rather than surfaced as an RPC error. + let status_response = rpc( + &harness.rpc_base, + 40_001, + "openhuman.medulla_local_status", + json!({}), + ) + .await; + let status = payload(&status_response, "medulla_local_status"); + assert_eq!( + status.get("enabled").and_then(Value::as_bool), + Some(true), + "status.enabled: {status_response}" + ); + assert_eq!( + status.get("running").and_then(Value::as_bool), + Some(false), + "status.running must be false without a serve child: {status_response}" + ); + assert!( + status.get("serve_version").is_some_and(Value::is_null), + "status.serve_version must be null when not connected: {status_response}" + ); + assert!( + status.get("session_id").is_some_and(Value::is_null), + "status.session_id must be null when not connected: {status_response}" + ); + assert_eq!( + status + .get("ports") + .and_then(Value::as_array) + .map(Vec::as_slice), + Some(&[][..]), + "status.ports must be an empty array when not connected: {status_response}" + ); + let message = status + .get("message") + .and_then(Value::as_str) + .unwrap_or_default(); + assert!( + message.contains(STARTUP_UNAVAILABLE_NEEDLE), + "status.message must carry the actionable startup error: {status_response}" + ); + + // `instruct` without the required `message` param fails at deserialization, + // before any supervisor involvement. + let missing_param = rpc( + &harness.rpc_base, + 40_002, + "openhuman.medulla_local_instruct", + json!({}), + ) + .await; + let missing_message = err_message(&missing_param, "medulla_local_instruct missing message"); + assert!( + missing_message.contains("message"), + "instruct without params should name the missing `message` field: {missing_param}" + ); + + // `instruct` with a message but no configured serve entry fails cleanly + // with the actionable configuration error (no hang, no Node spawn). + let instruct_response = rpc( + &harness.rpc_base, + 40_003, + "openhuman.medulla_local_instruct", + json!({ "message": "hello from the e2e suite" }), + ) + .await; + let instruct_error = err_message(&instruct_response, "medulla_local_instruct unconfigured"); + assert!( + instruct_error.contains(STARTUP_UNAVAILABLE_NEEDLE), + "instruct must surface the actionable startup error: {instruct_response}" + ); + + harness.join.abort(); +}