diff --git a/src/openhuman/socket/event_handlers.rs b/src/openhuman/socket/event_handlers.rs index 35dbd96b5..4595516f0 100644 --- a/src/openhuman/socket/event_handlers.rs +++ b/src/openhuman/socket/event_handlers.rs @@ -69,6 +69,10 @@ pub(super) fn handle_sio_event( "orch:register_tools", crate::openhuman::orchestration::effect_executor::device_tool_manifest(), ); + // Advertise this core's agent roster to the backend so a medulla + // operator can delegate `medulla:task_run` to a named agent. The + // backend clears the roster on socket disconnect. + super::medulla::emit_register_agents(); } "error" => { log::error!("[socket] Server error event: {}", data); @@ -479,6 +483,42 @@ pub(super) fn handle_sio_event( }); } + // ── Medulla harness plane ──────────────────────────────────────── + // A medulla operator (running in the backend) drives an openhuman agent + // session as a delegated sub-agent. See `socket::medulla`. + "medulla:task_run" => { + match serde_json::from_value::(data) { + Ok(run) => { + log::info!( + "[socket] medulla:task_run task_id={} cycle_id={} agent_id={:?}", + run.task_id, + run.cycle_id, + run.agent_id + ); + super::medulla::manager().start_task(run); + } + Err(e) => log::warn!("[socket] failed to parse medulla:task_run: {e}"), + } + } + "medulla:task_send" => { + match serde_json::from_value::(data) { + Ok(send) => { + log::info!("[socket] medulla:task_send task_id={}", send.task_id); + super::medulla::manager().steer_task(send); + } + Err(e) => log::warn!("[socket] failed to parse medulla:task_send: {e}"), + } + } + "medulla:task_abort" => { + match serde_json::from_value::(data) { + Ok(abort) => { + log::info!("[socket] medulla:task_abort task_id={}", abort.task_id); + super::medulla::manager().abort_task(abort); + } + Err(e) => log::warn!("[socket] failed to parse medulla:task_abort: {e}"), + } + } + // Channel inbound message — publish to event bus for ChannelInboundSubscriber _ if event_name.ends_with(":message") => { log::info!( diff --git a/src/openhuman/socket/medulla/envelope.rs b/src/openhuman/socket/medulla/envelope.rs new file mode 100644 index 000000000..067147d1f --- /dev/null +++ b/src/openhuman/socket/medulla/envelope.rs @@ -0,0 +1,262 @@ +//! Maps the agent's per-turn [`AgentProgress`] stream onto the +//! `tinyplace.harness.session.v2` envelope kinds openhuman already emits +//! through its tinyplace wrapper. +//! +//! We deliberately reuse the tinyplace v2 types (`SessionEnvelopeV2`, +//! `HarnessEvent`, `HarnessEventKind`, and the per-kind payload structs) rather +//! than re-deriving the envelope schema — the produced envelopes are wire- +//! identical to the ones a wrapped CLI session forwards, so a medulla operator +//! consumes one stream shape regardless of how the session was driven. + +use tinyplace::types::{ + ApprovalRequestPayload, HarnessEvent, HarnessEventKind, HarnessScope, SessionEnvelopeV2, + StatusPayload, TextPayload, ToolCallPayload, ToolResultPayload, SESSION_ENVELOPE_VERSION_V2, +}; + +use crate::openhuman::agent::progress::AgentProgress; + +/// `role` stamped on every openhuman-produced event: these are agent-side +/// stream frames (`owner` is reserved for `user_prompt`, which the agent never +/// emits about itself). +const AGENT_ROLE: &str = "agent"; + +/// Translate one [`AgentProgress`] event into a typed v2 event kind. +/// +/// Returns `None` for progress variants that carry no user-facing stream frame +/// (cost rollups, per-call token accounting, arg-delta fragments, …) so the +/// forwarded stream stays close to the `agent_message / agent_thinking / +/// tool_call / tool_result / status / approval_request / error` vocabulary the +/// spec enumerates. +pub fn progress_to_event_kind(progress: &AgentProgress) -> Option { + let kind = match progress { + AgentProgress::TurnStarted => HarnessEventKind::Status(StatusPayload { + state: "running".to_string(), + detail: "turn started".to_string(), + active_call_id: None, + }), + AgentProgress::IterationStarted { + iteration, + max_iterations, + } => HarnessEventKind::Status(StatusPayload { + state: "running".to_string(), + detail: format!("iteration {iteration}/{max_iterations}"), + active_call_id: None, + }), + AgentProgress::TextDelta { delta, .. } => HarnessEventKind::AgentMessage(TextPayload { + text: delta.clone(), + }), + AgentProgress::ThinkingDelta { delta, .. } => { + HarnessEventKind::AgentThinking(TextPayload { + text: delta.clone(), + }) + } + AgentProgress::ToolCallStarted { + call_id, + tool_name, + arguments, + display_label, + .. + } => HarnessEventKind::ToolCall(ToolCallPayload { + call_id: call_id.clone(), + tool_name: tool_name.clone(), + tool_kind: "other".to_string(), + display: display_label.clone().unwrap_or_else(|| tool_name.clone()), + input: arguments.clone(), + }), + AgentProgress::ToolCallCompleted { + call_id, + success, + output, + output_chars, + .. + } => HarnessEventKind::ToolResult(ToolResultPayload { + call_id: call_id.clone(), + ok: *success, + exit_code: None, + is_error: !*success, + output: output.clone(), + output_bytes: *output_chars as i64, + }), + AgentProgress::SubagentAwaitingUser { + task_id, question, .. + } => HarnessEventKind::ApprovalRequest(ApprovalRequestPayload { + call_id: Some(task_id.clone()), + tool_name: "subagent".to_string(), + display: question.clone(), + reason: None, + }), + AgentProgress::TurnCompleted { .. } => HarnessEventKind::Status(StatusPayload { + state: "idle".to_string(), + detail: "turn completed".to_string(), + active_call_id: None, + }), + // Everything else (arg deltas, cost/usage rollups, per-call model + // accounting, subagent-internal frames, task-board writes, raw + // TurnContent) carries no distinct stream frame in this vocabulary. + _ => return None, + }; + Some(kind) +} + +/// Wrap a typed [`HarnessEventKind`] in a full [`SessionEnvelopeV2`] anchored to +/// `session_id`, ready to serialize into a `medulla:task_envelope` frame. +/// +/// `seq` is the monotonic per-session ordering counter; `ts` is an ISO-8601 +/// timestamp. +pub fn envelope_for_kind(session_id: &str, seq: i64, kind: &HarnessEventKind) -> SessionEnvelopeV2 { + // `HarnessEventKind` is adjacently tagged (`{ "kind": .., "payload": .. }`), + // so serializing it yields exactly the `kind`/`payload` pair `HarnessEvent` + // stores — extract them rather than hand-writing the discriminator strings. + let tagged = serde_json::to_value(kind).unwrap_or_default(); + let kind_str = tagged + .get("kind") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + let payload = tagged + .get("payload") + .cloned() + .unwrap_or(serde_json::Value::Null); + + SessionEnvelopeV2 { + envelope_version: SESSION_ENVELOPE_VERSION_V2.to_string(), + version: 2, + scope: HarnessScope { + scope_type: "session".to_string(), + wrapper_session_id: session_id.to_string(), + harness_session_id: session_id.to_string(), + ..Default::default() + }, + event: HarnessEvent { + id: format!("{session_id}-{seq}"), + seq, + ts: chrono::Utc::now().to_rfc3339(), + role: AGENT_ROLE.to_string(), + kind: kind_str, + payload, + ..Default::default() + }, + ..Default::default() + } +} + +/// Convenience: build a bare `status` envelope (used to bookend a task run). +pub fn status_envelope(session_id: &str, seq: i64, state: &str, detail: &str) -> SessionEnvelopeV2 { + envelope_for_kind( + session_id, + seq, + &HarnessEventKind::Status(StatusPayload { + state: state.to_string(), + detail: detail.to_string(), + active_call_id: None, + }), + ) +} + +/// Convenience: build an `error` envelope. +pub fn error_envelope(session_id: &str, seq: i64, message: &str, fatal: bool) -> SessionEnvelopeV2 { + envelope_for_kind( + session_id, + seq, + &HarnessEventKind::Error(tinyplace::types::ErrorPayload { + message: message.to_string(), + fatal, + }), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn maps_text_delta_to_agent_message() { + let p = AgentProgress::TextDelta { + delta: "hello".to_string(), + iteration: 1, + }; + let kind = progress_to_event_kind(&p).expect("mapped"); + assert!(matches!(kind, HarnessEventKind::AgentMessage(ref t) if t.text == "hello")); + } + + #[test] + fn maps_thinking_delta_to_agent_thinking() { + let p = AgentProgress::ThinkingDelta { + delta: "hmm".to_string(), + iteration: 1, + }; + assert!(matches!( + progress_to_event_kind(&p), + Some(HarnessEventKind::AgentThinking(_)) + )); + } + + #[test] + fn drops_non_stream_variants() { + let p = AgentProgress::TurnContent { + input: Some("q".to_string()), + output: Some("a".to_string()), + }; + assert!(progress_to_event_kind(&p).is_none()); + } + + #[test] + fn envelope_is_valid_v2_and_round_trips_through_the_wire() { + let kind = HarnessEventKind::AgentMessage(TextPayload { + text: "done".to_string(), + }); + let env = envelope_for_kind("sess-1", 7, &kind); + assert!(env.is_valid_v2()); + assert_eq!(env.event.kind, "agent_message"); + assert_eq!(env.event.seq, 7); + + // Serialize → parse back through the tinyplace v2 decoder and confirm + // the typed kind survives the round-trip. + let wire = serde_json::to_string(&env).unwrap(); + let parsed = SessionEnvelopeV2::parse(&wire).expect("valid v2 wire"); + match parsed.event.decoded() { + HarnessEventKind::AgentMessage(t) => assert_eq!(t.text, "done"), + other => panic!("unexpected decoded kind: {other:?}"), + } + } + + #[test] + fn tool_call_and_result_map_to_their_kinds() { + let started = AgentProgress::ToolCallStarted { + call_id: "c1".to_string(), + tool_name: "Bash".to_string(), + arguments: serde_json::json!({ "cmd": "ls" }), + iteration: 1, + display_label: Some("List files".to_string()), + display_detail: None, + }; + match progress_to_event_kind(&started) { + Some(HarnessEventKind::ToolCall(tc)) => { + assert_eq!(tc.call_id, "c1"); + assert_eq!(tc.tool_name, "Bash"); + assert_eq!(tc.display, "List files"); + } + other => panic!("expected ToolCall, got {other:?}"), + } + + let completed = AgentProgress::ToolCallCompleted { + call_id: "c1".to_string(), + tool_name: "Bash".to_string(), + success: true, + output_chars: 3, + output: "ok\n".to_string(), + arguments: None, + elapsed_ms: 5, + iteration: 1, + failure: None, + }; + match progress_to_event_kind(&completed) { + Some(HarnessEventKind::ToolResult(tr)) => { + assert!(tr.ok); + assert!(!tr.is_error); + assert_eq!(tr.output, "ok\n"); + } + other => panic!("expected ToolResult, got {other:?}"), + } + } +} diff --git a/src/openhuman/socket/medulla/mod.rs b/src/openhuman/socket/medulla/mod.rs new file mode 100644 index 000000000..1a961e4c4 --- /dev/null +++ b/src/openhuman/socket/medulla/mod.rs @@ -0,0 +1,585 @@ +//! Medulla "harness plane" — binds the backend's `medulla:task_*` Socket.IO +//! protocol to an OpenHuman agent session so a medulla operator (running in the +//! backend) can drive an openhuman agent as a delegated sub-agent. +//! +//! This rides the *existing* authenticated backend socket owned by +//! [`crate::openhuman::socket::SocketManager`] — the transport, handshake auth +//! (`socket.handshake.auth.token`), and reconnection are already handled there, +//! so this module only adds the task/envelope binding on top: +//! +//! Down (backend → openhuman), handled in [`crate::openhuman::socket::event_handlers`]: +//! - `medulla:task_run` → [`MedullaTaskManager::start_task`] +//! - `medulla:task_send` → [`MedullaTaskManager::steer_task`] +//! - `medulla:task_abort` → [`MedullaTaskManager::abort_task`] +//! +//! Up (openhuman → backend): +//! - `medulla:task_envelope` — the live session stream, as +//! `tinyplace.harness.session.v2` envelopes (see [`envelope`]). +//! - `medulla:task_result` — explicit completion. +//! - `medulla:register_agents` — roster advertised on connect +//! ([`emit_register_agents`]); the backend clears it on disconnect. + +pub mod envelope; +pub mod payloads; + +use std::collections::HashMap; +use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::{Arc, OnceLock}; + +use parking_lot::Mutex; +use tokio::sync::mpsc; +use tokio::time::{Duration, Instant}; +use tokio_util::sync::CancellationToken; + +use crate::openhuman::agent::progress::AgentProgress; +use crate::openhuman::agent::turn_origin::{with_origin, AgentTurnOrigin}; +use crate::openhuman::agent::Agent; + +use payloads::{ + AgentDescriptor, RegisterAgents, TaskResult, EVENT_REGISTER_AGENTS, EVENT_TASK_ENVELOPE, + EVENT_TASK_RESULT, +}; + +/// Default agent an unspecified `medulla:task_run` runs as. +const DEFAULT_AGENT_ID: &str = "orchestrator"; + +/// How long we wait, after a turn settles, for a `medulla:task_send` follow-up +/// that arrived *during* the turn to be drained before declaring the task done. +/// Steering is inherently best-effort; this only catches input already queued. +const STEER_DRAIN_GRACE: Duration = Duration::from_millis(50); + +// ───────────────────────────────────────────────────────────────────────────── +// Global manager +// ───────────────────────────────────────────────────────────────────────────── + +static GLOBAL: OnceLock> = OnceLock::new(); + +/// The process-wide medulla task manager (lazily created). +pub fn manager() -> &'static Arc { + GLOBAL.get_or_init(|| Arc::new(MedullaTaskManager::new())) +} + +/// One in-flight task: a cooperative abort signal and a steering input channel. +struct RunningTask { + /// Fired by `medulla:task_abort` to cancel the task. A + /// [`CancellationToken`] *latches*: an abort that arrives before the run + /// path starts awaiting it (e.g. while the driver is still building the + /// agent) is still observed when the turn checks it, so no cancellation is + /// lost. + abort: CancellationToken, + /// Mid-task steering input (`medulla:task_send`) delivered as follow-up + /// turns on the same agent session. + steer_tx: mpsc::UnboundedSender, +} + +/// Tracks the openhuman side of every medulla-driven task. +pub struct MedullaTaskManager { + tasks: Mutex>, +} + +impl Default for MedullaTaskManager { + fn default() -> Self { + Self::new() + } +} + +impl MedullaTaskManager { + pub fn new() -> Self { + Self { + tasks: Mutex::new(HashMap::new()), + } + } + + /// Handle `medulla:task_run`: register the task and spawn its driver. + pub fn start_task(self: &Arc, run: payloads::TaskRun) { + let task_id = run.task_id.clone(); + if self.tasks.lock().contains_key(&task_id) { + log::warn!("[medulla] task_run for already-running task_id={task_id} — ignoring"); + return; + } + + let abort = CancellationToken::new(); + let (steer_tx, steer_rx) = mpsc::unbounded_channel::(); + self.tasks.lock().insert( + task_id.clone(), + RunningTask { + abort: abort.clone(), + steer_tx, + }, + ); + + let manager = Arc::clone(self); + tokio::spawn(async move { + manager.drive(run, abort, steer_rx).await; + }); + } + + /// Handle `medulla:task_send`: deliver steering input into the session. + pub fn steer_task(&self, send: payloads::TaskSend) { + match self.tasks.lock().get(&send.task_id) { + Some(task) => { + if task.steer_tx.send(send.input).is_err() { + log::warn!( + "[medulla] task_send for task_id={} whose driver has exited", + send.task_id + ); + } + } + None => log::warn!( + "[medulla] task_send for unknown task_id={} — dropping", + send.task_id + ), + } + } + + /// Handle `medulla:task_abort`: cancel the in-flight turn. + pub fn abort_task(&self, abort: payloads::TaskAbort) { + match self.tasks.lock().get(&abort.task_id) { + Some(task) => { + log::info!("[medulla] aborting task_id={}", abort.task_id); + task.abort.cancel(); + } + None => log::warn!( + "[medulla] task_abort for unknown task_id={} — dropping", + abort.task_id + ), + } + } + + /// Abort every in-flight task (used when the backend socket drops). + pub fn abort_all(&self) { + let tasks = self.tasks.lock(); + for (task_id, task) in tasks.iter() { + log::debug!("[medulla] socket down — aborting task_id={task_id}"); + task.abort.cancel(); + } + } + + fn finish(&self, task_id: &str) { + self.tasks.lock().remove(task_id); + } + + /// Drive a task to completion: build/resume an agent session, run the + /// instruction (plus any queued steering follow-ups) as turns, stream the + /// progress as `medulla:task_envelope` frames, and emit a terminal + /// `medulla:task_result`. + async fn drive( + &self, + run: payloads::TaskRun, + abort: CancellationToken, + mut steer_rx: mpsc::UnboundedReceiver, + ) { + let task_id = run.task_id.clone(); + // Session key: reuse the caller-supplied session id when resuming, else + // fall back to the task id so the envelope stream is still anchored. + let session_id = run.session_id.clone().unwrap_or_else(|| task_id.clone()); + let agent_id = run + .agent_id + .clone() + .unwrap_or_else(|| DEFAULT_AGENT_ID.to_string()); + let seq = Arc::new(AtomicI64::new(0)); + + let mut agent = match build_agent(&agent_id, &task_id, &session_id).await { + Ok(agent) => agent, + Err(err) => { + log::error!("[medulla] task_id={task_id} failed to build agent: {err}"); + emit_envelope( + &task_id, + envelope::error_envelope(&session_id, next_seq(&seq), &err, true), + ); + emit_result(TaskResult { + task_id: task_id.clone(), + ok: false, + reply: String::new(), + usage: None, + error: Some(err), + }); + self.finish(&task_id); + return; + } + }; + + // `timeout_ms` is a hard wall-clock budget for the WHOLE task, not + // per turn: anchor a single deadline now and charge every turn + // (initial + steering follow-ups) against the time remaining until it. + let deadline = + (run.timeout_ms > 0).then(|| Instant::now() + Duration::from_millis(run.timeout_ms)); + let mut next_input = run.instruction.clone(); + let result; + + 'outer: loop { + // Charge this turn against the remaining task budget; if it's + // already spent, settle the whole task as timed out. + let remaining = match remaining_budget(deadline, Instant::now()) { + Ok(remaining) => remaining, + Err(()) => { + result = timeout_result(&task_id, &session_id, &seq); + break 'outer; + } + }; + + let (progress_tx, progress_rx) = mpsc::channel::(256); + agent.set_on_progress(Some(progress_tx)); + let forwarder = spawn_forwarder( + task_id.clone(), + session_id.clone(), + Arc::clone(&seq), + progress_rx, + ); + + let origin = AgentTurnOrigin::ExternalChannel { + channel: "medulla_harness".to_string(), + sender: None, + reply_target: task_id.clone(), + message_id: uuid::Uuid::new_v4().to_string(), + }; + let turn = Box::pin(with_origin(origin, agent.run_single(&next_input))); + + let turn_result = run_with_optional_timeout(remaining, &abort, turn).await; + // The forwarder ends when `progress_tx` drops; make sure it's flushed. + agent.set_on_progress(None); + let _ = forwarder.await; + + match turn_result { + TurnOutcome::Aborted => { + result = TaskResult { + task_id: task_id.clone(), + ok: false, + reply: String::new(), + usage: None, + error: Some("aborted".to_string()), + }; + break 'outer; + } + TurnOutcome::TimedOut => { + result = timeout_result(&task_id, &session_id, &seq); + break 'outer; + } + TurnOutcome::Errored(err) => { + emit_envelope( + &task_id, + envelope::error_envelope(&session_id, next_seq(&seq), &err, true), + ); + result = TaskResult { + task_id: task_id.clone(), + ok: false, + reply: String::new(), + usage: None, + error: Some(err), + }; + break 'outer; + } + TurnOutcome::Completed(reply) => { + // Drain any steering input that arrived during the turn and, + // if present, run it as a follow-up turn on the same session. + match drain_steer(&mut steer_rx).await { + Some(next) => { + next_input = next; + continue 'outer; + } + None => { + let usage = agent.take_last_turn_usage_totals().map(usage_to_json); + result = TaskResult { + task_id: task_id.clone(), + ok: true, + reply, + usage, + error: None, + }; + break 'outer; + } + } + } + } + } + + emit_result(result); + self.finish(&task_id); + } +} + +/// Outcome of a single driven turn. +enum TurnOutcome { + Completed(String), + Errored(String), + Aborted, + TimedOut, +} + +/// Race the agent turn against the cooperative abort signal and the remaining +/// slice of the task's wall-clock budget. +/// +/// `abort` is a [`CancellationToken`], so a cancellation that landed *before* +/// this call starts polling is not lost: `cancelled()` resolves immediately for +/// an already-cancelled token, and the `biased` select settles the turn as +/// [`TurnOutcome::Aborted`] before the agent future is ever polled. +async fn run_with_optional_timeout( + remaining: Option, + abort: &CancellationToken, + turn: std::pin::Pin>>>, +) -> TurnOutcome { + let run = async { + tokio::select! { + biased; + _ = abort.cancelled() => TurnOutcome::Aborted, + res = turn => match res { + Ok(reply) => TurnOutcome::Completed(reply), + Err(err) => TurnOutcome::Errored(err.to_string()), + }, + } + }; + + match remaining { + Some(d) => match tokio::time::timeout(d, run).await { + Ok(outcome) => outcome, + Err(_) => TurnOutcome::TimedOut, + }, + None => run.await, + } +} + +/// Remaining wall-clock budget until the task `deadline`. +/// +/// `Ok(None)` = no deadline configured (run unbounded); `Ok(Some(d))` = `d` +/// left before the deadline; `Err(())` = the deadline has already passed, so +/// the caller must settle the task as timed out instead of starting a turn. +fn remaining_budget(deadline: Option, now: Instant) -> Result, ()> { + match deadline { + Some(d) if now >= d => Err(()), + Some(d) => Ok(Some(d - now)), + None => Ok(None), + } +} + +/// Build the terminal timeout [`TaskResult`], emitting the fatal `error` +/// envelope that bookends a timed-out task's stream. +fn timeout_result(task_id: &str, session_id: &str, seq: &AtomicI64) -> TaskResult { + emit_envelope( + task_id, + envelope::error_envelope(session_id, next_seq(seq), "task timed out", true), + ); + TaskResult { + task_id: task_id.to_string(), + ok: false, + reply: String::new(), + usage: None, + error: Some("timeout".to_string()), + } +} + +/// Return queued steering input (if any) after briefly waiting for input that +/// was sent while the turn was still in flight. +async fn drain_steer(steer_rx: &mut mpsc::UnboundedReceiver) -> Option { + if let Ok(msg) = steer_rx.try_recv() { + return Some(msg); + } + match tokio::time::timeout(STEER_DRAIN_GRACE, steer_rx.recv()).await { + Ok(msg) => msg, + Err(_) => None, + } +} + +/// Build (or resume) an agent session for a medulla task. +/// +/// The transcript identity is scoped by `session_id` (which the caller has +/// already resolved to the medulla `sessionId`, falling back to the `taskId`), +/// not the bare `agent_definition_name`. Without this, two `medulla:task_run`s +/// on the same `agentId` would collide onto one shared transcript and the +/// second would resume the first's history. +async fn build_agent(agent_id: &str, task_id: &str, session_id: &str) -> Result { + let config = crate::openhuman::config::rpc::load_config_with_timeout().await?; + crate::openhuman::agent::harness::AgentDefinitionRegistry::init_global(&config.workspace_dir) + .map_err(|err| format!("failed to init agent definition registry: {err}"))?; + let mut agent = Agent::from_config_for_agent(&config, agent_id) + .map_err(|err| format!("failed to build agent `{agent_id}`: {err}"))?; + agent.set_event_context(format!("medulla:{task_id}"), "medulla_harness"); + // Scope the transcript/session key per medulla session so distinct sessions + // on the same agent get isolated history (mirrors the web channel's + // per-thread `set_agent_definition_name`). + agent.set_agent_definition_name(medulla_session_key(agent_id, session_id)); + agent.fetch_connected_integrations().await; + let _ = agent.refresh_delegation_tools(); + Ok(agent) +} + +/// Derive a per-session agent-definition (transcript) key from the medulla +/// `session_id`, namespaced by `agent_id`. The session id is truncated on a +/// char boundary to keep transcript filenames bounded; the underlying +/// [`Agent::set_agent_definition_name`] sanitizes any remaining non-filename +/// characters. +fn medulla_session_key(agent_id: &str, session_id: &str) -> String { + let short: String = session_id.chars().take(32).collect(); + format!("{agent_id}_{short}") +} + +/// Spawn the per-turn progress → `medulla:task_envelope` forwarder. Returns its +/// join handle so the driver can flush it before the next turn. +fn spawn_forwarder( + task_id: String, + session_id: String, + seq: Arc, + mut progress_rx: mpsc::Receiver, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + while let Some(progress) = progress_rx.recv().await { + if let Some(kind) = envelope::progress_to_event_kind(&progress) { + let env = envelope::envelope_for_kind(&session_id, next_seq(&seq), &kind); + emit_envelope(&task_id, env); + } + } + }) +} + +fn next_seq(seq: &AtomicI64) -> i64 { + seq.fetch_add(1, Ordering::Relaxed) +} + +/// Project the crate-private [`LastTurnUsage`] into a compact JSON usage block +/// for `medulla:task_result` (the type itself isn't `Serialize`). +fn usage_to_json( + usage: crate::openhuman::agent::harness::turn_subagent_usage::LastTurnUsage, +) -> serde_json::Value { + serde_json::json!({ + "inputTokens": usage.input_tokens, + "outputTokens": usage.output_tokens, + "cachedInputTokens": usage.cached_input_tokens, + "costUsd": usage.cost_usd, + "contextWindow": usage.context_window, + }) +} + +/// Emit a `medulla:task_envelope` frame up the backend socket. +fn emit_envelope(task_id: &str, env: tinyplace::types::SessionEnvelopeV2) { + let envelope = match serde_json::to_value(&env) { + Ok(v) => v, + Err(err) => { + log::warn!("[medulla] failed to serialize envelope for task_id={task_id}: {err}"); + return; + } + }; + let frame = payloads::TaskEnvelope { + task_id: task_id.to_string(), + envelope, + }; + emit(EVENT_TASK_ENVELOPE, frame); +} + +/// Emit a terminal `medulla:task_result`. +fn emit_result(result: TaskResult) { + emit(EVENT_TASK_RESULT, result); +} + +/// Emit `medulla:register_agents` — the roster advertised on (re)connect. +/// +/// Built from the shipped default agent definitions. The backend clears the +/// roster on socket disconnect. +pub fn emit_register_agents() { + let agents: Vec = crate::openhuman::agent_registry::default_agents() + .into_iter() + .map(|entry| AgentDescriptor { + agent_id: entry.id, + name: entry.name, + description: entry.description, + }) + .collect(); + log::info!("[medulla] advertising {} agents to backend", agents.len()); + emit(EVENT_REGISTER_AGENTS, RegisterAgents { agents }); +} + +/// Serialize `payload` and emit it as a Socket.IO event on the global backend +/// socket. Best-effort: a missing/disconnected socket is logged, not fatal. +fn emit(event: &str, payload: T) { + let data = match serde_json::to_value(&payload) { + Ok(v) => v, + Err(err) => { + log::warn!("[medulla] failed to serialize payload for {event}: {err}"); + return; + } + }; + let Some(mgr) = crate::openhuman::socket::global_socket_manager() else { + log::debug!("[medulla] no socket manager — dropping {event}"); + return; + }; + let mgr = Arc::clone(mgr); + let event = event.to_string(); + tokio::spawn(async move { + if let Err(err) = mgr.emit(&event, data).await { + log::warn!("[medulla] failed to emit {event}: {err}"); + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unknown_task_send_and_abort_are_noops() { + let mgr = MedullaTaskManager::new(); + // Neither should panic when the task id is unknown. + mgr.steer_task(payloads::TaskSend { + task_id: "nope".into(), + input: "hi".into(), + }); + mgr.abort_task(payloads::TaskAbort { + task_id: "nope".into(), + }); + } + + #[test] + fn duplicate_task_registration_is_rejected() { + let mgr = Arc::new(MedullaTaskManager::new()); + // Manually seed a running task to simulate an in-flight run, then prove + // a second registration under the same id is ignored. + let abort = CancellationToken::new(); + let (steer_tx, _rx) = mpsc::unbounded_channel(); + mgr.tasks + .lock() + .insert("dup".to_string(), RunningTask { abort, steer_tx }); + assert!(mgr.tasks.lock().contains_key("dup")); + // A second start_task for "dup" must not overwrite / spawn. + mgr.start_task(payloads::TaskRun { + task_id: "dup".into(), + cycle_id: "c".into(), + session_id: None, + instruction: "x".into(), + agent_id: None, + timeout_ms: 0, + }); + assert_eq!(mgr.tasks.lock().len(), 1); + } + + #[test] + fn session_key_scopes_transcript_by_session_id() { + // Same agent id + distinct session ids => distinct transcript keys, so + // two medulla sessions can't collide onto one shared transcript. + let a = medulla_session_key("orchestrator", "sess-abc"); + let b = medulla_session_key("orchestrator", "sess-xyz"); + assert_ne!(a, b); + assert_eq!(a, "orchestrator_sess-abc"); + assert!(a.starts_with("orchestrator_")); + // Overlong session ids are truncated on a char boundary. + let long = "x".repeat(100); + let key = medulla_session_key("orchestrator", &long); + assert_eq!(key, format!("orchestrator_{}", "x".repeat(32))); + } + + #[test] + fn remaining_budget_reports_time_left_and_exhaustion() { + let now = Instant::now(); + // No deadline configured => unbounded. + assert_eq!(remaining_budget(None, now), Ok(None)); + // Deadline in the future => remaining time until it. + let future = now + Duration::from_secs(10); + match remaining_budget(Some(future), now) { + Ok(Some(d)) => assert!(d <= Duration::from_secs(10) && d > Duration::from_secs(9)), + other => panic!("expected some remaining budget, got {other:?}"), + } + // Deadline already reached / passed => exhausted. + assert_eq!(remaining_budget(Some(now), now), Err(())); + assert_eq!( + remaining_budget(Some(now), now + Duration::from_secs(1)), + Err(()) + ); + } +} diff --git a/src/openhuman/socket/medulla/payloads.rs b/src/openhuman/socket/medulla/payloads.rs new file mode 100644 index 000000000..9b8a6da00 --- /dev/null +++ b/src/openhuman/socket/medulla/payloads.rs @@ -0,0 +1,193 @@ +//! Wire payloads for the medulla "harness plane" — the `medulla:task_*` +//! Socket.IO protocol that lets a medulla operator (running in the backend) +//! drive an OpenHuman agent session as a delegated sub-agent. +//! +//! See `docs/specs/session-streaming-api-spec.md` §6 in the medulla repo. All +//! payloads are camelCase on the wire to match the backend's Socket.IO +//! conventions (the harness *envelope* they carry stays snake_case — that is +//! the tinyplace v2 wire format, decoded/encoded by [`super::envelope`]). + +use serde::{Deserialize, Serialize}; + +// ───────────────────────────────────────────────────────────────────────────── +// Down: backend / medulla → openhuman agent +// ───────────────────────────────────────────────────────────────────────────── + +/// `medulla:task_run` — start a task in an openhuman agent session. +/// +/// Creates (or resumes, when `session_id` is supplied) a session and sends +/// `instruction` as the opening prompt. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskRun { + pub task_id: String, + pub cycle_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_id: Option, + pub instruction: String, + /// Which openhuman agent to run the task as (defaults to the orchestrator). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agent_id: Option, + /// Hard wall-clock budget for the whole task, in milliseconds. + #[serde(default)] + pub timeout_ms: u64, +} + +/// `medulla:task_send` — mid-task steering (answer a question / approval +/// decision / follow-up); `input` is delivered into the running session. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskSend { + pub task_id: String, + pub input: String, +} + +/// `medulla:task_abort` — cancel the session/task. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskAbort { + pub task_id: String, +} + +// ───────────────────────────────────────────────────────────────────────────── +// Up: openhuman agent → backend / medulla +// ───────────────────────────────────────────────────────────────────────────── + +/// `medulla:task_envelope` — one live-stream frame for a task, carrying a +/// `tinyplace.harness.session.v2` envelope (see [`super::envelope`]). +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskEnvelope { + pub task_id: String, + /// A serialized [`tinyplace::types::SessionEnvelopeV2`]. Kept as raw JSON so + /// this struct stays a thin transport wrapper and never re-derives the + /// envelope kinds. + pub envelope: serde_json::Value, +} + +/// `medulla:task_result` — explicit completion (preferred over idle-detection). +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskResult { + pub task_id: String, + pub ok: bool, + #[serde(default)] + pub reply: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub usage: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +/// A single agent descriptor advertised in the roster. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentDescriptor { + pub agent_id: String, + pub name: String, + #[serde(default, skip_serializing_if = "String::is_empty")] + pub description: String, +} + +/// `medulla:register_agents` — roster advertisement sent on connect. The +/// backend clears the roster when this socket disconnects. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RegisterAgents { + pub agents: Vec, +} + +// ───────────────────────────────────────────────────────────────────────────── +// Socket.IO event names +// ───────────────────────────────────────────────────────────────────────────── + +/// Down events handled by openhuman. +pub const EVENT_TASK_RUN: &str = "medulla:task_run"; +pub const EVENT_TASK_SEND: &str = "medulla:task_send"; +pub const EVENT_TASK_ABORT: &str = "medulla:task_abort"; + +/// Up events emitted by openhuman. +pub const EVENT_TASK_ENVELOPE: &str = "medulla:task_envelope"; +pub const EVENT_TASK_RESULT: &str = "medulla:task_result"; +pub const EVENT_REGISTER_AGENTS: &str = "medulla:register_agents"; + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn task_run_round_trips_and_reads_camel_case_wire() { + let wire = json!({ + "taskId": "t1", + "cycleId": "c1", + "sessionId": "s1", + "instruction": "summarize the doc", + "agentId": "orchestrator", + "timeoutMs": 60000, + }); + let parsed: TaskRun = serde_json::from_value(wire.clone()).unwrap(); + assert_eq!(parsed.task_id, "t1"); + assert_eq!(parsed.cycle_id, "c1"); + assert_eq!(parsed.session_id.as_deref(), Some("s1")); + assert_eq!(parsed.agent_id.as_deref(), Some("orchestrator")); + assert_eq!(parsed.timeout_ms, 60000); + // Re-serialize and confirm it decodes back to the same value. + let again: TaskRun = + serde_json::from_value(serde_json::to_value(&parsed).unwrap()).unwrap(); + assert_eq!(parsed, again); + } + + #[test] + fn task_run_defaults_optional_fields() { + let wire = json!({ + "taskId": "t2", + "cycleId": "c2", + "instruction": "go", + }); + let parsed: TaskRun = serde_json::from_value(wire).unwrap(); + assert!(parsed.session_id.is_none()); + assert!(parsed.agent_id.is_none()); + assert_eq!(parsed.timeout_ms, 0); + } + + #[test] + fn task_send_and_abort_round_trip() { + let send: TaskSend = + serde_json::from_value(json!({ "taskId": "t", "input": "yes" })).unwrap(); + assert_eq!(send.input, "yes"); + let abort: TaskAbort = serde_json::from_value(json!({ "taskId": "t" })).unwrap(); + assert_eq!(abort.task_id, "t"); + } + + #[test] + fn task_result_omits_none_and_round_trips() { + let res = TaskResult { + task_id: "t".into(), + ok: true, + reply: "done".into(), + usage: None, + error: None, + }; + let v = serde_json::to_value(&res).unwrap(); + assert!(v.get("usage").is_none()); + assert!(v.get("error").is_none()); + assert_eq!(v["taskId"], "t"); + assert_eq!(v["ok"], true); + } + + #[test] + fn register_agents_round_trips() { + let roster = RegisterAgents { + agents: vec![AgentDescriptor { + agent_id: "orchestrator".into(), + name: "Orchestrator".into(), + description: "default".into(), + }], + }; + let wire = serde_json::to_value(&roster).unwrap(); + assert_eq!(wire["agents"][0]["agentId"], "orchestrator"); + let back: RegisterAgents = serde_json::from_value(wire).unwrap(); + assert_eq!(roster, back); + } +} diff --git a/src/openhuman/socket/mod.rs b/src/openhuman/socket/mod.rs index b20eacf1f..0a159ea36 100644 --- a/src/openhuman/socket/mod.rs +++ b/src/openhuman/socket/mod.rs @@ -6,6 +6,7 @@ mod event_handlers; pub mod manager; +pub mod medulla; mod schemas; pub(crate) mod token_provider; pub mod types;