diff --git a/hermes_state_common.py b/hermes_state_common.py index d9499eb913..fab7a1b205 100644 --- a/hermes_state_common.py +++ b/hermes_state_common.py @@ -38,6 +38,16 @@ def _sql_literal(text: str) -> str: return "'" + text.replace("'", "''") + "'" +def _sql_json_extract(expression: str, path: str) -> str: + """Build a non-throwing JSON marker lookup for a JSON TEXT column.""" + + safe_json = ( + f"(CASE WHEN json_valid({expression}) " + f"THEN {expression} ELSE json_object() END)" + ) + return f"json_extract({safe_json}, {_sql_literal(path)})" + + def _sql_ltrim_whitespace(expression: str) -> str: return f"LTRIM({expression}, {_SQL_WHITESPACE})" @@ -112,7 +122,7 @@ _PREVIEW_RAW_SUBQUERY_SQL = (f"COALESCE((SELECT {_PREVIEW_RAW_SELECT} FROM messa # ── Session lineage predicates ({a} = sessions alias) ─────────────────────── # /branch child (kept visible, never cascade-deleted): stable marker OR legacy end_reason heuristic. -_BRANCH_CHILD_SQL = ("json_extract(COALESCE({a}.model_config, '{{}}'), '$._branched_from') IS NOT NULL" +_BRANCH_CHILD_SQL = (f"{_sql_json_extract('{a}.model_config', '$._branched_from')} IS NOT NULL" " OR EXISTS (SELECT 1 FROM sessions p WHERE p.id = {a}.parent_session_id" " AND p.end_reason = 'branched' AND {a}.started_at >= p.ended_at)") _COMPRESSION_CHILD_SQL = ("EXISTS (SELECT 1 FROM sessions p WHERE p.id = {a}.parent_session_id" @@ -166,7 +176,7 @@ def _legacy_reset_child_sql(alias: str, reasons_sql: str) -> str: # A reset starts a separate user-visible conversation though rows keep parent_session_id for lineage. # Stable marker, or the same-key fallback for pre-marker rows (exact key keeps subagent children out). -_RESET_CHILD_SQL = ("json_extract(COALESCE({a}.model_config, '{{}}'), '$._reset_from') IS NOT NULL" +_RESET_CHILD_SQL = (f"{_sql_json_extract('{a}.model_config', '$._reset_from')} IS NOT NULL" " OR " + _legacy_reset_child_sql("{a}", _RESET_END_REASONS_SQL)) # Picker-visible rows: roots + branch/reset children (not subagent runs or compression continuations). diff --git a/hermes_state_compression.py b/hermes_state_compression.py index c902bfed1f..9f8415faba 100644 --- a/hermes_state_compression.py +++ b/hermes_state_compression.py @@ -12,7 +12,8 @@ import time from typing import Any, Dict, List, Optional, Tuple from hermes_state_common import ( - _BOUNDARY_END_REASONS, _COMPRESSION_LOCK_ROW_SQL as _LOCK_ROW_SQL, _ENDED_ROW_SQL, _ended_by_compression, _sql_session_last_active, is_automatic_end_reason) + _BOUNDARY_END_REASONS, _COMPRESSION_LOCK_ROW_SQL as _LOCK_ROW_SQL, _ENDED_ROW_SQL, _ended_by_compression, + _sql_json_extract, _sql_session_last_active, is_automatic_end_reason) # Log-record parity with the origin module (caplog tests pin "hermes_state"). logger = logging.getLogger("hermes_state") @@ -28,8 +29,8 @@ _CHAIN_STEP_SQL = f""" JOIN sessions child ON child.parent_session_id = parent.id WHERE parent.id = ? AND parent.end_reason = 'compression' - AND json_extract(COALESCE(child.model_config, '{{}}'), '$._branched_from') IS NULL - AND json_extract(COALESCE(child.model_config, '{{}}'), '$._delegate_from') IS NULL + AND {_sql_json_extract('child.model_config', '$._branched_from')} IS NULL + AND {_sql_json_extract('child.model_config', '$._delegate_from')} IS NULL AND COALESCE(child.source, '') != 'tool' ORDER BY CASE diff --git a/hermes_state_gateway.py b/hermes_state_gateway.py index 27f330badf..5133c5ed47 100644 --- a/hermes_state_gateway.py +++ b/hermes_state_gateway.py @@ -12,14 +12,15 @@ import time from pathlib import Path from typing import Any, Dict, List, Optional, Set, Tuple -from hermes_state_common import _RECOVERABLE_END_REASONS_SQL, _RESET_END_REASONS_SQL, _sql_session_last_active +from hermes_state_common import ( + _RECOVERABLE_END_REASONS_SQL, _RESET_END_REASONS_SQL, _sql_json_extract, _sql_session_last_active) # Log-record parity with the origin module (caplog tests pin "hermes_state"). logger = logging.getLogger("hermes_state") # Recursive CTE naming a session plus its compression ancestors (rows a # resume must keep on one routing peer); branch/delegate/tool rows stop it. -_COMPRESSION_LINEAGE_CTE = """ +_COMPRESSION_LINEAGE_CTE = f""" WITH RECURSIVE compression_lineage(id) AS ( SELECT ? UNION @@ -28,14 +29,8 @@ _COMPRESSION_LINEAGE_CTE = """ JOIN sessions child ON child.id = lineage.id JOIN sessions parent ON parent.id = child.parent_session_id WHERE parent.end_reason = 'compression' - AND json_extract( - COALESCE(child.model_config, '{}'), - '$._branched_from' - ) IS NULL - AND json_extract( - COALESCE(child.model_config, '{}'), - '$._delegate_from' - ) IS NULL + AND {_sql_json_extract('child.model_config', '$._branched_from')} IS NULL + AND {_sql_json_extract('child.model_config', '$._delegate_from')} IS NULL AND COALESCE(child.source, '') != 'tool' ) """ @@ -108,10 +103,8 @@ _ORPHANS_SQL = f""" 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 + AND {_sql_json_extract('o.model_config', '$._branched_from')} IS NULL + AND {_sql_json_extract('o.model_config', '$._delegate_from')} IS NULL ORDER BY o.started_at ASC """ _ORPHAN_LINEAGE_DONOR_SQL = f""" diff --git a/hermes_state_messages.py b/hermes_state_messages.py index 0f0d5466ca..74c1812fbb 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -14,7 +14,7 @@ from agent.memory_manager import sanitize_context from agent.message_sanitization import _sanitize_surrogates from hermes_state_common import ( _COMPRESSION_LOCK_ROW_SQL, _ENDED_ROW_SQL, _RESET_END_REASONS, _RESET_END_REASONS_SQL, _ended_by_compression, - _legacy_reset_child_sql, _placeholders) + _legacy_reset_child_sql, _placeholders, _sql_json_extract) logger = logging.getLogger("hermes_state") # caplog tests pin the origin module's name @@ -868,9 +868,9 @@ class SessionMessagesMixin: best = current child_row = conn.execute( "SELECT id FROM sessions AS child WHERE child.parent_session_id = ? " - " AND json_extract(COALESCE(child.model_config, '{}'), '$._branched_from') IS NULL " - " AND json_extract(COALESCE(child.model_config, '{}'), '$._delegate_from') IS NULL " - " AND json_extract(COALESCE(child.model_config, '{}'), '$._reset_from') IS NULL " + f" AND {_sql_json_extract('child.model_config', '$._branched_from')} IS NULL " + f" AND {_sql_json_extract('child.model_config', '$._delegate_from')} IS NULL " + f" AND {_sql_json_extract('child.model_config', '$._reset_from')} IS NULL " f" AND NOT {_legacy_reset_child_sql('child', _RESET_END_REASONS_SQL)} " " AND COALESCE(child.source, '') != 'tool' " "ORDER BY child.started_at DESC, child.id DESC LIMIT 1", (current,)).fetchone() diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py index 7974b3cccf..1e33f8d5a5 100644 --- a/hermes_state_sessions.py +++ b/hermes_state_sessions.py @@ -16,7 +16,7 @@ from agent.session_activity import ( from hermes_state_common import ( _LISTABLE_CHILD_SQL, _PREVIEW_ELIGIBLE_SQL, _PREVIEW_RAW_SELECT, _RECOVERABLE_END_REASONS, _RECOVERABLE_END_REASONS_SQL, _RESET_END_REASONS, _legacy_reset_child_sql, _shape_preview, - _sql_session_last_active, _sql_session_last_active_by_id, escape_like as _escape_like, + _sql_json_extract, _sql_session_last_active, _sql_session_last_active_by_id, escape_like as _escape_like, _placeholders as _session_ids_placeholders, ) @@ -31,7 +31,7 @@ def workspace_key(row: Dict[str, Any]) -> Optional[str]: def _delegate_from_json(col: str = "model_config") -> str: - return f"json_extract(COALESCE({col}, '{{}}'), '$._delegate_from')" + return _sql_json_extract(col, "$._delegate_from") # _merge_model_config_json's "no such row" result — distinct from the legal None @@ -420,10 +420,9 @@ class SessionSessionsMixin: # are bound to the queried parent id: continuations inherit model_config verbatim, so # presence-matching misclassified them as delegates. _NON_CONTINUATION_CHILD_FILTER_SQL = ( - " AND COALESCE(json_extract(COALESCE({alias}model_config, '{{}}')," - " '$._branched_from'), '') != ?\n" - " AND COALESCE(json_extract(COALESCE({alias}model_config, '{{}}')," - " '$._delegate_from'), '') != ?\n AND COALESCE({alias}source, '') != 'tool'\n" + f" AND COALESCE({_sql_json_extract('{alias}model_config', '$._branched_from')}, '') != ?\n" + f" AND COALESCE({_sql_json_extract('{alias}model_config', '$._delegate_from')}, '') != ?\n" + " AND COALESCE({alias}source, '') != 'tool'\n" ) def end_session(self, session_id: str, end_reason: str) -> None: @@ -1231,8 +1230,8 @@ class SessionSessionsMixin: JOIN sessions parent ON parent.id = c.cur_id JOIN sessions child ON child.parent_session_id = c.cur_id WHERE parent.end_reason = 'compression' - AND json_extract(COALESCE(child.model_config, '{{}}'), '$._branched_from') IS NULL - AND json_extract(COALESCE(child.model_config, '{{}}'), '$._delegate_from') IS NULL + AND {_sql_json_extract('child.model_config', '$._branched_from')} IS NULL + AND {_sql_json_extract('child.model_config', '$._delegate_from')} IS NULL AND COALESCE(child.source, '') != 'tool' ), chain_max AS ( diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 803a9c37b3..e0180a6c37 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -2417,6 +2417,59 @@ class TestListSessionsRich: assert len(sessions) == 1 assert "Help me refactor the auth module" in sessions[0]["preview"] + @pytest.mark.parametrize( + "unsafe_model_config", + ["{not-json", "[]", '"scalar"', "5", "null"], + ) + def test_unsafe_model_config_does_not_break_session_surfaces( + self, db, unsafe_model_config + ): + db.create_session("root", "telegram") + db.append_message("root", "user", "root message") + db.create_session("compression-parent", "telegram") + db.end_session("compression-parent", "compression") + db.create_session( + "compression-child", + "telegram", + parent_session_id="compression-parent", + ) + db.append_message("compression-child", "user", "child message") + db.create_session("routing-orphan", "telegram") + db.append_message("routing-orphan", "user", "orphan message") + db._conn.execute( + "UPDATE sessions SET model_config = ? " + "WHERE id IN (?, ?, ?)", + (unsafe_model_config, "root", "compression-child", "routing-orphan"), + ) + db._conn.commit() + + listed = db.list_sessions_rich(source="telegram") + ordered = db.list_sessions_rich( + source="telegram", order_by_last_active=True + ) + + assert "root" in {row["id"] for row in listed} + assert "root" in {row["id"] for row in ordered} + assert db.session_count(source="telegram", exclude_children=True) == 3 + assert db.session_count_by_source(exclude_children=True)["telegram"] == 3 + assert db.get_compression_chain("compression-parent") == [ + "compression-parent", + "compression-child", + ] + db.record_gateway_session_peer( + "compression-child", + source="telegram", + session_key="agent:main:telegram:dm:recovered", + include_compression_ancestors=True, + ) + assert db.get_session("compression-parent")["session_key"] == ( + "agent:main:telegram:dm:recovered" + ) + assert any( + row["orphan_id"] == "routing-orphan" + for row in db.find_orphaned_gateway_sessions() + ) +