diff --git a/src/openhuman/orchestration/cloud.rs b/src/openhuman/orchestration/cloud.rs index f578a0db29..290d445da3 100644 --- a/src/openhuman/orchestration/cloud.rs +++ b/src/openhuman/orchestration/cloud.rs @@ -40,7 +40,7 @@ const DEFAULT_BACKOFFS: [Duration; 3] = [ pub async fn push_event( config: &Config, envelope: &OrchestrationEventEnvelopeWire, -) -> Result<(), String> { +) -> Result, String> { let token = crate::openhuman::credentials::session_support::require_live_session_token(config)?; let api_url = effective_backend_api_url(&config.api_url); let client = BackendOAuthClient::new(&api_url).map_err(|e| e.to_string())?; @@ -70,6 +70,7 @@ pub async fn push_world_diff(config: &Config, batch: &WorldDiffBatchWire) -> Res ), ) .await + .map(|_| ()) } /// Inner push with an injectable client, token, and backoff schedule so the @@ -81,12 +82,12 @@ pub async fn push_event_with( token: &str, envelope: &OrchestrationEventEnvelopeWire, backoffs: &[Duration], -) -> Result<(), String> { +) -> Result, String> { let label = format!( "event session={} seq={}", envelope.session_id, envelope.event.seq ); - post_with_retry( + let resp = post_with_retry( client, token, EVENTS_PATH, @@ -94,7 +95,13 @@ pub async fn push_event_with( backoffs, &label, ) - .await + .await?; + // `202 { accepted, cycleId }` — return the cycleId so the caller can record + // the cycle's device-authoritative origin (see `exec_gate`). + Ok(resp + .get("cycleId") + .and_then(|v| v.as_str()) + .map(str::to_string)) } /// Generic authed POST with bounded jittered-backoff retry. Shared by every @@ -106,16 +113,16 @@ async fn post_with_retry( body: serde_json::Value, backoffs: &[Duration], label: &str, -) -> Result<(), String> { +) -> Result { let mut attempt: usize = 0; loop { match client .authed_json(token, Method::POST, path, Some(body.clone())) .await { - Ok(_) => { + Ok(data) => { log::debug!(target: LOG, "[orchestration] cloud.push.ok {label} attempt={}", attempt + 1); - return Ok(()); + return Ok(data); } Err(err) => { let msg = crate::api::flatten_authed_error(err); @@ -152,6 +159,7 @@ pub async fn push_world_diff_with( &label, ) .await + .map(|_| ()) } // ── Hosted read surface (GET) ───────────────────────────────────────────────── diff --git a/src/openhuman/orchestration/effect_executor.rs b/src/openhuman/orchestration/effect_executor.rs index 2f99dfec59..1e077a6a6f 100644 --- a/src/openhuman/orchestration/effect_executor.rs +++ b/src/openhuman/orchestration/effect_executor.rs @@ -53,6 +53,20 @@ pub fn device_tool_manifest() -> Value { "name": "device_status", "description": "Report this device's app version and platform.", "inputSchema": { "type": "object", "properties": {}, "additionalProperties": false } + }, + { + "name": "run_local_agent", + "description": "Spawn a local device sub-agent (e.g. code_executor for repo/shell/file work, researcher, tools_agent) on the user's own machine for background work, and return an acknowledgement. LOCAL-EXECUTION: only runs for a Master-chat cycle (the human ↔ their own OpenHuman); it is refused for any agent-to-agent cycle.", + "inputSchema": { + "type": "object", + "required": ["agent_id", "prompt"], + "properties": { + "agent_id": { "type": "string", "description": "Local sub-agent id, e.g. code_executor, researcher, tools_agent." }, + "prompt": { "type": "string", "description": "Clear, self-contained instruction for the sub-agent." }, + "context": { "type": "string", "description": "Optional context blob from prior results." } + }, + "additionalProperties": false + } } ] }) @@ -81,22 +95,162 @@ pub fn tool_result_frame(call_id: &str, ok: bool, result: Value, error: Option<& json!({ "callId": call_id, "ok": ok, "result": result, "error": error }) } -/// Run a device-declared tool locally. Read-only and side-effect-free for now; -/// local-workspace tools plug in here as they are added to the manifest. -pub fn dispatch_device_tool(name: &str, _args: &Value) -> Result { +/// Run a device-declared tool locally and return its result. `device_status` is +/// read-only; `run_local_agent` executes local workspace work and is therefore +/// **gated**: it runs only when `cycle_id` belongs to a Master-chat cycle +/// (device-authoritative origin, see [`super::exec_gate`]). The gate is enforced +/// here — on the device that holds the capability — so a prompt-injected or +/// compromised cloud brain cannot induce local execution for an A2A cycle. +pub async fn dispatch_device_tool( + name: &str, + args: &Value, + cycle_id: &str, +) -> Result { + if super::exec_gate::is_local_execution_tool(name) + && !super::exec_gate::cycle_is_master(cycle_id) + { + log::warn!( + target: LOG, + "[orchestration] device_tool.denied name={name} cycle={cycle_id} reason=non_master_origin" + ); + return Err(format!( + "device tool '{name}' denied: local execution is restricted to the Master chat" + )); + } match name { "device_status" => Ok(json!({ "version": env!("CARGO_PKG_VERSION"), "platform": std::env::consts::OS, })), + "run_local_agent" => run_local_agent(args, cycle_id).await, other => Err(format!("unknown device tool: {other}")), } } +/// The gated `run_local_agent` device tool. Reached only after the Master-chat +/// gate in [`dispatch_device_tool`] has passed. +/// +/// **Async model:** we do NOT block the wake cycle on a (potentially long) local +/// sub-agent. We fire it in the background and return an immediate `accepted` +/// ack — which the hosted brain sees as `orch:tool_result` well inside the +/// device-tool timeout. When the sub-agent finishes, [`run_local_agent_and_forward`] +/// pushes its result up as a fresh `tool_completion` event, which wakes a NEW +/// cycle that reasons over the result. So the original cycle is never blocked, +/// and the result still lands back in the brain via the follow-up cycle. +async fn run_local_agent(args: &Value, cycle_id: &str) -> Result { + let (counterpart, session_id) = super::exec_gate::cycle_target(cycle_id) + .ok_or_else(|| "run_local_agent: unknown cycle origin".to_string())?; + let agent_id = args + .get("agent_id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .trim() + .to_string(); + let prompt = args + .get("prompt") + .and_then(|v| v.as_str()) + .unwrap_or("") + .trim() + .to_string(); + if agent_id.is_empty() || prompt.is_empty() { + return Err("run_local_agent: `agent_id` and `prompt` are required".to_string()); + } + let task_id = cycle_id.to_string(); + let run_args = args.clone(); + let bg_task_id = task_id.clone(); + tokio::spawn(async move { + if let Err(e) = + run_local_agent_and_forward(&counterpart, &session_id, &bg_task_id, &agent_id, run_args) + .await + { + log::warn!( + target: LOG, + "[orchestration] run_local_agent.forward_failed task={bg_task_id}: {e}" + ); + } + }); + Ok(json!({ + "accepted": true, + "taskId": task_id, + "status": "running", + "note": "local sub-agent started; its result will arrive as a follow-up tool_completion event." + })) +} + +/// Background half of `run_local_agent`: run the local sub-agent to completion, +/// then forward its result up as a `tool_completion` event on the originating +/// session (which the backend wakes a fresh cycle for). +async fn run_local_agent_and_forward( + counterpart: &str, + session_id: &str, + task_id: &str, + agent_id: &str, + run_args: Value, +) -> Result<(), String> { + use crate::openhuman::tools::traits::Tool; + // 1. Run the local sub-agent synchronously to completion (real output). + let tool = crate::openhuman::agent_orchestration::tools::SpawnSubagentTool::new(); + // Convert an invocation error into a failure completion rather than bailing: + // the hosted brain must always learn the outcome (success OR failure) via the + // forwarded `tool_completion`, never be left with no follow-up at all. + let (ok, output) = match tool.execute(run_args).await { + Ok(result) => (!result.is_error, result.output()), + Err(e) => (false, format!("sub-agent invocation error: {e}")), + }; + let body = format!( + "[local sub-agent `{agent_id}` task {task_id} {}]\n{output}", + if ok { "completed" } else { "failed" } + ); + + // 2. Persist the completion into the render cache (allocates a monotonic seq) + // and forward it as a `tool_completion` event → backend wakes a new cycle. + let config = crate::openhuman::config::Config::load_or_init() + .await + .map_err(|e| format!("config load: {e}"))?; + let now = chrono::Utc::now().to_rfc3339(); + let seq = super::store::with_connection(&config.workspace_dir, |conn| { + let seq = super::store::next_session_seq(conn, counterpart, session_id)?; + super::store::insert_message( + conn, + &super::types::OrchestrationMessage { + id: format!("tool-completion:{task_id}:{seq}"), + agent_id: counterpart.to_string(), + session_id: session_id.to_string(), + chat_kind: super::types::ChatKind::Master, + role: "system".to_string(), + body: body.clone(), + timestamp: now.clone(), + seq, + ..Default::default() + }, + )?; + Ok(seq) + }) + .map_err(|e| format!("persist completion: {e}"))?; + + let ts = super::wire::parse_ts_ms(&now).unwrap_or(0); + let envelope = super::wire::OrchestrationEventEnvelopeWire::build( + counterpart, + session_id, + seq, + "system", + counterpart, + &body, + ts, + "tool_completion", + ); + super::cloud::push_event(&config, &envelope).await?; + log::debug!( + target: LOG, + "[orchestration] run_local_agent.forwarded task={task_id} session={session_id} seq={seq} ok={ok}" + ); + Ok(()) +} + /// Handle an inbound `orch:tool_call` frame end-to-end: parse → dispatch → /// build the result frame. Returns `(callId, resultFrame)` to emit, or `None` -/// when the frame is unparseable. -pub fn handle_tool_call(data: &Value) -> Option<(String, Value)> { +/// when the frame is unparseable. Async because a local sub-agent spawn is. +pub async fn handle_tool_call(data: &Value) -> Option<(String, Value)> { let frame = match parse_tool_call(data) { Ok(f) => f, Err(e) => { @@ -104,10 +258,11 @@ pub fn handle_tool_call(data: &Value) -> Option<(String, Value)> { return None; } }; - let (ok, result, error) = match dispatch_device_tool(&frame.name, &frame.args) { - Ok(value) => (true, value, None), - Err(e) => (false, Value::Null, Some(e)), - }; + let (ok, result, error) = + match dispatch_device_tool(&frame.name, &frame.args, &frame.cycle_id).await { + Ok(value) => (true, value, None), + Err(e) => (false, Value::Null, Some(e)), + }; Some(( frame.call_id.clone(), tool_result_frame(&frame.call_id, ok, result, error.as_deref()), diff --git a/src/openhuman/orchestration/exec_gate.rs b/src/openhuman/orchestration/exec_gate.rs new file mode 100644 index 0000000000..0a64326f51 --- /dev/null +++ b/src/openhuman/orchestration/exec_gate.rs @@ -0,0 +1,136 @@ +//! Device-authoritative trust gate for local-execution device tools. +//! +//! The hosted brain runs in the cloud and can push `orch:tool_call` frames for +//! device-declared tools. Some of those tools (`run_local_agent` → the local +//! `code_executor` / workspace workers) execute code and read files on the +//! user's machine. Those must run **only** for a Master-chat cycle (the human +//! talking to their own OpenHuman), never for an A2A cycle driven by another +//! agent's DM — otherwise a prompt-injected reasoning turn could induce local +//! code execution / file exfiltration (confused deputy). +//! +//! The gate is **device-authoritative**: it does not trust any origin the +//! backend asserts in the tool-call frame (an injected brain could lie). When +//! the device forwards an event to the hosted brain (`POST /events`) it already +//! knows the cycle's counterpart, and the backend returns the `cycleId` for +//! that trigger. We record `cycleId -> counterpart` from *our own* forward, so +//! at tool-call time we resolve the origin from a fact we established, not one +//! the backend supplied. Unknown cycles fail closed (denied). + +use std::collections::{HashMap, VecDeque}; +use std::sync::Mutex; + +use super::types::LOCAL_MASTER_AGENT; + +/// Bound on tracked cycles — cycles are short-lived, so a small ring is plenty +/// and keeps this from growing unbounded over a long session. +const MAX_TRACKED: usize = 512; + +/// The counterpart + session a forwarded cycle belongs to (device-recorded). +#[derive(Clone)] +struct CycleTarget { + counterpart: String, + session_id: String, +} + +struct CycleOrigins { + target_by_cycle: HashMap, + /// FIFO of cycle ids for bounded eviction of the oldest entries. + order: VecDeque, +} + +static CYCLE_ORIGINS: Mutex> = Mutex::new(None); + +/// Record the counterpart + session a forwarded cycle belongs to. Called at +/// forward time (`ingest` / master-send → `cloud::push_event`) with the +/// `cycleId` the backend returned and the counterpart/session *we* addressed the +/// event to. This is the device-authoritative fact the gate resolves against. +pub fn record_cycle_origin(cycle_id: &str, counterpart: &str, session_id: &str) { + if cycle_id.is_empty() { + return; + } + let mut guard = CYCLE_ORIGINS.lock().unwrap_or_else(|p| p.into_inner()); + let store = guard.get_or_insert_with(|| CycleOrigins { + target_by_cycle: HashMap::new(), + order: VecDeque::new(), + }); + let target = CycleTarget { + counterpart: counterpart.to_string(), + session_id: session_id.to_string(), + }; + if store + .target_by_cycle + .insert(cycle_id.to_string(), target) + .is_none() + { + store.order.push_back(cycle_id.to_string()); + while store.order.len() > MAX_TRACKED { + if let Some(old) = store.order.pop_front() { + store.target_by_cycle.remove(&old); + } + } + } +} + +/// True only when `cycle_id` was recorded as a **local Master-chat** cycle +/// (counterpart == [`LOCAL_MASTER_AGENT`]). Unknown or A2A cycles → false +/// (fail closed). This is the authorization for local-execution device tools. +pub fn cycle_is_master(cycle_id: &str) -> bool { + let guard = CYCLE_ORIGINS.lock().unwrap_or_else(|p| p.into_inner()); + guard + .as_ref() + .and_then(|s| s.target_by_cycle.get(cycle_id)) + .map(|t| t.counterpart == LOCAL_MASTER_AGENT) + .unwrap_or(false) +} + +/// The `(counterpart, session_id)` a cycle targets, for routing a local +/// sub-agent's completion back to the right session. `None` for unknown cycles. +pub fn cycle_target(cycle_id: &str) -> Option<(String, String)> { + let guard = CYCLE_ORIGINS.lock().unwrap_or_else(|p| p.into_inner()); + guard + .as_ref() + .and_then(|s| s.target_by_cycle.get(cycle_id)) + .map(|t| (t.counterpart.clone(), t.session_id.clone())) +} + +/// Device tools that touch local code / files / shell. Denied for any non-Master +/// cycle. Kept as an explicit allowlist so adding a tool to the manifest can't +/// silently escape the gate. +pub fn is_local_execution_tool(name: &str) -> bool { + matches!(name, "run_local_agent") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unknown_cycle_is_not_master() { + assert!(!cycle_is_master("never-seen")); + } + + #[test] + fn master_cycle_is_allowed_a2a_is_denied() { + record_cycle_origin("cyc-master", LOCAL_MASTER_AGENT, "master"); + record_cycle_origin("cyc-peer", "8xPeerBase58Addr", "sess-1"); + assert!(cycle_is_master("cyc-master")); + assert!(!cycle_is_master("cyc-peer")); + assert_eq!( + cycle_target("cyc-master"), + Some((LOCAL_MASTER_AGENT.to_string(), "master".to_string())) + ); + } + + #[test] + fn empty_cycle_id_is_ignored_and_denied() { + record_cycle_origin("", LOCAL_MASTER_AGENT, "master"); + assert!(!cycle_is_master("")); + assert_eq!(cycle_target(""), None); + } + + #[test] + fn local_execution_tools_are_gated() { + assert!(is_local_execution_tool("run_local_agent")); + assert!(!is_local_execution_tool("device_status")); + } +} diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index ab33094fd1..d9875acb66 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -601,8 +601,20 @@ fn forward_event( ); let config = config.clone(); tokio::spawn(async move { - if let Err(e) = super::cloud::push_event(&config, &envelope).await { - log::warn!(target: LOG, "[orchestration] cloud.shadow_push_failed: {e}"); + match super::cloud::push_event(&config, &envelope).await { + Ok(cycle_id) => { + // Device-authoritative origin: record the cycle → counterpart we + // forwarded, so the local-execution trust gate can resolve this + // cycle's origin without trusting the backend (see `exec_gate`). + if let Some(cid) = cycle_id { + super::exec_gate::record_cycle_origin( + &cid, + &envelope.counterpart_agent_id, + &envelope.session_id, + ); + } + } + Err(e) => log::warn!(target: LOG, "[orchestration] cloud.shadow_push_failed: {e}"), } }); } diff --git a/src/openhuman/orchestration/migrate_history.rs b/src/openhuman/orchestration/migrate_history.rs index daa5c1cab2..1ded5218e9 100644 --- a/src/openhuman/orchestration/migrate_history.rs +++ b/src/openhuman/orchestration/migrate_history.rs @@ -86,9 +86,19 @@ async fn resume_pending(config: &Config) -> Result { ts, msg.event_kind.as_deref().unwrap_or("message"), ); - super::cloud::push_event(config, &envelope) + let cycle_id = super::cloud::push_event(config, &envelope) .await .map_err(|e| format!("replay session={} seq={}: {e}", session.session_id, msg.seq))?; + // Record the resumed cycle's device-authoritative origin so a Master-chat + // turn resumed by the migration can still authorize `run_local_agent` + // (otherwise the gate sees an unknown cycle and denies local execution). + if let Some(cid) = cycle_id { + super::exec_gate::record_cycle_origin( + &cid, + &envelope.counterpart_agent_id, + &envelope.session_id, + ); + } resumed += 1; } Ok(resumed) diff --git a/src/openhuman/orchestration/mod.rs b/src/openhuman/orchestration/mod.rs index 8b9192e14d..b0d320d9f4 100644 --- a/src/openhuman/orchestration/mod.rs +++ b/src/openhuman/orchestration/mod.rs @@ -17,6 +17,7 @@ pub mod attention; pub mod bus; pub mod cloud; pub mod effect_executor; +pub mod exec_gate; pub mod ingest; pub mod migrate_history; pub mod ops; diff --git a/src/openhuman/orchestration/schemas.rs b/src/openhuman/orchestration/schemas.rs index e08816f37d..424c0a1523 100644 --- a/src/openhuman/orchestration/schemas.rs +++ b/src/openhuman/orchestration/schemas.rs @@ -642,8 +642,21 @@ fn handle_send_master_message(params: Map) -> ControllerFuture { ); let forward_cfg = config.clone(); tokio::spawn(async move { - if let Err(e) = super::cloud::push_event(&forward_cfg, &envelope).await { - log::warn!(target: LOG, "[orchestration_rpc] master_ask.forward_failed: {e}"); + match super::cloud::push_event(&forward_cfg, &envelope).await { + Ok(cycle_id) => { + // Record this Master-chat cycle's origin so the local-exec + // trust gate authorizes it (counterpart == LOCAL_MASTER_AGENT). + if let Some(cid) = cycle_id { + super::exec_gate::record_cycle_origin( + &cid, + &envelope.counterpart_agent_id, + &envelope.session_id, + ); + } + } + Err(e) => { + log::warn!(target: LOG, "[orchestration_rpc] master_ask.forward_failed: {e}") + } } }); log::debug!(target: LOG, "[orchestration_rpc] master_ask.forwarded id={message_id} seq={seq}"); diff --git a/src/openhuman/socket/event_handlers.rs b/src/openhuman/socket/event_handlers.rs index fdd0944ced..35dbd96b58 100644 --- a/src/openhuman/socket/event_handlers.rs +++ b/src/openhuman/socket/event_handlers.rs @@ -92,12 +92,15 @@ pub(super) fn handle_sio_event( // Hosted-brain device tool call: run a local (read-only) device tool and // return the result so the reasoning loop can continue. "orch:tool_call" => { - if let Some((call_id, result)) = - crate::openhuman::orchestration::effect_executor::handle_tool_call(&data) - { - log::debug!("[socket] orch:tool_call result call_id={call_id}"); - emit_via_channel(emit_tx, "orch:tool_result", result); - } + let tx = emit_tx.clone(); + tokio::spawn(async move { + if let Some((call_id, result)) = + crate::openhuman::orchestration::effect_executor::handle_tool_call(&data).await + { + log::debug!("[socket] orch:tool_call result call_id={call_id}"); + emit_via_channel(&tx, "orch:tool_result", result); + } + }); } // Hosted-brain context-guard eviction: fold the evicted compressed // summaries into local memory RAG so they stay retrievable offline, then diff --git a/tests/orchestration_exec_gate.rs b/tests/orchestration_exec_gate.rs new file mode 100644 index 0000000000..c61835af7e --- /dev/null +++ b/tests/orchestration_exec_gate.rs @@ -0,0 +1,45 @@ +//! Integration coverage for the local-execution trust gate (`exec_gate`). +//! +//! Links the compiled lib (the root crate's `cfg(test)` build is blocked by +//! unrelated stale test modules at this checkout — same reason the pushers are +//! tested from integration tests). Asserts the device-authoritative rule: a +//! local-execution device tool is authorized only for a Master-chat cycle. + +use openhuman_core::openhuman::orchestration::exec_gate::{ + cycle_is_master, cycle_target, is_local_execution_tool, record_cycle_origin, +}; +use openhuman_core::openhuman::orchestration::types::LOCAL_MASTER_AGENT; + +#[test] +fn master_cycle_is_authorized_a2a_and_unknown_are_denied() { + // A Master-chat forward records the sentinel counterpart + its session. + record_cycle_origin("gate-cyc-master", LOCAL_MASTER_AGENT, "master"); + // An A2A forward records the peer's real (base58) address + session. + record_cycle_origin("gate-cyc-peer", "8xPeerBase58AddressExample", "sess-peer"); + + assert!( + cycle_is_master("gate-cyc-master"), + "master cycle must authorize" + ); + assert!( + !cycle_is_master("gate-cyc-peer"), + "A2A cycle must be denied" + ); + // Fail closed: a cycle we never recorded (or a forged id) is not master. + assert!( + !cycle_is_master("gate-cyc-forged"), + "unknown cycle must be denied" + ); + // The completion router resolves the recorded (counterpart, session). + assert_eq!( + cycle_target("gate-cyc-master"), + Some((LOCAL_MASTER_AGENT.to_string(), "master".to_string())) + ); + assert_eq!(cycle_target("gate-cyc-forged"), None); +} + +#[test] +fn only_local_execution_tools_are_gated() { + assert!(is_local_execution_tool("run_local_agent")); + assert!(!is_local_execution_tool("device_status")); +}