Bremen
/
emma-hermes
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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.
fix/verification-admin-route-recovery
kshitijk4poor 2026-07-13 19:08:02 +05:30
parent a27d51ef46
commit 475922f2ce
6 changed files with 333 additions and 21 deletions
Show Stats 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().

57
agent/turn_context.py
View File

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

66
cli.py
View File

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

20
run_agent.py
View File

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

35
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,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

168
tests/cli/test_cli_shutdown_memory_messages.py
View File

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