diff --git a/gateway/session.py b/gateway/session.py index 42745a4675e3..b964adddb7aa 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -2507,7 +2507,13 @@ class SessionStore: if not self._db: return [] try: - return self._db.get_messages_as_conversation(session_id) + # repair_alternation: this load feeds LIVE REPLAY. A durable + # user;user wedge (e.g. a turn that persisted no assistant row) + # would otherwise re-trigger the pre-request repair on every + # request forever — heal it once at the restore boundary. + return self._db.get_messages_as_conversation( + session_id, repair_alternation=True + ) except Exception as e: logger.debug("Could not load messages from DB: %s", e) return [] diff --git a/hermes_cli/cli_agent_setup_mixin.py b/hermes_cli/cli_agent_setup_mixin.py index d3c967405ad1..8c25e7a1fca3 100644 --- a/hermes_cli/cli_agent_setup_mixin.py +++ b/hermes_cli/cli_agent_setup_mixin.py @@ -286,7 +286,9 @@ class CLIAgentSetupMixin: resolved_meta = self._session_db.get_session(self.session_id) if resolved_meta: session_meta = resolved_meta - restored = self._session_db.get_messages_as_conversation(self.session_id) + restored = self._session_db.get_messages_as_conversation( + self.session_id, repair_alternation=True + ) if restored: restored = [m for m in restored if m.get("role") != "session_meta"] self.conversation_history = restored @@ -484,7 +486,9 @@ class CLIAgentSetupMixin: if resolved_meta: session_meta = resolved_meta - restored = self._session_db.get_messages_as_conversation(self.session_id) + restored = self._session_db.get_messages_as_conversation( + self.session_id, repair_alternation=True + ) if restored: restored = [m for m in restored if m.get("role") != "session_meta"] self.conversation_history = restored diff --git a/hermes_state.py b/hermes_state.py index 74d2d6df50b3..0af1a0484711 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -4581,6 +4581,7 @@ class SessionDB: session_id: str, include_ancestors: bool = False, include_inactive: bool = False, + repair_alternation: bool = False, ) -> List[Dict[str, Any]]: """ Load messages in the OpenAI conversation format (role + content dicts). @@ -4589,6 +4590,16 @@ class SessionDB: By default only active messages are returned. Pass ``include_inactive=True`` to load soft-deleted (rewound) rows as well. See :meth:`rewind_to_message`. + + ``repair_alternation=True`` runs ``repair_message_sequence`` over the + loaded list before returning it. Callers that restore a session for + LIVE REPLAY should pass it: a durable alternation violation (e.g. a + ``user;user`` pair left by a turn that persisted no assistant row) + otherwise re-triggers the pre-request defensive repair on every + single request for the rest of the session's life — the repair + mutates only the per-request list, never the stored transcript. + Inspection/export consumers keep the default and see the transcript + verbatim. """ session_ids = [session_id] if include_ancestors: @@ -4685,6 +4696,21 @@ class SessionDB: # assistant reply immediately following it, so a polluted session # resumes clean even if stray rows exist. messages = _strip_background_review_harness(messages) + if repair_alternation and messages: + # Lazy import: hermes_state already depends on agent.* (see + # sanitize_context above), but keep this optional path from + # widening the import surface at module load. + from agent.agent_runtime_helpers import repair_message_sequence + + repaired = repair_message_sequence(None, messages) + if repaired: + logger.info( + "Repaired %d message-alternation violation(s) while " + "restoring session %s — durable transcript kept them, " + "see repair_message_sequence", + repaired, + session_id, + ) return messages def get_conversation_root(self, session_id: str) -> str: diff --git a/tests/hermes_state/test_restore_alternation_repair.py b/tests/hermes_state/test_restore_alternation_repair.py new file mode 100644 index 000000000000..90824c81a645 --- /dev/null +++ b/tests/hermes_state/test_restore_alternation_repair.py @@ -0,0 +1,72 @@ +"""get_messages_as_conversation(repair_alternation=True) — heal durable +alternation violations at the restore boundary. + +A turn that persists a user row but no assistant row (e.g. its reply was +suppressed, or two concurrent turns interleaved their flushes) leaves a +``user;user`` pair in state.db. Without repair at restore, the defensive +pre-request ``repair_message_sequence`` re-fires on EVERY request for the +rest of the session's life, because it mutates only the per-request list. + +Default (``repair_alternation=False``) must stay verbatim: inspection and +export consumers (trace upload, context guard) read the transcript as-is. +""" + +import pytest + +from hermes_state import SessionDB + + +@pytest.fixture() +def db(tmp_path): + db_path = tmp_path / "test_state.db" + session_db = SessionDB(db_path=db_path) + yield session_db + session_db.close() + + +def _seed_wedged_session(db, session_id="s1"): + """assistant → user → user (no assistant row between): the durable wedge.""" + db.create_session(session_id, "system prompt") + db.append_message(session_id=session_id, role="user", content="first ask") + db.append_message(session_id=session_id, role="assistant", content="first reply") + db.append_message(session_id=session_id, role="user", content="unanswered turn") + db.append_message(session_id=session_id, role="user", content="next turn") + db.append_message(session_id=session_id, role="assistant", content="next reply") + + +def test_default_load_is_verbatim(db): + _seed_wedged_session(db) + messages = db.get_messages_as_conversation("s1") + roles = [m["role"] for m in messages] + assert roles == ["user", "assistant", "user", "user", "assistant"] + + +def test_repair_alternation_merges_user_pair(db): + _seed_wedged_session(db) + messages = db.get_messages_as_conversation("s1", repair_alternation=True) + roles = [m["role"] for m in messages] + assert roles == ["user", "assistant", "user", "assistant"] + # Both user texts survive, merged in order — no user input is lost. + merged = messages[2]["content"] + assert "unanswered turn" in merged and "next turn" in merged + assert merged.index("unanswered turn") < merged.index("next turn") + + +def test_repaired_load_is_stable_under_prerequest_repair(db): + """The restored list must yield ZERO further repairs — this is the whole + point: the pre-request defensive repair stops firing every turn.""" + from agent.agent_runtime_helpers import repair_message_sequence + + _seed_wedged_session(db) + messages = db.get_messages_as_conversation("s1", repair_alternation=True) + assert repair_message_sequence(None, messages) == 0 + + +def test_repair_noop_on_clean_transcript(db): + db.create_session("s2", "system prompt") + db.append_message(session_id="s2", role="user", content="ask") + db.append_message(session_id="s2", role="assistant", content="reply") + verbatim = db.get_messages_as_conversation("s2") + repaired = db.get_messages_as_conversation("s2", repair_alternation=True) + assert [m["role"] for m in repaired] == [m["role"] for m in verbatim] + assert [m["content"] for m in repaired] == [m["content"] for m in verbatim]