From c790ed2a5d2482d812d75ccd5ed335a6ec7e5240 Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Sun, 9 Aug 2026 16:19:17 -0300 Subject: [PATCH] fix(state): recover gateway sessions stranded without a routing identity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When state.db's write path fails (corrupt FTS, or a crash landing between routing publication and row creation), the live gateway conversation can end up in a session row that never received its identity columns: session_key, chat_id, chat_type and origin_json are all NULL. In-memory routing hides the damage for as long as the gateway stays up. After a restart the chat is resolved from the DB, and find_latest_gateway_session_for_peer cannot see that row — both of its queries match on the very columns it lacks — so the chat resumes the last keyed sibling instead, days older. The messages were never lost, only unreachable. Hardening the write side cannot reach a row that is already damaged, so add the offline repair path the tracking issue asks for: - SessionDB.find_orphaned_gateway_sessions() reports message-bearing rows with no session_key, and names the predecessor each one continues only when the evidence is unambiguous — a recorded parent_session_id ("lineage"), or exactly one keyed row of the same source and compatible user_id that fell quiet within 15 minutes of the orphan's start ("contiguity"). Contested pairs are reported with a reason and left alone: a wrong adoption would splice one person's conversation into another person's chat. Branch, delegate and tool rows are excluded — they are unkeyed by design, not by damage. - SessionDB.adopt_orphaned_gateway_session() stamps the orphan from the predecessor (never overwriting a column that already has a value), records the lineage, and retires the predecessor under end_reason 'superseded_by_repair' — a reason recovery does not treat as resumable, so the repaired row wins the chat from then on. The pair is re-verified inside the write transaction, making a concurrent heal a no-op rather than a conflicting write. - `hermes sessions repair-routing` drives both. It reports without touching the database; --apply confirms first and warns that a running gateway still holds the old mapping in memory. Refs #82616. --- hermes_cli/main.py | 30 ++ hermes_cli/sessions_cmd.py | 46 +++ hermes_state.py | 229 +++++++++++++ .../test_orphan_gateway_session_repair.py | 310 ++++++++++++++++++ 4 files changed, 615 insertions(+) create mode 100644 tests/hermes_state/test_orphan_gateway_session_repair.py diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 9a560e9051..ba6dc07206 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -12327,6 +12327,36 @@ def main(): help="Skip the timestamped backup copy (not recommended)", ) + sessions_repair_routing = sessions_subparsers.add_parser( + "repair-routing", + help="Re-stamp gateway sessions that lost their routing identity", + description=( + "Find gateway conversations stranded in session rows whose " + "routing identity (session_key/chat_id/origin) was never " + "written — the damage a corrupt state.db write path leaves " + "behind (#82616). Such a row is invisible to restart recovery, " + "so the chat resumes an older session instead. Re-stamps each " + "orphan from the keyed predecessor it continues, and only when " + "that predecessor is unambiguous. Reports without touching the " + "database unless --apply is given." + ), + ) + sessions_repair_routing.add_argument( + "--apply", + action="store_true", + help="Perform the adoptions (default: report only)", + ) + sessions_repair_routing.add_argument( + "--max-gap-seconds", + type=float, + default=None, + help=( + "Window between a keyed predecessor's last activity and an " + "orphan's start for them to count as the same conversation " + "(default: 900)" + ), + ) + sessions_recover = sessions_subparsers.add_parser( "recover", help="Rebuild canonical session data into a separate clean database", diff --git a/hermes_cli/sessions_cmd.py b/hermes_cli/sessions_cmd.py index 18d457dc01..095ad5f9e7 100644 --- a/hermes_cli/sessions_cmd.py +++ b/hermes_cli/sessions_cmd.py @@ -1158,6 +1158,52 @@ def cmd_sessions(args, sessions_parser=None): print(" (VACUUM was skipped or failed — run " "`hermes sessions optimize` later to reclaim freed space.)") + elif action == "repair-routing": + records = db.find_orphaned_gateway_sessions( + max_gap_s=getattr(args, "max_gap_seconds", None) + ) + adoptable = [r for r in records if r["adoptable"]] + for record in records: + print(f"{record['orphan_id']} ({record['source']}, " + f"{record['message_count']} messages)") + if record["adoptable"]: + print(f" → adopt into {record['session_key']} " + f"(from {record['donor_id']}, " + f"evidence: {record['evidence']})") + else: + print(f" ✗ not repairable — {record['reason']}") + + if not records: + print("✓ No gateway sessions are missing their routing identity.") + elif not adoptable: + print(f"\n{len(records)} orphaned session(s) found, none " + "unambiguously repairable. Nothing to do.") + elif not getattr(args, "apply", False): + print(f"\n{len(adoptable)} of {len(records)} orphaned session(s) " + "can be repaired. Re-run with --apply to perform them.") + else: + # A running gateway holds the old routing mapping in memory and + # would write it back over the repair on its next save. + print("\nStop the gateway before applying — a running gateway " + "still holds the old routing mapping in memory.") + if _confirm_prompt( + f"Adopt {len(adoptable)} orphaned session(s)? [y/N] " + ): + repaired = 0 + for record in adoptable: + if db.adopt_orphaned_gateway_session( + record["orphan_id"], record["donor_id"] + ): + repaired += 1 + print(f"✓ {record['orphan_id']} now owns " + f"{record['session_key']}") + else: + print(f"✗ {record['orphan_id']} was not adopted " + "(the row changed since it was reported)") + print(f"\nRepaired {repaired} of {len(adoptable)} session(s).") + else: + print("Aborted — nothing was changed.") + elif action == "stats": total = db.session_count() msgs = db.message_count() diff --git a/hermes_state.py b/hermes_state.py index f8dd9eac5e..4733147521 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -4017,6 +4017,235 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) ).fetchone() return self._session_row_dict(row) if row else None + # ── Orphaned gateway-session repair (#82616) ────────────────────────── + # A write-path failure (corrupt FTS, crash between routing publication + # and row creation) can leave the live conversation in a session row + # that never received its identity columns. Both queries above require + # those columns, so the row holding the real transcript is invisible to + # recovery: the chat resolves to the last keyed row instead — days older + # — and the conversation time-travels. Hardening the write side cannot + # reach a row that is *already* damaged; these two methods are the + # offline repair path behind ``hermes sessions repair-routing``. + + # Widest plausible gap between a keyed predecessor going quiet and its + # unkeyed successor being minted. The reported incident gap was ~60s; + # 15 minutes stays generous without spanning unrelated conversations. + _ORPHAN_ADOPTION_MAX_GAP_S = 900.0 + + def find_orphaned_gateway_sessions( + self, *, max_gap_s: Optional[float] = None + ) -> List[Dict[str, Any]]: + """Report message-bearing session rows that lost their routing identity. + + A row is a candidate orphan when it has messages but no + ``session_key``. It is only *adoptable* when exactly one keyed + predecessor can be named as the conversation it continues: + + * ``lineage`` — ``parent_session_id`` points at a keyed row of the + same source. That is a recorded fact, so no time window applies. + * ``contiguity`` — exactly one keyed row of the same source (and + compatible ``user_id``) fell quiet within *max_gap_s* of the + orphan's start, and is older than the orphan's own last activity. + + Anything ambiguous is reported with ``adoptable=False`` and a reason + rather than guessed at: mis-adopting would splice one person's + conversation into another person's chat. Branch/delegate/tool rows + are excluded outright — they are unkeyed by design, not by damage. + """ + gap = ( + self._ORPHAN_ADOPTION_MAX_GAP_S + if max_gap_s is None + else float(max_gap_s) + ) + orphan_active = _sql_session_last_active("o") + donor_active = _sql_session_last_active("d") + donor_columns = ( + "d.id, d.session_key, d.chat_id, d.chat_type, d.thread_id, " + "d.user_id, d.origin_json, d.display_name, d.end_reason" + ) + records: List[Dict[str, Any]] = [] + + with self._lock: + orphans = self._conn.execute( + f""" + SELECT o.id, o.source, o.user_id, o.started_at, + o.parent_session_id, + {orphan_active} AS last_active, + (SELECT COUNT(*) FROM messages m + WHERE m.session_id = o.id) AS message_count + FROM sessions o + WHERE o.session_key IS NULL + AND EXISTS (SELECT 1 FROM messages m + WHERE m.session_id = o.id) + AND COALESCE(o.source, '') != 'tool' + AND json_extract(COALESCE(o.model_config, '{{}}'), + '$._branched_from') IS NULL + AND json_extract(COALESCE(o.model_config, '{{}}'), + '$._delegate_from') IS NULL + ORDER BY o.started_at ASC + """ + ).fetchall() + + for orphan in orphans: + donor = None + evidence = "" + reason = "" + + if orphan["parent_session_id"]: + evidence = "lineage" + donor = self._conn.execute( + f""" + SELECT {donor_columns} + FROM sessions d + WHERE d.id = ? + AND d.session_key IS NOT NULL + AND COALESCE(d.source, '') = COALESCE(?, '') + """, + (orphan["parent_session_id"], orphan["source"]), + ).fetchone() + if donor is None: + reason = ( + "parent session carries no gateway identity of " + "this source" + ) + else: + evidence = "contiguity" + candidates = self._conn.execute( + f""" + SELECT {donor_columns}, {donor_active} AS last_active + FROM sessions d + WHERE d.session_key IS NOT NULL + AND d.id != ? + AND COALESCE(d.source, '') = COALESCE(?, '') + AND (COALESCE(d.user_id, '') = '' + OR COALESCE(?, '') = '' + OR d.user_id = ?) + AND {donor_active} BETWEEN ? AND ? + AND {donor_active} < ? + ORDER BY last_active DESC + LIMIT 2 + """, + ( + orphan["id"], + orphan["source"], + orphan["user_id"], + orphan["user_id"], + (orphan["started_at"] or 0) - gap, + (orphan["started_at"] or 0) + gap, + orphan["last_active"], + ), + ).fetchall() + if not candidates: + reason = ( + f"no keyed predecessor fell quiet within {gap:.0f}s " + "of this session's start" + ) + elif len(candidates) > 1: + reason = ( + "ambiguous: more than one keyed predecessor " + "matches this window" + ) + else: + donor = candidates[0] + + records.append( + { + "orphan_id": orphan["id"], + "source": orphan["source"], + "message_count": orphan["message_count"], + "started_at": orphan["started_at"], + "last_active": orphan["last_active"], + "donor_id": donor["id"] if donor else None, + "session_key": donor["session_key"] if donor else None, + "evidence": evidence if donor else "", + "adoptable": donor is not None, + "reason": reason, + } + ) + + # Two unkeyed successors claiming the same predecessor means at most + # one of them continues that chat, and nothing here says which. + contested = { + r["donor_id"] + for r in records + if r["adoptable"] + and sum(1 for x in records if x["donor_id"] == r["donor_id"]) > 1 + } + for record in records: + if record["donor_id"] in contested: + record["adoptable"] = False + record["reason"] = ( + "ambiguous: more than one unkeyed session claims this " + "predecessor" + ) + return records + + def adopt_orphaned_gateway_session( + self, orphan_id: str, donor_id: str + ) -> bool: + """Stamp *orphan_id* with *donor_id*'s routing identity, retire *donor_id*. + + Re-verifies the pair inside the write transaction, so a concurrent + gateway that healed either row in the meantime turns this into a + no-op instead of a conflicting write. Existing non-NULL columns on + the orphan are preserved. Returns True when the adoption applied. + """ + if not orphan_id or not donor_id or orphan_id == donor_id: + return False + + def _do(conn): + donor = conn.execute( + "SELECT session_key, chat_id, chat_type, thread_id, user_id, " + "origin_json, display_name, source FROM sessions WHERE id = ?", + (donor_id,), + ).fetchone() + orphan = conn.execute( + "SELECT session_key, source FROM sessions WHERE id = ?", + (orphan_id,), + ).fetchone() + if donor is None or orphan is None: + return False + if not donor["session_key"] or orphan["session_key"]: + return False + if (donor["source"] or "") != (orphan["source"] or ""): + return False + + conn.execute( + """UPDATE sessions + SET session_key = ?, + chat_id = COALESCE(chat_id, ?), + chat_type = COALESCE(chat_type, ?), + thread_id = COALESCE(thread_id, ?), + user_id = COALESCE(user_id, ?), + origin_json = COALESCE(origin_json, ?), + display_name = COALESCE(display_name, ?), + parent_session_id = COALESCE(parent_session_id, ?) + WHERE id = ? AND session_key IS NULL""", + ( + donor["session_key"], + donor["chat_id"], + donor["chat_type"], + donor["thread_id"], + donor["user_id"], + donor["origin_json"], + donor["display_name"], + donor_id, + orphan_id, + ), + ) + # Retire the predecessor under a reason recovery does NOT treat + # as resumable — 'agent_close'/'ws_orphan_reap' would keep it in + # the running, and the newly keyed orphan could lose the chat + # again on the next restart. + conn.execute( + "UPDATE sessions SET ended_at = COALESCE(ended_at, ?), " + "end_reason = 'superseded_by_repair' WHERE id = ?", + (time.time(), donor_id), + ) + return True + + return self._execute_write(_do) + # Children that carry a ``parent_session_id`` but are NOT compression # continuations: branches, delegate/subagent runs, and tool sessions. # A marker only disqualifies a child when it points at the parent being diff --git a/tests/hermes_state/test_orphan_gateway_session_repair.py b/tests/hermes_state/test_orphan_gateway_session_repair.py new file mode 100644 index 0000000000..dabc4d2457 --- /dev/null +++ b/tests/hermes_state/test_orphan_gateway_session_repair.py @@ -0,0 +1,310 @@ +"""Repair of gateway sessions that lost their routing identity (#82616). + +Incident shape (production, Aug 2026): a state.db write-path failure left the +live Telegram DM in a session row that never received its identity columns. +That row is invisible to ``find_latest_gateway_session_for_peer`` — both of +its queries require the very columns the row lacks — so after a gateway +restart the chat resolved to a keyed sibling three days older and the +conversation time-travelled. The real messages were never lost, only +unreachable. + +These tests cover the offline repair path: detection must name the +predecessor only when the evidence is unambiguous, and adoption must make +the repaired row win recovery from then on. +""" + +import time + +import pytest + +from hermes_state import SessionDB + + +PEER = { + "source": "telegram", + "user_id": "6308981865", + "session_key": "agent:main:telegram:dm:6308981865", + "chat_id": "6308981865", + "chat_type": "dm", +} + + +@pytest.fixture +def db(tmp_path): + store = SessionDB(db_path=tmp_path / "state.db") + yield store + store.close() + + +def _mk_session( + db, + session_id, + *, + keyed=True, + source="telegram", + user_id=PEER["user_id"], + started_at=None, + last_activity_at=None, + ended_at=None, + end_reason=None, + parent_session_id=None, + model_config=None, + messages=0, +): + kwargs = {"user_id": user_id} + if keyed: + kwargs.update( + session_key=PEER["session_key"], + chat_id=PEER["chat_id"], + chat_type=PEER["chat_type"], + ) + if parent_session_id: + kwargs["parent_session_id"] = parent_session_id + if model_config: + kwargs["model_config"] = model_config + db.create_session(session_id, source, **kwargs) + for i in range(messages): + db.append_message( + session_id, "user" if i % 2 == 0 else "assistant", f"m{i}" + ) + with db._lock: + if keyed: + # ``create_session`` on main does not carry presentation/origin + # metadata into the INSERT — set it the way the gateway's + # per-turn peer refresh does, so the donor row is realistic. + db._conn.execute( + "UPDATE sessions SET origin_json = ?, display_name = ? " + "WHERE id = ?", + ( + '{"platform": "telegram", "chat_id": "6308981865"}', + "Teknium", + session_id, + ), + ) + if started_at is not None: + db._conn.execute( + "UPDATE sessions SET started_at = ? WHERE id = ?", + (started_at, session_id), + ) + if messages: + # Message timestamps feed the recency expression; pin them to the + # session window so tests control contiguity deterministically. + db._conn.execute( + "UPDATE messages SET timestamp = ? WHERE session_id = ?", + (last_activity_at or started_at, session_id), + ) + db._conn.execute( + "UPDATE sessions SET last_activity_at = ?, ended_at = ?, " + "end_reason = ? WHERE id = ?", + (last_activity_at, ended_at, end_reason, session_id), + ) + db._conn.commit() + + +def _incident(db, *, parent_link=False): + """Build the reported incident: keyed stale row + unkeyed live row.""" + now = time.time() + stale_last = now - 3 * 86400 + _mk_session( + db, + "20260803_120103_37976afb", + keyed=True, + started_at=now - 6 * 86400, + last_activity_at=stale_last, + end_reason="agent_close", + messages=4, + ) + _mk_session( + db, + "20260806_161836_04c1d6c6", + keyed=False, + started_at=stale_last + 60, + last_activity_at=now - 3600, + parent_session_id=( + "20260803_120103_37976afb" if parent_link else None + ), + messages=6, + ) + return "20260803_120103_37976afb", "20260806_161836_04c1d6c6" + + +def _last_activity(db, session_id): + with db._lock: + row = db._conn.execute( + "SELECT last_activity_at FROM sessions WHERE id = ?", (session_id,) + ).fetchone() + return row["last_activity_at"] + + +class TestDetection: + def test_incident_orphan_is_adoptable_by_contiguity(self, db): + stale, orphan = _incident(db) + records = db.find_orphaned_gateway_sessions() + assert len(records) == 1 + record = records[0] + assert record["orphan_id"] == orphan + assert record["donor_id"] == stale + assert record["session_key"] == PEER["session_key"] + assert record["evidence"] == "contiguity" + assert record["adoptable"] is True + assert record["message_count"] == 6 + + def test_parent_link_is_used_without_a_time_window(self, db): + stale, orphan = _incident(db, parent_link=True) + # Push the predecessor far outside the contiguity window: a recorded + # lineage is a fact, not a guess, so it must still resolve. + long_ago = time.time() - 400 * 86400 + with db._lock: + db._conn.execute( + "UPDATE sessions SET last_activity_at = ? WHERE id = ?", + (long_ago, stale), + ) + db._conn.execute( + "UPDATE messages SET timestamp = ? WHERE session_id = ?", + (long_ago, stale), + ) + db._conn.commit() + record = db.find_orphaned_gateway_sessions()[0] + assert record["orphan_id"] == orphan + assert record["donor_id"] == stale + assert record["evidence"] == "lineage" + assert record["adoptable"] is True + + def test_sessions_without_messages_are_not_reported(self, db): + _mk_session(db, "empty", keyed=False, started_at=time.time()) + assert db.find_orphaned_gateway_sessions() == [] + + def test_two_predecessors_in_the_window_fail_closed(self, db): + stale, _ = _incident(db) + quiet_at = _last_activity(db, stale) + # A second chat on the same platform fell quiet at the same moment. + _mk_session( + db, + "20260803_120104_other", + keyed=True, + user_id=None, + started_at=quiet_at - 3600, + last_activity_at=quiet_at, + end_reason="agent_close", + messages=2, + ) + with db._lock: + db._conn.execute( + "UPDATE sessions SET session_key = ? WHERE id = ?", + ("agent:main:telegram:dm:999", "20260803_120104_other"), + ) + db._conn.commit() + record = db.find_orphaned_gateway_sessions()[0] + assert record["adoptable"] is False + assert "ambiguous" in record["reason"] + + def test_two_orphans_claiming_one_predecessor_fail_closed(self, db): + stale, _ = _incident(db) + _mk_session( + db, + "20260806_161840_second", + keyed=False, + started_at=_last_activity(db, stale) + 90, + last_activity_at=time.time() - 3600, + messages=3, + ) + records = db.find_orphaned_gateway_sessions() + assert len(records) == 2 + assert not any(r["adoptable"] for r in records) + assert all("ambiguous" in r["reason"] for r in records) + + def test_delegate_children_are_not_orphans(self, db): + stale, orphan = _incident(db) + with db._lock: + db._conn.execute( + "DELETE FROM messages WHERE session_id = ?", (orphan,) + ) + db._conn.execute("DELETE FROM sessions WHERE id = ?", (orphan,)) + db._conn.commit() + _mk_session( + db, + "20260806_161836_delegate", + keyed=False, + started_at=_last_activity(db, stale) + 60, + last_activity_at=time.time() - 3600, + model_config={"_delegate_from": stale}, + messages=3, + ) + assert db.find_orphaned_gateway_sessions() == [] + + def test_a_predecessor_of_another_platform_is_not_a_donor(self, db): + stale, _ = _incident(db) + with db._lock: + db._conn.execute( + "UPDATE sessions SET source = 'discord' WHERE id = ?", (stale,) + ) + db._conn.commit() + record = db.find_orphaned_gateway_sessions()[0] + assert record["adoptable"] is False + assert record["donor_id"] is None + + +class TestAdoption: + def _resolve(self, db): + return db.find_latest_gateway_session_for_peer( + source=PEER["source"], + user_id=PEER["user_id"], + session_key=PEER["session_key"], + chat_id=PEER["chat_id"], + chat_type=PEER["chat_type"], + ) + + def test_adoption_moves_recovery_to_the_live_conversation(self, db): + stale, orphan = _incident(db) + # Before: recovery hands the chat to the three-day-old row. + assert self._resolve(db)["id"] == stale + + assert db.adopt_orphaned_gateway_session(orphan, stale) is True + + after = self._resolve(db) + assert after["id"] == orphan + assert after["chat_id"] == PEER["chat_id"] + assert after["origin_json"] + assert after["parent_session_id"] == stale + + def test_predecessor_is_retired_under_a_non_resumable_reason(self, db): + stale, orphan = _incident(db) + db.adopt_orphaned_gateway_session(orphan, stale) + row = db.get_session(stale) + assert row["end_reason"] == "superseded_by_repair" + assert row["ended_at"] is not None + + def test_adoption_is_idempotent(self, db): + stale, orphan = _incident(db) + assert db.adopt_orphaned_gateway_session(orphan, stale) is True + # The orphan is keyed now, so a replay must not re-retire anything. + assert db.adopt_orphaned_gateway_session(orphan, stale) is False + assert db.find_orphaned_gateway_sessions() == [] + + def test_existing_columns_are_never_overwritten(self, db): + stale, orphan = _incident(db) + with db._lock: + db._conn.execute( + "UPDATE sessions SET display_name = ? WHERE id = ?", + ("Renamed", orphan), + ) + db._conn.commit() + db.adopt_orphaned_gateway_session(orphan, stale) + assert db.get_session(orphan)["display_name"] == "Renamed" + + def test_cross_source_adoption_is_refused(self, db): + stale, orphan = _incident(db) + with db._lock: + db._conn.execute( + "UPDATE sessions SET source = 'discord' WHERE id = ?", (orphan,) + ) + db._conn.commit() + assert db.adopt_orphaned_gateway_session(orphan, stale) is False + assert db.get_session(orphan)["session_key"] is None + + def test_unkeyed_donor_is_refused(self, db): + _, orphan = _incident(db) + _mk_session( + db, "no_key_donor", keyed=False, started_at=time.time() - 100 + ) + assert db.adopt_orphaned_gateway_session(orphan, "no_key_donor") is False