fix(orchestration): reliable agent-to-agent DM replies (surface, correct content, resilience) — Closes #4583 (#4599)

src/openhuman/orchestration/ingest.rs

@@ -92,53 +92,51 @@ fn persist_message(
     now: &str,
 ) -> Result<bool, String> {
     store::with_connection(workspace_dir, |c| {
-        // Invariant guard (session windows only): the wake cursor keys on `session_id`
-        // while `seq` (= `env.message.line`) is monotonic only within one emitting
-        // harness. `upsert_session` clamps `last_seq` with `MAX(..)` and the wake fires
-        // only on `last_seq > cursor`, so a non-monotonic inbound `seq` (a peer reusing
-        // one `wrapper_session_id` across harness sessions, or a plugin-composed message
-        // whose line is 0) is persisted + acked but can SILENTLY skip the graph. Log it
-        // so the break surfaces in prod instead of dropping quietly. Robust fix (a
-        // per-wrapper monotonic ingest cursor) tracked in #4583.
-        if classified.chat_kind == ChatKind::Session {
-            if let Ok(Some(existing_last_seq)) =
-                store::session_last_seq(c, agent_id, &classified.session_id)
-            {
-                if classified.seq <= existing_last_seq {
-                    log::warn!(
-                        target: LOG,
-                        "[orchestration] wrapper session {} got seq {} <= last_seq {} — possible cross-harness reuse; wake may skip",
-                        classified.session_id, classified.seq, existing_last_seq
-                    );
-                }
-            }
-        }
-        store::upsert_session(
-            c,
-            &OrchestrationSession {
-                session_id: classified.session_id.clone(),
-                agent_id: agent_id.to_string(),
-                source: classified.source.clone(),
-                label: classified.label.clone(),
-                workspace: classified.workspace.clone(),
-                last_seq: classified.seq,
-                created_at: now.to_string(),
-                last_message_at: classified.timestamp.clone(),
-            },
-        )?;
-        store::insert_message(
-            c,
-            &OrchestrationMessage {
-                id: msg_id.to_string(),
-                agent_id: agent_id.to_string(),
-                session_id: classified.session_id.clone(),
-                chat_kind: classified.chat_kind,
-                role: classified.role.clone(),
-                body: classified.body.clone(),
-                timestamp: classified.timestamp.clone(),
-                seq: classified.seq,
-            },
-        )
+        // Wake idempotence keys on a per-session `seq` being monotonic, but the
+        // harness `message.line` we classify into `seq` is NOT reliable: a wrapped
+        // Claude harness stamps `line = 0` on every DM, and a peer reusing one
+        // `wrapper_session_id` across harness sessions can reset it. Under the
+        // shared per-pair session key that collapses every message into one
+        // session whose `last_seq`/wake cursor then pins at 0, so after the first
+        // message the graph is skipped and the DM is silently dropped (#4583).
+        //
+        // Fix: ignore the wire `line` for ordering and stamp a store-assigned,
+        // strictly-increasing per-(agent, session) ingest ordinal. Messages are
+        // append-only and deduped-by-id upstream, so `MAX(seq)+1` is monotonic and
+        // every genuinely-new DM advances `last_seq` past the cursor → wakes the graph.
+        //
+        // Allocate the ordinal and write both rows in one IMMEDIATE txn so a
+        // concurrent writer on the same session (the drain here vs the graph's
+        // `send_dm` reply persist) can't read the same `MAX(seq)` and duplicate it.
+        store::in_immediate_txn(c, |c| {
+            let ingest_seq = store::next_session_seq(c, agent_id, &classified.session_id)?;
+            store::upsert_session(
+                c,
+                &OrchestrationSession {
+                    session_id: classified.session_id.clone(),
+                    agent_id: agent_id.to_string(),
+                    source: classified.source.clone(),
+                    label: classified.label.clone(),
+                    workspace: classified.workspace.clone(),
+                    last_seq: ingest_seq,
+                    created_at: now.to_string(),
+                    last_message_at: classified.timestamp.clone(),
+                },
+            )?;
+            store::insert_message(
+                c,
+                &OrchestrationMessage {
+                    id: msg_id.to_string(),
+                    agent_id: agent_id.to_string(),
+                    session_id: classified.session_id.clone(),
+                    chat_kind: classified.chat_kind,
+                    role: classified.role.clone(),
+                    body: classified.body.clone(),
+                    timestamp: classified.timestamp.clone(),
+                    seq: ingest_seq,
+                },
+            )
+        })
     })
     .map_err(|e| format!("persist: {e}"))
 }
