diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index 76cb67dfc34ee..a570ec9ee4882 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -577,6 +577,7 @@ def _prune_unanswered_tool_calls(messages: List[Dict]) -> Tuple[List[Dict], int] def _merge_consecutive_users(messages: List[Dict]) -> Tuple[List[Dict], int]: """Pass 3: merge consecutive plain-text user messages (no user input lost).""" from agent.context_compressor import _DB_PERSISTED_MARKER, split_user_originated_turn + from agent.session_persistence import _is_ephemeral_scaffolding repairs = 0 merged: List[Dict] = [] @@ -585,6 +586,9 @@ def _merge_consecutive_users(messages: List[Dict]) -> Tuple[List[Dict], int]: if ( prev is not None and prev.get("role") == "user" and isinstance(msg, dict) and msg.get("role") == "user" + # Request-only scaffolding merged with a real row would make its text durable (or the + # real row's text unpersisted); the wire copy merges them instead. + and not _is_ephemeral_scaffolding(prev) and not _is_ephemeral_scaffolding(msg) # A summary carrier followed by a new user row is a deliberate durable shape after # retry/rewind; never mutate the persisted carrier (sanitizers merge copies later). and split_user_originated_turn(prev)[0] is None diff --git a/agent/session_persistence.py b/agent/session_persistence.py index 4025256c01e0d..805f733a662f9 100644 --- a/agent/session_persistence.py +++ b/agent/session_persistence.py @@ -39,6 +39,7 @@ "_pre_verify_synthetic", "_kanban_stop_synthetic", # kanban worker stop-guard "_dropped_toolcall_nudge", # internal retry instruction; must not replay as user context + "_length_continuation_synthetic", # continuation nudge right after a user row (turn_truncation) ) _IMAGE_PART_TYPES = {"image", "image_url", "input_image"} @@ -277,6 +278,10 @@ def _db_flush_write(agent, batch_rows: List[Dict[str, Any]], batch_msgs: List[Di turn_lease_ttl_seconds=getattr(agent, "_active_session_turn_lease_ttl_seconds", 300.0) or 300.0, ) sync_flushed_message_markers(batch_msgs, batch_rows) + for msg in batch_msgs: + # One write only: left set, the flag keeps exempting the carried row from the history skip, + # so every later in-place repair of it (a merge with the next prompt) is written again. + msg.pop(_PERSIST_AFTER_ADMISSION_INTERRUPT, None) if _newest_checkpoint_carrier(batch_msgs, "codex_reasoning_items") >= 0: # The insert already rewrote the older rows (SessionDB._drop_shadowed_checkpoint_rows); mirror it on # the live transcript so forks/compaction built from memory carry one checkpoint too. Markers stay: diff --git a/agent/turn_final_response.py b/agent/turn_final_response.py index 6ebc40503d33d..18933e6baa770 100644 --- a/agent/turn_final_response.py +++ b/agent/turn_final_response.py @@ -28,7 +28,7 @@ # Ephemeral retry scaffolding rows popped before the final answer becomes durable. _EPHEMERAL_SCAFFOLDING_FLAGS = ( "_thinking_prefill", "_empty_recovery_synthetic", "_empty_terminal_sentinel", - "_dropped_toolcall_nudge", + "_dropped_toolcall_nudge", "_length_continuation_synthetic", ) diff --git a/agent/turn_truncation.py b/agent/turn_truncation.py index dbcdb42233d35..d8d3dc3318176 100644 --- a/agent/turn_truncation.py +++ b/agent/turn_truncation.py @@ -330,10 +330,16 @@ def _continue_text(st: _Trunc, _retry: TurnRetryState, assistant_message: Any) - agent._vprint(f"{agent.log_prefix}↻ Stream interrupted — requesting continuation ({n}/4)...", diagnostic=True) else: agent._vprint(f"{agent.log_prefix}↻ Requesting continuation ({n}/4)...", diagnostic=True) - append_message(messages, { + nudge = { "role": "user", "content": _get_continuation_prompt(st.is_stub, _dropped_tools), "_length_continuation_nudge": True, - }) + } + if messages and isinstance(messages[-1], dict) and messages[-1].get("role") == "user": + # No visible fragment: the nudge would follow this turn's already-written user row, and the + # pre-request repair would merge it INTO that row, saved and replayed as the user's own + # words. Keep it request-only; the wire copy merges it for alternation. + nudge["_length_continuation_synthetic"] = True + append_message(messages, nudge) agent._session_messages = messages _retry.restart_with_length_continuation = True return st.done("break") diff --git a/tests/agent/test_session_persistence_repaired_user_row.py b/tests/agent/test_session_persistence_repaired_user_row.py new file mode 100644 index 0000000000000..5b96472d19cd5 --- /dev/null +++ b/tests/agent/test_session_persistence_repaired_user_row.py @@ -0,0 +1,119 @@ +"""A user row repaired in place after it was written must not reach state.db a second time. + +The pre-request alternation repair merges a following user row into an already-written one and +pops its persist marker; the flush only appends, so the merged row used to land again, and with the +length-continuation nudge Hermes's internal instruction was saved as the user's own words and replayed +on every later turn. Both flows drive the real turn loop against a real SessionDB (LLM faked). +""" + +from __future__ import annotations + +import threading +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from hermes_constants import FINISH_REASON_LENGTH +from hermes_state import SessionDB + + +def _response(content, finish_reason="stop", reasoning_content=None): + message = SimpleNamespace(content=content, tool_calls=None) + if reasoning_content is not None: + message.reasoning_content = reasoning_content + return SimpleNamespace( + id="chatcmpl-test", model="test/model", usage=None, + choices=[SimpleNamespace(index=0, message=message, finish_reason=finish_reason)], + ) + + +@pytest.fixture() +def session_agent(tmp_path, monkeypatch): + # No request may leave the machine: any real HTTP client dies on a dead local proxy. + for key in ("HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "https_proxy", "http_proxy", "all_proxy"): + monkeypatch.setenv(key, "http://127.0.0.1:9") + monkeypatch.setenv("NO_PROXY", "") + monkeypatch.setenv("no_proxy", "") + monkeypatch.setattr("agent.turn_context._maybe_title_session_at_turn_start", lambda *a, **k: None) + monkeypatch.setattr("agent.turn_context.start_deferred_title_upgrade", lambda *a, **k: None) + + from run_agent import AIAgent + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "repaired-user-row" + with ( + patch("model_tools.get_tool_definitions", return_value=[]), + patch("model_tools.check_toolset_requirements", return_value={}), + patch("agent.process_bootstrap.OpenAI"), + ): + agent = AIAgent( + api_key="test-key-1234567890", base_url="https://openrouter.ai/api/v1", model="test/model", + quiet_mode=True, skip_context_files=True, skip_memory=True, + session_db=db, session_id=session_id, + ) + agent.client = MagicMock() + agent._cached_system_prompt = "You are helpful." + agent._use_prompt_caching = False + agent.compression_enabled = False + agent.save_trajectories = False + try: + yield agent, db, session_id + finally: + db.close() + + +def _answer_with(agent, *responses): + agent.client.chat.completions.create.side_effect = list(responses) + + +def _saved_user_texts(db, session_id): + return [m["content"] for m in db.get_messages_as_conversation(session_id) if m["role"] == "user"] + + +def test_thinking_only_length_retry_saves_the_prompt_once_and_never_the_nudge(session_agent): + agent, db, session_id = session_agent + _answer_with(agent, _response("Answer one.")) + first = agent.run_conversation("first question") + + # A reasoning model spends the whole output cap on thinking, then answers on the retry. + _answer_with( + agent, + _response("", finish_reason=FINISH_REASON_LENGTH, reasoning_content="thinking " * 50), + _response("Here is the full report."), + ) + second = agent.run_conversation("write me a long report", conversation_history=first["messages"]) + + assert second["completed"] is True + prompts = ["first question", "write me a long report"] + assert _saved_user_texts(db, session_id) == prompts + # A CLI host replays result["messages"] as the next turn's history. + assert [m["content"] for m in second["messages"] if m["role"] == "user"] == prompts + + +def test_prompt_carried_past_an_interrupted_lease_wait_is_saved_once(session_agent): + agent, db, session_id = session_agent + _answer_with(agent, _response("Answer one.")) + first = agent.run_conversation("first question") + + # Another Hermes process holds the session; the user sends B while A waits for it. + other_holder = "pid=0:turn=other:platform=desktop" + assert db.acquire_session_turn_lease(session_id, other_holder, ttl_seconds=300.0, wait_seconds=0.0) + waiting = threading.Event() + agent.status_callback = lambda *_: waiting.set() + outcome = {} + waiter = threading.Thread(target=lambda: outcome.setdefault( + "result", agent.run_conversation("message A", conversation_history=first["messages"]))) + waiter.start() + assert waiting.wait(30) + agent.interrupt("message B") + waiter.join(30) + assert outcome["result"]["interrupted"] is True + agent.status_callback = None + db.release_session_turn_lease(session_id, other_holder) + + _answer_with(agent, _response("Answer to A and B.")) + follow_up = agent.run_conversation("message B", conversation_history=outcome["result"]["messages"]) + + assert follow_up["completed"] is True + assert _saved_user_texts(db, session_id) == ["first question", "message A", "message B"]