From 057c3ffbf5c397bd8eeb3c5bf23cd735a4e847e2 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Mon, 6 Jul 2026 21:43:11 +0530 Subject: [PATCH] fix(orchestration): key inbound DMs on the shared per-pair session id (#4582) --- .github/workflows/pr-quality.yml | 2 + src/openhuman/orchestration/ingest.rs | 58 +++++++++++++++++++++++++-- src/openhuman/orchestration/store.rs | 17 ++++++++ src/openhuman/orchestration/types.rs | 31 ++++++++++++++ 4 files changed, 105 insertions(+), 3 deletions(-) diff --git a/.github/workflows/pr-quality.yml b/.github/workflows/pr-quality.yml index 7b0ac82a3..35b24d1dd 100644 --- a/.github/workflows/pr-quality.yml +++ b/.github/workflows/pr-quality.yml @@ -70,6 +70,8 @@ jobs: --exclude '^https://github\.com/tinyhumansai/homebrew-openhuman' --exclude '^https://www\.reddit\.com/r/tinyhumansai/?$' --exclude '^https://www\.star-history\.com/#tinyhumansai/openhuman&type=date&legend=top-left$' + --exclude '^https://github\.com/tinyhumansai/openhuman/stargazers' + --exclude '^https://api\.star-history\.com/' 'docs/**/*.md' 'src/**/README.md' '.github/PULL_REQUEST_TEMPLATE.md' diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index 1d55300d2..a2a0a94af 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -42,6 +42,9 @@ fn is_dm_stream(kind: &str, stream_id: &str) -> bool { fn classify_message(plaintext: String, fallback_timestamp: &str) -> ClassifiedMessage { match SessionEnvelopeV1::parse(&plaintext) { Some(env) => { + // Compute the session key while `env` is still fully intact (before any + // field moves below), since `session_key` borrows `&env`. + let session_id = env.session_key(); let label = (env.scope.scope_type == "folder").then(|| env.scope.key.clone()); let workspace = (!env.scope.cwd.is_empty()).then(|| env.scope.cwd.clone()); let timestamp = if env.message.timestamp.is_empty() { @@ -51,7 +54,11 @@ fn classify_message(plaintext: String, fallback_timestamp: &str) -> ClassifiedMe }; ClassifiedMessage { chat_kind: ChatKind::Session, - session_id: env.scope.harness_session_id, + // Key on the single per-pair session id (the shared `wrapper_session_id` + // both peers put on every message for a thread), so a reply threads back + // into the same session. Falls back to `harness_session_id` for a legacy + // envelope with no per-pair id. See `SessionEnvelopeV1::session_key`. + session_id, role: env.message.role, source: env.harness.provider, label, @@ -85,6 +92,27 @@ 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 + ); + } + } + } store::upsert_session( c, &OrchestrationSession { @@ -273,7 +301,7 @@ mod tests { fn classifies_harness_envelope_as_session() { let c = classify_message(ENVELOPE.to_string(), "2026-07-02T09:00:00Z"); assert_eq!(c.chat_kind, ChatKind::Session); - assert_eq!(c.session_id, "h1"); + assert_eq!(c.session_id, "w1"); // keyed on the shared per-pair wrapper_session_id assert_eq!(c.role, "user"); assert_eq!(c.source, "codex"); assert_eq!(c.label.as_deref(), Some("my-repo")); // folder scope → label @@ -307,10 +335,34 @@ mod tests { assert!(persist_message(tmp.path(), "m2", "@peer", &master, "now").unwrap()); store::with_connection(tmp.path(), |c| { - assert_eq!(store::count_messages(c, "@peer", "h1")?, 1); + assert_eq!(store::count_messages(c, "@peer", "w1")?, 1); // per-pair wrapper id assert_eq!(store::count_messages(c, "@peer", "master")?, 1); Ok(()) }) .unwrap(); } + + #[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. + 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 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()); + + store::with_connection(tmp.path(), |c| { + assert_eq!(store::count_messages(c, "@peer", "w1")?, 2); // both stored despite the warn + Ok(()) + }) + .unwrap(); + } } diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index 42d680ae5..22a40f1ea 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -213,6 +213,23 @@ pub fn count_messages(conn: &Connection, agent_id: &str, session_id: &str) -> Re )?) } +/// 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(...)`. +pub fn session_last_seq( + conn: &Connection, + agent_id: &str, + session_id: &str, +) -> Result> { + Ok(conn + .query_row( + "SELECT last_seq FROM sessions WHERE agent_id = ?1 AND session_id = ?2", + params![agent_id, session_id], + |row| row.get(0), + ) + .optional()?) +} + /// List every persisted session row, newest activity first (stage-7 read surface). pub fn list_sessions(conn: &Connection) -> Result> { let mut stmt = conn.prepare( diff --git a/src/openhuman/orchestration/types.rs b/src/openhuman/orchestration/types.rs index 806e143f7..21db30862 100644 --- a/src/openhuman/orchestration/types.rs +++ b/src/openhuman/orchestration/types.rs @@ -109,6 +109,19 @@ impl SessionEnvelopeV1 { envelope.is_valid_v1().then_some(envelope) } + /// The single per-pair session id to bucket an inbound message under. Every + /// message — whoever sent it — carries exactly one id: the shared conversation + /// id in `scope.wrapper_session_id`. Both peers put the SAME id there for a given + /// thread (the peer reuses it on reply), so it is the sole routing key. Falls + /// back to `harness_session_id` only for a legacy envelope that predates the + /// per-pair id. + pub fn session_key(&self) -> String { + if !self.scope.wrapper_session_id.is_empty() { + return self.scope.wrapper_session_id.clone(); + } + self.scope.harness_session_id.clone() + } + /// Build an outgoing v1 session envelope carrying `body` under `session_id`, /// so a compliant peer harness threads its reply under the same session. pub fn outgoing(session_id: &str, body: &str, message_id: &str, timestamp: &str) -> Self { @@ -240,4 +253,22 @@ mod tests { ) .is_none()); } + + #[test] + fn session_key_is_the_shared_wrapper_id_then_harness_fallback() { + // The single per-pair id lives in `wrapper_session_id`. + assert_eq!( + SessionEnvelopeV1::parse(SAMPLE).unwrap().session_key(), + "w1" + ); + // Legacy envelope with no per-pair id: fall back to the harness id. + let env = SessionEnvelopeV1::parse( + r#"{ + "envelope_version": "tinyplace.harness.session.v1", + "scope": { "harness_session_id": "h-only" } + }"#, + ) + .expect("valid v1"); + assert_eq!(env.session_key(), "h-only"); + } }