fix(agent): stop re-appending repaired rows and saving the length nudge as user text

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

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:

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",
 )
 
 

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

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