@@ -343,24 +341,38 @@ mod tests {
     }
 
     #[test]
-    fn persist_still_stores_a_lower_seq_in_the_same_wrapper_session() {
-        // The non-monotonic-seq guard only WARNS — it must never block persistence.
-        // A later message on the same wrapper session with a LOWER seq (e.g. a
-        // plugin-composed line 0, or a different emitting harness) still lands.
+    fn persist_stamps_monotonic_ingest_seq_so_line_zero_dms_still_wake() {
+        // Regression for the silent drop (#4583). A wrapped Claude harness stamps
+        // `line = 0` on EVERY DM, so pre-fix both messages classified to seq 0;
+        // under the shared wrapper-session key the wake cursor pinned at 0 and the
+        // second message was persisted + acked but never woke the graph (no reply).
+        // Persist must ignore the wire `line` and stamp a strictly-increasing
+        // per-(agent, session) ingest ordinal so `last_seq` advances past the cursor.
         let tmp = tempfile::tempdir().unwrap();
-        let hi = classify_message(ENVELOPE.to_string(), "2026-07-02T09:00:00Z"); // seq 7
-        assert!(persist_message(tmp.path(), "m1", "@peer", &hi, "now").unwrap());
+        let line_zero = || {
+            classify_message(
+                ENVELOPE.replace("\"line\": 7", "\"line\": 0"),
+                "2026-07-02T09:00:00Z",
+            )
+        };
+        let first = line_zero();
+        let second = line_zero();
+        assert_eq!(first.seq, 0); // wire line is 0 for both …
+        assert_eq!(second.seq, 0);
 
-        let lo = classify_message(
-            ENVELOPE.replace("\"line\": 7", "\"line\": 3"),
-            "2026-07-02T09:00:00Z",
-        );
-        assert_eq!(lo.session_id, "w1");
-        assert_eq!(lo.seq, 3); // lower than the stored last_seq (7) → guard warns
-        assert!(persist_message(tmp.path(), "m9", "@peer", &lo, "now").unwrap());
+        assert!(persist_message(tmp.path(), "mA", "@peer", &first, "now").unwrap());
+        assert!(persist_message(tmp.path(), "mB", "@peer", &second, "now").unwrap());
 
         store::with_connection(tmp.path(), |c| {
-            assert_eq!(store::count_messages(c, "@peer", "w1")?, 2); // both stored despite the warn
+            // … but the persisted seqs are monotonic ingest ordinals 1 and 2, and
+            // last_seq advanced to 2 — so a wake cursor left at 1 sees new work.
+            assert_eq!(store::count_messages(c, "@peer", "w1")?, 2);
+            assert_eq!(store::session_last_seq(c, "@peer", "w1")?, Some(2));
+            let seqs: Vec<i64> = store::list_recent_messages(c, "@peer", "w1", 10)?
+                .iter()
+                .map(|m| m.seq)
+                .collect();
+            assert_eq!(seqs, vec![1, 2]);
             Ok(())
         })
         .unwrap();

src/openhuman/orchestration/ops.rs

@@ -111,8 +111,46 @@ pub async fn schedule_wake(agent_id: String, session_id: String, chat_kind: Stri
             log::debug!(target: LOG, "[orchestration] wake.coalesced key={key} gen={gen}");
             return;
         }
-        if let Err(e) = invoke_orchestration_graph(&config, &agent_id, &session_id).await {
-            log::warn!(target: LOG, "[orchestration] wake.run_failed session={session_id}: {e}");
+        // Retry on failure with backoff. A graph-run error (e.g. a transient
+        // relay HTTP 400 / rate-limit / Signal-session hiccup on send_dm) leaves
+        // the idempotence cursor unmoved, but the wake is one-shot and the DM was
+        // already acked from the relay — so without this the message is orphaned
+        // in silence with nothing to re-trigger it. The graph checkpoints every
+        // super-step under `orchestration:<session>`, so a retry resumes from the
+        // last good boundary (execute/compress already cached) and just re-attempts
+        // the failed tail — cheap, no repeated LLM work. Bail if a newer wake
+        // supersedes this one; it will reprocess the same (or a wider) window.
+        const WAKE_RETRY_BACKOFF_MS: [u64; 3] = [5_000, 15_000, 45_000];
+        let mut attempt = 0usize;
+        loop {
+            match invoke_orchestration_graph(&config, &agent_id, &session_id).await {
+                Ok(()) => break,
+                Err(e) => {
+                    if attempt >= WAKE_RETRY_BACKOFF_MS.len() {
+                        log::warn!(
+                            target: LOG,
+                            "[orchestration] wake.run_failed session={session_id} gave up after {} attempts: {e}",
+                            attempt + 1,
+                        );
+                        break;
+                    }
+                    let backoff = WAKE_RETRY_BACKOFF_MS[attempt];
+                    attempt += 1;
+                    log::warn!(
+                        target: LOG,
+                        "[orchestration] wake.run_failed session={session_id} attempt={attempt}/{} retrying in {backoff}ms: {e}",
+                        WAKE_RETRY_BACKOFF_MS.len() + 1,
+                    );
+                    tokio::time::sleep(std::time::Duration::from_millis(backoff)).await;
+                    if !is_latest_generation(&key, gen) {
+                        log::debug!(
+                            target: LOG,
+                            "[orchestration] wake.retry_superseded key={key} gen={gen}"
+                        );
+                        break;
+                    }
+                }
+            }
         }
     });
 }
