mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-21 16:18:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Merge pull request #64004 from kshitijk4poor/salvage/63274-cli-close-persist

Browse source 
fix(cli): persist close transcript without history alias
This commit is contained in:
kshitij 2026-07-14 16:45:35 +05:30 • committed by GitHub
commit 0684506072
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 1431 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

8
agent/agent_init.py
View file

@ -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().

4
agent/conversation_loop.py
View file

@ -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]:

61
agent/turn_context.py
View file

@ -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,
@ -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.

9
agent/turn_finalizer.py
View file

@ -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}")

114
cli.py
View file

@ -12176,7 +12176,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).
@ -12264,8 +12267,31 @@ 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)
# 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)
@ -12402,13 +12428,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):
@ -12871,20 +12904,75 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
if not agent or not hasattr(agent, "_persist_session"):
return
messages = getattr(agent, "_session_messages", None)
if not isinstance(messages, list):
messages = getattr(self, "conversation_history", None)
if not isinstance(messages, list) or not messages:
return
persist_lock = getattr(agent, "_session_persist_lock", None)
conversation_history = getattr(self, "conversation_history", None)
if not isinstance(conversation_history, list):
conversation_history = messages
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
try:
# 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):
self.session_id = agent.session_id
try:
if persist_lock is None:
_snapshot_and_persist()
else:
with persist_lock:
_snapshot_and_persist()
except (Exception, KeyboardInterrupt) as e:
logger.debug("Could not persist active CLI session before close: %s", e)

4
gateway/run.py
View file

@ -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]:
"""

79
run_agent.py
View file

@ -1665,14 +1665,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
@ -1691,10 +1691,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.
@ -1754,7 +1766,23 @@ 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:
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: Optional[List[Dict]] = None,
):
"""Persist any un-flushed messages to the SQLite session store.
Deduplicates via an intrinsic ``_DB_PERSISTED_MARKER`` stamped on each
@ -1861,11 +1889,22 @@ 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.
if _ov_idx == _msg_idx and msg.get("role") == "user":
if _ov_content is not None and not isinstance(content, list):
# (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.
# 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)
):
content = _ov_content
if _ov_timestamp is not None:
_row_timestamp = _ov_timestamp
@ -5788,12 +5827,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]:

53
tests/agent/test_turn_context.py
View file

@ -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 = "<unset>"
@ -206,6 +210,55 @@ 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_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

90
tests/agent/test_turn_finalizer_final_response_persistence.py
View file

@ -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")
@ -51,7 +55,15 @@ 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):
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
@ -69,6 +81,80 @@ 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 is not None
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 is not None
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.

467
tests/cli/test_cli_interrupt_ack_race.py
View file

@ -26,7 +26,9 @@ from __future__ import annotations
import importlib
import queue
import sys
import threading
import time
import types
from unittest.mock import MagicMock, patch
@ -192,3 +194,468 @@ 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"
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()
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",
]
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",
]

498
tests/cli/test_cli_shutdown_memory_messages.py
View file

@ -16,6 +16,9 @@ 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
@ -149,7 +152,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, None)
def test_cli_close_persist_skips_empty_transcripts():
@ -167,3 +170,496 @@ def test_cli_close_persist_skips_empty_transcripts():
cli._persist_active_session_before_close()
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 = "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
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
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 = _real_agent(db, session_id, transcript)
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
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
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] == [
"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",
]
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_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"))
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"))
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"
]

95
tests/run_agent/test_run_agent.py
View file

@ -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
@ -143,6 +144,100 @@ 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 = [
{"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"
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'."""