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

fix(gateway): make session identity durable so chat continuity survives crashes and restarts

Browse Source 
Root cause of #82616: gateway session identity (session_key/chat_id/
origin_json) was written best-effort in a separate UPDATE after row
creation, both reset-path DB writes swallowed failures silently
(logger.debug / bare print), transcript reads ignored the reroute map
that writes follow, and restart recovery ranked candidate rows by
started_at while hard-rejecting empty rows. A single failed write could
therefore strand the live conversation in an unroutable orphan row while
a days-old zombie kept the routing key — after any gateway restart the
chat silently resumed the zombie (user-visible context loss, 5 confirmed
incidents on one install since June).

Four class fixes:

1. Identity lands atomically in the session INSERT: origin_json and
   display_name join _insert_session_row's column list + COALESCE
   backfill; both gateway creation paths (get_or_create + reset) pass
   full identity including parent_session_id lineage (fixes #12857).

2. record_gateway_session_peer self-heals: when the target row is
   missing (failed/deferred create, crash window) it INSERTs the row
   with full identity instead of silently no-opping — every per-turn
   peer refresh is now a repair opportunity, and an identity-less lazy
   writer (update_token_counts/record_auxiliary_usage) can never leave
   a gateway session permanently unroutable.

3. load_transcript follows the write-side reroute chain and the durable
   compression tip before querying, so reads can no longer return 0
   rows for a session whose messages live under its compression child;
   read exceptions are WARNING, distinguishable from an empty result.

4. find_latest_gateway_session_for_peer ranks by
   COALESCE(last_activity_at, started_at) (message-bearing rows first)
   and returns an empty-but-keyed row instead of None — a zombie
   predecessor can no longer beat the live conversation, and recovery
   never mints a fresh id when a keyed row exists.

Reset-path DB write failures now log at WARNING with the routing
consequence spelled out.

Tests: tests/gateway/test_session_continuity_82616.py (11 tests) —
sabotage-verified: 6/11 fail without the fixes. E2E incident replay
(real SessionDB, temp HERMES_HOME) confirms the production shape now
resolves to the live session.

Fixes #82616. Related: #12857, #78182 (read-path half), #79576.
This commit is contained in:
Teknium
2026-08-09 10:57:09 -07:00
parent 0e2b5f3835
commit d2a4d373eb
3 changed files with 401 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/session.py
+83 -5
View File
@@ -2678,6 +2678,10 @@ class SessionStore:
entry = published
_needs_save = True
if entry is candidate:
try:
_origin_json = json.dumps(source.to_dict())
except Exception:
_origin_json = None
db_create_kwargs = {
"session_id": session_id,
"source": source.platform.value,
@@ -2687,6 +2691,12 @@ class SessionStore:
"chat_type": source.chat_type,
"thread_id": source.thread_id,
"profile_name": source.profile,
# Identity lands atomically in the INSERT (#82616): a
# crash after this write can no longer strand the row
# unroutable, and lineage survives resets (#12857).
"origin_json": _origin_json,
"display_name": source.chat_name,
"parent_session_id": prev_session_id,
}
if _needs_save:
@@ -2713,7 +2723,16 @@ class SessionStore:
else:
self._db.end_session(db_end_session_id, _db_end_reason)
except Exception as e:
logger.debug("Session DB operation failed: %s", e)
# A failed end-write leaves a zombie open row still holding
# this chat's session_key: restart recovery will resolve the
# chat to it and time-travel the conversation (#82616). Say
# so loudly — this was a silent logger.debug for months.
logger.warning(
"Failed to end predecessor session row %s for %s: %s — "
"the old row remains open and may win restart recovery "
"until the next successful peer refresh",
db_end_session_id, session_key, e,
)
if self._db and db_create_kwargs:
try:
@@ -2725,7 +2744,15 @@ class SessionStore:
display_name=entry.display_name,
)
except Exception as e:
print(f"[gateway] Warning: Failed to create SQLite session: {e}")
# The row will be self-healed with full identity by the next
# per-turn peer refresh (record_gateway_session_peer now
# INSERTs on missing row, #82616) — but the failure itself is
# a routing hazard and must be visible, not a bare print.
logger.warning(
"Failed to create session row %s for %s: %s — deferring "
"to the self-healing peer refresh on the next turn",
db_create_kwargs.get("session_id"), session_key, e,
)
return entry
@@ -3145,6 +3172,12 @@ class SessionStore:
self._entries[session_key] = new_entry
self._save()
_reset_origin_json = None
if old_entry.origin is not None:
try:
_reset_origin_json = json.dumps(old_entry.origin.to_dict())
except Exception:
_reset_origin_json = None
db_create_kwargs = {
"session_id": session_id,
"source": old_entry.platform.value if old_entry.platform else "unknown",
@@ -3154,6 +3187,11 @@ class SessionStore:
"chat_type": old_entry.origin.chat_type if old_entry.origin else None,
"thread_id": old_entry.origin.thread_id if old_entry.origin else None,
"profile_name": old_entry.origin.profile if old_entry.origin else None,
# Identity + lineage land atomically in the INSERT (#82616,
# #12857) — see the get_or_create twin path.
"origin_json": _reset_origin_json,
"display_name": old_entry.display_name,
"parent_session_id": db_end_session_id,
}
if self._db and db_end_session_id:
@@ -3168,7 +3206,13 @@ class SessionStore:
else:
self._db.end_session(db_end_session_id, "session_reset")
except Exception as e:
logger.debug("Session DB operation failed: %s", e)
# Zombie hazard — see the get_or_create twin path (#82616).
logger.warning(
"Failed to end predecessor session row %s for %s during "
"reset: %s — the old row remains open and may win restart "
"recovery until the next successful peer refresh",
db_end_session_id, session_key, e,
)
if self._db and db_create_kwargs:
try:
@@ -3180,7 +3224,12 @@ class SessionStore:
display_name=new_entry.display_name if new_entry else None,
)
except Exception as e:
logger.debug("Session DB operation failed: %s", e)
logger.warning(
"Failed to create session row %s for %s during reset: %s "
"— deferring to the self-healing peer refresh on the next "
"turn",
session_id, session_key, e,
)
return new_entry
@@ -3626,9 +3675,31 @@ class SessionStore:
state.db is the canonical store. The legacy JSONL fallback was removed
in spec 002 — pre-DB sessions on existing disks have already been
migrated (their DB row holds the full message history).
Reads follow the same routing writes use (#82616): the in-memory
reroute map installed after a compression rotation, then the durable
compression tip in state.db. Before this, writes followed the reroute
chain while reads queried the stale id directly — the transcript
"vanished" (disk=0) even though every message sat healthy under the
child session.
"""
if not self._db:
return []
# Follow the write-side reroute chain (cycle-guarded, same shape as
# append_to_transcript).
reroutes = getattr(self, "_transcript_reroutes", None) or {}
seen = set()
while session_id in reroutes and session_id not in seen:
seen.add(session_id)
session_id = reroutes[session_id]
try:
# Durable successor: a compression child published to state.db
# survives restart even though the in-memory reroute map doesn't.
tip = self._db.get_compression_tip(session_id)
if tip:
session_id = tip
except Exception:
pass
try:
# repair_alternation: this load feeds LIVE REPLAY. A durable
# user;user wedge (e.g. a turn that persisted no assistant row)
@@ -3638,7 +3709,14 @@ class SessionStore:
session_id, repair_alternation=True
)
except Exception as e:
logger.debug("Could not load messages from DB: %s", e)
# A failed read must be distinguishable from an empty transcript:
# downstream guards treat [] as "nothing persisted" and may make
# routing decisions on it (#82616). WARNING, not DEBUG.
logger.warning(
"Transcript read failed for session %s (returning empty; "
"downstream must not treat this as data loss): %s",
session_id, e,
)
return []
def rewind_session(self, session_id: str, n: int = 1) -> Optional[Dict[str, Any]]:
hermes_state.py
+76 -10
View File
@@ -3453,6 +3453,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
cwd: str = None,
profile_name: str = None,
git_repo_root: str = None,
origin_json: str = None,
display_name: str = None,
) -> None:
"""Insert a session row, enriching NULL metadata on conflict.
@@ -3495,9 +3497,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
"""INSERT INTO sessions (
id, source, user_id, session_key, chat_id, chat_type, thread_id,
model, model_config, system_prompt, system_prompt_hash,
parent_session_id, cwd, profile_name, git_repo_root, started_at
parent_session_id, cwd, profile_name, git_repo_root,
origin_json, display_name, started_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
model = COALESCE(sessions.model, excluded.model),
model_config = COALESCE(sessions.model_config, excluded.model_config),
@@ -3518,7 +3521,9 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
parent_session_id = COALESCE(sessions.parent_session_id, excluded.parent_session_id),
cwd = COALESCE(sessions.cwd, excluded.cwd),
profile_name = COALESCE(sessions.profile_name, excluded.profile_name),
git_repo_root = COALESCE(sessions.git_repo_root, excluded.git_repo_root)""",
git_repo_root = COALESCE(sessions.git_repo_root, excluded.git_repo_root),
origin_json = COALESCE(sessions.origin_json, excluded.origin_json),
display_name = COALESCE(sessions.display_name, excluded.display_name)""",
(
session_id,
source,
@@ -3534,6 +3539,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
cwd,
profile_name,
git_repo_root,
origin_json,
display_name,
time.time(),
),
)
@@ -3635,6 +3642,15 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
on one routing peer when an explicit gateway resume moves its tip to a
different lane. Normal per-turn metadata refreshes update only the
supplied row.
Self-healing (#82616): when the target row does not exist yet — the
gateway's ``create_session`` write failed and was deferred, or a
crash landed between routing publication and row creation — this
recorder INSERTs the row with the full identity instead of silently
no-opping. Every per-turn peer refresh is therefore a repair
opportunity: a gateway session row can no longer be first-created by
an identity-less lazy writer (``update_token_counts`` /
``record_auxiliary_usage``) and stay unroutable forever.
"""
if not session_id or not session_key:
return
@@ -3690,6 +3706,43 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
{target_clause}""",
query_params,
)
# Self-heal (#82616): the UPDATE is a silent no-op when the row
# is missing (create_session failed earlier, or a crash landed
# between routing publication and row creation). Insert it with
# the full identity so the session is durably routable — never
# leave first-creation to an identity-less lazy writer.
if not include_compression_ancestors:
cur = conn.execute(
"SELECT 1 FROM sessions WHERE id = ? LIMIT 1", (session_id,)
)
if cur.fetchone() is None:
conn.execute(
"""INSERT INTO sessions (
id, source, user_id, session_key, chat_id,
chat_type, thread_id, display_name, origin_json,
started_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
session_key = COALESCE(sessions.session_key, excluded.session_key),
chat_id = COALESCE(sessions.chat_id, excluded.chat_id),
chat_type = COALESCE(sessions.chat_type, excluded.chat_type),
thread_id = COALESCE(sessions.thread_id, excluded.thread_id),
display_name = COALESCE(sessions.display_name, excluded.display_name),
origin_json = COALESCE(sessions.origin_json, excluded.origin_json)""",
(
session_id,
source,
user_id,
session_key,
chat_id,
chat_type,
thread_id,
display_name,
origin_json,
time.time(),
),
)
self._execute_write(_do)
@@ -3898,6 +3951,15 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
(dashboard viewer disconnect before #60609) are treated as recoverable;
explicit conversation boundaries such as /new, /resume switches, and
compression splits are not.
Ordering and emptiness (#82616): candidates are ranked by actual
conversation recency (``last_activity_at``, falling back to
``started_at``) — ``started_at`` alone resurrected days-old zombie
rows over the live conversation. Rows with messages are preferred,
but an empty keyed row is still returned rather than ``None``:
returning ``None`` mints a brand-new session id, which is a worse
outcome than resuming an empty-but-correctly-keyed row (and "empty"
may just mean the transcript lives under a compression child).
"""
if not session_key:
return None
@@ -3906,16 +3968,17 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
"""
SELECT s.*,
COALESCE(sp.prompt, s.system_prompt)
AS _system_prompt_resolved
AS _system_prompt_resolved,
(COALESCE(s.message_count, 0) > 0 OR EXISTS (
SELECT 1 FROM messages WHERE messages.session_id = s.id LIMIT 1
)) AS _has_messages
FROM sessions s
LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash
WHERE s.session_key = ?
AND s.source = ?
AND (s.ended_at IS NULL OR s.end_reason IN ('agent_close', 'ws_orphan_reap'))
AND (COALESCE(s.message_count, 0) > 0 OR EXISTS (
SELECT 1 FROM messages WHERE messages.session_id = s.id LIMIT 1
))
ORDER BY s.started_at DESC
ORDER BY _has_messages DESC,
COALESCE(s.last_activity_at, s.started_at) DESC
LIMIT 1
""",
(session_key, source),
@@ -3932,7 +3995,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
"""
SELECT s.*,
COALESCE(sp.prompt, s.system_prompt)
AS _system_prompt_resolved
AS _system_prompt_resolved,
(COALESCE(s.message_count, 0) > 0 OR EXISTS (
SELECT 1 FROM messages WHERE messages.session_id = s.id LIMIT 1
)) AS _has_messages
FROM sessions s
LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash
WHERE s.source = ?
@@ -3944,7 +4010,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
AND (COALESCE(s.message_count, 0) > 0 OR EXISTS (
SELECT 1 FROM messages WHERE messages.session_id = s.id LIMIT 1
))
ORDER BY s.started_at DESC
ORDER BY COALESCE(s.last_activity_at, s.started_at) DESC
LIMIT 1
""",
(source, user_id, chat_id, chat_type, thread_id),
tests/gateway/test_session_continuity_82616.py
+242
View File
@@ -0,0 +1,242 @@
"""Regression tests for gateway session continuity (#82616).
Incident shape (Aug 2026, production): a /new reset's DB writes both failed
silently → the routing index moved to a new session id that had NO row →
the row was later lazily materialized with no identity columns → a gateway
crash pruned sessions.json → restart recovery resolved the chat to the
days-old zombie predecessor (still open, still keyed) → the user's DM
time-traveled three days back.
Four fixes under test:
1. Identity lands atomically in the session INSERT (origin_json,
display_name, parent_session_id in db_create_kwargs).
2. record_gateway_session_peer self-heals a missing row (INSERT on
absent id) so identity-less lazy writers can never be first.
3. load_transcript follows the write-side reroute chain + durable
compression tip; read exceptions are WARNING not silent [].
4. find_latest_gateway_session_for_peer ranks by last_activity_at and
returns an empty-but-keyed row instead of None.
"""
import json
import time
import uuid
from pathlib import Path
import pytest
from hermes_state import SessionDB
@pytest.fixture
def db(tmp_path):
d = SessionDB(db_path=tmp_path / "state.db")
yield d
try:
d.close()
except Exception:
pass
PEER = dict(
source="telegram",
user_id="6308981865",
session_key="agent:main:telegram:dm:6308981865",
chat_id="6308981865",
chat_type="dm",
thread_id=None,
)
def _mk_session(db, session_id, *, key=True, started_at=None, last_activity_at=None,
ended_at=None, end_reason=None, msgs=0):
kwargs = {"user_id": PEER["user_id"]}
if key:
kwargs.update(
session_key=PEER["session_key"], chat_id=PEER["chat_id"],
chat_type=PEER["chat_type"],
)
db.create_session(session_id, "telegram", **kwargs)
for i in range(msgs):
db.append_message(session_id, "user" if i % 2 == 0 else "assistant", f"m{i}")
with db._lock:
if started_at is not None:
db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (started_at, session_id))
if last_activity_at is not None:
db._conn.execute("UPDATE sessions SET last_activity_at=? WHERE id=?", (last_activity_at, session_id))
if ended_at is not None:
db._conn.execute(
"UPDATE sessions SET ended_at=?, end_reason=? WHERE id=?",
(ended_at, end_reason, session_id),
)
db._conn.commit()
class TestIdentityAtInsert:
def test_insert_session_row_persists_origin_and_display_name(self, db):
origin = json.dumps({"platform": "telegram", "chat_id": "6308981865"})
db.create_session(
"s1", "telegram",
session_key=PEER["session_key"], chat_id=PEER["chat_id"],
chat_type="dm", origin_json=origin, display_name="Teknium",
)
row = db.get_session("s1")
assert row["origin_json"] == origin
assert row["display_name"] == "Teknium"
assert row["session_key"] == PEER["session_key"]
def test_conflict_backfills_origin_without_overwriting(self, db):
db.create_session("s2", "telegram", session_key=PEER["session_key"])
# Second insert (agent-side upsert) backfills origin_json on the
# existing row...
db.create_session("s2", "telegram", origin_json='{"a":1}', display_name="X")
row = db.get_session("s2")
assert row["origin_json"] == '{"a":1}'
# ...but never overwrites a value already set.
db.create_session("s2", "telegram", origin_json='{"b":2}', display_name="Y")
row = db.get_session("s2")
assert row["origin_json"] == '{"a":1}'
assert row["display_name"] == "X"
class TestPeerRecorderSelfHeal:
def test_recorder_inserts_missing_row_with_full_identity(self, db):
"""The incident's core gap: routing points at a session id whose
create_session write failed. The peer refresh must create the row
with identity, not silently no-op."""
sid = "20260806_161836_" + uuid.uuid4().hex[:8]
assert db.get_session(sid) is None
db.record_gateway_session_peer(
sid,
source="telegram",
user_id=PEER["user_id"],
session_key=PEER["session_key"],
chat_id=PEER["chat_id"],
chat_type="dm",
display_name="Teknium",
origin_json='{"platform": "telegram"}',
)
row = db.get_session(sid)
assert row is not None, "peer refresh must self-heal the missing row"
assert row["session_key"] == PEER["session_key"]
assert row["chat_id"] == PEER["chat_id"]
assert row["origin_json"] == '{"platform": "telegram"}'
def test_lazy_writer_then_peer_refresh_repairs_identity(self, db):
"""Orphan-factory shape: update_token_counts materializes the row
identity-less first; the next peer refresh must stamp it."""
sid = "lazy_" + uuid.uuid4().hex[:8]
db.update_token_counts(sid, input_tokens=10, output_tokens=5)
row = db.get_session(sid)
assert row is not None and row["session_key"] is None # the orphan
db.record_gateway_session_peer(
sid,
source="telegram",
user_id=PEER["user_id"],
session_key=PEER["session_key"],
chat_id=PEER["chat_id"],
chat_type="dm",
)
row = db.get_session(sid)
assert row["session_key"] == PEER["session_key"]
assert row["chat_id"] == PEER["chat_id"]
class TestPeerResolutionRecency:
def test_prefers_recent_activity_over_started_at(self, db):
"""The Aug 9 misroute: zombie predecessor (older start, no recent
activity, still open+keyed) must lose to the keyed row with newer
activity."""
now = time.time()
_mk_session(db, "zombie", started_at=now - 6 * 86400,
last_activity_at=now - 3 * 86400, msgs=4)
_mk_session(db, "live", started_at=now - 3 * 86400,
last_activity_at=now - 600, msgs=6)
found = db.find_latest_gateway_session_for_peer(**PEER)
assert found is not None
assert found["id"] == "live"
def test_empty_keyed_row_returned_not_none(self, db):
"""Returning None mints a fresh id (worse than an empty resume);
an empty keyed row must be returned."""
_mk_session(db, "emptyrow", msgs=0)
found = db.find_latest_gateway_session_for_peer(**PEER)
assert found is not None
assert found["id"] == "emptyrow"
def test_rows_with_messages_beat_empty_rows(self, db):
now = time.time()
_mk_session(db, "hascontent", last_activity_at=now - 86400, msgs=4)
_mk_session(db, "emptynewer", last_activity_at=now - 60, msgs=0)
found = db.find_latest_gateway_session_for_peer(**PEER)
assert found["id"] == "hascontent"
def test_explicit_reset_rows_stay_unrecoverable(self, db):
now = time.time()
_mk_session(db, "resetold", last_activity_at=now - 60,
ended_at=now - 30, end_reason="session_reset", msgs=4)
found = db.find_latest_gateway_session_for_peer(**PEER)
assert found is None, "/new boundaries must never be resurrected"
def test_incident_shape_still_resolves_to_best_available(self, db):
"""Pre-fix production shape: zombie keyed+open, real session
UNSTAMPED. The resolver can only see the zombie — proving stamping
(fixes 1-2) is load-bearing; but with the real session stamped by
the self-heal, the resolver must pick it."""
now = time.time()
_mk_session(db, "zombie2", started_at=now - 6 * 86400,
last_activity_at=now - 3 * 86400, msgs=4)
_mk_session(db, "real", key=False, started_at=now - 3 * 86400,
last_activity_at=now - 600, msgs=8)
found = db.find_latest_gateway_session_for_peer(**PEER)
assert found["id"] == "zombie2" # unstamped rows are invisible
# Self-heal stamps the real row (any later peer refresh):
db.record_gateway_session_peer(
"real", source="telegram", user_id=PEER["user_id"],
session_key=PEER["session_key"], chat_id=PEER["chat_id"],
chat_type="dm",
)
found = db.find_latest_gateway_session_for_peer(**PEER)
assert found["id"] == "real"
class TestLoadTranscriptReroutes:
def test_load_transcript_follows_reroute_chain(self, tmp_path):
from gateway.session import SessionStore
from gateway.config import GatewayConfig
store = SessionStore(sessions_dir=tmp_path / "gw", config=GatewayConfig())
db = store._db
assert db is not None
db.create_session("parent", "telegram", session_key=PEER["session_key"])
db.create_session("child", "telegram", session_key=PEER["session_key"])
db.append_message("child", "user", "hello from the child")
# Write-side reroute installed after a compression rotation:
store._transcript_reroutes["parent"] = "child"
msgs = store.load_transcript("parent")
assert any("hello from the child" in str(m.get("content", "")) for m in msgs), (
"reads must follow the same reroute chain writes use"
)
def test_load_transcript_follows_durable_compression_tip(self, tmp_path):
from gateway.session import SessionStore
from gateway.config import GatewayConfig
store = SessionStore(sessions_dir=tmp_path / "gw2", config=GatewayConfig())
db = store._db
assert db is not None
db.create_session("p2", "telegram", session_key=PEER["session_key"])
db.append_message("p2", "user", "old")
db.publish_compression_child(
parent_session_id="p2",
child_session_id="c2",
source="telegram",
messages=[{"role": "user", "content": "compressed history"}],
require_compression_lease=False,
)
# No in-memory reroute (fresh store after restart) — durable tip only.
store._transcript_reroutes.clear()
msgs = store.load_transcript("p2")
assert any("compressed history" in str(m.get("content", "")) for m in msgs)