@@ -588,6 +626,82 @@ impl ProductionRuntime {
             .await
             .map_err(|e| anyhow::anyhow!("{agent_id} run: {e}"))
     }
+
+    /// Persist the agent's own outgoing reply into the orchestration store so it
+    /// surfaces in the chat window (`orchestration_messages_list`) alongside the
+    /// inbound DMs. Ingest only persists inbound messages, so without this the
+    /// agent's replies never appear in the UI. Best-effort: a store error is
+    /// logged, never fails the (already-sent) DM. Does not trigger a wake — the
+    /// wake fires only on ingest's `OrchestrationSessionMessage`, not this write.
+    fn persist_outgoing_reply(&self, body: &str) {
+        let chat_kind = match self.session_id.as_str() {
+            "master" => ChatKind::Master,
+            "subconscious" => ChatKind::Subconscious,
+            _ => ChatKind::Session,
+        };
+        let now = chrono::Utc::now().to_rfc3339();
+        let msg_id = format!("orch-reply:{}", uuid::Uuid::new_v4());
+        // Allocate the ordinal + write both rows in one IMMEDIATE txn so this
+        // reply persist can't race the drain's inbound persist on the same
+        // session and duplicate `seq` (see `store::in_immediate_txn`).
+        let result = store::with_connection(&self.config.workspace_dir, |c| {
+            store::in_immediate_txn(c, |c| {
+                let seq = store::next_session_seq(c, &self.agent_id, &self.session_id)?;
+                store::upsert_session(
+                    c,
+                    &OrchestrationSession {
+                        session_id: self.session_id.clone(),
+                        agent_id: self.agent_id.clone(),
+                        source: String::new(),
+                        label: None,
+                        workspace: None,
+                        // Do NOT advance the wake-driven `last_seq` for our own
+                        // outbound reply: the wake cursor only tracks inbound seqs,
+                        // so bumping it here would make `ingest_cursor_lag` (and
+                        // `orchestration.status`) falsely report pending work until
+                        // the next inbound DM. `upsert_session` clamps with
+                        // `MAX(..)`, so 0 refreshes `last_message_at` only.
+                        last_seq: 0,
+                        created_at: now.clone(),
+                        last_message_at: now.clone(),
+                    },
+                )?;
+                store::insert_message(
+                    c,
+                    &OrchestrationMessage {
+                        id: msg_id.clone(),
+                        agent_id: self.agent_id.clone(),
+                        session_id: self.session_id.clone(),
+                        chat_kind,
+                        role: "owner".to_string(),
+                        body: body.to_string(),
+                        timestamp: now.clone(),
+                        seq,
+                    },
+                )
+            })
+        });
+        match result {
+            Ok(_) => {
+                // Fan the reply out to the renderer socket so the chat window
+                // live-refetches it. Without this the row lands in the store but
+                // the UI (which only refetches on `orchestration:message`) never
+                // surfaces it. Mirrors the inbound path and the send_master RPC.
+                super::bus::notify_orchestration_message(
+                    &self.agent_id,
+                    &self.session_id,
+                    chat_kind.as_str(),
+                );
+            }
+            Err(e) => {
+                log::warn!(
+                    target: LOG,
+                    "[orchestration] persist_outgoing_reply failed session={}: {e}",
+                    self.session_id
+                );
+            }
+        }
+    }
 }
 
 #[async_trait]
