fix(orchestration): key inbound DMs on the shared per-pair session id (#4582)

.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'

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<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 {
@@ -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();
+    }
 }

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<Option<i64>> {
+    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<Vec<OrchestrationSession>> {
     let mut stmt = conn.prepare(

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