diff --git a/gateway/session.py b/gateway/session.py index e4953cac53..c760520dce 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -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]]: diff --git a/hermes_state.py b/hermes_state.py index c580be0dcc..f8dd9eac5e 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -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), diff --git a/tests/gateway/test_session_continuity_82616.py b/tests/gateway/test_session_continuity_82616.py new file mode 100644 index 0000000000..d99bccfa77 --- /dev/null +++ b/tests/gateway/test_session_continuity_82616.py @@ -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)