@@ -599,8 +713,17 @@ impl OrchestrationRuntime for ProductionRuntime {
              macro-instructions for the reasoning core.",
             render_transcript(state),
         );
-        self.run_agent_turn("frontend_agent", "hint:chat", "frontend", prompt)
-            .await
+        // Prefer the `defer_to_orchestrator` / `reply_to_channel` argument the
+        // model actually passed over `run_single`'s trailing narration.
+        let (raw, decision) = super::tools::with_decision_capture(self.run_agent_turn(
+            "frontend_agent",
+            "hint:chat",
+            "frontend",
+            prompt,
+        ))
+        .await;
+        let raw = raw?;
+        Ok(decision.unwrap_or(raw))
     }
 
     async fn frontend_compile(&self, state: &OrchestrationState) -> anyhow::Result<String> {
@@ -612,8 +735,18 @@ impl OrchestrationRuntime for ProductionRuntime {
             render_transcript(state),
             reply,
         );
-        self.run_agent_turn("frontend_agent", "hint:chat", "frontend", prompt)
-            .await
+        // The finished reply is the `reply_to_channel` argument the model passed,
+        // NOT the trailing "Done — sent to the session" narration `run_single`
+        // returns. Fall back to the raw text only if no decision tool fired.
+        let (raw, decision) = super::tools::with_decision_capture(self.run_agent_turn(
+            "frontend_agent",
+            "hint:chat",
+            "frontend",
+            prompt,
+        ))
+        .await;
+        let raw = raw?;
+        Ok(decision.unwrap_or(raw))
     }
 
     async fn execute(&self, state: &OrchestrationState) -> anyhow::Result<ExecuteOutcome> {
@@ -811,8 +944,10 @@ impl OrchestrationRuntime for ProductionRuntime {
         params.insert("plaintext".to_string(), Value::from(plaintext));
         crate::openhuman::tinyplace::handle_tinyplace_signal_send_message(params)
             .await
-            .map(|_| ())
-            .map_err(|e| anyhow::anyhow!("signal send: {e}"))
+            .map_err(|e| anyhow::anyhow!("signal send: {e}"))?;
+        // Record our own reply in the chat model so it shows in the UI.
+        self.persist_outgoing_reply(body);
+        Ok(())
     }
 }
 

src/openhuman/orchestration/store.rs

@@ -106,12 +106,45 @@ pub fn with_connection<T>(
     }
     let conn = Connection::open(&db_path)
         .with_context(|| format!("open orchestration DB: {}", db_path.display()))?;
+    // Concurrent writers (the drain ingesting an inbound DM vs the graph's
+    // `send_dm` persisting a reply) each open their own connection. Wait for a
+    // held write lock instead of erroring `SQLITE_BUSY`; paired with the
+    // IMMEDIATE txn in the seq-allocating writers this serialises
+    // `MAX(seq)+1 → INSERT` so `seq` stays unique per `(agent_id, session_id)`.
+    conn.busy_timeout(std::time::Duration::from_secs(5))
+        .context("set orchestration busy_timeout")?;
     conn.execute_batch(SCHEMA_DDL)
         .context("initialise orchestration schema")?;
     migrate(&conn).context("migrate orchestration schema")?;
     f(&conn)
 }
 
