diff --git a/src/embed/medulla.rs b/src/embed/medulla.rs new file mode 100644 index 000000000..6b8965630 --- /dev/null +++ b/src/embed/medulla.rs @@ -0,0 +1,199 @@ +//! Medulla sub-facade — typed access to the Medulla orchestration backend. +//! +//! Follows the shape [`super::config`] established: a borrowed newtype over the +//! runtime, and two-line methods delegating to [`call`](super::call::call). +//! +//! # Types are re-exported, not redefined +//! +//! [`MedullaStatus`], [`SessionSummary`] and [`RosterWorker`] come straight from +//! the domain rather than being mirrored here. A parallel set of facade structs +//! would be one more thing to keep in step with the wire contract, and the whole +//! point of the domain owning them is that there is a single definition. +//! +//! # Gating +//! +//! Compiled only with the `medulla` feature, like the domain it wraps. With the +//! feature off `Core::medulla()` does not exist, so a host that cannot use it +//! fails to compile against it rather than discovering an error at runtime. + +use std::sync::Arc; + +use super::call::call; +use super::error::CoreError; +use crate::core::runtime::CoreRuntime; + +pub use crate::openhuman::medulla::client::{ + AbortResult, EventEnvelope, Message, RosterWorker, SendResult, SessionCreated, SessionDetail, + SessionSummary, +}; +pub use crate::openhuman::medulla::ops::MedullaStatus; + +/// Typed access to the Medulla backend. +/// +/// Obtained from [`Core::medulla`](super::Core::medulla); never constructed +/// directly. +pub struct Medulla<'a>(pub(super) &'a Arc); + +impl Medulla<'_> { + /// Whether the integration is configured and signed in. + /// + /// Makes no network call and does not fail on an unconfigured host — the + /// result carries a `configured` flag and a stable reason instead. A host + /// polls this to decide whether to show the Medulla surface at all, so + /// "not set up" has to be a value it can render, not an error it must + /// special-case. + pub async fn status(&self) -> Result { + call(self.0, "openhuman.medulla_status", serde_json::json!({})).await + } + + /// List the operator's durable sessions. + /// + /// # Errors + /// + /// [`CoreError::Domain`] with `kind = "MedullaNoBaseUrl"` or + /// `"MedullaNoSessionToken"` when the integration is not usable, both + /// flagged `expected_user_state` so a host renders a notice rather than a + /// failure. Backend rejections carry the backend's own `errorCode` as + /// `kind`, and HTTP 401/403 are likewise `expected_user_state`. + pub async fn list_sessions(&self) -> Result, CoreError> { + call( + self.0, + "openhuman.medulla_list_sessions", + serde_json::json!({}), + ) + .await + } + + /// Create a durable session. + /// + /// `title` is optional — the backend names an untitled session itself + /// rather than the host inventing one. + pub async fn create_session(&self, title: Option<&str>) -> Result { + call( + self.0, + "openhuman.medulla_create_session", + serde_json::json!({ "title": title }), + ) + .await + } + + /// Fetch one session's state. + pub async fn get_session(&self, session_id: &str) -> Result { + call( + self.0, + "openhuman.medulla_get_session", + serde_json::json!({ "sessionId": session_id }), + ) + .await + } + + /// Send a message to a session. + /// + /// `sync = false` returns as soon as the backend accepts the turn, leaving + /// the reply to arrive over the event stream; `true` blocks until it + /// replies. A UI wants the former so it can render progress, a scripted + /// caller usually wants the latter. + pub async fn send_message( + &self, + session_id: &str, + body: &str, + sync: bool, + ) -> Result { + call( + self.0, + "openhuman.medulla_send_message", + serde_json::json!({ "sessionId": session_id, "body": body, "sync": sync }), + ) + .await + } + + /// Abort a session's running cycle. + pub async fn abort(&self, session_id: &str) -> Result { + call( + self.0, + "openhuman.medulla_abort", + serde_json::json!({ "sessionId": session_id }), + ) + .await + } + + /// Replay a session's messages after `after`. + /// + /// `after` is a cursor, not a page offset: passing the last seq already + /// seen returns only what is new, which is what makes a reconnect cheap. + pub async fn list_messages( + &self, + session_id: &str, + after: Option, + ) -> Result, CoreError> { + call( + self.0, + "openhuman.medulla_list_messages", + serde_json::json!({ "sessionId": session_id, "after": after }), + ) + .await + } + + /// Replay a session's events after `after`. + /// + /// Same cursor semantics as [`list_messages`](Self::list_messages). + pub async fn list_events( + &self, + session_id: &str, + after: Option, + ) -> Result, CoreError> { + call( + self.0, + "openhuman.medulla_list_events", + serde_json::json!({ "sessionId": session_id, "after": after }), + ) + .await + } + + /// Read the roster of workers currently connected to the backend. + /// + /// # Errors + /// + /// Same shape as [`list_sessions`](Self::list_sessions). + pub async fn roster(&self) -> Result, CoreError> { + call(self.0, "openhuman.medulla_roster", serde_json::json!({})).await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::medulla::all_medulla_registered_controllers; + + /// Every method this facade dispatches must name a registered controller. + /// + /// The facade's method names are strings; a typo or a renamed controller + /// would otherwise surface as `CoreError::Unavailable` at runtime, which is + /// indistinguishable from a domain the host gated off on purpose. Pinning + /// them here turns that into a test failure. + #[test] + fn every_dispatched_method_is_registered() { + let registered: Vec = all_medulla_registered_controllers() + .iter() + .map(|c| c.rpc_method_name()) + .collect(); + + for method in [ + "openhuman.medulla_status", + "openhuman.medulla_list_sessions", + "openhuman.medulla_create_session", + "openhuman.medulla_get_session", + "openhuman.medulla_send_message", + "openhuman.medulla_abort", + "openhuman.medulla_list_messages", + "openhuman.medulla_list_events", + "openhuman.medulla_roster", + ] { + assert!( + registered.iter().any(|m| m == method), + "facade dispatches `{method}`, which no controller registers. \ + Registered: {registered:?}" + ); + } + } +} diff --git a/src/embed/mod.rs b/src/embed/mod.rs index 45a7821e2..186ea9d57 100644 --- a/src/embed/mod.rs +++ b/src/embed/mod.rs @@ -53,9 +53,16 @@ mod call; mod config; mod error; +#[cfg(feature = "medulla")] +mod medulla; pub use config::{Config, RuntimeFlags}; pub use error::CoreError; +#[cfg(feature = "medulla")] +pub use medulla::{ + AbortResult, EventEnvelope, Medulla, MedullaStatus, Message, RosterWorker, SendResult, + SessionCreated, SessionDetail, SessionSummary, +}; use std::sync::Arc; @@ -86,6 +93,15 @@ impl Core { Config(&self.rt) } + /// Typed access to the Medulla orchestration backend. + /// + /// Absent unless the `medulla` feature is on, so a host built without it + /// fails to compile against this rather than meeting a runtime error. + #[cfg(feature = "medulla")] + pub fn medulla(&self) -> Medulla<'_> { + Medulla(&self.rt) + } + /// The underlying runtime, for anything this facade does not yet model. /// /// An escape hatch, not the intended path — every use is a candidate for a diff --git a/src/openhuman/medulla/client/program/types.rs b/src/openhuman/medulla/client/program/types.rs index 8f8c9d4f6..c876948b4 100644 --- a/src/openhuman/medulla/client/program/types.rs +++ b/src/openhuman/medulla/client/program/types.rs @@ -25,7 +25,7 @@ pub struct RosterBudget { } /// Connected worker returned by `GET /medulla/v1/roster`. -#[derive(Debug, Clone, Deserialize, Serialize)] +#[derive(Debug, Clone, Deserialize, Serialize, Default)] #[serde(rename_all = "camelCase")] pub struct RosterWorker { /// Stable worker identifier in the manager registry. diff --git a/src/openhuman/medulla/client/types/session.rs b/src/openhuman/medulla/client/types/session.rs index 9762d6ec0..a6e8725ba 100644 --- a/src/openhuman/medulla/client/types/session.rs +++ b/src/openhuman/medulla/client/types/session.rs @@ -7,13 +7,17 @@ use serde::{Deserialize, Serialize}; /// Session lifecycle status. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] #[serde(rename_all = "snake_case")] pub enum SessionStatus { Active, Idle, Archived, /// Any status not yet modelled by this client. + /// + /// Also the `Default`: a value constructed rather than decoded has not + /// declared a status, and defaulting to `Active` would assert one. + #[default] #[serde(other)] Other, } @@ -36,7 +40,7 @@ pub struct SessionCreated { } /// Item in the session list (`GET /medulla/v1/sessions`). -#[derive(Debug, Clone, Deserialize, Serialize)] +#[derive(Debug, Clone, Deserialize, Serialize, Default)] #[serde(rename_all = "camelCase")] pub struct SessionSummary { pub session_id: String, diff --git a/src/openhuman/medulla/ops.rs b/src/openhuman/medulla/ops.rs index 349ffbd78..11108cde6 100644 --- a/src/openhuman/medulla/ops.rs +++ b/src/openhuman/medulla/ops.rs @@ -5,16 +5,24 @@ //! failures become [`StructuredRpcError`]s so a host can branch on a stable //! `data.kind` instead of matching on message text. -use serde::Serialize; +use serde::{Deserialize, Serialize}; use crate::openhuman::config::Config; use crate::rpc::{RpcOutcome, StructuredRpcError}; -use super::client::{ClientError, MedullaClient, RosterWorker, SessionSummary}; +use super::client::{ + AbortResult, ClientError, EventEnvelope, MedullaClient, Message, RosterWorker, SendResult, + SessionCreated, SessionDetail, SessionSummary, +}; use super::resolve::{self, NotConfigured}; /// Whether the Medulla integration is usable, and why not when it isn't. -#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +/// +/// `Deserialize` as well as `Serialize` because this type round-trips: ops +/// serializes it onto the RPC boundary and the embed facade deserializes it +/// back on the other side. An output-only derive compiles until the facade +/// tries to read it. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "camelCase")] pub struct MedullaStatus { /// True when a base URL and a session token are both available. @@ -67,6 +75,107 @@ pub async fn list_sessions(config: &Config) -> Result, +) -> Result, String> { + let client = resolved(config)?; + let created = call(client.create_session(title).await, "medulla_create_session")?; + log::debug!("[medulla] create_session id={}", created.session_id); + Ok(RpcOutcome::new(created, Vec::new())) +} + +/// Fetch one session's state. +pub async fn get_session( + config: &Config, + session_id: &str, +) -> Result, String> { + let client = resolved(config)?; + let detail = call(client.get_session(session_id).await, "medulla_get_session")?; + Ok(RpcOutcome::new(detail, Vec::new())) +} + +/// Send a message to a session. +/// +/// `sync = false` returns as soon as the backend accepts the turn; `true` +/// blocks until it replies. The caller chooses, because a TUI wants the former +/// (so it can render streaming progress) while a scripted client wants the +/// latter. +pub async fn send_message( + config: &Config, + session_id: &str, + body: &str, + sync: bool, +) -> Result, String> { + let client = resolved(config)?; + let result = call( + client.send_message(session_id, body, sync).await, + "medulla_send_message", + )?; + log::debug!( + "[medulla] send_message session={session_id} sync={sync} cycle={} seq={}", + result.cycle_id, + result.seq + ); + Ok(RpcOutcome::new(result, Vec::new())) +} + +/// Abort a session's running cycle. +pub async fn abort(config: &Config, session_id: &str) -> Result, String> { + let client = resolved(config)?; + let result = call(client.abort(session_id).await, "medulla_abort")?; + log::debug!( + "[medulla] abort session={session_id} aborted={}", + result.aborted + ); + Ok(RpcOutcome::new(result, Vec::new())) +} + +/// Replay a session's messages after `after`. +/// +/// `after` is a cursor, not a page offset: passing the last seq already seen +/// returns only what is new, which is what makes a reconnect cheap. +pub async fn list_messages( + config: &Config, + session_id: &str, + after: Option, +) -> Result>, String> { + let client = resolved(config)?; + let messages = call( + client.list_messages(session_id, after).await, + "medulla_list_messages", + )?; + log::debug!( + "[medulla] list_messages session={session_id} after={after:?} count={}", + messages.len() + ); + Ok(RpcOutcome::new(messages, Vec::new())) +} + +/// Replay a session's events after `after`. +/// +/// Same cursor semantics as [`list_messages`]. +pub async fn list_events( + config: &Config, + session_id: &str, + after: Option, +) -> Result>, String> { + let client = resolved(config)?; + let events = call( + client.list_events(session_id, after).await, + "medulla_list_events", + )?; + log::debug!( + "[medulla] list_events session={session_id} after={after:?} count={}", + events.len() + ); + Ok(RpcOutcome::new(events, Vec::new())) +} + /// Read the connected worker roster. pub async fn roster(config: &Config) -> Result>, String> { let client = resolved(config)?; diff --git a/src/openhuman/medulla/schemas.rs b/src/openhuman/medulla/schemas.rs index 9c1b5eaae..e3827c596 100644 --- a/src/openhuman/medulla/schemas.rs +++ b/src/openhuman/medulla/schemas.rs @@ -5,6 +5,8 @@ //! the single site in `src/core/all.rs`, so a host that switches the family off //! sees these methods as unknown rather than as failing. +use serde::de::DeserializeOwned; +use serde::Deserialize; use serde_json::{Map, Value}; use crate::core::all::{ControllerFuture, RegisteredController}; @@ -12,11 +14,50 @@ use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; use super::ops; +#[derive(Debug, Deserialize, Default)] +#[serde(rename_all = "camelCase")] +struct CreateSessionParams { + #[serde(default)] + title: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct SessionIdParams { + session_id: String, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct SendMessageParams { + session_id: String, + body: String, + /// Block until the backend replies. Defaults to false so a caller that + /// omits it gets the non-blocking behaviour a UI wants. + #[serde(default)] + sync: bool, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct ReplayParams { + session_id: String, + /// Replay cursor: the last seq already seen. Absent replays from the start. + #[serde(default)] + after: Option, +} + /// Every schema in the namespace, for `/schema` introspection. pub fn all_medulla_controller_schemas() -> Vec { vec![ medulla_schemas("medulla_status"), medulla_schemas("medulla_list_sessions"), + medulla_schemas("medulla_create_session"), + medulla_schemas("medulla_get_session"), + medulla_schemas("medulla_send_message"), + medulla_schemas("medulla_abort"), + medulla_schemas("medulla_list_messages"), + medulla_schemas("medulla_list_events"), medulla_schemas("medulla_roster"), ] } @@ -32,6 +73,30 @@ pub fn all_medulla_registered_controllers() -> Vec { schema: medulla_schemas("medulla_list_sessions"), handler: handle_list_sessions, }, + RegisteredController { + schema: medulla_schemas("medulla_create_session"), + handler: handle_create_session, + }, + RegisteredController { + schema: medulla_schemas("medulla_get_session"), + handler: handle_get_session, + }, + RegisteredController { + schema: medulla_schemas("medulla_send_message"), + handler: handle_send_message, + }, + RegisteredController { + schema: medulla_schemas("medulla_abort"), + handler: handle_abort, + }, + RegisteredController { + schema: medulla_schemas("medulla_list_messages"), + handler: handle_list_messages, + }, + RegisteredController { + schema: medulla_schemas("medulla_list_events"), + handler: handle_list_events, + }, RegisteredController { schema: medulla_schemas("medulla_roster"), handler: handle_roster, @@ -66,6 +131,97 @@ pub fn medulla_schemas(function: &str) -> ControllerSchema { required: true, }], }, + "medulla_create_session" => ControllerSchema { + namespace: "medulla", + function: "create_session", + description: "Create a durable Medulla session.", + inputs: vec![FieldSchema { + name: "title", + ty: TypeSchema::Option(Box::new(TypeSchema::String)), + comment: "Optional title. Omitted lets the backend name the session.", + required: false, + }], + outputs: vec![FieldSchema { + name: "session", + ty: TypeSchema::Json, + comment: "The created session's identifier.", + required: true, + }], + }, + "medulla_get_session" => ControllerSchema { + namespace: "medulla", + function: "get_session", + description: "Fetch one session's current state.", + inputs: vec![session_id_input()], + outputs: vec![FieldSchema { + name: "session", + ty: TypeSchema::Json, + comment: "Full session detail.", + required: true, + }], + }, + "medulla_send_message" => ControllerSchema { + namespace: "medulla", + function: "send_message", + description: "Send a message to a session, optionally blocking until it replies.", + inputs: vec![ + session_id_input(), + FieldSchema { + name: "body", + ty: TypeSchema::String, + comment: "Message text.", + required: true, + }, + FieldSchema { + name: "sync", + ty: TypeSchema::Bool, + comment: "Block until the backend replies. Defaults to false.", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "Cycle id and sequence, plus the reply when sync was set.", + required: true, + }], + }, + "medulla_abort" => ControllerSchema { + namespace: "medulla", + function: "abort", + description: "Abort a session's running cycle.", + inputs: vec![session_id_input()], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "Whether a cycle was actually aborted.", + required: true, + }], + }, + "medulla_list_messages" => ControllerSchema { + namespace: "medulla", + function: "list_messages", + description: "Replay a session's messages after a sequence cursor.", + inputs: vec![session_id_input(), after_input()], + outputs: vec![FieldSchema { + name: "messages", + ty: TypeSchema::Array(Box::new(TypeSchema::Json)), + comment: "Messages newer than the cursor.", + required: true, + }], + }, + "medulla_list_events" => ControllerSchema { + namespace: "medulla", + function: "list_events", + description: "Replay a session's events after a sequence cursor.", + inputs: vec![session_id_input(), after_input()], + outputs: vec![FieldSchema { + name: "events", + ty: TypeSchema::Array(Box::new(TypeSchema::Json)), + comment: "Sequenced events newer than the cursor.", + required: true, + }], + }, "medulla_roster" => ControllerSchema { namespace: "medulla", function: "roster", @@ -82,6 +238,26 @@ pub fn medulla_schemas(function: &str) -> ControllerSchema { } } +/// The session identifier every per-session method takes. +fn session_id_input() -> FieldSchema { + FieldSchema { + name: "sessionId", + ty: TypeSchema::String, + comment: "Target session identifier.", + required: true, + } +} + +/// The replay cursor shared by the two list methods. +fn after_input() -> FieldSchema { + FieldSchema { + name: "after", + ty: TypeSchema::Option(Box::new(TypeSchema::I64)), + comment: "Last sequence already seen. Absent replays from the start.", + required: false, + } +} + fn handle_status(_params: Map) -> ControllerFuture { Box::pin(async move { to_json(ops::status(&load_config().await?).await?) }) } @@ -90,6 +266,48 @@ fn handle_list_sessions(_params: Map) -> ControllerFuture { Box::pin(async move { to_json(ops::list_sessions(&load_config().await?).await?) }) } +fn handle_create_session(params: Map) -> ControllerFuture { + Box::pin(async move { + let p: CreateSessionParams = deserialize_params(params)?; + to_json(ops::create_session(&load_config().await?, p.title.as_deref()).await?) + }) +} + +fn handle_get_session(params: Map) -> ControllerFuture { + Box::pin(async move { + let p: SessionIdParams = deserialize_params(params)?; + to_json(ops::get_session(&load_config().await?, &p.session_id).await?) + }) +} + +fn handle_send_message(params: Map) -> ControllerFuture { + Box::pin(async move { + let p: SendMessageParams = deserialize_params(params)?; + to_json(ops::send_message(&load_config().await?, &p.session_id, &p.body, p.sync).await?) + }) +} + +fn handle_abort(params: Map) -> ControllerFuture { + Box::pin(async move { + let p: SessionIdParams = deserialize_params(params)?; + to_json(ops::abort(&load_config().await?, &p.session_id).await?) + }) +} + +fn handle_list_messages(params: Map) -> ControllerFuture { + Box::pin(async move { + let p: ReplayParams = deserialize_params(params)?; + to_json(ops::list_messages(&load_config().await?, &p.session_id, p.after).await?) + }) +} + +fn handle_list_events(params: Map) -> ControllerFuture { + Box::pin(async move { + let p: ReplayParams = deserialize_params(params)?; + to_json(ops::list_events(&load_config().await?, &p.session_id, p.after).await?) + }) +} + fn handle_roster(_params: Map) -> ControllerFuture { Box::pin(async move { to_json(ops::roster(&load_config().await?).await?) }) } @@ -99,6 +317,11 @@ async fn load_config() -> Result { crate::openhuman::config::ops::load_config_with_timeout().await } +/// Decode a controller's params into its typed shape. +fn deserialize_params(params: Map) -> Result { + serde_json::from_value(Value::Object(params)).map_err(|e| format!("invalid params: {e}")) +} + /// Serialize an outcome through the shared CLI-compatible envelope. fn to_json(outcome: crate::rpc::RpcOutcome) -> Result { outcome.into_cli_compatible_json() @@ -142,6 +365,12 @@ mod tests { vec![ "openhuman.medulla_status", "openhuman.medulla_list_sessions", + "openhuman.medulla_create_session", + "openhuman.medulla_get_session", + "openhuman.medulla_send_message", + "openhuman.medulla_abort", + "openhuman.medulla_list_messages", + "openhuman.medulla_list_events", "openhuman.medulla_roster", ] );