fix(agent): preserve later user message boundaries

agent/agent_runtime_helpers.py

@@ -37,7 +37,13 @@
 from agent.trajectory import convert_scratchpad_to_think
 from agent.credential_pool import STATUS_EXHAUSTED
 from agent.error_classifier import FailoverReason
-from utils import base_url_host_matches, base_url_hostname, env_var_enabled, atomic_json_write
+from utils import (
+    atomic_json_write,
+    base_url_host_matches,
+    base_url_hostname,
+    env_var_enabled,
+    merge_later_user_text,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -362,8 +368,9 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
     Repairs applied:
       1. Stray ``tool`` messages whose ``tool_call_id`` doesn't match
          any preceding assistant tool_call — dropped.
-      2. Consecutive ``user`` messages — merged with newline separator
-         so no user input is lost.
+      2. Consecutive ``user`` messages — merged with an explicit
+         later-message boundary so no user input is lost and the model
+         can still see where one user thought ended and the next began.
 
     Deliberately does NOT rewind orphan ``assistant(tool_calls)+tool``
     pairs that precede a user message — that pattern IS valid when the
@@ -428,11 +435,7 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
             # content alone — collapsing image/audio blocks risks
             # mangling the attachment structure.
             if isinstance(prev_content, str) and isinstance(new_content, str):
-                prev["content"] = (
-                    (prev_content + "\n\n" + new_content)
-                    if prev_content and new_content
-                    else (prev_content or new_content)
-                )
+                prev["content"] = merge_later_user_text(prev_content, new_content)
                 repairs += 1
                 continue
         merged.append(msg)

agent/anthropic_adapter.py

@@ -23,7 +23,11 @@
 
 from hermes_constants import get_hermes_home
 from typing import Any, Dict, List, Optional, Tuple
-from utils import base_url_host_matches, normalize_proxy_env_vars
+from utils import (
+    base_url_host_matches,
+    merge_later_user_text,
+    normalize_proxy_env_vars,
+)
 
 # NOTE: `import anthropic` is deliberately NOT at module top — the SDK pulls
 # ~220 ms of imports (anthropic.types, anthropic.lib.tools._beta_runner, etc.)
@@ -2008,7 +2012,7 @@ def _merge_consecutive_roles(result: List[Dict[str, Any]]) -> List[Dict[str, Any
                 prev_content = fixed[-1]["content"]
                 curr_content = m["content"]
                 if isinstance(prev_content, str) and isinstance(curr_content, str):
-                    fixed[-1]["content"] = prev_content + "\n" + curr_content
+                    fixed[-1]["content"] = merge_later_user_text(prev_content, curr_content)
                 elif isinstance(prev_content, list) and isinstance(curr_content, list):
                     fixed[-1]["content"] = prev_content + curr_content
                 else:

gateway/platforms/base.py

@@ -20,7 +20,7 @@
 from abc import ABC, abstractmethod
 from urllib.parse import urlsplit
 
-from utils import normalize_proxy_url
+from utils import merge_later_user_text, normalize_proxy_url
 
 logger = logging.getLogger(__name__)
 
@@ -1617,10 +1617,11 @@ def merge_pending_message_event(
     events. Merge those into the existing queued event so the next turn sees
     the whole burst.
 
-    When ``merge_text`` is enabled, rapid follow-up TEXT events are appended
+    When ``merge_text`` is enabled, rapid follow-up TEXT events are merged
     instead of replacing the pending turn. This is used for Telegram bursty
     follow-ups so a multi-part user thought is not silently truncated to only
-    the last queued fragment.
+    the last queued fragment, while still preserving a visible boundary between
+    later messages.
     """
     existing = pending_messages.get(session_key)
     if existing:
@@ -1660,7 +1661,7 @@ def merge_pending_message_event(
             and event.message_type == MessageType.TEXT
         ):
             if event.text:
-                existing.text = f"{existing.text}\n{event.text}" if existing.text else event.text
+                existing.text = merge_later_user_text(existing.text or "", event.text)
             return
 
     pending_messages[session_key] = event
@@ -3598,11 +3599,7 @@ async def _queue_text_debounce(self, session_key: str, event: MessageEvent) -> N
             store[session_key] = state
         else:
             if event.text:
-                state.event.text = (
-                    f"{state.event.text}\n{event.text}"
-                    if state.event.text
-                    else event.text
-                )
+                state.event.text = merge_later_user_text(state.event.text or "", event.text)
             latest_message_id = getattr(event, "message_id", None)
             latest_anchor = latest_message_id or getattr(event, "reply_to_message_id", None)
             if latest_message_id is not None:

tests/agent/test_anthropic_adapter.py

@@ -27,6 +27,7 @@
     run_oauth_setup_token,
 )
 from agent.transports import get_transport
+from utils import merge_later_user_text
 
 
 # ---------------------------------------------------------------------------
@@ -1638,8 +1639,7 @@ def test_merges_consecutive_user_messages(self):
         _, result = convert_messages_to_anthropic(messages)
         assert len(result) == 1
         assert result[0]["role"] == "user"
-        assert "Hello" in result[0]["content"]
-        assert "World" in result[0]["content"]
+        assert result[0]["content"] == merge_later_user_text("Hello", "World")
 
     def test_preserves_proper_alternation(self):
         messages = [

tests/gateway/test_active_session_text_merge.py

@@ -37,6 +37,7 @@
     SendResult,
 )
 from gateway.session import SessionSource, build_session_key
+from utils import merge_later_user_text
 
 
 def _make_event(
@@ -128,7 +129,7 @@ async def test_rapid_text_followups_accumulate_instead_of_replacing():
     await adapter.handle_message(_make_event("part three"))
 
     pending = adapter._pending_messages[session_key]
-    assert pending.text == "part two\npart three"
+    assert pending.text == merge_later_user_text("part two", "part three")
     assert not adapter._active_sessions[session_key].is_set()
 
 
@@ -147,12 +148,18 @@ async def test_debounce_buffers_rapid_text_then_flushes_to_pending():
     assert session_key not in adapter._pending_messages
 
     await adapter.handle_message(_make_event("part three"))
-    assert _debounced_event(adapter, session_key).text == "part two\npart three"
+    assert _debounced_event(adapter, session_key).text == merge_later_user_text(
+        "part two",
+        "part three",
+    )
 
     await asyncio.sleep(0.15)
 
     assert session_key not in adapter._text_debounce
-    assert adapter._pending_messages[session_key].text == "part two\npart three"
+    assert adapter._pending_messages[session_key].text == merge_later_user_text(
+        "part two",
+        "part three",
+    )
 
 
 @pytest.mark.asyncio
@@ -184,7 +191,10 @@ async def test_debounce_resets_timer_on_new_arrival():
 
     await asyncio.sleep(0.2)
     assert session_key not in adapter._text_debounce
-    assert adapter._pending_messages[session_key].text == "one\ntwo\nthree"
+    assert adapter._pending_messages[session_key].text == merge_later_user_text(
+        merge_later_user_text("one", "two"),
+        "three",
+    )
 
 
 @pytest.mark.asyncio

tests/run_agent/test_message_sequence_repair.py

@@ -10,6 +10,7 @@
 """
 
 from run_agent import AIAgent
+from utils import merge_later_user_text
 
 
 def _bare_agent():
@@ -90,7 +91,7 @@ def test_repair_merges_consecutive_user_messages():
     assert repairs == 1
     assert len(messages) == 1
     assert messages[0]["role"] == "user"
-    assert messages[0]["content"] == "first\n\nsecond"
+    assert messages[0]["content"] == merge_later_user_text("first", "second")
 
 
 def test_repair_preserves_user_content_when_one_side_empty():

utils.py

@@ -17,6 +17,7 @@
 
 
 TRUTHY_STRINGS = frozenset({"1", "true", "yes", "on"})
+LATER_USER_MESSAGE_SEPARATOR = "\n\n[Later user message]\n\n"
 
 
 def is_truthy_value(value: Any, default: bool = False) -> bool:
@@ -35,6 +36,15 @@ def env_var_enabled(name: str, default: str = "") -> bool:
     return is_truthy_value(os.getenv(name, default), default=False)
 
 
+def merge_later_user_text(existing_text: str, new_text: str) -> str:
+    """Join forced same-turn user text with an explicit later-message boundary."""
+    if not existing_text:
+        return new_text
+    if not new_text:
+        return existing_text
+    return f"{existing_text}{LATER_USER_MESSAGE_SEPARATOR}{new_text}"
+
+
 def _preserve_file_mode(path: Path) -> "int | None":
     """Capture the permission bits of *path* if it exists, else ``None``."""
     try: