From b0c376df4f2151cd627f026b414308f4c741c4b1 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Tue, 7 Jul 2026 00:39:32 +0530 Subject: [PATCH 1/3] fix(orchestration): stamp a monotonic ingest seq so line=0 DMs aren't dropped MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A wrapped Claude harness stamps `message.line = 0` on every DM, so after #4582 keyed the session on the shared `wrapper_session_id` the wake idempotence cursor (`latest_seq > cursor`) pinned at 0 after the first message — every later DM was persisted + acked but silently skipped the graph (no reply). Replace the diagnostic-only WARN guard with a store-assigned, strictly-increasing per-`(agent, session)` ingest ordinal (`store::next_session_seq`), so every new DM advances `last_seq` past the cursor and wakes the graph. Self-migrating and also fixes the secondary `cycle_id` collision. Refs #4583 --- src/openhuman/orchestration/ingest.rs | 80 ++++++++++++++------------- src/openhuman/orchestration/store.rs | 15 +++++ 2 files changed, 58 insertions(+), 37 deletions(-) diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index a2a0a94afc..73ed1ff7de 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -92,27 +92,19 @@ fn persist_message( now: &str, ) -> Result { 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 - ); - } - } - } + // 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. + let ingest_seq = store::next_session_seq(c, agent_id, &classified.session_id)?; store::upsert_session( c, &OrchestrationSession { @@ -121,7 +113,7 @@ fn persist_message( source: classified.source.clone(), label: classified.label.clone(), workspace: classified.workspace.clone(), - last_seq: classified.seq, + last_seq: ingest_seq, created_at: now.to_string(), last_message_at: classified.timestamp.clone(), }, @@ -136,7 +128,7 @@ fn persist_message( role: classified.role.clone(), body: classified.body.clone(), timestamp: classified.timestamp.clone(), - seq: classified.seq, + seq: ingest_seq, }, ) }) @@ -343,24 +335,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 = store::list_recent_messages(c, "@peer", "w1", 10)? + .iter() + .map(|m| m.seq) + .collect(); + assert_eq!(seqs, vec![1, 2]); Ok(()) }) .unwrap(); diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index 22a40f1ea2..8835cfa381 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -213,6 +213,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 { + 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(...)`. From a06cedb4389cf0f4a8b5b5bf5d94c3dc9beaab20 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Tue, 7 Jul 2026 00:39:32 +0530 Subject: [PATCH 2/3] fix(orchestration): deliver the real reply, surface it in the UI, retry on failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three outbound-path fixes on top of #4582: - Reply content: the front end's channel response used `run_single`'s trailing narration ("Done — sent to the session") and discarded the `reply_to_channel` argument (the real answer). Capture the decision-tool payload via `with_decision_capture` and send that instead (same for `defer_to_orchestrator` → reasoning instructions). - Surface in UI: persist the agent's own outbound reply (`role = owner`) and emit `notify_orchestration_message`, so it appears in the orchestration chat window. Ingest only persisted inbound DMs and the UI live-refetches only on that socket event, so replies never showed. - Resilience: a transient send/relay/provider error (e.g. a staging `HTTP 400: body must be encrypted ciphertext` on `send_dm`) made the node throw → the whole wake failed, the cursor stayed put, and the one-shot wake left the message orphaned in silence. Retry the wake with backoff (5s/15s/45s); it resumes from the graph checkpoint so retries only re-attempt the failed tail. Refs #4583 --- src/openhuman/orchestration/ops.rs | 140 +++++++++++++++++++++++++-- src/openhuman/orchestration/tools.rs | 70 +++++++++++++- 2 files changed, 200 insertions(+), 10 deletions(-) diff --git a/src/openhuman/orchestration/ops.rs b/src/openhuman/orchestration/ops.rs index b087b9bd5a..c456320257 100644 --- a/src/openhuman/orchestration/ops.rs +++ b/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:`, 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,71 @@ 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()); + let result = store::with_connection(&self.config.workspace_dir, |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, + last_seq: seq, + 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 +702,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 { @@ -612,8 +724,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 { @@ -811,8 +933,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(()) } } diff --git a/src/openhuman/orchestration/tools.rs b/src/openhuman/orchestration/tools.rs index 3543f55075..e985872959 100644 --- a/src/openhuman/orchestration/tools.rs +++ b/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>>; +} + +/// 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(fut: F) -> (F::Output, Option) { + 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 { 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 { 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); + } } From 2c477580d86e751acd6893b44c30b83db5cf4a82 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Tue, 7 Jul 2026 01:03:16 +0530 Subject: [PATCH 3/3] =?UTF-8?q?fix(orchestration):=20address=20review=20?= =?UTF-8?q?=E2=80=94=20atomic=20seq=20allocation=20+=20don't=20bump=20wake?= =?UTF-8?q?=20last=5Fseq?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Wrap `MAX(seq)+1` → `INSERT` in a `BEGIN IMMEDIATE` txn (`store::in_immediate_txn`) with a `busy_timeout`, so the drain's inbound persist and the graph's `send_dm` reply persist can't race on the same `(agent, session)` and write a duplicate `seq` (which would break the monotonic wake cursor). Adds a multi-thread concurrency test. [CodeRabbit] - `persist_outgoing_reply` no longer advances `sessions.last_seq` for our own outbound reply — the wake cursor only tracks inbound seqs, so bumping it made `ingest_cursor_lag` / `orchestration.status` falsely report pending work until the next inbound DM. `upsert_session`'s `MAX(..)` clamp keeps the inbound `last_seq`; passing 0 refreshes `last_message_at` only. [Codex] Refs #4583 --- src/openhuman/orchestration/ingest.rs | 60 ++++++++++++---------- src/openhuman/orchestration/ops.rs | 65 ++++++++++++++---------- src/openhuman/orchestration/store.rs | 73 +++++++++++++++++++++++++++ 3 files changed, 144 insertions(+), 54 deletions(-) diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index 73ed1ff7de..6b74fb5e96 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -104,33 +104,39 @@ fn persist_message( // 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. - 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, - }, - ) + // + // 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}")) } diff --git a/src/openhuman/orchestration/ops.rs b/src/openhuman/orchestration/ops.rs index c456320257..fb42ffc342 100644 --- a/src/openhuman/orchestration/ops.rs +++ b/src/openhuman/orchestration/ops.rs @@ -641,34 +641,45 @@ impl ProductionRuntime { }; 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| { - 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, - last_seq: seq, - 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, - }, - ) + 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(_) => { diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index 8835cfa381..f494b38108 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -106,12 +106,45 @@ pub fn with_connection( } 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( + conn: &Connection, + f: impl FnOnce(&Connection) -> Result, +) -> Result { + 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<()> { @@ -1021,4 +1054,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 = 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::>(), + "seqs must be a contiguous 1..=n range" + ); + } }