From 7659c75dc5ef55ad87dba3fee84626046c288632 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Thu, 9 Jul 2026 00:28:04 -0700 Subject: [PATCH] =?UTF-8?q?feat(orchestration):=20thin=20hosted-brain=20cl?= =?UTF-8?q?ient=20=E2=80=94=20pushers,=20effect/tool=20executors,=20wire?= =?UTF-8?q?=20allowlist=20(#4725)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/openhuman/config/schema/orchestration.rs | 9 + src/openhuman/orchestration/cloud.rs | 159 +++++++++++++ .../orchestration/effect_executor.rs | 199 ++++++++++++++++ src/openhuman/orchestration/ingest.rs | 141 +++++++++-- src/openhuman/orchestration/mod.rs | 3 + src/openhuman/orchestration/ops.rs | 2 +- src/openhuman/orchestration/wire.rs | 154 ++++++++++++ src/openhuman/socket/event_handlers.rs | 34 ++- tests/orchestration_effect_executor_e2e.rs | 99 ++++++++ tests/orchestration_shadow_push_e2e.rs | 221 ++++++++++++++++++ 10 files changed, 997 insertions(+), 24 deletions(-) create mode 100644 src/openhuman/orchestration/cloud.rs create mode 100644 src/openhuman/orchestration/effect_executor.rs create mode 100644 src/openhuman/orchestration/wire.rs create mode 100644 tests/orchestration_effect_executor_e2e.rs create mode 100644 tests/orchestration_shadow_push_e2e.rs diff --git a/src/openhuman/config/schema/orchestration.rs b/src/openhuman/config/schema/orchestration.rs index 43f6514d6..7b090fe76 100644 --- a/src/openhuman/config/schema/orchestration.rs +++ b/src/openhuman/config/schema/orchestration.rs @@ -38,6 +38,14 @@ pub struct OrchestrationConfig { #[serde(default = "default_enabled")] pub enabled: bool, + /// Phase 0 shadow migration: in addition to running the local wake graph, + /// mirror each ingested event up to the hosted brain via + /// `POST /orchestration/v1/events` (best-effort, fire-and-forget). The local + /// brain stays authoritative; this only dual-writes server-side state. + /// Default: `false` (opt-in until the hosted graph is the source of truth). + #[serde(default)] + pub cloud_shadow: bool, + /// Coalesce a burst of DMs for one session into a single graph run: after a /// session message lands, wait this many milliseconds for the burst to /// settle before invoking the wake graph. Default: `750`. @@ -96,6 +104,7 @@ impl Default for OrchestrationConfig { fn default() -> Self { Self { enabled: default_enabled(), + cloud_shadow: false, debounce_ms: default_debounce_ms(), max_supersteps: default_max_supersteps(), message_window: default_message_window(), diff --git a/src/openhuman/orchestration/cloud.rs b/src/openhuman/orchestration/cloud.rs new file mode 100644 index 000000000..579652870 --- /dev/null +++ b/src/openhuman/orchestration/cloud.rs @@ -0,0 +1,159 @@ +//! Cloud event pusher — forwards sanitized orchestration events up to the +//! hosted brain (`POST /orchestration/v1/events`). +//! +//! Phase 0 runs this in **shadow mode**: the local wake graph remains +//! authoritative and the push is an additional, best-effort, fire-and-forget +//! side effect gated behind [`OrchestrationConfig::cloud_shadow`] +//! (default off). It never blocks or fails ingest. +//! +//! Auth + base-URL plumbing mirrors the other hosted-API adapters +//! (`announcements/ops.rs`, `billing/ops.rs`): an app-session JWT via +//! `require_live_session_token`, the backend base via `effective_backend_api_url`, +//! and one shared `BackendOAuthClient`. + +use std::time::Duration; + +use reqwest::Method; + +use crate::api::config::effective_backend_api_url; +use crate::api::BackendOAuthClient; +use crate::openhuman::config::Config; + +use super::wire::{OrchestrationEventEnvelopeWire, WorldDiffBatchWire}; + +const LOG: &str = "orchestration"; +const EVENTS_PATH: &str = "/orchestration/v1/events"; +const WORLD_DIFF_PATH: &str = "/orchestration/v1/world-diff"; + +/// Jittered retry schedule for a transient push failure (3 retries after the +/// first attempt). Matches the plan's 1s/4s/10s cadence. +const DEFAULT_BACKOFFS: [Duration; 3] = [ + Duration::from_secs(1), + Duration::from_secs(4), + Duration::from_secs(10), +]; + +/// Push one sanitized event to the hosted brain. Resolves the app-session JWT +/// and backend base, then POSTs with bounded retry. Returns `Err` only after +/// the retry budget is exhausted (or the session is signed out). +pub async fn push_event( + config: &Config, + envelope: &OrchestrationEventEnvelopeWire, +) -> 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())?; + push_event_with(&client, &token, envelope, &DEFAULT_BACKOFFS).await +} + +/// Upload a batch of world-diff entries — the subconscious tier's primary +/// trigger. Same auth/base/retry plumbing as [`push_event`]. Returns `Err` only +/// after the retry budget is exhausted (or the session is signed out). +pub async fn push_world_diff(config: &Config, batch: &WorldDiffBatchWire) -> Result<(), String> { + if batch.entries.is_empty() { + return Ok(()); + } + 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())?; + post_with_retry( + &client, + &token, + WORLD_DIFF_PATH, + batch.to_value(), + &DEFAULT_BACKOFFS, + &format!( + "world-diff session={} entries={}", + batch.session_id, + batch.entries.len() + ), + ) + .await +} + +/// Inner push with an injectable client, token, and backoff schedule so the +/// transport can be exercised against a mock server without real credentials or +/// real sleeps (`backoffs = &[]` → single attempt). Public for integration +/// tests (`tests/orchestration_shadow_push_e2e.rs`). +pub async fn push_event_with( + client: &BackendOAuthClient, + token: &str, + envelope: &OrchestrationEventEnvelopeWire, + backoffs: &[Duration], +) -> Result<(), String> { + let label = format!( + "event session={} seq={}", + envelope.session_id, envelope.event.seq + ); + post_with_retry( + client, + token, + EVENTS_PATH, + envelope.to_value(), + backoffs, + &label, + ) + .await +} + +/// Generic authed POST with bounded jittered-backoff retry. Shared by every +/// orchestration uplink (`events`, `world-diff`). `backoffs = &[]` → one attempt. +async fn post_with_retry( + client: &BackendOAuthClient, + token: &str, + path: &str, + body: serde_json::Value, + backoffs: &[Duration], + label: &str, +) -> Result<(), String> { + let mut attempt: usize = 0; + loop { + match client + .authed_json(token, Method::POST, path, Some(body.clone())) + .await + { + Ok(_) => { + log::debug!(target: LOG, "[orchestration] cloud.push.ok {label} attempt={}", attempt + 1); + return Ok(()); + } + Err(err) => { + let msg = crate::api::flatten_authed_error(err); + if attempt >= backoffs.len() { + log::warn!(target: LOG, "[orchestration] cloud.push.give_up {label} attempts={} err={msg}", attempt + 1); + return Err(msg); + } + log::warn!(target: LOG, "[orchestration] cloud.push.retry {label} attempt={} err={msg}", attempt + 1); + tokio::time::sleep(backoffs[attempt]).await; + attempt += 1; + } + } + } +} + +/// World-diff uploader with injectable client/token/backoffs for tests. +pub async fn push_world_diff_with( + client: &BackendOAuthClient, + token: &str, + batch: &WorldDiffBatchWire, + backoffs: &[Duration], +) -> Result<(), String> { + let label = format!( + "world-diff session={} entries={}", + batch.session_id, + batch.entries.len() + ); + post_with_retry( + client, + token, + WORLD_DIFF_PATH, + batch.to_value(), + backoffs, + &label, + ) + .await +} + +// Transport tests live in `tests/orchestration_shadow_push_e2e.rs` (integration +// crate): the root crate's `cfg(test)` build is currently blocked by unrelated +// stale test modules at this checkout, so the pusher is exercised over wiremock +// from an integration test that links the compiled lib instead. diff --git a/src/openhuman/orchestration/effect_executor.rs b/src/openhuman/orchestration/effect_executor.rs new file mode 100644 index 000000000..bfc1a6017 --- /dev/null +++ b/src/openhuman/orchestration/effect_executor.rs @@ -0,0 +1,199 @@ +//! Client-side executor for device-bound orchestration effects pushed by the +//! hosted brain over the socket. +//! +//! The backend computes a wake cycle and, when it wants the device to act, +//! delivers an effect frame (`orch:effect:send_dm`, `orch:effect:evict`, …). The +//! device runs the effect against its local Signal keys / memory and acks with +//! `orch:effect:result { callId, ok, error? }`. Delivery is at-least-once and the +//! client dedupes on `callId` (a `send_dm` whose ack was already latched +//! server-side is a no-op here — see [`is_duplicate_call`]). + +use std::collections::HashSet; +use std::sync::Mutex; + +use serde::Deserialize; +use serde_json::{json, Value}; + +const LOG: &str = "orchestration"; + +/// A `send_dm` effect: relay `body` to `counterpartAgentId` over Signal, wrapping +/// it in `sessionId`'s session envelope so the peer threads the reply. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SendDmEffect { + #[serde(default)] + pub cycle_id: String, + pub call_id: String, + pub counterpart_agent_id: String, + #[serde(default)] + pub session_id: String, + #[serde(default)] + pub body: String, +} + +/// Parse an `orch:effect:send_dm` frame. Pure. +pub fn parse_send_dm(data: &Value) -> Result { + serde_json::from_value(data.clone()).map_err(|e| format!("parse send_dm: {e}")) +} + +/// Build the `orch:effect:result` ack frame the device sends back. Pure. +pub fn effect_result_frame(call_id: &str, ok: bool, error: Option<&str>) -> Value { + json!({ "callId": call_id, "ok": ok, "error": error }) +} + +/// The device-tool manifest declared to the hosted brain on socket connect +/// (`orch:register_tools`). These are **queryable** tools the reasoning loop may +/// call mid-cycle (results feed back), distinct from the terminal `send_dm` +/// effect. Phase 2 seeds it with a read-only status probe; local-workspace tools +/// grow this list as they are wired to the device tool dispatcher. +pub fn device_tool_manifest() -> Value { + json!({ + "tools": [ + { + "name": "device_status", + "description": "Report this device's app version and platform.", + "inputSchema": { "type": "object", "properties": {}, "additionalProperties": false } + } + ] + }) +} + +/// A device tool call pushed by the hosted brain (`orch:tool_call`). Run it +/// locally and return the result over `orch:tool_result`. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ToolCallFrame { + #[serde(default)] + pub cycle_id: String, + pub call_id: String, + pub name: String, + #[serde(default)] + pub args: Value, +} + +/// Parse an `orch:tool_call` frame. Pure. +pub fn parse_tool_call(data: &Value) -> Result { + serde_json::from_value(data.clone()).map_err(|e| format!("parse tool_call: {e}")) +} + +/// Build the `orch:tool_result` frame returned to the hosted brain. Pure. +pub fn tool_result_frame(call_id: &str, ok: bool, result: Value, error: Option<&str>) -> Value { + 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 { + match name { + "device_status" => Ok(json!({ + "version": env!("CARGO_PKG_VERSION"), + "platform": std::env::consts::OS, + })), + other => Err(format!("unknown device tool: {other}")), + } +} + +/// 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)> { + let frame = match parse_tool_call(data) { + Ok(f) => f, + Err(e) => { + log::warn!(target: LOG, "[orchestration] tool_call.parse_failed: {e}"); + 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)), + }; + Some(( + frame.call_id.clone(), + tool_result_frame(&frame.call_id, ok, result, error.as_deref()), + )) +} + +// ── callId dedupe (at-least-once delivery guard) ────────────────────────────── + +static SEEN_CALL_IDS: Mutex>> = Mutex::new(None); + +/// Record a `callId` and report whether it was already executed on this device. +/// Guards the at-least-once socket delivery so a redelivered effect is acked +/// (idempotently) but not executed twice. +pub fn is_duplicate_call(call_id: &str) -> bool { + let mut guard = SEEN_CALL_IDS.lock().unwrap_or_else(|p| p.into_inner()); + let set = guard.get_or_insert_with(HashSet::new); + !set.insert(call_id.to_string()) +} + +/// Wrap the reply body for `session_id`. Empty/`master`/`subconscious` sessions +/// send the plain body; a real harness session is wrapped in a v1 envelope so +/// the peer threads it. Delegates to the shared orchestration helper. +fn outgoing_plaintext(session_id: &str, body: &str) -> Result { + if session_id.is_empty() { + return Ok(body.to_string()); + } + super::ops::session_send_plaintext(session_id, body).map_err(|e| e.to_string()) +} + +/// Execute a `send_dm` effect: wrap + send over Signal via the existing +/// tinyplace transport. The device's Signal keys never leave the machine. +pub async fn execute_send_dm(effect: &SendDmEffect) -> Result<(), String> { + let plaintext = outgoing_plaintext(&effect.session_id, &effect.body)?; + let mut params = serde_json::Map::new(); + params.insert( + "recipient".to_string(), + Value::from(effect.counterpart_agent_id.clone()), + ); + params.insert("plaintext".to_string(), Value::from(plaintext)); + crate::openhuman::tinyplace::handle_tinyplace_signal_send_message(params) + .await + .map_err(|e| format!("signal send: {e}"))?; + Ok(()) +} + +/// Handle an inbound `orch:effect:send_dm` frame end-to-end: parse → dedupe → +/// send → produce the ack frame. Returns `(callId, ackFrame)` for the caller to +/// emit, or `None` when the frame is unparseable (nothing to ack). +pub async fn handle_send_dm(data: &Value) -> Option<(String, Value)> { + let effect = match parse_send_dm(data) { + Ok(e) => e, + Err(e) => { + log::warn!(target: LOG, "[orchestration] effect.send_dm.parse_failed: {e}"); + return None; + } + }; + + if is_duplicate_call(&effect.call_id) { + log::debug!( + target: LOG, + "[orchestration] effect.send_dm.duplicate call_id={} (re-acking)", + effect.call_id + ); + return Some(( + effect.call_id.clone(), + effect_result_frame(&effect.call_id, true, None), + )); + } + + let (ok, error) = match execute_send_dm(&effect).await { + Ok(()) => { + log::debug!( + target: LOG, + "[orchestration] effect.send_dm.sent call_id={} session={}", + effect.call_id, + effect.session_id + ); + (true, None) + } + Err(e) => { + log::warn!(target: LOG, "[orchestration] effect.send_dm.failed call_id={}: {e}", effect.call_id); + (false, Some(e)) + } + }; + Some(( + effect.call_id.clone(), + effect_result_frame(&effect.call_id, ok, error.as_deref()), + )) +} diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index af55d06f8..08ddec987 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -456,13 +456,17 @@ fn non_empty(s: String) -> Option { /// Persist a classified message + its session row. Idempotent by `msg_id`; /// returns true if a new message row landed. Testable with a tempdir DB. +/// Persist a classified message. Returns `Some(ingest_seq)` when a new row +/// landed (the store-assigned per-session ordinal), or `None` when the message +/// was a duplicate and nothing was written. The seq is the idempotency key the +/// shadow cloud push (`cloud::push_event`) uploads with. fn persist_message( workspace_dir: &Path, msg_id: &str, agent_id: &str, classified: &ClassifiedMessage, now: &str, -) -> Result { +) -> Result, String> { store::with_connection(workspace_dir, |c| { // Wake idempotence keys on a per-session `seq` being monotonic, but the // harness `message.line` we classify into `seq` is NOT reliable: a wrapped @@ -552,12 +556,58 @@ fn persist_message( if landed && classified.event_kind.as_deref() == Some("error") { store::kv_set(c, "orchestration:last_error", &classified.body)?; } - Ok(landed) + Ok(landed.then_some(ingest_seq)) }) }) .map_err(|e| format!("persist: {e}")) } +/// Fire-and-forget the Phase 0 shadow push for a freshly-landed event. No-op +/// unless `orchestration.cloud_shadow` is set. Builds the sanitized wire +/// envelope (the security allowlist lives in [`super::wire`]) and spawns the +/// upload so ingest never blocks on the network. +fn maybe_shadow_push( + config: &Config, + agent_id: &str, + ingest_seq: i64, + classified: &ClassifiedMessage, + now: &str, +) { + if !config.orchestration.cloud_shadow { + return; + } + // Content events carry the real per-session ordinal; a status/lifecycle + // event stamps seq 0 and would collide on the backend's idempotency key, so + // only mirror seq-advancing events in shadow mode. + if ingest_seq <= 0 { + return; + } + let ts = super::wire::parse_ts_ms(&classified.timestamp) + .or_else(|| super::wire::parse_ts_ms(now)) + .unwrap_or(0); + let kind = classified + .event_kind + .as_deref() + .unwrap_or_else(|| classified.chat_kind.as_str()); + let envelope = super::wire::OrchestrationEventEnvelopeWire::build( + agent_id, + &classified.session_id, + ingest_seq, + &classified.role, + // For an inbound DM the sender is the counterpart agent itself. + agent_id, + &classified.body, + ts, + kind, + ); + 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}"); + } + }); +} + /// Entry point from the bus subscriber. Cheap no-op when orchestration is /// disabled or the stream is not a DM stream. pub async fn ingest_stream_message( @@ -657,17 +707,28 @@ async fn ingest_one( let landed = persist_message(&workspace_dir, &msg_id, &agent_id, &classified, &now)?; // 3. Acknowledge (consume once) + fan out for stages 4/7. - if landed { + if let Some(ingest_seq) = landed { if let Err(e) = acknowledge_message(&msg_id).await { log::warn!(target: LOG, "[orchestration] ingest.ack_failed id={msg_id}: {e}"); } + + // Phase 0 shadow migration: mirror the sanitized event up to the hosted + // brain. Best-effort and fire-and-forget — the local wake graph below + // stays authoritative, so a push failure (or offline) never affects + // ingest. Default-off (`orchestration.cloud_shadow`). + maybe_shadow_push(config, &agent_id, ingest_seq, &classified, &now); + publish_global(DomainEvent::OrchestrationSessionMessage { agent_id, session_id: classified.session_id, chat_kind: classified.chat_kind.as_str().to_string(), }); } - log::debug!(target: LOG, "[orchestration] ingest.exit id={msg_id} landed={landed}"); + log::debug!( + target: LOG, + "[orchestration] ingest.exit id={msg_id} landed={}", + landed.is_some() + ); Ok(()) } @@ -1092,9 +1153,13 @@ mod tests { ), "2026-07-05T09:00:00Z", ); - assert!(persist_message(tmp.path(), "si1", "@peer", &si, "now").unwrap()); + assert!(persist_message(tmp.path(), "si1", "@peer", &si, "now") + .unwrap() + .is_some()); // Re-persisting the SAME event id is idempotent (dedup on the relay id). - assert!(!persist_message(tmp.path(), "si1", "@peer", &si, "now").unwrap()); + assert!(persist_message(tmp.path(), "si1", "@peer", &si, "now") + .unwrap() + .is_none()); store::with_connection(tmp.path(), |c| { let s = store::load_session(c, "@peer", "w2")?.expect("session lazy-created"); @@ -1117,7 +1182,9 @@ mod tests { v2_env("agent_message", r#"{ "text": "working" }"#, "w2", "agent"), "2026-07-05T09:01:00Z", ); - assert!(persist_message(tmp.path(), "m1", "@peer", &content, "now").unwrap()); + assert!(persist_message(tmp.path(), "m1", "@peer", &content, "now") + .unwrap() + .is_some()); let resumed = classify_message( v2_env( "session_info", @@ -1128,7 +1195,9 @@ mod tests { ), "2026-07-05T09:02:00Z", ); - assert!(persist_message(tmp.path(), "si2", "@peer", &resumed, "now").unwrap()); + assert!(persist_message(tmp.path(), "si2", "@peer", &resumed, "now") + .unwrap() + .is_some()); store::with_connection(tmp.path(), |c| { let s = store::load_session(c, "@peer", "w2")?.expect("session"); @@ -1163,8 +1232,12 @@ mod tests { assert_eq!(v1.session_id, "w1"); assert_eq!(v2.session_id, "w1"); - assert!(persist_message(tmp.path(), "m-v1", "@peer", &v1, "now").unwrap()); - assert!(persist_message(tmp.path(), "m-v2", "@peer", &v2, "now").unwrap()); + assert!(persist_message(tmp.path(), "m-v1", "@peer", &v1, "now") + .unwrap() + .is_some()); + assert!(persist_message(tmp.path(), "m-v2", "@peer", &v2, "now") + .unwrap() + .is_some()); store::with_connection(tmp.path(), |c| { // Both rows land in the single "w1" session, seqs monotonic 1,2. assert_eq!(store::count_messages(c, "@peer", "w1")?, 2); @@ -1186,7 +1259,9 @@ mod tests { v2_env("agent_message", r#"{ "text": "working" }"#, "w2", "agent"), "2026-07-05T09:00:00Z", ); - assert!(persist_message(tmp.path(), "m1", "@peer", &msg, "now").unwrap()); + assert!(persist_message(tmp.path(), "m1", "@peer", &msg, "now") + .unwrap() + .is_some()); // A status event follows: it lands a (deduped) row + updates the session // status columns, but last_seq must stay at 1 (no spurious wake). let status = classify_message( @@ -1198,7 +1273,9 @@ mod tests { ), "2026-07-05T09:00:00Z", ); - assert!(persist_message(tmp.path(), "m2", "@peer", &status, "now").unwrap()); + assert!(persist_message(tmp.path(), "m2", "@peer", &status, "now") + .unwrap() + .is_some()); store::with_connection(tmp.path(), |c| { let session = store::load_session(c, "@peer", "w2")?.expect("session exists"); @@ -1231,7 +1308,9 @@ mod tests { ), "2026-07-05T09:00:00Z", ); - assert!(persist_message(tmp.path(), "s1", "@peer", &running, "now").unwrap()); + assert!(persist_message(tmp.path(), "s1", "@peer", &running, "now") + .unwrap() + .is_some()); store::with_connection(tmp.path(), |c| { let s = store::load_session(c, "@peer", "w2")?.expect("session"); assert_eq!(s.current_detail.as_deref(), Some("compiling")); @@ -1247,7 +1326,9 @@ mod tests { v2_env("status", r#"{ "state": "idle" }"#, "w2", "agent"), "2026-07-05T09:01:00Z", ); - assert!(persist_message(tmp.path(), "s2", "@peer", &idle, "now").unwrap()); + assert!(persist_message(tmp.path(), "s2", "@peer", &idle, "now") + .unwrap() + .is_some()); store::with_connection(tmp.path(), |c| { let s = store::load_session(c, "@peer", "w2")?.expect("session"); assert_eq!(s.status_state.as_deref(), Some("idle")); @@ -1274,7 +1355,9 @@ mod tests { ), "2026-07-05T09:00:00Z", ); - assert!(persist_message(tmp.path(), "s1", "@peer", &running, "now").unwrap()); + assert!(persist_message(tmp.path(), "s1", "@peer", &running, "now") + .unwrap() + .is_some()); // A content event (agent_message) carries no run-state — it must COALESCE, // preserving the live status rather than nulling it. let msg = classify_message( @@ -1286,7 +1369,9 @@ mod tests { ), "2026-07-05T09:00:30Z", ); - assert!(persist_message(tmp.path(), "m1", "@peer", &msg, "now").unwrap()); + assert!(persist_message(tmp.path(), "m1", "@peer", &msg, "now") + .unwrap() + .is_some()); store::with_connection(tmp.path(), |c| { let s = store::load_session(c, "@peer", "w2")?.expect("session"); assert_eq!(s.status_state.as_deref(), Some("running_tool")); @@ -1309,7 +1394,9 @@ mod tests { ), "2026-07-05T09:00:00Z", ); - assert!(persist_message(tmp.path(), "e1", "@peer", &err, "now").unwrap()); + assert!(persist_message(tmp.path(), "e1", "@peer", &err, "now") + .unwrap() + .is_some()); store::with_connection(tmp.path(), |c| { assert_eq!( store::kv_get(c, "orchestration:last_error")?.as_deref(), @@ -1328,10 +1415,16 @@ mod tests { let session = classify_message(ENVELOPE.to_string(), "2026-07-02T09:00:00Z"); let master = classify_message("hi".to_string(), "2026-07-02T09:00:00Z"); - assert!(persist_message(tmp.path(), "m1", "@peer", &session, "now").unwrap()); + assert!(persist_message(tmp.path(), "m1", "@peer", &session, "now") + .unwrap() + .is_some()); // Replay of the same relay id does not double-insert. - assert!(!persist_message(tmp.path(), "m1", "@peer", &session, "now").unwrap()); - assert!(persist_message(tmp.path(), "m2", "@peer", &master, "now").unwrap()); + assert!(persist_message(tmp.path(), "m1", "@peer", &session, "now") + .unwrap() + .is_none()); + assert!(persist_message(tmp.path(), "m2", "@peer", &master, "now") + .unwrap() + .is_some()); store::with_connection(tmp.path(), |c| { assert_eq!(store::count_messages(c, "@peer", "w1")?, 1); // per-pair wrapper id @@ -1361,8 +1454,12 @@ mod tests { assert_eq!(first.seq, 0); // wire line is 0 for both … assert_eq!(second.seq, 0); - assert!(persist_message(tmp.path(), "mA", "@peer", &first, "now").unwrap()); - assert!(persist_message(tmp.path(), "mB", "@peer", &second, "now").unwrap()); + assert!(persist_message(tmp.path(), "mA", "@peer", &first, "now") + .unwrap() + .is_some()); + assert!(persist_message(tmp.path(), "mB", "@peer", &second, "now") + .unwrap() + .is_some()); store::with_connection(tmp.path(), |c| { // … but the persisted seqs are monotonic ingest ordinals 1 and 2, and diff --git a/src/openhuman/orchestration/mod.rs b/src/openhuman/orchestration/mod.rs index 94ff84e5a..e62cc3b48 100644 --- a/src/openhuman/orchestration/mod.rs +++ b/src/openhuman/orchestration/mod.rs @@ -12,6 +12,8 @@ pub mod attention; pub mod bus; +pub mod cloud; +pub mod effect_executor; pub mod frontend_agent; pub mod graph; pub mod ingest; @@ -25,6 +27,7 @@ pub mod steering; pub mod store; pub mod tools; pub mod types; +pub mod wire; pub use bus::{ notify_orchestration_message, register_orchestration_ingest_subscriber, diff --git a/src/openhuman/orchestration/ops.rs b/src/openhuman/orchestration/ops.rs index e5e8fecc7..9ee64b30b 100644 --- a/src/openhuman/orchestration/ops.rs +++ b/src/openhuman/orchestration/ops.rs @@ -1343,7 +1343,7 @@ impl OrchestrationRuntime for ProductionRuntime { /// Wire body for an agent reply into `session_id`: a v1 session envelope for a /// real harness session (so the peer threads its reply under the same id), or /// the plain body for the pinned Master / subconscious windows. -fn session_send_plaintext(session_id: &str, body: &str) -> anyhow::Result { +pub(crate) fn session_send_plaintext(session_id: &str, body: &str) -> anyhow::Result { if session_id == "master" || session_id == "subconscious" { return Ok(body.to_string()); } diff --git a/src/openhuman/orchestration/wire.rs b/src/openhuman/orchestration/wire.rs new file mode 100644 index 000000000..989a27e1a --- /dev/null +++ b/src/openhuman/orchestration/wire.rs @@ -0,0 +1,154 @@ +//! Sanitized wire DTOs for the hosted orchestration brain. +//! +//! # Security choke point +//! +//! This file is the trust-boundary allowlist between the device and the backend. +//! Every field that crosses to `POST /orchestration/v1/events` is constructed +//! **explicitly** here — there is no `#[derive(Serialize)]` over an internal +//! state struct that could leak a newly-added field. Signal key material, +//! credentials, local filesystem paths, and workspace locations are NEVER part +//! of these structs, and the golden key-set test below asserts the exact JSON +//! shape so a future field addition fails loudly in review. + +use serde::Serialize; +use serde_json::Value; + +/// Wire-protocol version this client speaks. Must fall within the backend's +/// advertised `[min, max]`; a mismatch yields `409 ORCH_PROTOCOL_MISMATCH`. +pub const ORCH_WIRE_PROTOCOL: u8 = 1; + +/// The inner `event` object of an ingest upload. Field-for-field the backend's +/// `orchestrationEventSchema`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct OrchestrationEventWire { + /// Store-assigned monotonic per-session ordinal (the idempotency key). + pub seq: i64, + /// One of `user` | `assistant` | `system` (clamped on build). + pub role: String, + /// The event's originator — for an inbound DM, the counterpart agent id. + pub sender: String, + /// Decrypted plaintext body. Never key material or a local path. + pub body: String, + /// Client event timestamp, epoch milliseconds. + pub ts: i64, + /// Event kind (`dm`, a v2 harness `event.kind`, …). Free-form, ≤64 chars. + pub kind: String, +} + +/// The full `POST /orchestration/v1/events` request body. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct OrchestrationEventEnvelopeWire { + pub protocol: u8, + /// The paired agent this session is with. Not a credential — a public handle. + pub counterpart_agent_id: String, + pub session_id: String, + pub event: OrchestrationEventWire, +} + +/// Clamp an arbitrary role string to the backend's accepted enum. Unknown roles +/// (or empty) default to `user` so a malformed harness role never trips a 400. +fn sanitize_role(role: &str) -> String { + match role { + "user" | "assistant" | "system" => role.to_string(), + _ => "user".to_string(), + } +} + +impl OrchestrationEventEnvelopeWire { + /// Build a sanitized envelope from primitive fields. Constructing from + /// primitives (rather than an internal state struct) is deliberate: it keeps + /// the crossed field set an explicit, auditable allowlist. + #[allow(clippy::too_many_arguments)] + pub fn build( + counterpart_agent_id: &str, + session_id: &str, + seq: i64, + role: &str, + sender: &str, + body: &str, + ts: i64, + kind: &str, + ) -> Self { + Self { + protocol: ORCH_WIRE_PROTOCOL, + counterpart_agent_id: counterpart_agent_id.to_string(), + session_id: session_id.to_string(), + event: OrchestrationEventWire { + seq: seq.max(0), + role: sanitize_role(role), + sender: sender.to_string(), + body: body.to_string(), + ts: ts.max(0), + kind: if kind.is_empty() { + "message".to_string() + } else { + kind.to_string() + }, + }, + } + } + + /// Serialize to a JSON value for the HTTP body. + pub fn to_value(&self) -> Value { + // Infallible for this closed struct of primitives. + serde_json::to_value(self).expect("orchestration wire envelope serializes") + } +} + +/// Parse an RFC3339 timestamp into epoch milliseconds. Pure; returns `None` on +/// a malformed input so the caller can fall back (e.g. to the ingest clock). +pub fn parse_ts_ms(ts: &str) -> Option { + chrono::DateTime::parse_from_rfc3339(ts) + .ok() + .map(|dt| dt.timestamp_millis()) +} + +/// One world-state-diff entry crossing to `POST /orchestration/v1/world-diff`. +/// `note` is a short derived observation (never a local path or key material). +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct WorldDiffEntryWire { + pub seq: i64, + pub note: String, + pub ts: i64, +} + +impl WorldDiffEntryWire { + pub fn build(seq: i64, note: &str, ts: i64) -> Self { + Self { + seq: seq.max(0), + note: note.to_string(), + ts: ts.max(0), + } + } +} + +/// The full `POST /orchestration/v1/world-diff` request body. Same allowlist +/// discipline as the event envelope — only these fields are constructed. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WorldDiffBatchWire { + pub protocol: u8, + pub session_id: String, + pub entries: Vec, +} + +impl WorldDiffBatchWire { + pub fn build(session_id: &str, entries: Vec) -> Self { + Self { + protocol: ORCH_WIRE_PROTOCOL, + session_id: session_id.to_string(), + entries, + } + } + + pub fn to_value(&self) -> Value { + serde_json::to_value(self).expect("orchestration world-diff batch serializes") + } +} + +// The wire allowlist is guarded by the golden key-set test in +// `tests/orchestration_shadow_push_e2e.rs` (integration crate): the root +// crate's `cfg(test)` build is currently blocked by unrelated stale test +// modules at this checkout, so the security-critical assertions live where they +// can actually run — against the compiled lib. diff --git a/src/openhuman/socket/event_handlers.rs b/src/openhuman/socket/event_handlers.rs index 461efd82f..f4e02ff4e 100644 --- a/src/openhuman/socket/event_handlers.rs +++ b/src/openhuman/socket/event_handlers.rs @@ -23,7 +23,7 @@ use super::manager::{emit_server_event, emit_state_change, SharedState}; pub(super) fn handle_sio_event( event_name: &str, data: serde_json::Value, - _emit_tx: &mpsc::UnboundedSender, + emit_tx: &mpsc::UnboundedSender, shared: &Arc, ) { // Log every incoming event for observability. @@ -61,12 +61,44 @@ pub(super) fn handle_sio_event( log::info!("[socket] Server ready — auth successful"); *shared.status.write() = ConnectionStatus::Connected; emit_state_change(shared); + // Declare the device-tool manifest so the hosted brain knows which + // tool calls to round-trip to this device (Phase 2). Sent every + // (re)connect so the server's view is rebuilt from scratch. + emit_via_channel( + emit_tx, + "orch:register_tools", + crate::openhuman::orchestration::effect_executor::device_tool_manifest(), + ); } "error" => { log::error!("[socket] Server error event: {}", data); *shared.status.write() = ConnectionStatus::Error; emit_state_change(shared); } + // Hosted-brain device effect: relay a reply over Signal, then ack. Runs + // async so the recv loop isn't blocked on the send; the ack rides back + // over the same socket. Device Signal keys never leave the machine. + "orch:effect:send_dm" => { + let tx = emit_tx.clone(); + tokio::spawn(async move { + if let Some((call_id, ack)) = + crate::openhuman::orchestration::effect_executor::handle_send_dm(&data).await + { + log::debug!("[socket] orch:effect:send_dm acked call_id={call_id}"); + emit_via_channel(&tx, "orch:effect:result", ack); + } + }); + } + // 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); + } + } // Webhook tunnel — publish to event bus for routing by WebhookRequestSubscriber "webhook:request" => { log::info!("[socket] Publishing webhook:request to event bus"); diff --git a/tests/orchestration_effect_executor_e2e.rs b/tests/orchestration_effect_executor_e2e.rs new file mode 100644 index 000000000..2bbbe5ade --- /dev/null +++ b/tests/orchestration_effect_executor_e2e.rs @@ -0,0 +1,99 @@ +//! Phase 1 client effect-executor coverage: parsing the hosted brain's +//! `orch:effect:send_dm` frame, the ack frame it returns, the at-least-once +//! callId dedupe, and the device-tool manifest. Lives in an integration crate +//! (links the compiled lib) because the root cfg(test) build is blocked by +//! unrelated stale test modules at this checkout. + +use openhuman_core::openhuman::orchestration::effect_executor::{ + device_tool_manifest, dispatch_device_tool, effect_result_frame, handle_tool_call, + is_duplicate_call, parse_send_dm, parse_tool_call, tool_result_frame, +}; +use serde_json::json; + +#[test] +fn parses_a_well_formed_send_dm_frame() { + let frame = json!({ + "cycleId": "cyc:agent-alice:sess-1:3", + "callId": "cyc:agent-alice:sess-1:3:send_dm:0", + "counterpartAgentId": "agent-alice", + "sessionId": "sess-1", + "body": "on it" + }); + let effect = parse_send_dm(&frame).expect("parse"); + assert_eq!(effect.call_id, "cyc:agent-alice:sess-1:3:send_dm:0"); + assert_eq!(effect.counterpart_agent_id, "agent-alice"); + assert_eq!(effect.session_id, "sess-1"); + assert_eq!(effect.body, "on it"); +} + +#[test] +fn rejects_a_frame_missing_required_fields() { + let frame = json!({ "cycleId": "c", "body": "hi" }); // no callId / counterpartAgentId + assert!(parse_send_dm(&frame).is_err()); +} + +#[test] +fn ack_frame_shapes_ok_and_error_cases() { + assert_eq!( + effect_result_frame("call-1", true, None), + json!({ "callId": "call-1", "ok": true, "error": null }) + ); + assert_eq!( + effect_result_frame("call-2", false, Some("device offline")), + json!({ "callId": "call-2", "ok": false, "error": "device offline" }) + ); +} + +#[test] +fn dedupe_reports_first_call_new_and_repeat_duplicate() { + // Unique id so the process-global dedupe set can't collide with other tests. + let id = "dedupe-test-unique-call-id-abc123"; + assert!(!is_duplicate_call(id), "first sighting is not a duplicate"); + assert!(is_duplicate_call(id), "second sighting is a duplicate"); +} + +#[test] +fn manifest_declares_a_queryable_device_tool() { + let manifest = device_tool_manifest(); + let tools = manifest["tools"].as_array().expect("tools array"); + assert!(tools.iter().any(|t| t["name"] == "device_status")); +} + +#[test] +fn parses_a_tool_call_frame() { + let frame = json!({ + "cycleId": "c", + "callId": "c:tool_call:0", + "name": "device_status", + "args": {} + }); + let parsed = parse_tool_call(&frame).expect("parse"); + assert_eq!(parsed.call_id, "c:tool_call:0"); + assert_eq!(parsed.name, "device_status"); +} + +#[test] +fn dispatches_device_status_and_rejects_unknown_tools() { + let status = dispatch_device_tool("device_status", &json!({})).expect("ok"); + assert!(status["version"].is_string()); + assert!(status["platform"].is_string()); + + assert!(dispatch_device_tool("rm_rf", &json!({})).is_err()); +} + +#[test] +fn handle_tool_call_builds_result_frame() { + let frame = json!({ "callId": "c:tool_call:0", "name": "device_status", "args": {} }); + let (call_id, result) = handle_tool_call(&frame).expect("handled"); + assert_eq!(call_id, "c:tool_call:0"); + assert_eq!(result["ok"], json!(true)); + assert!(result["result"]["platform"].is_string()); +} + +#[test] +fn tool_result_frame_shapes_error_case() { + assert_eq!( + tool_result_frame("c1", false, json!(null), Some("boom")), + json!({ "callId": "c1", "ok": false, "result": null, "error": "boom" }) + ); +} diff --git a/tests/orchestration_shadow_push_e2e.rs b/tests/orchestration_shadow_push_e2e.rs new file mode 100644 index 000000000..efeb36877 --- /dev/null +++ b/tests/orchestration_shadow_push_e2e.rs @@ -0,0 +1,221 @@ +//! Phase 0 shadow-migration coverage: the sanitized wire DTO (the security +//! allowlist gate) and the cloud event pusher's transport. +//! +//! These live in an integration crate rather than inline `#[cfg(test)]` modules +//! because the root crate's test build is currently blocked by unrelated stale +//! test modules at this checkout. An integration test links the *compiled* lib +//! (no `cfg(test)` siblings), so it runs cleanly here. + +use std::time::Duration; + +use openhuman_core::api::rest::BackendOAuthClient; +use openhuman_core::openhuman::orchestration::cloud::{push_event_with, push_world_diff_with}; +use openhuman_core::openhuman::orchestration::wire::{ + parse_ts_ms, OrchestrationEventEnvelopeWire, WorldDiffBatchWire, WorldDiffEntryWire, + ORCH_WIRE_PROTOCOL, +}; +use wiremock::matchers::{body_json, header, method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +fn envelope() -> OrchestrationEventEnvelopeWire { + OrchestrationEventEnvelopeWire::build( + "agent-alice", + "sess-1", + 3, + "user", + "agent-alice", + "hey there", + 1_700_000_000_000, + "dm", + ) +} + +// ── Wire DTO: the security allowlist gate ────────────────────────────────── + +#[test] +fn wire_envelope_has_exactly_the_allowlisted_keys() { + let value = envelope().to_value(); + let obj = value.as_object().expect("object"); + + let mut top: Vec<&str> = obj.keys().map(String::as_str).collect(); + top.sort_unstable(); + assert_eq!( + top, + ["counterpartAgentId", "event", "protocol", "sessionId"] + ); + + let event = obj["event"].as_object().expect("event object"); + let mut event_keys: Vec<&str> = event.keys().map(String::as_str).collect(); + event_keys.sort_unstable(); + assert_eq!(event_keys, ["body", "kind", "role", "sender", "seq", "ts"]); + + assert_eq!(obj["protocol"], serde_json::json!(ORCH_WIRE_PROTOCOL)); +} + +#[test] +fn wire_envelope_never_carries_credential_or_path_keys() { + let value = envelope().to_value(); + let obj = value.as_object().unwrap(); + let event = obj["event"].as_object().unwrap(); + for forbidden in [ + "path", + "cwd", + "workspace", + "workspaceDir", + "credentials", + "token", + "apiKey", + "key", + "signalKey", + "identityKey", + "ratchet", + "secret", + ] { + assert!(!obj.contains_key(forbidden), "top-level leaked {forbidden}"); + assert!(!event.contains_key(forbidden), "event leaked {forbidden}"); + } +} + +#[test] +fn wire_build_clamps_unknown_role_and_negative_scalars() { + let env = OrchestrationEventEnvelopeWire::build("a", "s", -5, "robot", "a", "", -1, ""); + assert_eq!(env.event.role, "user"); + assert_eq!(env.event.seq, 0); + assert_eq!(env.event.ts, 0); + assert_eq!(env.event.kind, "message"); + assert_eq!(env.protocol, 1); +} + +#[test] +fn wire_build_preserves_valid_roles() { + for role in ["user", "assistant", "system"] { + let env = OrchestrationEventEnvelopeWire::build("a", "s", 1, role, "a", "b", 1, "dm"); + assert_eq!(env.event.role, role); + } +} + +#[test] +fn parse_ts_ms_parses_rfc3339_and_rejects_garbage() { + assert_eq!(parse_ts_ms("1970-01-01T00:00:01Z"), Some(1000)); + assert_eq!(parse_ts_ms("not-a-timestamp"), None); +} + +// ── Cloud pusher transport ───────────────────────────────────────────────── + +#[tokio::test] +async fn push_posts_sanitized_event_with_bearer_and_accepts_202() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/orchestration/v1/events")) + .and(header("authorization", "Bearer test-token")) + .and(body_json(serde_json::json!({ + "protocol": 1, + "counterpartAgentId": "agent-alice", + "sessionId": "sess-1", + "event": { + "seq": 3, + "role": "user", + "sender": "agent-alice", + "body": "hey there", + "ts": 1_700_000_000_000i64, + "kind": "dm" + } + }))) + .respond_with(ResponseTemplate::new(202).set_body_json(serde_json::json!({ + "success": true, + "data": { "accepted": true, "cycleId": "cyc:agent-alice:sess-1:3" } + }))) + .expect(1) + .mount(&server) + .await; + + let client = BackendOAuthClient::new(&server.uri()).unwrap(); + let res = push_event_with(&client, "test-token", &envelope(), &[]).await; + assert!(res.is_ok(), "expected ok, got {res:?}"); +} + +#[test] +fn world_diff_batch_has_exactly_the_allowlisted_keys() { + let batch = WorldDiffBatchWire::build( + "sess-1", + vec![WorldDiffEntryWire::build(0, "peer online", 1)], + ); + let value = batch.to_value(); + let obj = value.as_object().unwrap(); + let mut top: Vec<&str> = obj.keys().map(String::as_str).collect(); + top.sort_unstable(); + assert_eq!(top, ["entries", "protocol", "sessionId"]); + + let entry = obj["entries"][0].as_object().unwrap(); + let mut ek: Vec<&str> = entry.keys().map(String::as_str).collect(); + ek.sort_unstable(); + assert_eq!(ek, ["note", "seq", "ts"]); + assert_eq!(obj["protocol"], serde_json::json!(ORCH_WIRE_PROTOCOL)); +} + +#[tokio::test] +async fn push_world_diff_posts_the_batch() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/orchestration/v1/world-diff")) + .and(header("authorization", "Bearer test-token")) + .and(body_json(serde_json::json!({ + "protocol": 1, + "sessionId": "sess-1", + "entries": [{ "seq": 0, "note": "peer online", "ts": 1i64 }] + }))) + .respond_with(ResponseTemplate::new(202).set_body_json(serde_json::json!({ + "success": true, "data": { "accepted": 1, "duplicates": 0, "tickScheduled": false } + }))) + .expect(1) + .mount(&server) + .await; + + let client = BackendOAuthClient::new(&server.uri()).unwrap(); + let batch = WorldDiffBatchWire::build( + "sess-1", + vec![WorldDiffEntryWire::build(0, "peer online", 1)], + ); + let res = push_world_diff_with(&client, "test-token", &batch, &[]).await; + assert!(res.is_ok(), "expected ok, got {res:?}"); +} + +#[tokio::test] +async fn push_returns_err_when_backend_5xxs_and_retries_exhausted() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/orchestration/v1/events")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + + let client = BackendOAuthClient::new(&server.uri()).unwrap(); + // No backoffs → single attempt, fails fast (no real sleeps). + let res = push_event_with(&client, "test-token", &envelope(), &[]).await; + assert!(res.is_err(), "expected err on 500"); +} + +#[tokio::test] +async fn push_retries_then_succeeds() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/orchestration/v1/events")) + .respond_with(ResponseTemplate::new(503)) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/orchestration/v1/events")) + .respond_with(ResponseTemplate::new(202).set_body_json(serde_json::json!({ + "success": true, "data": { "accepted": true, "cycleId": "c" } + }))) + .expect(1) + .mount(&server) + .await; + + let client = BackendOAuthClient::new(&server.uri()).unwrap(); + // Zero-duration backoff → immediate retry, still exercises the loop. + let res = push_event_with(&client, "test-token", &envelope(), &[Duration::ZERO]).await; + assert!(res.is_ok(), "expected ok after retry, got {res:?}"); +}