feat(orchestration): local-exec device tool + trust gate + async completion forward (stacked on #4738) (#4753)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

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<Option<String>, 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<Option<String>, 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<Value, String> {
     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) ─────────────────────────────────────────────────

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<Value, String> {
+/// 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<Value, String> {
+    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<Value, String> {
+    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()),

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<String, CycleTarget>,
+    /// FIFO of cycle ids for bounded eviction of the oldest entries.
+    order: VecDeque<String>,
+}
+
+static CYCLE_ORIGINS: Mutex<Option<CycleOrigins>> = 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"));
+    }
+}

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}"),
         }
     });
 }

src/openhuman/orchestration/migrate_history.rs

@@ -86,9 +86,19 @@ async fn resume_pending(config: &Config) -> Result<usize, String> {
             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)

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;

src/openhuman/orchestration/schemas.rs

@@ -642,8 +642,21 @@ fn handle_send_master_message(params: Map<String, Value>) -> 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}");

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

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"));
+}