+/// Run `f` inside a single `BEGIN IMMEDIATE` transaction, rolling back on error.
+/// Use for read-then-write allocations (`MAX(seq)+1` then `INSERT`) so two
+/// concurrent writers on the same `(agent_id, session_id)` cannot read the same
+/// max and persist a duplicate `seq` (which would break the monotonic wake
+/// cursor). `IMMEDIATE` takes the write lock up front; the `busy_timeout` set in
+/// [`with_connection`] makes the loser wait for the holder to commit rather than
+/// fail.
+pub fn in_immediate_txn<T>(
+    conn: &Connection,
+    f: impl FnOnce(&Connection) -> Result<T>,
+) -> Result<T> {
+    conn.execute_batch("BEGIN IMMEDIATE")
+        .context("begin orchestration immediate txn")?;
+    match f(conn) {
+        Ok(value) => {
+            conn.execute_batch("COMMIT")
+                .context("commit orchestration txn")?;
+            Ok(value)
+        }
+        Err(e) => {
+            let _ = conn.execute_batch("ROLLBACK");
+            Err(e)
+        }
+    }
+}
+
 /// One-time, `user_version`-gated migrations. Runs after the idempotent
 /// `SCHEMA_DDL`; each version block executes exactly once per DB.
 fn migrate(conn: &Connection) -> Result<()> {
@@ -213,6 +246,21 @@ pub fn count_messages(conn: &Connection, agent_id: &str, session_id: &str) -> Re
     )?)
 }
 
+/// The next monotonic per-session ingest ordinal: `MAX(seq) + 1` over the
+/// session's messages (`1` for the first message). Stamped at persist time so
+/// the wake idempotence cursor rides a strictly-increasing value instead of the
+/// harness `message.line`, which is unreliable (a wrapped Claude harness stamps
+/// `line = 0` on every DM, and a peer can reuse/reset it across harness sessions
+/// under one shared `wrapper_session_id`). Messages are append-only and
+/// deduped-by-id before persist, so this is strictly increasing. (#4583)
+pub fn next_session_seq(conn: &Connection, agent_id: &str, session_id: &str) -> Result<i64> {
+    Ok(conn.query_row(
+        "SELECT COALESCE(MAX(seq), 0) + 1 FROM messages WHERE agent_id = ?1 AND session_id = ?2",
+        params![agent_id, session_id],
+        |row| row.get(0),
+    )?)
+}
+
 /// The current `last_seq` for a session, or `None` if the session row does not
 /// exist yet. Used to detect a non-monotonic inbound `seq` before the upsert
 /// clamps it away via `MAX(...)`.
@@ -1051,4 +1099,44 @@ mod tests {
         })
         .unwrap();
     }
+
+    #[test]
+    fn in_immediate_txn_serialises_concurrent_seq_allocation() {
+        // Two writers on the same (agent, session) — the drain's inbound persist
+        // and the graph's send_dm reply persist — must not read the same
+        // MAX(seq) and duplicate it. Allocate + insert under `in_immediate_txn`
+        // from several threads and assert every seq is distinct and contiguous.
+        use std::sync::Arc;
+        let tmp = Arc::new(tempfile::tempdir().unwrap());
+        let n = 8usize;
+        let handles: Vec<_> = (0..n)
+            .map(|i| {
+                let tmp = Arc::clone(&tmp);
+                std::thread::spawn(move || {
+                    with_connection(tmp.path(), |c| {
+                        in_immediate_txn(c, |c| {
+                            let seq = next_session_seq(c, "@peer", "s1")?;
+                            insert_message(c, &msg(&format!("m{i}"), "@peer", "s1", seq))?;
+                            Ok(seq)
+                        })
+                    })
+                    .expect("txn ok")
+                })
+            })
+            .collect();
+        let mut seqs: Vec<i64> = handles.into_iter().map(|h| h.join().unwrap()).collect();
+        seqs.sort_unstable();
+        let mut unique = seqs.clone();
+        unique.dedup();
+        assert_eq!(
+            unique.len(),
+            n,
+            "concurrent seq allocation must not duplicate: {seqs:?}"
+        );
+        assert_eq!(
+            seqs,
+            (1..=n as i64).collect::<Vec<_>>(),
+            "seqs must be a contiguous 1..=n range"
+        );
+    }
 }

src/openhuman/orchestration/tools.rs

