From 35ebf6ba679f3b3e57e79b7c9ddb7a48c9c33646 Mon Sep 17 00:00:00 2001 From: dsad Date: Sun, 12 Jul 2026 18:58:01 +0300 Subject: [PATCH 01/12] fix(cli): persist close transcript without history alias --- cli.py | 9 ++- .../cli/test_cli_shutdown_memory_messages.py | 58 ++++++++++++++++++- 2 files changed, 60 insertions(+), 7 deletions(-) diff --git a/cli.py b/cli.py index 9887bb029780..ac07ff7ba531 100644 --- a/cli.py +++ b/cli.py @@ -12830,12 +12830,11 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): if not isinstance(messages, list) or not messages: return - conversation_history = getattr(self, "conversation_history", None) - if not isinstance(conversation_history, list): - conversation_history = messages - try: - agent._persist_session(messages, conversation_history) + # Do not pass CLI conversation_history here: during interrupted + # shutdown it can alias _session_messages, which makes the DB flush + # treat every live message as already durable. + agent._persist_session(messages) if getattr(agent, "session_id", None): self.session_id = agent.session_id except (Exception, KeyboardInterrupt) as e: diff --git a/tests/cli/test_cli_shutdown_memory_messages.py b/tests/cli/test_cli_shutdown_memory_messages.py index 87df42f337f5..4a6a23900bbd 100644 --- a/tests/cli/test_cli_shutdown_memory_messages.py +++ b/tests/cli/test_cli_shutdown_memory_messages.py @@ -131,7 +131,7 @@ def test_cli_close_persists_agent_session_messages_before_end_session(): cli._persist_active_session_before_close() - agent._persist_session.assert_called_once_with(transcript, conversation_history) + agent._persist_session.assert_called_once_with(transcript) assert cli.session_id == "live-session" @@ -149,7 +149,7 @@ def test_cli_close_persist_falls_back_to_conversation_history(): cli._persist_active_session_before_close() - agent._persist_session.assert_called_once_with(conversation_history, conversation_history) + agent._persist_session.assert_called_once_with(conversation_history) def test_cli_close_persist_skips_empty_transcripts(): @@ -167,3 +167,57 @@ def test_cli_close_persist_skips_empty_transcripts(): cli._persist_active_session_before_close() agent._persist_session.assert_not_called() + + +def test_cli_close_persist_real_db_survives_history_alias(tmp_path, monkeypatch): + """CLI close safety-net must persist even when history aliases messages. + + In the real CLI, ``conversation_history`` and ``agent._session_messages`` can + point at the same live list during interrupted shutdown. Passing that list + as ``conversation_history`` makes ``_flush_messages_to_session_db`` treat + every message as already durable and write zero rows. The close safety-net + should use marker-based dedup instead. + """ + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + + import cli as cli_mod + from hermes_state import SessionDB + from run_agent import AIAgent + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "cli-close-alias" + db.create_session(session_id=session_id, source="cli") + + transcript = [ + {"role": "user", "content": "long task"}, + {"role": "assistant", "content": "partial answer"}, + ] + + agent = object.__new__(AIAgent) + agent._session_db = db + agent._session_db_created = True + agent.session_id = session_id + agent.platform = "cli" + agent.model = "test-model" + agent._session_messages = transcript + agent._last_flushed_db_idx = 0 + agent._flushed_db_message_ids = set() + agent._flushed_db_message_session_id = None + agent._persist_disabled = False + agent._cached_system_prompt = None + agent._session_init_model_config = None + agent._parent_session_id = None + agent._session_json_enabled = False + + cli = object.__new__(cli_mod.HermesCLI) + cli.conversation_history = transcript + cli.session_id = "old-session" + cli.agent = agent + + assert db.get_messages_as_conversation(session_id) == [] + + cli._persist_active_session_before_close() + + stored = db.get_messages_as_conversation(session_id) + assert [m["content"] for m in stored] == ["long task", "partial answer"] + assert cli.session_id == session_id From a27d51ef467c4a5c16b08494741a04e7865fa454 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Mon, 13 Jul 2026 18:13:21 +0530 Subject: [PATCH 02/12] fix(cli): preserve resumed history during close flush Retain a distinct CLI history baseline during the signal window before a turn's normal persistence flush. When CLI history aliases the live agent list, use marker-only persistence so a genuinely unflushed tail is written. --- cli.py | 16 +- .../cli/test_cli_shutdown_memory_messages.py | 176 ++++++++++++++++-- 2 files changed, 170 insertions(+), 22 deletions(-) diff --git a/cli.py b/cli.py index ac07ff7ba531..4d5dfafc1a32 100644 --- a/cli.py +++ b/cli.py @@ -12830,11 +12830,19 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): if not isinstance(messages, list) or not messages: return + # A normal turn builds a new list that reuses the resumed-history dicts. + # Keep that CLI history as the baseline so a signal between assigning + # ``_session_messages`` and the turn's DB flush cannot append its durable + # prefix a second time. Once the CLI takes the turn result, however, both + # names can point at the same live list; passing that alias would mark an + # unflushed tail durable without writing it. Marker-only persistence is + # correct only in that alias case. + conversation_history = getattr(self, "conversation_history", None) + if not isinstance(conversation_history, list) or conversation_history is messages: + conversation_history = None + try: - # Do not pass CLI conversation_history here: during interrupted - # shutdown it can alias _session_messages, which makes the DB flush - # treat every live message as already durable. - agent._persist_session(messages) + agent._persist_session(messages, conversation_history) if getattr(agent, "session_id", None): self.session_id = agent.session_id except (Exception, KeyboardInterrupt) as e: diff --git a/tests/cli/test_cli_shutdown_memory_messages.py b/tests/cli/test_cli_shutdown_memory_messages.py index 4a6a23900bbd..b56b0d85f5ee 100644 --- a/tests/cli/test_cli_shutdown_memory_messages.py +++ b/tests/cli/test_cli_shutdown_memory_messages.py @@ -16,6 +16,8 @@ other tests keep their existing no-arg behaviour. from __future__ import annotations +import threading +from typing import Any from unittest.mock import MagicMock, patch @@ -131,7 +133,7 @@ def test_cli_close_persists_agent_session_messages_before_end_session(): cli._persist_active_session_before_close() - agent._persist_session.assert_called_once_with(transcript) + agent._persist_session.assert_called_once_with(transcript, conversation_history) assert cli.session_id == "live-session" @@ -149,7 +151,7 @@ def test_cli_close_persist_falls_back_to_conversation_history(): cli._persist_active_session_before_close() - agent._persist_session.assert_called_once_with(conversation_history) + agent._persist_session.assert_called_once_with(conversation_history, None) def test_cli_close_persist_skips_empty_transcripts(): @@ -169,6 +171,47 @@ def test_cli_close_persist_skips_empty_transcripts(): agent._persist_session.assert_not_called() +def test_cli_close_uses_distinct_history_as_baseline(): + """A pre-flush shutdown keeps the distinct CLI prefix as a DB baseline.""" + import cli as cli_mod + + history = [{"role": "user", "content": "resumed prompt"}] + live_messages = history + [{"role": "assistant", "content": "partial response"}] + cli = object.__new__(cli_mod.HermesCLI) + cli.conversation_history = history + cli.session_id = "session-id" + agent = MagicMock() + agent.session_id = "session-id" + agent._session_messages = live_messages + cli.agent = agent + + cli._persist_active_session_before_close() + + agent._persist_session.assert_called_once_with(live_messages, history) + + +def _real_agent(db, session_id, session_messages): + """Build the real persistence seam without the heavyweight LLM client.""" + from run_agent import AIAgent + + agent = object.__new__(AIAgent) + agent._session_db = db + agent._session_db_created = True + agent.session_id = session_id + agent.platform = "cli" + agent.model = "test-model" + agent._session_messages = session_messages + agent._last_flushed_db_idx = 0 + agent._flushed_db_message_ids = set() + agent._flushed_db_message_session_id = None + agent._persist_disabled = False + agent._cached_system_prompt = None + agent._session_init_model_config = None + agent._parent_session_id = None + agent._session_json_enabled = False + return agent + + def test_cli_close_persist_real_db_survives_history_alias(tmp_path, monkeypatch): """CLI close safety-net must persist even when history aliases messages. @@ -182,7 +225,6 @@ def test_cli_close_persist_real_db_survives_history_alias(tmp_path, monkeypatch) import cli as cli_mod from hermes_state import SessionDB - from run_agent import AIAgent db = SessionDB(db_path=tmp_path / "state.db") session_id = "cli-close-alias" @@ -193,21 +235,7 @@ def test_cli_close_persist_real_db_survives_history_alias(tmp_path, monkeypatch) {"role": "assistant", "content": "partial answer"}, ] - agent = object.__new__(AIAgent) - agent._session_db = db - agent._session_db_created = True - agent.session_id = session_id - agent.platform = "cli" - agent.model = "test-model" - agent._session_messages = transcript - agent._last_flushed_db_idx = 0 - agent._flushed_db_message_ids = set() - agent._flushed_db_message_session_id = None - agent._persist_disabled = False - agent._cached_system_prompt = None - agent._session_init_model_config = None - agent._parent_session_id = None - agent._session_json_enabled = False + agent = _real_agent(db, session_id, transcript) cli = object.__new__(cli_mod.HermesCLI) cli.conversation_history = transcript @@ -221,3 +249,115 @@ def test_cli_close_persist_real_db_survives_history_alias(tmp_path, monkeypatch) stored = db.get_messages_as_conversation(session_id) assert [m["content"] for m in stored] == ["long task", "partial answer"] assert cli.session_id == session_id + + +def test_cli_close_preflush_resumed_prefix_is_not_duplicated(tmp_path, monkeypatch): + """A signal during the turn-start flush preserves the old DB prefix once. + + The pause is after ``_persist_session`` records its live snapshot but before + its normal DB flush. The close helper must retain the distinct CLI baseline. + """ + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + + import cli as cli_mod + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "cli-close-preflush-resume" + db.create_session(session_id=session_id, source="cli") + loaded = [ + {"role": "user", "content": "old prompt"}, + {"role": "assistant", "content": "old answer"}, + ] + for message in loaded: + db.append_message( + session_id=session_id, + role=message["role"], + content=message["content"], + ) + + live_messages = list(loaded) + [{"role": "user", "content": "new prompt"}] + agent = _real_agent(db, session_id, []) + entered_flush = threading.Event() + release_flush = threading.Event() + flush_calls = 0 + + def _pause_before_flush( + messages: list[dict[str, Any]], + conversation_history: list[dict[str, Any]] | None = None, + ) -> None: + nonlocal flush_calls + flush_calls += 1 + if flush_calls == 1: + # The worker has assigned its snapshot and is now paused before its + # regular DB write. The concurrent close call must stay live. + agent._session_messages = messages + entered_flush.set() + assert release_flush.wait(timeout=5) + from run_agent import AIAgent + + # Runtime accepts None; the stub keeps that optional contract explicit. + return AIAgent._flush_messages_to_session_db( + agent, + messages, + conversation_history if conversation_history is not None else [], + ) + + agent._flush_messages_to_session_db = _pause_before_flush + worker = threading.Thread( + target=lambda: agent._persist_session(live_messages, loaded), + daemon=True, + ) + worker.start() + assert entered_flush.wait(timeout=5) + + cli = object.__new__(cli_mod.HermesCLI) + cli.conversation_history = list(loaded) + [{"role": "user", "content": "ui prompt"}] + cli.session_id = session_id + cli.agent = agent + cli._persist_active_session_before_close() + + release_flush.set() + worker.join(timeout=5) + assert not worker.is_alive() + + stored = db.get_messages_as_conversation(session_id) + assert [m["content"] for m in stored] == [ + "old prompt", + "old answer", + "new prompt", + ] + + +def test_cli_close_preserves_unflushed_tail_after_prior_prefix_flush(tmp_path, monkeypatch): + """Marker-only alias close writes only a new tail after a prior flush.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + + import cli as cli_mod + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "cli-close-tail" + db.create_session(session_id=session_id, source="cli") + prefix = [ + {"role": "user", "content": "old prompt"}, + {"role": "assistant", "content": "old answer"}, + ] + agent = _real_agent(db, session_id, prefix) + agent._flush_messages_to_session_db(prefix, []) + live_messages = prefix + [{"role": "assistant", "content": "new tail"}] + agent._session_messages = live_messages + + cli = object.__new__(cli_mod.HermesCLI) + cli.conversation_history = live_messages + cli.session_id = session_id + cli.agent = agent + + cli._persist_active_session_before_close() + + stored = db.get_messages_as_conversation(session_id) + assert [m["content"] for m in stored] == [ + "old prompt", + "old answer", + "new tail", + ] From 475922f2ce125290559b86f9a82363c1f6c2639f Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Mon, 13 Jul 2026 19:08:02 +0530 Subject: [PATCH 03/12] fix(cli): serialize close persistence handoff Preserve one durable staged input across terminal close and the worker's early turn flush, without duplicating resumed transcripts or creating a session with a null prompt. Fixes #63766. --- agent/agent_init.py | 8 + agent/turn_context.py | 57 +++++- cli.py | 66 ++++++- run_agent.py | 20 ++- tests/agent/test_turn_context.py | 35 ++++ .../cli/test_cli_shutdown_memory_messages.py | 168 +++++++++++++++++- 6 files changed, 333 insertions(+), 21 deletions(-) diff --git a/agent/agent_init.py b/agent/agent_init.py index c5826b4b9475..f9acb3c982ab 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -1311,6 +1311,14 @@ def init_agent( # SQLite session store (optional -- provided by CLI or gateway) agent._session_db = session_db agent._parent_session_id = parent_session_id + # A close flush and the worker's turn-start flush can overlap. The durable + # marker is attached to each in-memory message dict, so its test-and-append + # sequence must be serialized per agent rather than relying on SQLite alone. + agent._session_persist_lock = threading.RLock() + # CLI retains its just-accepted user dict until turn setup can reuse it. + # This preserves the message-local durable marker if close persistence wins + # the race before the agent's normal early turn flush. + agent._pending_cli_user_message = None agent._last_flushed_db_idx = 0 # tracks DB-write cursor to prevent duplicate writes agent._session_db_created = False # DB row deferred to run_conversation() # Most agents own their session row and should finalize it on close(). diff --git a/agent/turn_context.py b/agent/turn_context.py index a5e738588e48..cf1cdff47517 100644 --- a/agent/turn_context.py +++ b/agent/turn_context.py @@ -271,6 +271,29 @@ def build_turn_context( # Initialize conversation (copy to avoid mutating the caller's list). messages = list(conversation_history) if conversation_history else [] + # The CLI may already have staged this input outside the history passed to + # ``run_conversation``. Reuse it only when its clean transcript text matches + # this turn; a stale handoff from a failed prior turn must not replace a + # later, different user input. Voice turns compare against their explicit + # clean persistence override rather than the API-only prefixed payload. + pending_cli_message = getattr(agent, "_pending_cli_user_message", None) + expected_persist_content = ( + persist_user_message if persist_user_message is not None else user_message + ) + if ( + isinstance(pending_cli_message, dict) + and pending_cli_message.get("content") == expected_persist_content + ): + user_msg = pending_cli_message + # The CLI-staged value is the clean transcript text. Restore the + # API-facing variant (for example, a voice-mode prefix) while retaining + # the same dict and any close-path durable marker. + user_msg["content"] = user_message + else: + user_msg = {"role": "user", "content": user_message} + if isinstance(pending_cli_message, dict): + agent._pending_cli_user_message = None + # Hydrate todo store from conversation history. if conversation_history and not agent._todo_store.has_items(): agent._hydrate_todo_store(conversation_history) @@ -285,6 +308,13 @@ def build_turn_context( if agent._memory_nudge_interval > 0 and agent._turns_since_memory == 0: agent._turns_since_memory = prior_user_turns % agent._memory_nudge_interval + # Add the current user message after the prompt/session setup has made + # close persistence safe. The handoff above preserves any marker already + # stamped by an earlier close flush. + messages.append(user_msg) + current_turn_user_idx = len(messages) - 1 + agent._persist_user_message_idx = current_turn_user_idx + # Track user turns for memory flush and periodic nudge logic. agent._user_turn_count += 1 # Copilot x-initiator: the first API call of this user turn is @@ -313,12 +343,6 @@ def build_turn_context( should_review_memory = True agent._turns_since_memory = 0 - # Add user message. - user_msg = {"role": "user", "content": user_message} - messages.append(user_msg) - current_turn_user_idx = len(messages) - 1 - agent._persist_user_message_idx = current_turn_user_idx - # Cosmetic side-signal: detect an affection "reaction" (ily / <3 / good bot) # and notify the host so it can play hearts. Token-free, never touches the # conversation, and never fatal — a purely optional UI beat. @@ -348,18 +372,33 @@ def build_turn_context( # Create the DB session row now that _cached_system_prompt is populated, so # the persisted snapshot is written non-NULL on the first turn (Issue - # #45499). Idempotent: _ensure_db_session() no-ops once the row exists. - agent._ensure_db_session() + # #45499). Keep row creation and the marker-based append in the same + # per-agent critical section as CLI close persistence. + persist_lock = getattr(agent, "_session_persist_lock", None) + + def _ensure_and_persist() -> None: + agent._ensure_db_session() + agent._persist_session(messages, conversation_history) # Crash-resilience: persist the inbound user turn as soon as the session row exists. try: - agent._persist_session(messages, conversation_history) + if persist_lock is None: + _ensure_and_persist() + else: + with persist_lock: + _ensure_and_persist() except Exception: logger.warning( "Early turn-start session persistence failed for session=%s", agent.session_id or "none", exc_info=True, ) + finally: + # Keep an unmarked staged input available to a later close retry if the + # normal persistence attempt failed. Once the marker is present, the + # close path must no longer treat it as a pre-worker UI input. + if not isinstance(pending_cli_message, dict) or pending_cli_message.get("_db_persisted"): + agent._pending_cli_user_message = None # ── Preflight context compression ── # Gate the (expensive) full token estimate behind a cheap pre-check. diff --git a/cli.py b/cli.py index 4d5dfafc1a32..0b4c4b194286 100644 --- a/cli.py +++ b/cli.py @@ -12129,7 +12129,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): request_overrides=turn_route.get("request_overrides"), ): return None - + agent = self.agent + if agent is None: + return None + # Route image attachments based on the active model's vision capability. # "native" → pass pixels as OpenAI-style content parts (adapters # translate for Anthropic/Gemini/Bedrock). @@ -12217,8 +12220,15 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): from run_agent import _sanitize_surrogates message = _sanitize_surrogates(message) - # Add user message to history - self.conversation_history.append({"role": "user", "content": message}) + # Keep the exact CLI input dict available until turn-start persistence. + # Copy the completed agent transcript before appending: otherwise this + # UI-only staging step mutates ``agent._session_messages`` and exposes a + # duplicate-prone intermediate snapshot to terminal-close persistence. + if self.conversation_history is getattr(agent, "_session_messages", None): + self.conversation_history = list(self.conversation_history) + staged_user_message = {"role": "user", "content": message} + agent._pending_cli_user_message = staged_user_message + self.conversation_history.append(staged_user_message) ChatConsole().print(f"[{_accent_hex()}]{'─' * 40}[/]") print(flush=True) @@ -12825,9 +12835,19 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): return messages = getattr(agent, "_session_messages", None) + pending_cli_message = getattr(agent, "_pending_cli_user_message", None) if not isinstance(messages, list): messages = getattr(self, "conversation_history", None) - if not isinstance(messages, list) or not messages: + if not isinstance(messages, list): + return + if isinstance(pending_cli_message, dict) and not any( + message is pending_cli_message for message in messages + ): + # The UI has accepted a new input but the worker still exposes its + # prior snapshot. Include only that staged dict; the baseline below + # keeps any durable resumed prefix from being re-appended. + messages = [*messages, pending_cli_message] + if not messages: return # A normal turn builds a new list that reuses the resumed-history dicts. @@ -12838,13 +12858,47 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): # unflushed tail durable without writing it. Marker-only persistence is # correct only in that alias case. conversation_history = getattr(self, "conversation_history", None) - if not isinstance(conversation_history, list) or conversation_history is messages: + pending_cli_message = getattr(agent, "_pending_cli_user_message", None) + if ( + isinstance(conversation_history, list) + and conversation_history + and conversation_history[-1] is pending_cli_message + ): + # The UI accepted this user message before the agent finished its + # early persistence. Its dict can already be in ``messages`` but is + # not durable yet, so exclude it from the resumed-history baseline. + conversation_history = conversation_history[:-1] + elif not isinstance(conversation_history, list) or conversation_history is messages: conversation_history = None - try: + # A first-turn close can arrive before the worker builds its cached + # prompt. Build or restore it before the DB row is created so the + # durable transcript never leaves a NULL system_prompt cache entry. + if getattr(agent, "_cached_system_prompt", None) is None: + try: + from agent.conversation_loop import _restore_or_build_system_prompt + + _restore_or_build_system_prompt(agent, None, conversation_history) + except Exception: + logger.debug("Could not build system prompt during CLI close", exc_info=True) + return + if getattr(agent, "_cached_system_prompt", None) is None: + return + + persist_lock = getattr(agent, "_session_persist_lock", None) + + def _ensure_and_persist() -> None: + agent._ensure_db_session() agent._persist_session(messages, conversation_history) if getattr(agent, "session_id", None): self.session_id = agent.session_id + + try: + if persist_lock is None: + _ensure_and_persist() + else: + with persist_lock: + _ensure_and_persist() except (Exception, KeyboardInterrupt) as e: logger.debug("Could not persist active CLI session before close: %s", e) diff --git a/run_agent.py b/run_agent.py index fe378f396ae9..2d710d77e2d9 100644 --- a/run_agent.py +++ b/run_agent.py @@ -1689,10 +1689,22 @@ class AIAgent: """ # Scaffolding removal mutates the live list (desired — ephemeral # retry/failure sentinels must not survive into the real transcript). - self._drop_trailing_empty_response_scaffolding(messages) - self._session_messages = messages - self._save_session_log(messages) - self._flush_messages_to_session_db(messages, conversation_history) + # Close and turn-start persistence can run on separate CLI threads; the + # marker test-and-append below must be one critical section or both can + # observe the same unmarked dict and write duplicate durable rows. + persist_lock = getattr(self, "_session_persist_lock", None) + if persist_lock is None: + self._drop_trailing_empty_response_scaffolding(messages) + self._session_messages = messages + self._save_session_log(messages) + self._flush_messages_to_session_db(messages, conversation_history) + return + + with persist_lock: + self._drop_trailing_empty_response_scaffolding(messages) + self._session_messages = messages + self._save_session_log(messages) + self._flush_messages_to_session_db(messages, conversation_history) def _drop_trailing_empty_response_scaffolding(self, messages: List[Dict]) -> None: """Remove private empty-response retry/failure scaffolding from transcript tails. diff --git a/tests/agent/test_turn_context.py b/tests/agent/test_turn_context.py index 0cd0e91caa7b..d024eca56fc8 100644 --- a/tests/agent/test_turn_context.py +++ b/tests/agent/test_turn_context.py @@ -8,6 +8,7 @@ confirm the prologue produces the right ``TurnContext`` and applies the from __future__ import annotations +import threading import types from unittest.mock import MagicMock, patch @@ -73,6 +74,9 @@ class _FakeAgent: self._invalid_tool_retries = -1 self._vision_supported = None self._persist_calls = 0 + self._session_messages = [] + self._pending_cli_user_message = None + self._session_persist_lock = threading.RLock() # Records _cached_system_prompt at the moment _ensure_db_session() # is called (regression guard for #45499 turn-setup ordering). self._ensure_db_prompt_at_call = "" @@ -206,6 +210,37 @@ def test_persist_user_message_becomes_original(): assert ctx.messages[-1]["content"] == "api-prefixed" +def test_pending_cli_message_carries_durable_marker_to_new_turn_dict(): + """A close-persisted CLI input must not be written again by turn start.""" + agent = _FakeAgent() + staged = {"role": "user", "content": "already durable", "_db_persisted": True} + agent._pending_cli_user_message = staged + + ctx = _build(agent, user_message="already durable") + + assert ctx.messages[-1] is staged + assert ctx.messages[-1]["content"] == "already durable" + assert ctx.messages[-1]["_db_persisted"] is True + assert agent._pending_cli_user_message is None + + +def test_stale_pending_cli_message_does_not_replace_new_turn_input(): + """A failed prior persistence handoff cannot substitute later user input.""" + agent = _FakeAgent() + agent._pending_cli_user_message = {"role": "user", "content": "old prompt"} + + stale = agent._pending_cli_user_message + ctx = _build( + agent, + user_message="new prompt", + conversation_history=[{"role": "assistant", "content": "old answer"}], + ) + + assert ctx.messages[-1]["content"] == "new prompt" + assert ctx.messages[-1] is not stale + assert agent._pending_cli_user_message is None + + def test_memory_nudge_fires_at_interval(): agent = _FakeAgent() agent._memory_nudge_interval = 1 diff --git a/tests/cli/test_cli_shutdown_memory_messages.py b/tests/cli/test_cli_shutdown_memory_messages.py index b56b0d85f5ee..85787a72e4a7 100644 --- a/tests/cli/test_cli_shutdown_memory_messages.py +++ b/tests/cli/test_cli_shutdown_memory_messages.py @@ -205,10 +205,12 @@ def _real_agent(db, session_id, session_messages): agent._flushed_db_message_ids = set() agent._flushed_db_message_session_id = None agent._persist_disabled = False - agent._cached_system_prompt = None + agent._cached_system_prompt = "test system prompt" agent._session_init_model_config = None agent._parent_session_id = None agent._session_json_enabled = False + agent._pending_cli_user_message = None + agent._session_persist_lock = threading.RLock() return agent @@ -315,11 +317,26 @@ def test_cli_close_preflush_resumed_prefix_is_not_duplicated(tmp_path, monkeypat cli.conversation_history = list(loaded) + [{"role": "user", "content": "ui prompt"}] cli.session_id = session_id cli.agent = agent - cli._persist_active_session_before_close() + close_started = threading.Event() + close_finished = threading.Event() + + def _close_while_worker_flushes(): + close_started.set() + cli._persist_active_session_before_close() + close_finished.set() + + close_worker = threading.Thread(target=_close_while_worker_flushes, daemon=True) + close_worker.start() + assert close_started.wait(timeout=5) + # The per-agent persistence lock holds the close flush until the normal + # turn-start write has stamped its durable markers. + assert not close_finished.wait(timeout=0.1) release_flush.set() worker.join(timeout=5) + close_worker.join(timeout=5) assert not worker.is_alive() + assert not close_worker.is_alive() stored = db.get_messages_as_conversation(session_id) assert [m["content"] for m in stored] == [ @@ -361,3 +378,150 @@ def test_cli_close_preserves_unflushed_tail_after_prior_prefix_flush(tmp_path, m "old answer", "new tail", ] + + +def test_cli_close_hands_staged_user_marker_to_turn_start(tmp_path, monkeypatch): + """A close before turn setup does not duplicate the CLI-staged user row.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + + import cli as cli_mod + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "cli-close-staged-user" + db.create_session(session_id=session_id, source="cli") + prefix = [ + {"role": "user", "content": "old prompt"}, + {"role": "assistant", "content": "old answer"}, + ] + agent = _real_agent(db, session_id, prefix) + agent._flush_messages_to_session_db(prefix, []) + staged = {"role": "user", "content": "new prompt"} + # `chat()` copies a completed agent transcript before it stages the next + # user input, so close initially sees the prior agent snapshot only. + cli_history = list(prefix) + [staged] + agent._pending_cli_user_message = staged + + cli = object.__new__(cli_mod.HermesCLI) + cli.conversation_history = cli_history + cli.session_id = session_id + cli.agent = agent + + # Close appends only the pending UI dict, while treating the durable prefix + # as its baseline. Turn setup then reuses the marked dict without re-writing. + cli._persist_active_session_before_close() + assert staged["_db_persisted"] is True + + worker_messages = list(prefix) + [staged] + agent._persist_session(worker_messages, prefix) + + stored = db.get_messages_as_conversation(session_id) + assert [m["content"] for m in stored] == [ + "old prompt", + "old answer", + "new prompt", + ] + + +def test_cli_chat_staging_does_not_mutate_live_agent_snapshot(): + """The next CLI input must be outside the prior live agent transcript.""" + import cli as cli_mod + + previous = [{"role": "assistant", "content": "done"}] + agent = MagicMock() + agent._session_messages = previous + agent._pending_cli_user_message = None + + cli = object.__new__(cli_mod.HermesCLI) + cli.agent = agent + cli.conversation_history = previous + + # Model the narrow staging operation in ``chat`` without starting a provider. + if cli.conversation_history is agent._session_messages: + cli.conversation_history = list(cli.conversation_history) + staged = {"role": "user", "content": "next"} + agent._pending_cli_user_message = staged + cli.conversation_history.append(staged) + + assert agent._session_messages == [{"role": "assistant", "content": "done"}] + assert cli.conversation_history == [ + {"role": "assistant", "content": "done"}, + {"role": "user", "content": "next"}, + ] + + +def test_cli_close_persists_pending_user_when_agent_snapshot_is_empty(tmp_path, monkeypatch): + """Close before worker startup persists only the CLI-staged user input.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + + import cli as cli_mod + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "cli-close-before-worker" + db.create_session(session_id=session_id, source="cli") + prefix = [ + {"role": "user", "content": "old prompt"}, + {"role": "assistant", "content": "old answer"}, + ] + for message in prefix: + db.append_message( + session_id=session_id, + role=message["role"], + content=message["content"], + ) + + agent = _real_agent(db, session_id, []) + staged = {"role": "user", "content": "new prompt"} + agent._pending_cli_user_message = staged + + cli = object.__new__(cli_mod.HermesCLI) + cli.conversation_history = list(prefix) + [staged] + cli.session_id = session_id + cli.agent = agent + + cli._persist_active_session_before_close() + + stored = db.get_messages_as_conversation(session_id) + assert [m["content"] for m in stored] == [ + "old prompt", + "old answer", + "new prompt", + ] + assert staged["_db_persisted"] is True + + +def test_cli_close_builds_prompt_before_creating_first_session_row(tmp_path, monkeypatch): + """First-turn close persistence must not leave a NULL prompt snapshot.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + + import agent.conversation_loop as loop_mod + import cli as cli_mod + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "cli-close-first-turn" + agent = _real_agent(db, session_id, []) + agent._session_db_created = False + agent._cached_system_prompt = None + staged = {"role": "user", "content": "first prompt"} + agent._pending_cli_user_message = staged + + def _build_prompt(target, _system_message, _history): + target._cached_system_prompt = "close-built-system-prompt" + + monkeypatch.setattr(loop_mod, "_restore_or_build_system_prompt", _build_prompt) + + cli = object.__new__(cli_mod.HermesCLI) + cli.conversation_history = [staged] + cli.session_id = session_id + cli.agent = agent + + cli._persist_active_session_before_close() + + session = db.get_session(session_id) + assert session is not None + assert session["system_prompt"] == "close-built-system-prompt" + assert [m["content"] for m in db.get_messages_as_conversation(session_id)] == [ + "first prompt" + ] From a22a1079a39183cee3f5509c443990a203289b0c Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Mon, 13 Jul 2026 23:37:03 +0530 Subject: [PATCH 04/12] fix(cli): preserve noted staged input on close --- cli.py | 9 +- tests/agent/test_turn_context.py | 18 ++++ tests/cli/test_cli_interrupt_ack_race.py | 39 ++++++++ .../cli/test_cli_shutdown_memory_messages.py | 93 +++++++++++++++++++ 4 files changed, 158 insertions(+), 1 deletion(-) diff --git a/cli.py b/cli.py index 0b4c4b194286..45e32090febd 100644 --- a/cli.py +++ b/cli.py @@ -12365,13 +12365,20 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): self._pending_moa_config = None if _moa_cfg is None: _moa_cfg = None + # Model/skill notes and voice instructions are API-local. Keep + # the original staged input as the durable transcript value so a + # close-path marker follows the same dict into turn setup rather + # than producing a second noted user row (#63766). + _persist_clean_user_message = ( + message if (_voice_prefix or agent_message != message) else None + ) try: result = self.agent.run_conversation( user_message=agent_message, conversation_history=self.conversation_history[:-1], # Exclude the message we just added stream_callback=stream_callback, task_id=self.session_id, - persist_user_message=message if _voice_prefix else None, + persist_user_message=_persist_clean_user_message, moa_config=_moa_cfg, ) if getattr(self, "_pending_moa_disable_after_turn", False): diff --git a/tests/agent/test_turn_context.py b/tests/agent/test_turn_context.py index d024eca56fc8..e0642c5f11bf 100644 --- a/tests/agent/test_turn_context.py +++ b/tests/agent/test_turn_context.py @@ -241,6 +241,24 @@ def test_stale_pending_cli_message_does_not_replace_new_turn_input(): assert agent._pending_cli_user_message is None +def test_pending_cli_message_uses_clean_override_for_api_local_note(): + """A noted API message reuses the clean staged dict and its DB marker.""" + agent = _FakeAgent() + staged = {"role": "user", "content": "clean prompt", "_db_persisted": True} + agent._pending_cli_user_message = staged + + ctx = _build( + agent, + user_message="[MODEL NOTE]\n\nclean prompt", + persist_user_message="clean prompt", + ) + + assert ctx.messages[-1] is staged + assert ctx.messages[-1]["content"] == "[MODEL NOTE]\n\nclean prompt" + assert ctx.messages[-1]["_db_persisted"] is True + assert agent._pending_cli_user_message is None + + def test_memory_nudge_fires_at_interval(): agent = _FakeAgent() agent._memory_nudge_interval = 1 diff --git a/tests/cli/test_cli_interrupt_ack_race.py b/tests/cli/test_cli_interrupt_ack_race.py index a55fe43d2022..dbcd7a15bd09 100644 --- a/tests/cli/test_cli_interrupt_ack_race.py +++ b/tests/cli/test_cli_interrupt_ack_race.py @@ -192,3 +192,42 @@ def test_acknowledged_interrupt_still_requeues_message(): queued.append(cli._pending_input.get_nowait()) assert any("redirect please" in str(q) for q in queued) assert cli._last_turn_interrupted is True + + +def test_chat_persists_clean_input_when_a_queued_note_changes_api_message(): + """Queued notes remain API-local and preserve close-handoff marker identity.""" + cli = _make_cli() + + class _NoteAgent(_StubAgent): + def __init__(self, session_id): + super().__init__(session_id, turn_seconds=0) + self.captured = None + + def run_conversation(self, **kwargs): + self.captured = kwargs + return { + "final_response": "done", + "messages": [{"role": "assistant", "content": "done"}], + "api_calls": 1, + "completed": True, + "partial": True, + "response_previewed": True, + } + + agent = _NoteAgent(cli.session_id) + cli.agent = agent + cli._interrupt_queue = queue.Queue() + cli._pending_input = queue.Queue() + cli._pending_model_switch_note = "[MODEL SWITCH NOTE]" + + with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \ + patch.object(cli, "_resolve_turn_agent_config", return_value={ + "signature": cli._active_agent_route_signature, + "model": None, "runtime": None, "request_overrides": None, + }), \ + patch.object(cli, "_init_agent", return_value=True): + cli.chat("clean prompt") + + assert agent.captured is not None + assert agent.captured["user_message"] == "[MODEL SWITCH NOTE]\n\nclean prompt" + assert agent.captured["persist_user_message"] == "clean prompt" diff --git a/tests/cli/test_cli_shutdown_memory_messages.py b/tests/cli/test_cli_shutdown_memory_messages.py index 85787a72e4a7..63ce29cc033f 100644 --- a/tests/cli/test_cli_shutdown_memory_messages.py +++ b/tests/cli/test_cli_shutdown_memory_messages.py @@ -17,6 +17,7 @@ other tests keep their existing no-arg behaviour. from __future__ import annotations import threading +import types from typing import Any from unittest.mock import MagicMock, patch @@ -491,6 +492,98 @@ def test_cli_close_persists_pending_user_when_agent_snapshot_is_empty(tmp_path, assert staged["_db_persisted"] is True +def test_cli_close_preserves_clean_staged_user_across_noted_worker_turn(tmp_path, monkeypatch): + """A noted API-only turn reuses the close-marked clean staged user row.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + + import cli as cli_mod + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "cli-close-noted-staged-user" + db.create_session(session_id=session_id, source="cli") + prefix = [ + {"role": "user", "content": "old prompt"}, + {"role": "assistant", "content": "old answer"}, + ] + agent = _real_agent(db, session_id, prefix) + agent._flush_messages_to_session_db(prefix, []) + staged = {"role": "user", "content": "new prompt"} + agent._pending_cli_user_message = staged + + cli = object.__new__(cli_mod.HermesCLI) + cli.conversation_history = list(prefix) + [staged] + cli.session_id = session_id + cli.agent = agent + + cli._persist_active_session_before_close() + assert staged["_db_persisted"] is True + + # A queued model/skills note changes only the API message. The worker + # reuses the marked clean dict, so the normal persistence seam cannot append + # a second noted user row. + from agent.turn_context import build_turn_context + + agent.quiet_mode = True + agent.max_iterations = 1 + agent.provider = "test" + agent.base_url = "" + agent.api_key = "" + agent.api_mode = "chat_completions" + agent.tools = [] + agent.valid_tool_names = set() + agent.enabled_toolsets = None + agent.disabled_toolsets = None + agent._skip_mcp_refresh = True + agent.compression_enabled = False + agent.context_compressor = types.SimpleNamespace(protect_first_n=2, protect_last_n=2) + agent._memory_store = None + agent._memory_manager = None + agent._memory_nudge_interval = 0 + agent._turns_since_memory = 0 + agent._user_turn_count = 0 + agent._todo_store = types.SimpleNamespace(has_items=lambda: True) + agent._tool_guardrails = types.SimpleNamespace(reset_for_turn=lambda: None) + agent._compression_warning = None + agent._interrupt_requested = False + agent._memory_write_origin = "assistant_tool" + agent._stream_context_scrubber = None + agent._stream_think_scrubber = None + agent._restore_primary_runtime = lambda: None + agent._cleanup_dead_connections = lambda: False + agent._emit_status = lambda _message: None + agent._replay_compression_warning = lambda: None + agent._hydrate_todo_store = lambda *_args: None + agent._safe_print = lambda *_args: None + + worker = build_turn_context( + agent, + "[MODEL SWITCH NOTE]\n\nnew prompt", + None, + prefix, + "task", + None, + "new prompt", + None, + restore_or_build_system_prompt=lambda *_args: None, + install_safe_stdio=lambda: None, + sanitize_surrogates=lambda value: value, + summarize_user_message_for_log=lambda value: value, + set_session_context=lambda _session_id: None, + set_current_write_origin=lambda _origin: None, + ra=lambda: types.SimpleNamespace(_set_interrupt=lambda *_args: None), + ) + assert worker.messages[-1] is staged + assert worker.messages[-1]["content"] == "[MODEL SWITCH NOTE]\n\nnew prompt" + + stored = db.get_messages_as_conversation(session_id) + assert [m["content"] for m in stored] == [ + "old prompt", + "old answer", + "new prompt", + ] + + def test_cli_close_builds_prompt_before_creating_first_session_row(tmp_path, monkeypatch): """First-turn close persistence must not leave a NULL prompt snapshot.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) From 0b422559f3b21b6dc2c94039b70df8a5b8a103cf Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Mon, 13 Jul 2026 23:52:32 +0530 Subject: [PATCH 05/12] fix(session): preserve clean multimodal persistence override --- agent/conversation_loop.py | 4 ++-- agent/turn_context.py | 4 ++-- gateway/run.py | 4 ++-- run_agent.py | 15 +++++++++------ tests/run_agent/test_run_agent.py | 25 +++++++++++++++++++++++++ 5 files changed, 40 insertions(+), 12 deletions(-) diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 75c07540d885..16083c7cab6d 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -522,12 +522,12 @@ def _sync_failover_system_message(agent, api_messages, active_system_prompt): def run_conversation( agent, - user_message: str, + user_message: Any, system_message: str = None, conversation_history: List[Dict[str, Any]] = None, task_id: str = None, stream_callback: Optional[callable] = None, - persist_user_message: Optional[str] = None, + persist_user_message: Optional[Any] = None, persist_user_timestamp: Optional[float] = None, moa_config: Optional[dict[str, Any]] = None, ) -> Dict[str, Any]: diff --git a/agent/turn_context.py b/agent/turn_context.py index cf1cdff47517..ea150ff30a7c 100644 --- a/agent/turn_context.py +++ b/agent/turn_context.py @@ -118,12 +118,12 @@ class TurnContext: def build_turn_context( agent, - user_message: str, + user_message: Any, system_message: Optional[str], conversation_history: Optional[List[Dict[str, Any]]], task_id: Optional[str], stream_callback, - persist_user_message: Optional[str], + persist_user_message: Optional[Any], persist_user_timestamp: Optional[float] = None, *, restore_or_build_system_prompt, diff --git a/gateway/run.py b/gateway/run.py index d5b3fbff4743..9b19190338d4 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -17123,7 +17123,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew event_message_id: Optional[str] = None, channel_prompt: Optional[str] = None, moa_config: Optional[dict] = None, - persist_user_message: Optional[str] = None, + persist_user_message: Optional[Any] = None, persist_user_timestamp: Optional[float] = None, ) -> Dict[str, Any]: """Profile-scoping wrapper around the agent run. @@ -17184,7 +17184,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew event_message_id: Optional[str] = None, channel_prompt: Optional[str] = None, moa_config: Optional[dict] = None, - persist_user_message: Optional[str] = None, + persist_user_message: Optional[Any] = None, persist_user_timestamp: Optional[float] = None, ) -> Dict[str, Any]: """ diff --git a/run_agent.py b/run_agent.py index 2d710d77e2d9..82e93a5dc04a 100644 --- a/run_agent.py +++ b/run_agent.py @@ -1871,11 +1871,14 @@ class AIAgent: content = msg.get("content") _row_timestamp = msg.get("timestamp") # Apply the persist override to THIS row's written values only - # (never to the live dict). Match the original guard: text-only - # content is replaced; multimodal (list) content is left intact - # so image/audio blocks aren't clobbered by the text override. + # (never to the live dict). A multimodal override is a complete + # clean replacement for an API-local noted payload. Preserve the + # historical text-only guard for a list payload, though: a plain + # text override must not erase its image/audio transcript summary. if _ov_idx == _msg_idx and msg.get("role") == "user": - if _ov_content is not None and not isinstance(content, list): + if _ov_content is not None and ( + not isinstance(content, list) or isinstance(_ov_content, list) + ): content = _ov_content if _ov_timestamp is not None: _row_timestamp = _ov_timestamp @@ -5798,12 +5801,12 @@ class AIAgent: def run_conversation( self, - user_message: str, + user_message: Any, system_message: str = None, conversation_history: List[Dict[str, Any]] = None, task_id: str = None, stream_callback: Optional[callable] = None, - persist_user_message: Optional[str] = None, + persist_user_message: Optional[Any] = None, persist_user_timestamp: Optional[float] = None, moa_config: Optional[dict[str, Any]] = None, ) -> Dict[str, Any]: diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 396657ea4e85..07e54a50ded4 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -143,6 +143,31 @@ def test_persist_user_message_override_preserves_multimodal_turns(agent): assert messages == [{"role": "user", "content": multimodal_content}] +def test_flush_persist_override_replaces_api_local_multimodal_note(agent): + """A note-added multimodal API payload stores the original clean content.""" + clean_content = [ + {"type": "text", "text": "Describe this screenshot"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}, + ] + api_content = [ + {"type": "text", "text": "[MODEL SWITCH NOTE]\n\nDescribe this screenshot"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}, + ] + agent._session_db = MagicMock() + agent._session_db_created = True + agent.session_id = "session-123" + agent._last_flushed_db_idx = 0 + agent._persist_user_message_idx = 0 + agent._persist_user_message_override = clean_content + agent._persist_user_message_timestamp = None + + agent._flush_messages_to_session_db([{"role": "user", "content": api_content}], []) + + db_write = agent._session_db.append_message.call_args.kwargs + assert db_write["content"] == "Describe this screenshot\n[screenshot]" + assert api_content[0]["text"] == "[MODEL SWITCH NOTE]\n\nDescribe this screenshot" + + @pytest.fixture() def agent_with_memory_tool(): """Agent whose valid_tool_names includes 'memory'.""" From 69fd846ef86c034c74a855e98733f7edd3ce433e Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 14 Jul 2026 00:09:41 +0530 Subject: [PATCH 06/12] fix(session): serialize direct persistence flushes --- run_agent.py | 12 +++++++ tests/run_agent/test_run_agent.py | 52 +++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+) diff --git a/run_agent.py b/run_agent.py index 82e93a5dc04a..e815971d35f5 100644 --- a/run_agent.py +++ b/run_agent.py @@ -1765,6 +1765,18 @@ class AIAgent: return repair_message_sequence(self, messages) def _flush_messages_to_session_db(self, messages: List[Dict], conversation_history: List[Dict] = None): + """Serialize direct and turn-boundary session flushes per agent.""" + persist_lock = getattr(self, "_session_persist_lock", None) + if persist_lock is None: + return self._flush_messages_to_session_db_unlocked(messages, conversation_history) + with persist_lock: + return self._flush_messages_to_session_db_unlocked(messages, conversation_history) + + def _flush_messages_to_session_db_unlocked( + self, + messages: List[Dict], + conversation_history: List[Dict] = None, + ): """Persist any un-flushed messages to the SQLite session store. Deduplicates via an intrinsic ``_DB_PERSISTED_MARKER`` stamped on each diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 07e54a50ded4..b0b9a2b93162 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -11,6 +11,7 @@ import io import json import logging import re +import threading import uuid from logging.handlers import RotatingFileHandler from pathlib import Path @@ -168,6 +169,57 @@ def test_flush_persist_override_replaces_api_local_multimodal_note(agent): assert api_content[0]["text"] == "[MODEL SWITCH NOTE]\n\nDescribe this screenshot" +def test_direct_session_db_flushes_share_marker_claim(agent): + """A direct flush cannot interleave its marker check with `_persist_session`.""" + class _BarrierDB: + def __init__(self): + self.rows = [] + self.entered = threading.Event() + self.release = threading.Event() + self.calls = 0 + self._lock = threading.Lock() + + def append_message(self, **kwargs): + with self._lock: + self.calls += 1 + first = self.calls == 1 + if first: + self.entered.set() + assert self.release.wait(timeout=5) + self.rows.append(kwargs["content"]) + + db = _BarrierDB() + agent._session_db = db + agent._session_db_created = True + agent.session_id = "session-123" + agent._last_flushed_db_idx = 0 + agent._flushed_db_message_ids = set() + agent._flushed_db_message_session_id = None + agent._persist_user_message_idx = None + agent._persist_user_message_override = None + agent._persist_user_message_timestamp = None + agent._persist_disabled = False + agent._session_persist_lock = threading.RLock() + agent._session_json_enabled = False + + message = {"role": "user", "content": "exactly once"} + normal = threading.Thread(target=lambda: agent._persist_session([message], [])) + direct = threading.Thread(target=lambda: agent._flush_messages_to_session_db([message], [])) + normal.start() + assert db.entered.wait(timeout=5) + direct.start() + # Direct flush is blocked by the agent-wide persistence lock until the + # normal writer stamps the message's durable marker. + assert db.calls == 1 + db.release.set() + normal.join(timeout=5) + direct.join(timeout=5) + + assert not normal.is_alive() + assert not direct.is_alive() + assert db.rows == ["exactly once"] + + @pytest.fixture() def agent_with_memory_tool(): """Agent whose valid_tool_names includes 'memory'.""" From 50aebcbcffada3c8e7c3e7a0dd2181ee80c43e3d Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 14 Jul 2026 00:23:08 +0530 Subject: [PATCH 07/12] fix(session): preserve clean shortened close snapshots --- run_agent.py | 10 ++++- .../cli/test_cli_shutdown_memory_messages.py | 45 +++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/run_agent.py b/run_agent.py index e815971d35f5..89c4ce633733 100644 --- a/run_agent.py +++ b/run_agent.py @@ -1887,7 +1887,15 @@ class AIAgent: # clean replacement for an API-local noted payload. Preserve the # historical text-only guard for a list payload, though: a plain # text override must not erase its image/audio transcript summary. - if _ov_idx == _msg_idx and msg.get("role") == "user": + # The close safety-net may flush a shortened snapshot while + # turn setup still owns its staged CLI dict. In that shape the + # normal turn index refers to the full history, not this list; + # preserve the API-local override by recognizing the same dict. + pending_cli_message = getattr(self, "_pending_cli_user_message", None) + is_current_turn_user = ( + _ov_idx == _msg_idx or msg is pending_cli_message + ) + if is_current_turn_user and msg.get("role") == "user": if _ov_content is not None and ( not isinstance(content, list) or isinstance(_ov_content, list) ): diff --git a/tests/cli/test_cli_shutdown_memory_messages.py b/tests/cli/test_cli_shutdown_memory_messages.py index 63ce29cc033f..aec69e8c4692 100644 --- a/tests/cli/test_cli_shutdown_memory_messages.py +++ b/tests/cli/test_cli_shutdown_memory_messages.py @@ -492,6 +492,51 @@ def test_cli_close_persists_pending_user_when_agent_snapshot_is_empty(tmp_path, assert staged["_db_persisted"] is True +def test_cli_close_uses_clean_override_for_shortened_pending_snapshot(tmp_path, monkeypatch): + """Close retains the clean user text when its snapshot omits the prefix.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + + import cli as cli_mod + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "cli-close-shortened-noted-pending" + db.create_session(session_id=session_id, source="cli") + prefix = [ + {"role": "user", "content": "old prompt"}, + {"role": "assistant", "content": "old answer"}, + ] + for message in prefix: + db.append_message( + session_id=session_id, + role=message["role"], + content=message["content"], + ) + + agent = _real_agent(db, session_id, []) + staged = {"role": "user", "content": "[MODEL NOTE]\n\nnew prompt"} + agent._pending_cli_user_message = staged + # The normal worker index is relative to the full resumed history, while a + # close before its first persistence flush sees only this staged dict. + agent._persist_user_message_idx = len(prefix) + agent._persist_user_message_override = "new prompt" + agent._persist_user_message_timestamp = None + + cli = object.__new__(cli_mod.HermesCLI) + cli.conversation_history = list(prefix) + [staged] + cli.session_id = session_id + cli.agent = agent + + cli._persist_active_session_before_close() + + assert [m["content"] for m in db.get_messages_as_conversation(session_id)] == [ + "old prompt", + "old answer", + "new prompt", + ] + assert staged["_db_persisted"] is True + + def test_cli_close_preserves_clean_staged_user_across_noted_worker_turn(tmp_path, monkeypatch): """A noted API-only turn reuses the close-marked clean staged user row.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) From 962189d9ea66326d0e40672cbb7fd6110452cc30 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 14 Jul 2026 00:35:56 +0530 Subject: [PATCH 08/12] fix(cli): clear stale persistence override before staging --- cli.py | 22 +++- run_agent.py | 8 +- tests/cli/test_cli_interrupt_ack_race.py | 131 +++++++++++++++++++++++ 3 files changed, 156 insertions(+), 5 deletions(-) diff --git a/cli.py b/cli.py index 45e32090febd..99b910b23fb8 100644 --- a/cli.py +++ b/cli.py @@ -12226,9 +12226,25 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): # duplicate-prone intermediate snapshot to terminal-close persistence. if self.conversation_history is getattr(agent, "_session_messages", None): self.conversation_history = list(self.conversation_history) - staged_user_message = {"role": "user", "content": message} - agent._pending_cli_user_message = staged_user_message - self.conversation_history.append(staged_user_message) + # The prior turn's override applies only to its own user dict. Clear it + # before exposing the next staged input to close persistence; otherwise + # a shutdown before the worker prologue can write old API-local text as + # this new user message (#63766). + persist_lock = getattr(agent, "_session_persist_lock", None) + + def _stage_user_message() -> None: + agent._persist_user_message_idx = None + agent._persist_user_message_override = None + agent._persist_user_message_timestamp = None + staged_user_message = {"role": "user", "content": message} + agent._pending_cli_user_message = staged_user_message + self.conversation_history.append(staged_user_message) + + if persist_lock is None: + _stage_user_message() + else: + with persist_lock: + _stage_user_message() ChatConsole().print(f"[{_accent_hex()}]{'─' * 40}[/]") print(flush=True) diff --git a/run_agent.py b/run_agent.py index 89c4ce633733..cf8e63ff9b52 100644 --- a/run_agent.py +++ b/run_agent.py @@ -1764,7 +1764,11 @@ class AIAgent: from agent.agent_runtime_helpers import repair_message_sequence return repair_message_sequence(self, messages) - def _flush_messages_to_session_db(self, messages: List[Dict], conversation_history: List[Dict] = None): + def _flush_messages_to_session_db( + self, + messages: List[Dict], + conversation_history: Optional[List[Dict]] = None, + ): """Serialize direct and turn-boundary session flushes per agent.""" persist_lock = getattr(self, "_session_persist_lock", None) if persist_lock is None: @@ -1775,7 +1779,7 @@ class AIAgent: def _flush_messages_to_session_db_unlocked( self, messages: List[Dict], - conversation_history: List[Dict] = None, + conversation_history: Optional[List[Dict]] = None, ): """Persist any un-flushed messages to the SQLite session store. diff --git a/tests/cli/test_cli_interrupt_ack_race.py b/tests/cli/test_cli_interrupt_ack_race.py index dbcd7a15bd09..ac465cb20ca1 100644 --- a/tests/cli/test_cli_interrupt_ack_race.py +++ b/tests/cli/test_cli_interrupt_ack_race.py @@ -26,6 +26,7 @@ from __future__ import annotations import importlib import queue import sys +import threading import time from unittest.mock import MagicMock, patch @@ -231,3 +232,133 @@ def test_chat_persists_clean_input_when_a_queued_note_changes_api_message(): assert agent.captured is not None assert agent.captured["user_message"] == "[MODEL SWITCH NOTE]\n\nclean prompt" assert agent.captured["persist_user_message"] == "clean prompt" + + +def test_chat_clears_previous_turn_persistence_override_before_staging(): + """A close before the next worker starts cannot reuse a stale override.""" + cli = _make_cli() + + class _StagingAgent(_StubAgent): + def __init__(self, session_id): + super().__init__(session_id, turn_seconds=0) + self.staged_override = None + self.staged_message = None + self._session_messages = [] + self._persist_user_message_idx = 7 + self._persist_user_message_override = "previous clean prompt" + self._persist_user_message_timestamp = 123.0 + + def run_conversation(self, **kwargs): + self.staged_override = self._persist_user_message_override + self.staged_message = self._pending_cli_user_message + return { + "final_response": "done", + "messages": [{"role": "assistant", "content": "done"}], + "api_calls": 1, + "completed": True, + "partial": True, + "response_previewed": True, + } + + agent = _StagingAgent(cli.session_id) + cli.agent = agent + cli._interrupt_queue = queue.Queue() + cli._pending_input = queue.Queue() + + with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \ + patch.object(cli, "_resolve_turn_agent_config", return_value={ + "signature": cli._active_agent_route_signature, + "model": None, "runtime": None, "request_overrides": None, + }), \ + patch.object(cli, "_init_agent", return_value=True): + cli.chat("new prompt") + + assert agent.staged_override is None + assert agent._persist_user_message_idx is None + assert agent._persist_user_message_timestamp is None + assert agent.staged_message == {"role": "user", "content": "new prompt"} + + +def test_chat_close_does_not_persist_previous_turn_override(tmp_path, monkeypatch): + """A close after input staging writes the new prompt, not old API-only text.""" + from hermes_state import SessionDB + from run_agent import AIAgent + + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + cli = _make_cli() + session_id = cli.session_id + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session(session_id=session_id, source="cli") + prefix = [ + {"role": "user", "content": "old prompt"}, + {"role": "assistant", "content": "old answer"}, + ] + for message in prefix: + db.append_message( + session_id=session_id, + role=message["role"], + content=message["content"], + ) + + agent = object.__new__(AIAgent) + agent._session_db = db + agent._session_db_created = True + agent.session_id = session_id + agent.platform = "cli" + agent.model = "test-model" + agent._session_messages = [] + agent._last_flushed_db_idx = 0 + agent._flushed_db_message_ids = set() + agent._flushed_db_message_session_id = None + agent._persist_disabled = False + agent._cached_system_prompt = "test system prompt" + agent._session_init_model_config = None + agent._parent_session_id = None + agent._session_json_enabled = False + agent._pending_cli_user_message = None + agent._session_persist_lock = threading.RLock() + agent._persist_user_message_idx = len(prefix) + agent._persist_user_message_override = "previous clean prompt" + agent._persist_user_message_timestamp = 123.0 + agent._active_children = [] + agent._interrupt_requested = False + entered = threading.Event() + release = threading.Event() + + def _block_run(**_kwargs): + entered.set() + assert release.wait(timeout=5) + return { + "final_response": "done", + "messages": prefix + [{"role": "assistant", "content": "done"}], + "api_calls": 1, + "completed": True, + "partial": True, + "response_previewed": True, + } + + agent.run_conversation = _block_run + cli.agent = agent + cli.conversation_history = list(prefix) + cli._interrupt_queue = queue.Queue() + cli._pending_input = queue.Queue() + + with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \ + patch.object(cli, "_resolve_turn_agent_config", return_value={ + "signature": cli._active_agent_route_signature, + "model": None, "runtime": None, "request_overrides": None, + }), \ + patch.object(cli, "_init_agent", return_value=True): + chat_thread = threading.Thread(target=lambda: cli.chat("new prompt")) + chat_thread.start() + assert entered.wait(timeout=5) + cli._persist_active_session_before_close() + release.set() + chat_thread.join(timeout=10) + + assert not chat_thread.is_alive() + assert [m["content"] for m in db.get_messages_as_conversation(session_id)] == [ + "old prompt", + "old answer", + "new prompt", + ] From 32bdc67e104934ca7324df430437e0cd839e01c3 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:24:46 +0530 Subject: [PATCH 09/12] fix(cli): snapshot close state under staging lock --- cli.py | 112 +++++++++++---------- tests/cli/test_cli_interrupt_ack_race.py | 122 +++++++++++++++++++++++ 2 files changed, 180 insertions(+), 54 deletions(-) diff --git a/cli.py b/cli.py index 99b910b23fb8..92010a1000e8 100644 --- a/cli.py +++ b/cli.py @@ -12857,60 +12857,64 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): if not agent or not hasattr(agent, "_persist_session"): return - messages = getattr(agent, "_session_messages", None) - pending_cli_message = getattr(agent, "_pending_cli_user_message", None) - if not isinstance(messages, list): - messages = getattr(self, "conversation_history", None) - if not isinstance(messages, list): - return - if isinstance(pending_cli_message, dict) and not any( - message is pending_cli_message for message in messages - ): - # The UI has accepted a new input but the worker still exposes its - # prior snapshot. Include only that staged dict; the baseline below - # keeps any durable resumed prefix from being re-appended. - messages = [*messages, pending_cli_message] - if not messages: - return - - # A normal turn builds a new list that reuses the resumed-history dicts. - # Keep that CLI history as the baseline so a signal between assigning - # ``_session_messages`` and the turn's DB flush cannot append its durable - # prefix a second time. Once the CLI takes the turn result, however, both - # names can point at the same live list; passing that alias would mark an - # unflushed tail durable without writing it. Marker-only persistence is - # correct only in that alias case. - conversation_history = getattr(self, "conversation_history", None) - pending_cli_message = getattr(agent, "_pending_cli_user_message", None) - if ( - isinstance(conversation_history, list) - and conversation_history - and conversation_history[-1] is pending_cli_message - ): - # The UI accepted this user message before the agent finished its - # early persistence. Its dict can already be in ``messages`` but is - # not durable yet, so exclude it from the resumed-history baseline. - conversation_history = conversation_history[:-1] - elif not isinstance(conversation_history, list) or conversation_history is messages: - conversation_history = None - - # A first-turn close can arrive before the worker builds its cached - # prompt. Build or restore it before the DB row is created so the - # durable transcript never leaves a NULL system_prompt cache entry. - if getattr(agent, "_cached_system_prompt", None) is None: - try: - from agent.conversation_loop import _restore_or_build_system_prompt - - _restore_or_build_system_prompt(agent, None, conversation_history) - except Exception: - logger.debug("Could not build system prompt during CLI close", exc_info=True) - return - if getattr(agent, "_cached_system_prompt", None) is None: - return - persist_lock = getattr(agent, "_session_persist_lock", None) - def _ensure_and_persist() -> None: + def _snapshot_and_persist() -> None: + # This snapshot must share the staging lock with ``chat()``. Without + # it, close can retain a mutable history baseline just before chat + # appends its pending dict; the later flush then mistakes that dict + # for durable history and stamps it without writing a row (#63766). + messages = getattr(agent, "_session_messages", None) + pending_cli_message = getattr(agent, "_pending_cli_user_message", None) + if not isinstance(messages, list): + messages = getattr(self, "conversation_history", None) + if not isinstance(messages, list): + return + if isinstance(pending_cli_message, dict) and not any( + message is pending_cli_message for message in messages + ): + # The UI has accepted a new input but the worker still exposes its + # prior snapshot. Include only that staged dict; the baseline below + # keeps any durable resumed prefix from being re-appended. + messages = [*messages, pending_cli_message] + if not messages: + return + + # A normal turn builds a new list that reuses the resumed-history dicts. + # Keep that CLI history as the baseline so a signal between assigning + # ``_session_messages`` and the turn's DB flush cannot append its durable + # prefix a second time. Once the CLI takes the turn result, however, both + # names can point at the same live list; passing that alias would mark an + # unflushed tail durable without writing it. Marker-only persistence is + # correct only in that alias case. + conversation_history = getattr(self, "conversation_history", None) + pending_cli_message = getattr(agent, "_pending_cli_user_message", None) + if ( + isinstance(conversation_history, list) + and conversation_history + and conversation_history[-1] is pending_cli_message + ): + # The UI accepted this user message before the agent finished its + # early persistence. Its dict can already be in ``messages`` but is + # not durable yet, so exclude it from the resumed-history baseline. + conversation_history = conversation_history[:-1] + elif not isinstance(conversation_history, list) or conversation_history is messages: + conversation_history = None + + # A first-turn close can arrive before the worker builds its cached + # prompt. Build or restore it before the DB row is created so the + # durable transcript never leaves a NULL system_prompt cache entry. + if getattr(agent, "_cached_system_prompt", None) is None: + try: + from agent.conversation_loop import _restore_or_build_system_prompt + + _restore_or_build_system_prompt(agent, None, conversation_history) + except Exception: + logger.debug("Could not build system prompt during CLI close", exc_info=True) + return + if getattr(agent, "_cached_system_prompt", None) is None: + return + agent._ensure_db_session() agent._persist_session(messages, conversation_history) if getattr(agent, "session_id", None): @@ -12918,10 +12922,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): try: if persist_lock is None: - _ensure_and_persist() + _snapshot_and_persist() else: with persist_lock: - _ensure_and_persist() + _snapshot_and_persist() except (Exception, KeyboardInterrupt) as e: logger.debug("Could not persist active CLI session before close: %s", e) diff --git a/tests/cli/test_cli_interrupt_ack_race.py b/tests/cli/test_cli_interrupt_ack_race.py index ac465cb20ca1..c4e8036d67dd 100644 --- a/tests/cli/test_cli_interrupt_ack_race.py +++ b/tests/cli/test_cli_interrupt_ack_race.py @@ -362,3 +362,125 @@ def test_chat_close_does_not_persist_previous_turn_override(tmp_path, monkeypatc "old answer", "new prompt", ] + + +def test_close_waits_for_atomic_cli_staging_before_snapshot(tmp_path, monkeypatch): + """Close cannot retain the mutable pre-append history as its DB baseline.""" + from hermes_state import SessionDB + from run_agent import AIAgent + + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + cli = _make_cli() + session_id = cli.session_id + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session(session_id=session_id, source="cli") + prefix = [ + {"role": "user", "content": "old prompt"}, + {"role": "assistant", "content": "old answer"}, + ] + for message in prefix: + db.append_message( + session_id=session_id, + role=message["role"], + content=message["content"], + ) + + agent = object.__new__(AIAgent) + agent._session_db = db + agent._session_db_created = True + agent.session_id = session_id + agent.platform = "cli" + agent.model = "test-model" + # Deliberately distinct from CLI history: this is the normal pre-worker + # state that used to let close retain the wrong mutable baseline. + agent._session_messages = list(prefix) + agent._last_flushed_db_idx = 0 + agent._flushed_db_message_ids = set() + agent._flushed_db_message_session_id = None + agent._persist_disabled = False + agent._cached_system_prompt = "test system prompt" + agent._session_init_model_config = None + agent._parent_session_id = None + agent._session_json_enabled = False + agent._pending_cli_user_message = None + agent._session_persist_lock = threading.RLock() + agent._persist_user_message_idx = None + agent._persist_user_message_override = None + agent._persist_user_message_timestamp = None + agent._active_children = [] + agent._interrupt_requested = False + + staging_entered = threading.Event() + release_staging = threading.Event() + run_entered = threading.Event() + release_run = threading.Event() + + class _BlockingHistory(list): + def __init__(self, values): + super().__init__(values) + self._block_next_append = True + + def append(self, value): + if self._block_next_append: + self._block_next_append = False + staging_entered.set() + assert release_staging.wait(timeout=5) + return super().append(value) + + def _block_run(**_kwargs): + run_entered.set() + assert release_run.wait(timeout=5) + return { + "final_response": "done", + "messages": prefix + [{"role": "assistant", "content": "done"}], + "api_calls": 1, + "completed": True, + "partial": True, + "response_previewed": True, + } + + agent.run_conversation = _block_run + cli.agent = agent + cli.conversation_history = _BlockingHistory(prefix) + cli._interrupt_queue = queue.Queue() + cli._pending_input = queue.Queue() + + with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \ + patch.object(cli, "_resolve_turn_agent_config", return_value={ + "signature": cli._active_agent_route_signature, + "model": None, "runtime": None, "request_overrides": None, + }), \ + patch.object(cli, "_init_agent", return_value=True): + chat_thread = threading.Thread(target=lambda: cli.chat("new prompt")) + chat_thread.start() + assert staging_entered.wait(timeout=5) + + close_started = threading.Event() + close_finished = threading.Event() + + def _close(): + close_started.set() + cli._persist_active_session_before_close() + close_finished.set() + + close_thread = threading.Thread(target=_close) + close_thread.start() + assert close_started.wait(timeout=5) + # The close snapshot must wait for the locked pending-pointer/history + # handoff; otherwise the subsequent append poisons its DB baseline. + assert not close_finished.wait(timeout=0.1) + + release_staging.set() + assert run_entered.wait(timeout=5) + assert close_finished.wait(timeout=5) + release_run.set() + chat_thread.join(timeout=10) + close_thread.join(timeout=10) + + assert not chat_thread.is_alive() + assert not close_thread.is_alive() + assert [m["content"] for m in db.get_messages_as_conversation(session_id)] == [ + "old prompt", + "old answer", + "new prompt", + ] From 8341d775a97f7dfda65827617564d703e27719cb Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:37:53 +0530 Subject: [PATCH 10/12] fix(session): restore clean API-local turn content --- agent/turn_finalizer.py | 9 +++ run_agent.py | 16 ++-- ...rn_finalizer_final_response_persistence.py | 81 ++++++++++++++++++- tests/run_agent/test_run_agent.py | 18 +++++ 4 files changed, 115 insertions(+), 9 deletions(-) diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py index 1adc3c6c3632..fdf5babe1aea 100644 --- a/agent/turn_finalizer.py +++ b/agent/turn_finalizer.py @@ -228,6 +228,15 @@ def finalize_turn( if _tail_role != "assistant": messages.append({"role": "assistant", "content": final_response}) + # The model has completed its request, so replace API-local + # voice/model/skill guidance with the clean user input before writing the + # final durable snapshot and returning the continuation history. Earlier + # turn-start flushes use the DB-only override because their messages are + # still needed for the API request; this finalizer runs after that request + # is complete (#48677 / #63766). + _apply_override = getattr(agent, "_apply_persist_user_message_override", None) + if callable(_apply_override): + _apply_override(messages) agent._persist_session(messages, conversation_history) except Exception as _persist_err: _cleanup_errors.append(f"persist_session: {_persist_err}") diff --git a/run_agent.py b/run_agent.py index cf8e63ff9b52..19f4639b4940 100644 --- a/run_agent.py +++ b/run_agent.py @@ -1663,14 +1663,14 @@ class AIAgent: msg = messages[idx] if isinstance(msg, dict) and msg.get("role") == "user": # Text-only call paths may pass a synthetic API-facing prompt - # and a cleaner transcript string separately. Multimodal - # turns, however, keep image/audio blocks in the live - # messages list that is still used for the API request after - # early crash-resilience persistence. Do not replace those - # blocks with the text-only persistence override before the - # model call is built. The paired timestamp override still - # applies — it is metadata, not content. - if override is not None and not isinstance(msg.get("content"), list): + # and a cleaner transcript string separately. Before the API + # call, a plain-text override must not replace native image/audio + # blocks. A list override, however, is the original clean + # multimodal payload (for example before a queued /model note) + # and must replace the API-local list once the turn is final. + if override is not None and ( + not isinstance(msg.get("content"), list) or isinstance(override, list) + ): msg["content"] = override if timestamp is not None: msg["timestamp"] = timestamp diff --git a/tests/agent/test_turn_finalizer_final_response_persistence.py b/tests/agent/test_turn_finalizer_final_response_persistence.py index 2a54fd6e837f..9c6089aacbd8 100644 --- a/tests/agent/test_turn_finalizer_final_response_persistence.py +++ b/tests/agent/test_turn_finalizer_final_response_persistence.py @@ -51,7 +51,14 @@ class FakeAgent: pass def _persist_session(self, messages, conversation_history): - self.persisted_messages = list(messages) + # Capture the durable write before finalization restores API-local + # guidance to the returned/live transcript. + self.persisted_messages = [dict(message) for message in messages] + + def _apply_persist_user_message_override(self, messages): + from run_agent import AIAgent + + return AIAgent._apply_persist_user_message_override(self, messages) def _file_mutation_verifier_enabled(self): return False @@ -69,6 +76,78 @@ class FakeAgent: pass +def test_finalizer_restores_clean_api_local_text_before_return(monkeypatch): + """One-shot CLI notes do not replay through same-process history.""" + monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: []) + agent = FakeAgent() + messages = [ + {"role": "user", "content": "[MODEL SWITCH NOTE]\n\nclean prompt"}, + {"role": "assistant", "content": "Done."}, + ] + agent._persist_user_message_idx = 0 + agent._persist_user_message_override = "clean prompt" + agent._persist_user_message_timestamp = None + + result = finalize_turn( + agent, + final_response="Done.", + api_call_count=1, + interrupted=False, + failed=False, + messages=messages, + conversation_history=[], + effective_task_id="task", + turn_id="turn", + user_message="[MODEL SWITCH NOTE]\n\nclean prompt", + original_user_message="clean prompt", + _should_review_memory=False, + _turn_exit_reason="text_response(finish_reason=stop)", + ) + + assert agent.persisted_messages[0]["content"] == "clean prompt" + assert result["messages"][0]["content"] == "clean prompt" + + +def test_finalizer_restores_clean_api_local_multimodal_before_return(monkeypatch): + """A queued note does not remain in the next-turn native image payload.""" + monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: []) + agent = FakeAgent() + clean_content = [ + {"type": "text", "text": "Describe the image"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}, + ] + api_content = [ + {"type": "text", "text": "[MODEL SWITCH NOTE]\n\nDescribe the image"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}, + ] + messages = [ + {"role": "user", "content": api_content}, + {"role": "assistant", "content": "Done."}, + ] + agent._persist_user_message_idx = 0 + agent._persist_user_message_override = clean_content + agent._persist_user_message_timestamp = None + + result = finalize_turn( + agent, + final_response="Done.", + api_call_count=1, + interrupted=False, + failed=False, + messages=messages, + conversation_history=[], + effective_task_id="task", + turn_id="turn", + user_message=api_content, + original_user_message=clean_content, + _should_review_memory=False, + _turn_exit_reason="text_response(finish_reason=stop)", + ) + + assert agent.persisted_messages[0]["content"] == clean_content + assert result["messages"][0]["content"] == clean_content + + def test_final_response_closes_tool_tail_before_persistence(monkeypatch): """A recovered/previewed final response must be durable in session history. diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index b0b9a2b93162..85572ef4fc2b 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -144,6 +144,24 @@ def test_persist_user_message_override_preserves_multimodal_turns(agent): assert messages == [{"role": "user", "content": multimodal_content}] +def test_persist_user_message_override_restores_clean_multimodal_note(agent): + clean_content = [ + {"type": "text", "text": "Describe this screenshot"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}, + ] + api_content = [ + {"type": "text", "text": "[MODEL SWITCH NOTE]\n\nDescribe this screenshot"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}, + ] + messages = [{"role": "user", "content": api_content}] + agent._persist_user_message_idx = 0 + agent._persist_user_message_override = clean_content + + agent._apply_persist_user_message_override(messages) + + assert messages == [{"role": "user", "content": clean_content}] + + def test_flush_persist_override_replaces_api_local_multimodal_note(agent): """A note-added multimodal API payload stores the original clean content.""" clean_content = [ From b708d10db0afe3c3772fa0ad070ae8032b190535 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:42:23 +0530 Subject: [PATCH 11/12] test(session): type finalizer clean-history assertions --- ...t_turn_finalizer_final_response_persistence.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/tests/agent/test_turn_finalizer_final_response_persistence.py b/tests/agent/test_turn_finalizer_final_response_persistence.py index 9c6089aacbd8..a007e4732644 100644 --- a/tests/agent/test_turn_finalizer_final_response_persistence.py +++ b/tests/agent/test_turn_finalizer_final_response_persistence.py @@ -1,4 +1,5 @@ from types import SimpleNamespace +from typing import Any from agent.turn_finalizer import finalize_turn @@ -30,7 +31,10 @@ class FakeAgent: self._skill_nudge_interval = 0 self._iters_since_skill = 0 self.valid_tool_names = [] - self.persisted_messages = None + self.persisted_messages: list[dict[str, Any]] | None = None + self._persist_user_message_idx: int | None = None + self._persist_user_message_override: Any = None + self._persist_user_message_timestamp: float | None = None def _handle_max_iterations(self, messages, api_call_count): raise AssertionError("not expected") @@ -56,9 +60,10 @@ class FakeAgent: self.persisted_messages = [dict(message) for message in messages] def _apply_persist_user_message_override(self, messages): - from run_agent import AIAgent - - return AIAgent._apply_persist_user_message_override(self, messages) + idx = self._persist_user_message_idx + override = self._persist_user_message_override + if idx is not None and override is not None: + messages[idx]["content"] = override def _file_mutation_verifier_enabled(self): return False @@ -104,6 +109,7 @@ def test_finalizer_restores_clean_api_local_text_before_return(monkeypatch): _turn_exit_reason="text_response(finish_reason=stop)", ) + assert agent.persisted_messages is not None assert agent.persisted_messages[0]["content"] == "clean prompt" assert result["messages"][0]["content"] == "clean prompt" @@ -144,6 +150,7 @@ def test_finalizer_restores_clean_api_local_multimodal_before_return(monkeypatch _turn_exit_reason="text_response(finish_reason=stop)", ) + assert agent.persisted_messages is not None assert agent.persisted_messages[0]["content"] == clean_content assert result["messages"][0]["content"] == clean_content From ff52dce1faed6e4d74ce8341ba13dfe76b8c2311 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 14 Jul 2026 02:05:41 +0530 Subject: [PATCH 12/12] test(cli): cover noted multimodal persistence handoff --- tests/cli/test_cli_interrupt_ack_race.py | 175 +++++++++++++++++++++++ 1 file changed, 175 insertions(+) diff --git a/tests/cli/test_cli_interrupt_ack_race.py b/tests/cli/test_cli_interrupt_ack_race.py index c4e8036d67dd..0e2c21b6059c 100644 --- a/tests/cli/test_cli_interrupt_ack_race.py +++ b/tests/cli/test_cli_interrupt_ack_race.py @@ -28,6 +28,7 @@ import queue import sys import threading import time +import types from unittest.mock import MagicMock, patch @@ -234,6 +235,180 @@ def test_chat_persists_clean_input_when_a_queued_note_changes_api_message(): assert agent.captured["persist_user_message"] == "clean prompt" +def test_chat_preserves_clean_multimodal_input_when_note_changes_api_message(): + """A queued note forwards original native parts as the persistence override.""" + cli = _make_cli() + + class _NoteAgent(_StubAgent): + def __init__(self, session_id): + super().__init__(session_id, turn_seconds=0) + self.captured = None + + def run_conversation(self, **kwargs): + self.captured = kwargs + return { + "final_response": "done", + "messages": [{"role": "assistant", "content": "done"}], + "api_calls": 1, + "completed": True, + "partial": True, + "response_previewed": True, + } + + clean_parts = [ + {"type": "text", "text": "Describe this screenshot"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}, + ] + agent = _NoteAgent(cli.session_id) + cli.agent = agent + cli._interrupt_queue = queue.Queue() + cli._pending_input = queue.Queue() + cli._pending_model_switch_note = "[MODEL SWITCH NOTE]" + + with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \ + patch.object(cli, "_resolve_turn_agent_config", return_value={ + "signature": cli._active_agent_route_signature, + "model": None, "runtime": None, "request_overrides": None, + }), \ + patch.object(cli, "_init_agent", return_value=True): + cli.chat(clean_parts) + + assert agent.captured is not None + assert agent.captured["persist_user_message"] == clean_parts + assert agent.captured["persist_user_message"] is not agent.captured["user_message"] + api_parts = agent.captured["user_message"] + assert api_parts[0]["text"] == "[MODEL SWITCH NOTE]\n\nDescribe this screenshot" + assert api_parts[1] == clean_parts[1] + + +def test_chat_multimodal_note_persists_clean_input_once(tmp_path, monkeypatch): + """The real CLI-to-agent path stores clean image parts, never the queued note.""" + from hermes_state import SessionDB + from run_agent import AIAgent + + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + cli = _make_cli() + session_id = cli.session_id + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session(session_id=session_id, source="cli") + + agent = object.__new__(AIAgent) + agent._session_db = db + agent._session_db_created = True + agent.session_id = session_id + agent.platform = "cli" + agent.model = "test-model" + agent.provider = "test" + agent.base_url = "" + agent.api_key = "" + agent.api_mode = "chat_completions" + agent._session_messages = [] + agent._last_flushed_db_idx = 0 + agent._flushed_db_message_ids = set() + agent._flushed_db_message_session_id = None + agent._persist_disabled = False + agent._cached_system_prompt = "test system prompt" + agent._session_init_model_config = None + agent._parent_session_id = None + agent._session_json_enabled = False + agent._pending_cli_user_message = None + agent._session_persist_lock = threading.RLock() + agent._persist_user_message_idx = None + agent._persist_user_message_override = None + agent._persist_user_message_timestamp = None + agent._active_children = [] + agent._interrupt_requested = False + + clean_parts = [ + {"type": "text", "text": "Describe this screenshot"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}, + ] + captured = {} + + def _realish_run(**kwargs): + captured.update(kwargs) + # Drive production turn setup and the real SQLite persistence seam, + # then return a normal CLI result without starting a provider loop. + from agent.turn_context import build_turn_context + + agent.quiet_mode = True + agent.max_iterations = 1 + agent.tools = [] + agent.valid_tool_names = set() + agent.enabled_toolsets = None + agent.disabled_toolsets = None + agent._skip_mcp_refresh = True + agent.compression_enabled = False + agent.context_compressor = types.SimpleNamespace(protect_first_n=2, protect_last_n=2) + agent._memory_store = None + agent._memory_manager = None + agent._memory_nudge_interval = 0 + agent._turns_since_memory = 0 + agent._user_turn_count = 0 + agent._todo_store = types.SimpleNamespace(has_items=lambda: True) + agent._tool_guardrails = types.SimpleNamespace(reset_for_turn=lambda: None) + agent._compression_warning = None + agent._memory_write_origin = "assistant_tool" + agent._stream_context_scrubber = None + agent._stream_think_scrubber = None + agent._restore_primary_runtime = lambda: None + agent._cleanup_dead_connections = lambda: False + agent._emit_status = lambda _message: None + agent._replay_compression_warning = lambda: None + agent._hydrate_todo_store = lambda *_args: None + agent._safe_print = lambda *_args: None + + context = build_turn_context( + agent, + kwargs["user_message"], + None, + kwargs["conversation_history"], + kwargs["task_id"], + None, + kwargs["persist_user_message"], + None, + restore_or_build_system_prompt=lambda *_args: None, + install_safe_stdio=lambda: None, + sanitize_surrogates=lambda value: value, + summarize_user_message_for_log=lambda value: ( + value if isinstance(value, str) else "[multimodal test message]" + ), + set_session_context=lambda _session_id: None, + set_current_write_origin=lambda _origin: None, + ra=lambda: types.SimpleNamespace(_set_interrupt=lambda *_args: None), + ) + agent._apply_persist_user_message_override(context.messages) + agent._persist_session(context.messages, kwargs["conversation_history"]) + return { + "final_response": "done", + "messages": context.messages + [{"role": "assistant", "content": "done"}], + "api_calls": 1, + "completed": True, + "partial": True, + "response_previewed": True, + } + + agent.run_conversation = _realish_run + cli.agent = agent + cli._interrupt_queue = queue.Queue() + cli._pending_input = queue.Queue() + cli._pending_model_switch_note = "[MODEL SWITCH NOTE]" + + with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \ + patch.object(cli, "_resolve_turn_agent_config", return_value={ + "signature": cli._active_agent_route_signature, + "model": None, "runtime": None, "request_overrides": None, + }), \ + patch.object(cli, "_init_agent", return_value=True): + cli.chat(clean_parts) + + assert captured["persist_user_message"] == clean_parts + assert captured["user_message"][0]["text"] == "[MODEL SWITCH NOTE]\n\nDescribe this screenshot" + assert [m["content"] for m in db.get_messages_as_conversation(session_id)] == [ + "Describe this screenshot\n[screenshot]" + ] + + def test_chat_clears_previous_turn_persistence_override_before_staging(): """A close before the next worker starts cannot reuse a stale override.""" cli = _make_cli()