feat(orchestration): thin hosted-brain client — pushers, effect/tool executors, wire allowlist (#4725)

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(),

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.

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<SendDmEffect, String> {
+    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<ToolCallFrame, String> {
+    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<Value, String> {
+    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<Option<HashSet<String>>> = 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<String, String> {
+    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()),
+    ))
+}

src/openhuman/orchestration/ingest.rs

@@ -456,13 +456,17 @@ fn non_empty(s: String) -> Option<String> {
 
 /// 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<bool, String> {
+) -> Result<Option<i64>, 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

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,

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<String> {
+pub(crate) fn session_send_plaintext(session_id: &str, body: &str) -> anyhow::Result<String> {
     if session_id == "master" || session_id == "subconscious" {
         return Ok(body.to_string());
     }

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<i64> {
+    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<WorldDiffEntryWire>,
+}
+
+impl WorldDiffBatchWire {
+    pub fn build(session_id: &str, entries: Vec<WorldDiffEntryWire>) -> 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.

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<String>,
+    emit_tx: &mpsc::UnboundedSender<String>,
     shared: &Arc<SharedState>,
 ) {
     // 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");

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

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