@@ -13,11 +13,45 @@
 //! hook captures the tool name + argument. They carry no external effect — the
 //! actual DM send is the graph's `send_dm` node — so they stay `ReadOnly`.
 
+use std::future::Future;
+use std::sync::{Arc, Mutex};
+
 use async_trait::async_trait;
 use serde_json::{json, Value};
 
 use crate::openhuman::tools::{Tool, ToolResult};
 
+tokio::task_local! {
+    /// Task-local capture of the front end's decision payload. The decision
+    /// tools echo their argument as a `ToolResult`, but the split-brain graph
+    /// needs the exact `text` / `instructions` the model passed — NOT the
+    /// trailing narration the agent loop returns after the tool call (which is
+    /// what `run_single` yields). Each decision tool records its payload here.
+    static DECISION_CAPTURE: Arc<Mutex<Option<String>>>;
+}
+
+/// Scope a front-end decision capture around one front-end agent turn `fut`,
+/// returning `(turn_output, captured_payload)`. `captured_payload` is the
+/// argument the model passed to `reply_to_channel` / `defer_to_orchestrator`
+/// (the authoritative channel response / macro-instructions), or `None` when the
+/// turn ended without calling a decision tool (caller falls back to the raw text).
+pub async fn with_decision_capture<F: Future>(fut: F) -> (F::Output, Option<String>) {
+    let cell = Arc::new(Mutex::new(None));
+    let out = DECISION_CAPTURE.scope(cell.clone(), Box::pin(fut)).await;
+    let captured = cell.lock().ok().and_then(|mut slot| slot.take());
+    (out, captured)
+}
+
+/// Record a front-end decision payload from a decision tool. Last write wins
+/// (the turn's terminal decision). No-op outside a [`with_decision_capture`] scope.
+fn record_decision(payload: &str) {
+    let _ = DECISION_CAPTURE.try_with(|cell| {
+        if let Ok(mut slot) = cell.lock() {
+            *slot = Some(payload.to_string());
+        }
+    });
+}
+
 /// `reply_to_channel` — the front end's pass-2 terminal decision.
 pub struct ReplyToChannelTool;
 
@@ -58,7 +92,10 @@ impl Tool for ReplyToChannelTool {
 
     async fn execute(&self, args: Value) -> anyhow::Result<ToolResult> {
         match required_str(&args, "text") {
-            Ok(text) => Ok(ToolResult::success(text)),
+            Ok(text) => {
+                record_decision(&text);
+                Ok(ToolResult::success(text))
+            }
             Err(e) => Ok(e),
         }
     }
@@ -91,7 +128,10 @@ impl Tool for DeferToOrchestratorTool {
 
     async fn execute(&self, args: Value) -> anyhow::Result<ToolResult> {
         match required_str(&args, "instructions") {
-            Ok(instructions) => Ok(ToolResult::success(instructions)),
+            Ok(instructions) => {
+                record_decision(&instructions);
+                Ok(ToolResult::success(instructions))
+            }
             Err(e) => Ok(e),
         }
     }
@@ -123,4 +163,30 @@ mod tests {
         let bad = t.execute(json!({})).await.unwrap();
         assert!(bad.is_error);
     }
+
+    #[tokio::test]
+    async fn decision_capture_surfaces_tool_payload_not_turn_narration() {
+        // The runtime must send the `reply_to_channel` argument (the real reply),
+        // not the model's trailing "Done — sent to the session" narration that
+        // `run_single` returns. Reproduces the reply-plumbing bug.
+        let reply = ReplyToChannelTool;
+        let (turn_text, captured) = with_decision_capture(async {
+            let _ = reply
+                .execute(json!({"text": "the actual email summary"}))
+                .await
+                .unwrap();
+            "Done — the reply has been sent to the session".to_string()
+        })
+        .await;
+        assert_eq!(turn_text, "Done — the reply has been sent to the session");
+        assert_eq!(captured.as_deref(), Some("the actual email summary"));
+    }
+
+    #[tokio::test]
+    async fn decision_capture_is_none_without_a_decision_tool() {
+        let (turn_text, captured) =
+            with_decision_capture(async { "just narration".to_string() }).await;
+        assert_eq!(turn_text, "just narration");
+        assert_eq!(captured, None);
+    }
 }