diff --git a/hermes_state_schema.py b/hermes_state_schema.py index ffd06529fbb34..c1e9881c422c6 100644 --- a/hermes_state_schema.py +++ b/hermes_state_schema.py @@ -910,6 +910,50 @@ def _heal_session_model_usage_pk(self, cursor: sqlite3.Cursor) -> None: finally: cursor.execute("PRAGMA foreign_keys=ON") + def _heal_polluted_gateway_delegate_markers(self, cursor: sqlite3.Cursor) -> None: + """Strip ``_delegate_from`` from gateway main rows (``session_key`` set). + + A main gateway session must never carry the delegate marker — the + marker excludes the row from every picker (``list_sessions_rich``, + ``list_recent_sessions_bounded``) while the gateway keeps routing + into it by ``session_key``. The polluted state is the opposite of + #103789 (children leaking into the sidebar); here the main chat + vanishes while messages keep flowing (#109073). The pollution + survives via historic upsert/merge paths that copied a delegate + child's ``model_config`` onto the main row. Idempotent and safe to + run on every open (no version gate). + """ + safe_mc = ( + "CASE WHEN json_valid(COALESCE(model_config, '{}')) " + "THEN COALESCE(model_config, '{}') ELSE json_object() END" + ) + delegate_present = ( + f"{_sql_json_extract('model_config', '$._delegate_from')} IS NOT NULL" + ) + try: + # Probe first: the UPDATE takes the write lock even when it + # matches no rows and would block every open behind a sibling's + # transaction. + if cursor.execute( + "SELECT 1 FROM sessions WHERE session_key IS NOT NULL AND session_key != '' " + f"AND {delegate_present} LIMIT 1" + ).fetchone() is None: + return + cur = cursor.execute( + "UPDATE sessions SET model_config = " + f"CASE WHEN json_remove({safe_mc}, '$._delegate_from') = '{{}}' " + f"THEN NULL ELSE json_remove({safe_mc}, '$._delegate_from') END " + "WHERE session_key IS NOT NULL AND session_key != '' " + f"AND {delegate_present}" + ) + if cur.rowcount: + logger.warning( + "Healed %d polluted gateway session(s) carrying _delegate_from (session_key set) (#109073)", + cur.rowcount, + ) + except sqlite3.OperationalError as exc: + logger.debug("gateway delegate-marker heal skipped: %s", exc) + # ── _init_schema ─────────────────────────────────────────────────────── def _init_schema(self): @@ -937,6 +981,15 @@ def _init_schema(self): # already at v22+ when the column landed — the version-gated rebuild is unreachable there, #73823). # Same PK-rebuild constraint as gateway_routing above. self._heal_session_model_usage_pk(cursor) + # Heal polluted gateway sessions: a main gateway row (session_key set) + # must never carry _delegate_from — it hides the row from every picker + # while the gateway keeps routing into it (#109073). This is the + # opposite direction of #103789/PR#105284 (which stripped markers from + # children); here the marker leaked ONTO the main row via upsert/ + # merge paths and via historic delegation upserts. Idempotent, runs on + # every open so already-versioned DBs are repaired without a version + # bump. + self._heal_polluted_gateway_delegate_markers(cursor) # Indexes referencing reconciler-added columns must be created AFTER _reconcile_columns # (in SCHEMA_SQL the executescript would fail on legacy DBs). diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py index 77400e01d8251..8af2043a8411b 100644 --- a/hermes_state_sessions.py +++ b/hermes_state_sessions.py @@ -51,6 +51,30 @@ def _parse_model_config(raw: Any) -> Dict[str, Any]: return dict(raw) if isinstance(raw, dict) else {} +def _gateway_main_session_key(session_key: Optional[str]) -> bool: + return session_key is not None and session_key != "" + + +def _strip_delegate_from_gateway_config( + config: Optional[Dict[str, Any]], + session_key: Optional[str], + *, + session_id: str, + context: str, +) -> Optional[Dict[str, Any]]: + """Remove ``_delegate_from`` from gateway main rows (``session_key`` set, #109073).""" + if not config or not _gateway_main_session_key(session_key) or "_delegate_from" not in config: + return config if config else None + stripped = {k: v for k, v in config.items() if k != "_delegate_from"} + logger.warning( + "Stripped _delegate_from from gateway session %s (session_key=%r) at %s", + session_id, + session_key, + context, + ) + return stripped if stripped else None + + def _cwd_prefix_clause(cwd_prefix: str) -> Tuple[str, List[str]]: prefix = cwd_prefix.rstrip("/\\") or cwd_prefix # ``_``/``%`` are LIKE wildcards but ordinary path characters: unescaped, a @@ -333,6 +357,9 @@ def _insert_session_row( """ if not (profile_name or "").strip(): profile_name = self._own_profile_name() + model_config = _strip_delegate_from_gateway_config( # type: ignore[assignment] + model_config, session_key, session_id=session_id, context="insert", + ) def _do(conn): system_prompt_hash = self._store_system_prompt(conn, system_prompt) conn.execute( @@ -646,10 +673,24 @@ def update_session_meta( ) -> None: """Update model_config and (COALESCE) optionally model.""" self.flush_token_counts() # barrier against queued token deltas — see update_session_model - self._write_sql( - "UPDATE sessions SET model_config = ?, model = COALESCE(?, model) WHERE id = ?", - (model_config_json, model, session_id), - ) + def _do(conn): + row = conn.execute( + "SELECT session_key FROM sessions WHERE id = ?", (session_id,), + ).fetchone() + if row is None: + return + config = _strip_delegate_from_gateway_config( + _parse_model_config(model_config_json), + row["session_key"], + session_id=session_id, + context="replace", + ) + stored = json.dumps(config) if config else None + conn.execute( + "UPDATE sessions SET model_config = ?, model = COALESCE(?, model) WHERE id = ?", + (stored, model, session_id), + ) + self._execute_write(_do) def update_system_prompt(self, session_id: str, system_prompt: Optional[str]) -> None: """Store the full assembled system prompt snapshot.""" @@ -725,7 +766,7 @@ def _merge_model_config_json( """SELECT + tolerant-parse + merge ``patch`` into model_config (the one place that keeps ``_branched_from``/``_delegate_from`` alive); ``None`` deletes a key. Returns serialized JSON (``None`` when empty) or ``_MODEL_CONFIG_ROW_MISSING`` (``on_missing="raise"`` → ValueError).""" - row = conn.execute("SELECT model_config FROM sessions WHERE id = ?", (session_id,)).fetchone() + row = conn.execute("SELECT model_config, session_key FROM sessions WHERE id = ?", (session_id,)).fetchone() if row is None: if on_missing == "raise": raise ValueError(f"Session not found: {session_id}") @@ -736,6 +777,9 @@ def _merge_model_config_json( config.pop(key, None) else: config[key] = value + config = _strip_delegate_from_gateway_config( + config, row["session_key"], session_id=session_id, context="merge", + ) return json.dumps(config) if config else None def patch_session_model_config(self, session_id: str, patch: Dict[str, Any]) -> None: diff --git a/tests/hermes_state/test_gateway_delegate_marker_strip.py b/tests/hermes_state/test_gateway_delegate_marker_strip.py new file mode 100644 index 0000000000000..bba855334b2a0 --- /dev/null +++ b/tests/hermes_state/test_gateway_delegate_marker_strip.py @@ -0,0 +1,376 @@ +"""Gateway main sessions must never carry ``_delegate_from`` (#109073). + +A row with ``session_key`` set is a gateway main lane (e.g. Telegram DM). +That marker excludes the row from every picker (``list_sessions_rich``, +``list_recent_sessions_bounded``, /resume) while the gateway keeps routing +into it by ``session_key`` — chat works, the conversation vanishes from +desktop. Historic insert/merge paths could copy a delegate child's +``model_config`` onto the main row (often with the contradictory +``_reset_from`` pair). These tests lock the write-time strip + startup heal +salvaged from PR #109081. +""" + +from __future__ import annotations + +import json + +import pytest + +from hermes_state import SessionDB + + +PARENT = "parent_for_delegate_strip" +SK = "agent:main:telegram:dm:410541755" + + +@pytest.fixture +def db(tmp_path): + store = SessionDB(db_path=tmp_path / "state.db") + store.create_session(PARENT, "cli") + yield store + store.close() + + +def _mc(session: dict) -> dict: + raw = session.get("model_config") + if raw is None: + return {} + if isinstance(raw, dict): + return raw + return json.loads(raw) + + +def _listed_ids(db: SessionDB) -> set[str]: + return { + s["id"] + for s in db.list_sessions_rich( + min_message_count=1, include_archived=True, limit=100 + ) + } + + +def test_insert_strips_delegate_from_when_session_key_set(db): + """create_session with session_key must not persist _delegate_from.""" + sid = "gw_insert_strip" + db.create_session( + sid, + "telegram", + user_id="410541755", + session_key=SK, + chat_id="410541755", + chat_type="dm", + parent_session_id=PARENT, + model_config={ + "_delegate_from": PARENT, + "_reset_from": PARENT, + "_usage_anchor": {"model": "x"}, + }, + ) + db.append_message(sid, "user", "still here") + db.append_message(sid, "assistant", "ok") + + cfg = _mc(db.get_session(sid)) + assert "_delegate_from" not in cfg + assert cfg.get("_reset_from") == PARENT + assert sid in _listed_ids(db) + + +def test_merge_strips_delegate_from_when_session_key_set(db): + """patch_session_model_config must not plant _delegate_from on a gateway row.""" + sid = "gw_merge_strip" + db.create_session( + sid, + "telegram", + user_id="410541755", + session_key=SK + ":merge", + chat_id="410541755", + chat_type="dm", + model_config={"_reset_from": PARENT}, + ) + db.append_message(sid, "user", "clean") + db.append_message(sid, "assistant", "ok") + + db.patch_session_model_config( + sid, {"_delegate_from": PARENT, "_usage_anchor": {"model": "y"}} + ) + + cfg = _mc(db.get_session(sid)) + assert "_delegate_from" not in cfg + assert cfg.get("_usage_anchor", {}).get("model") == "y" + assert sid in _listed_ids(db) + + +def test_insert_keeps_delegate_from_on_child_without_session_key(db): + """Delegate children (no session_key) must keep the marker unchanged.""" + sid = "delegate_child_keep" + db.create_session( + sid, + "delegate", + parent_session_id=PARENT, + model_config={"_delegate_from": PARENT}, + ) + cfg = _mc(db.get_session(sid)) + assert cfg.get("_delegate_from") == PARENT + + +def test_startup_heal_strips_polluted_gateway_row(tmp_path): + """Historic pollution is repaired on the next SessionDB open.""" + db_path = tmp_path / "heal_state.db" + store = SessionDB(db_path=db_path) + store.create_session(PARENT, "cli") + sid = "gw_heal_strip" + store.create_session( + sid, + "telegram", + user_id="410541755", + session_key=SK + ":heal", + chat_id="410541755", + chat_type="dm", + model_config={"_reset_from": PARENT}, + ) + store.append_message(sid, "user", "pollute me") + store.append_message(sid, "assistant", "ok") + # Bypass write guards the way a historic bad upsert would: raw SQL. + with store._lock: + store._conn.execute( + "UPDATE sessions SET model_config = ? WHERE id = ?", + ( + json.dumps( + { + "_delegate_from": PARENT, + "_reset_from": PARENT, + "_usage_anchor": {"model": "z"}, + } + ), + sid, + ), + ) + store._conn.commit() + assert "_delegate_from" in _mc(store.get_session(sid)) + assert sid not in _listed_ids(store) + store.close() + + healed = SessionDB(db_path=db_path) + try: + cfg = _mc(healed.get_session(sid)) + assert "_delegate_from" not in cfg + assert cfg.get("_reset_from") == PARENT + assert sid in _listed_ids(healed) + finally: + healed.close() + + +def test_insert_does_not_mutate_caller_model_config(db): + """Strip must copy; callers may reuse the same dict for a child spawn.""" + sid = "gw_insert_no_mutate" + caller_cfg = { + "_delegate_from": PARENT, + "_reset_from": PARENT, + } + db.create_session( + sid, + "telegram", + user_id="410541755", + session_key=SK + ":nomutate", + chat_id="410541755", + chat_type="dm", + model_config=caller_cfg, + ) + assert caller_cfg.get("_delegate_from") == PARENT + assert "_delegate_from" not in _mc(db.get_session(sid)) + + +def test_insert_empty_session_key_keeps_delegate_from(db): + """Empty string is not a gateway key (same as heal: session_key != '').""" + sid = "empty_key_keep" + db.create_session( + sid, + "delegate", + parent_session_id=PARENT, + session_key="", + model_config={"_delegate_from": PARENT}, + ) + assert _mc(db.get_session(sid)).get("_delegate_from") == PARENT + + +def test_merge_sole_delegate_from_becomes_null(db): + """Stripping the only key must store NULL, not '{}'.""" + sid = "gw_merge_empty" + db.create_session( + sid, + "telegram", + user_id="410541755", + session_key=SK + ":merge_empty", + chat_id="410541755", + chat_type="dm", + ) + db.patch_session_model_config(sid, {"_delegate_from": PARENT}) + raw = db.get_session(sid).get("model_config") + assert raw is None + assert "_delegate_from" not in _mc(db.get_session(sid)) + + +def test_merge_missing_row_is_noop(db): + """patch on a missing id must not raise (on_missing=skip).""" + db.patch_session_model_config("no_such_session", {"_delegate_from": PARENT}) + assert db.get_session("no_such_session") is None + + +def test_startup_heal_sole_marker_becomes_null(tmp_path): + """Heal must store NULL when json_remove leaves only '{}'.""" + db_path = tmp_path / "heal_null.db" + store = SessionDB(db_path=db_path) + store.create_session(PARENT, "cli") + sid = "gw_heal_null" + store.create_session( + sid, + "telegram", + user_id="410541755", + session_key=SK + ":heal_null", + chat_id="410541755", + chat_type="dm", + ) + with store._lock: + store._conn.execute( + "UPDATE sessions SET model_config = ? WHERE id = ?", + (json.dumps({"_delegate_from": PARENT}), sid), + ) + store._conn.commit() + store.close() + + healed = SessionDB(db_path=db_path) + try: + row = healed.get_session(sid) + raw = row.get("model_config") + assert raw is None + assert "_delegate_from" not in _mc(row) + finally: + healed.close() + + +def test_update_session_meta_strips_delegate_from_when_session_key_set(db): + """Full model_config replacement must not plant _delegate_from on a gateway row.""" + sid = "gw_replace_strip" + db.create_session( + sid, + "telegram", + user_id="410541755", + session_key=SK + ":replace", + chat_id="410541755", + chat_type="dm", + model_config={"_reset_from": PARENT}, + ) + db.append_message(sid, "user", "hi") + db.append_message(sid, "assistant", "ok") + + db.update_session_meta( + sid, + json.dumps({"_delegate_from": PARENT, "_usage_anchor": {"model": "r"}}), + ) + + cfg = _mc(db.get_session(sid)) + assert "_delegate_from" not in cfg + assert cfg.get("_usage_anchor", {}).get("model") == "r" + assert sid in _listed_ids(db) + + +def test_update_session_meta_sole_delegate_from_becomes_null(db): + sid = "gw_replace_empty" + db.create_session( + sid, + "telegram", + user_id="410541755", + session_key=SK + ":replace_empty", + chat_id="410541755", + chat_type="dm", + ) + db.update_session_meta(sid, json.dumps({"_delegate_from": PARENT})) + assert db.get_session(sid).get("model_config") is None + + +def test_startup_heal_with_malformed_sibling_still_heals_gateway(tmp_path): + """Malformed model_config on another row must not abort heal for valid polluted rows.""" + db_path = tmp_path / "heal_malformed.db" + store = SessionDB(db_path=db_path) + store.create_session(PARENT, "cli") + sid = "gw_heal_malformed_sibling" + store.create_session( + sid, + "telegram", + user_id="410541755", + session_key=SK + ":heal_malformed", + chat_id="410541755", + chat_type="dm", + ) + store.append_message(sid, "user", "pollute") + store.append_message(sid, "assistant", "ok") + bad = "bad_json_row" + store.create_session(bad, "cli") + with store._lock: + store._conn.execute( + "UPDATE sessions SET model_config = ? WHERE id = ?", + ( + json.dumps( + { + "_delegate_from": PARENT, + "_reset_from": PARENT, + } + ), + sid, + ), + ) + store._conn.execute( + "UPDATE sessions SET model_config = ? WHERE id = ?", + ("{not valid json", bad), + ) + store._conn.commit() + assert sid not in _listed_ids(store) + store.close() + + healed = SessionDB(db_path=db_path) + try: + assert "_delegate_from" not in _mc(healed.get_session(sid)) + assert sid in _listed_ids(healed) + assert healed.get_session(bad).get("model_config") == "{not valid json" + finally: + healed.close() + + +def test_startup_heal_leaves_child_without_session_key(tmp_path): + """Heal must not strip _delegate_from from delegate children.""" + db_path = tmp_path / "heal_child.db" + store = SessionDB(db_path=db_path) + store.create_session(PARENT, "cli") + sid = "delegate_child_heal_keep" + store.create_session( + sid, + "delegate", + parent_session_id=PARENT, + model_config={"_delegate_from": PARENT}, + ) + # Also plant a polluted gateway row so the heal UPDATE actually runs + # (probe returns a hit) rather than short-circuiting before children + # could be touched by a buggy WHERE clause. + gw = "gw_heal_sibling" + store.create_session( + gw, + "telegram", + user_id="410541755", + session_key=SK + ":heal_sibling", + chat_id="410541755", + chat_type="dm", + ) + with store._lock: + store._conn.execute( + "UPDATE sessions SET model_config = ? WHERE id = ?", + (json.dumps({"_delegate_from": PARENT, "_reset_from": PARENT}), gw), + ) + store._conn.commit() + store.close() + + healed = SessionDB(db_path=db_path) + try: + assert _mc(healed.get_session(sid)).get("_delegate_from") == PARENT + assert "_delegate_from" not in _mc(healed.get_session(gw)) + finally: + healed.close()