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

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