ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

fix(state): recover gateway sessions stranded without a routing identity

Browse Source 
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.
This commit is contained in:
joaomarcos
2026-08-09 16:19:17 -03:00
committed by Teknium
parent 408634c2e1
commit c790ed2a5d
4 changed files with 615 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/main.py
+30
View File
@@ -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",
hermes_cli/sessions_cmd.py
+46
View File
@@ -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()
hermes_state.py
+229
View File
@@ -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
tests/hermes_state/test_orphan_gateway_session_repair.py
+310
View File
@@ -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