fix: preserve queued prompt boundaries end to end

agent/agent_runtime_helpers.py

@@ -8,7 +8,7 @@
 Methods covered:
 * ``convert_to_trajectory_format`` — internal -> trajectory-file format
 * ``sanitize_tool_call_arguments`` — repair corrupted JSON in tool_calls
-* ``repair_message_sequence`` — enforce alternation invariants
+* ``repair_message_sequence`` — repair canonical assistant/tool structure
 * ``strip_think_blocks`` — remove inline reasoning from stored content
 * ``recover_with_credential_pool`` — rotate pool entries on 429
 * ``try_recover_primary_transport`` — re-create OpenAI client after rate-limit
@@ -560,19 +560,12 @@ def note_turn_persisted(agent):
 
 
 def repair_message_sequence(agent, messages: List[Dict]) -> int:
-    """Collapse malformed role-alternation left in the live history.
+    """Repair malformed assistant/tool structure in canonical history.
 
-    Providers (OpenAI, OpenRouter, Anthropic) expect strict alternation:
-    after the system message, user/tool alternates with assistant, with
-    no two consecutive user messages and no tool-result that doesn't
-    follow an assistant-with-tool_calls. Violations cause silent empty
-    responses on most providers, which triggers the empty-retry loop.
-
-    This runs right before the API call as a defensive belt — by the
-    time it fires, the scaffolding strip should already have prevented
-    most shapes, but external callers (gateway multi-queue replay,
-    session resume, cron, explicit conversation_history passed in by
-    host code) can feed in already-broken histories.
+    This canonical-history repair deliberately preserves adjacent ``user``
+    messages as distinct source turns. Provider role alternation is repaired
+    later on the per-request ``api_messages`` copy by
+    :func:`drop_thinking_only_and_merge_users`.
 
     Repairs applied:
       0. Consecutive ``assistant`` messages with no intervening
@@ -589,8 +582,6 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
          resumed histories. Refs #29148, #49147.
       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.
 
     Deliberately does NOT rewind orphan ``assistant(tool_calls)+tool``
     pairs that precede a user message — that pattern IS valid when the
@@ -759,54 +750,14 @@ def _is_verification_candidate(m: Dict) -> bool:
                 matched_tool_groups = set()
             filtered.append(msg)
 
-    # Pass 2: merge consecutive user messages. Preserves all user input
-    # so nothing the user typed is lost.
-    merged: List[Dict] = []
-    for msg in filtered:
-        if (
-            merged
-            and isinstance(msg, dict)
-            and msg.get("role") == "user"
-            and isinstance(merged[-1], dict)
-            and merged[-1].get("role") == "user"
-        ):
-            prev = merged[-1]
-            # A summary carrier followed by a new user row is a deliberate
-            # durable shape after retry/rewind.  Do not absorb the fresh ask
-            # into the already-persisted carrier: mutating that dict can make
-            # the only in-memory copy diverge from its durable row.  Provider
-            # sanitizers merge copies later when strict alternation requires
-            # it, without rewriting either durable message.
-            from agent.context_compressor import split_user_originated_turn
-
-            handoff, _ = split_user_originated_turn(prev)
-            if handoff is not None:
-                merged.append(msg)
-                continue
-
-            prev_content = prev.get("content", "")
-            new_content = msg.get("content", "")
-            # Only merge plain-text content; leave multimodal (list)
-            # 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)
-                )
-                # Merged content invalidates the api_content sidecar (exact
-                # bytes previously sent for the pre-merge message) — drop it
-                # so replay can't substitute stale bytes.
-                drop_stale_api_content(prev)
-                repairs += 1
-                continue
-        merged.append(msg)
+    # Adjacent user messages are canonical source boundaries, not malformed
+    # history. Keep them distinct here; the per-request wire copy is merged
+    # later by ``drop_thinking_only_and_merge_users`` for strict providers.
 
     if repairs > 0:
         # Rewrite in place so downstream paths (persistence, return
         # value, session DB flush) see the repaired sequence.
-        messages[:] = merged
+        messages[:] = filtered
 
     return repairs
 
@@ -815,11 +766,11 @@ def repair_message_sequence_with_cursor(agent, messages: List[Dict]) -> int:
     """Run :func:`repair_message_sequence` and keep the SessionDB flush
     cursor consistent with the compacted list (#44837).
 
-    ``repair_message_sequence`` merges/drops messages in place, shrinking
-    the list. ``_last_flushed_db_idx`` (the DB-write cursor) indexes into
-    that list, so after compaction it can point past the new end — the
-    turn-end flush would then skip the assistant/tool chain entirely — or
-    past unflushed messages shifted to lower indexes.
+    ``repair_message_sequence`` merges assistant messages and drops orphaned
+    tool messages in place, shrinking the list. ``_last_flushed_db_idx`` (the
+    DB-write cursor) indexes into that list, so after compaction it can point
+    past the new end — the turn-end flush would then skip the assistant/tool
+    chain entirely — or past unflushed messages shifted to lower indexes.
 
     Repair preserves object identity for surviving messages, so counting
     the survivors from the previously-flushed prefix gives the exact new
@@ -1413,13 +1364,12 @@ def drop_thinking_only_and_merge_users(
     *,
     drop_codex_reasoning_items: bool = True,
 ) -> List[Dict[str, Any]]:
-    """Drop thinking-only assistant turns; merge any adjacent user messages left behind.
+    """Drop thinking-only turns and merge adjacent users on the wire copy.
 
     Runs on the per-call ``api_messages`` copy only. The stored
-    conversation history (``agent.messages``) is never mutated, so the
-    user still sees the thinking block in the CLI/gateway transcript and
-    session persistence keeps the full trace. Only the wire copy sent to
-    the provider is cleaned.
+    conversation history (``agent.messages``) is never mutated, so canonical
+    user source boundaries and thinking blocks remain available to the UI and
+    session persistence. Only the wire copy sent to the provider is cleaned.
 
     Why drop-and-merge rather than inject stub text:
     - Fabricating ``"."`` / ``"(continued)"`` text lies in the history
@@ -1441,8 +1391,18 @@ def drop_thinking_only_and_merge_users(
         )
     ]
     dropped = len(messages) - len(kept)
+    has_adjacent_users = any(
+        previous.get("role") == "user" and current.get("role") == "user"
+        for previous, current in zip(kept, kept[1:])
+    )
+    if dropped == 0 and not has_adjacent_users:
+        return messages
 
-    # Pass 2: merge any newly-adjacent user messages.
+    # Pass 2: merge adjacent source turns for provider compatibility while
+    # retaining an explicit semantic boundary in the transient wire content.
+    # This marker never reaches canonical history or SessionDB.
+    boundary_text = "[Next user message]"
+    boundary_block = {"type": "text", "text": boundary_text}
     merged: List[Dict[str, Any]] = []
     merges = 0
     for m in kept:
@@ -1463,21 +1423,25 @@ def drop_thinking_only_and_merge_users(
             # purposes. If either side is a list (multimodal), append as a
             # separate block rather than collapsing.
             if isinstance(prev_content, str) and isinstance(cur_content, str):
-                sep = "\n\n" if prev_content and cur_content else ""
-                prev_copy["content"] = prev_content + sep + cur_content
+                prev_copy["content"] = "\n\n".join(
+                    part
+                    for part in (prev_content, boundary_text, cur_content)
+                    if part
+                )
             elif isinstance(prev_content, list) and isinstance(cur_content, list):
-                prev_copy["content"] = list(prev_content) + list(cur_content)
+                prev_copy["content"] = (
+                    list(prev_content) + [dict(boundary_block)] + list(cur_content)
+                )
             elif isinstance(prev_content, list) and isinstance(cur_content, str):
+                new_blocks = list(prev_content) + [dict(boundary_block)]
                 if cur_content:
-                    prev_copy["content"] = list(prev_content) + [
-                        {"type": "text", "text": cur_content}
-                    ]
-                else:
-                    prev_copy["content"] = list(prev_content)
+                    new_blocks.append({"type": "text", "text": cur_content})
+                prev_copy["content"] = new_blocks
             elif isinstance(prev_content, str) and isinstance(cur_content, list):
                 new_blocks: List[Dict[str, Any]] = []
                 if prev_content:
                     new_blocks.append({"type": "text", "text": prev_content})
+                new_blocks.append(dict(boundary_block))
                 new_blocks.extend(cur_content)
                 prev_copy["content"] = new_blocks
             else:
@@ -3713,6 +3677,21 @@ def repair_empty_non_final_messages(
         return repaired
     return messages
 
+_API_SOURCE_METADATA_KEYS = (
+    "timestamp",
+    "message_id",
+    "platform_message_id",
+    "_source_message_id",
+)
+
+
+def copy_message_for_api(message: Dict[str, Any]) -> Dict[str, Any]:
+    """Copy one canonical message without transcript-only source metadata."""
+    api_message = message.copy()
+    for key in _API_SOURCE_METADATA_KEYS:
+        api_message.pop(key, None)
+    return api_message
+
 
 def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     """Fix orphaned tool_call / tool_result pairs before every LLM call.
@@ -4644,6 +4623,7 @@ def force_close_tcp_sockets(client: Any) -> int:
     "invoke_tool",
     "repair_tool_call",
     "sanitize_api_messages",
+    "copy_message_for_api",
     "looks_like_codex_intermediate_ack",
     "copy_reasoning_content_for_api",
     "cleanup_dead_connections",

agent/chat_completion_helpers.py

@@ -2924,26 +2924,25 @@ def _managed_summary_call(request, callback, *, retry_count: int):
     append_message(messages, {"role": "user", "content": summary_request})
 
     try:
+        from agent.agent_runtime_helpers import copy_message_for_api
+
         # Build API messages, stripping internal-only fields
-        # (finish_reason, reasoning) that strict APIs like Mistral reject with 422
+        # (finish_reason, reasoning) that strict APIs like Mistral reject with 422.
         _needs_sanitize = agent._should_sanitize_tool_calls()
         api_messages = []
         for msg in messages:
-            api_msg = msg.copy()
+            api_msg = copy_message_for_api(msg)
             agent._copy_reasoning_content_for_api(msg, api_msg)
             for internal_field in ("reasoning", "finish_reason"):
                 api_msg.pop(internal_field, None)
             # Strict OpenAI-compatible gateways (Fireworks-backed OpenCode Go,
             # Mistral, Moonshot/Kimi) reject any message key outside the Chat
             # Completions schema. The main loop drops these via
             # ChatCompletionsTransport.convert_messages(), but the summary path
-            # hand-builds messages and calls chat.completions.create() directly,
-            # bypassing the transport — so mirror that sanitization here:
-            # tool_name (SQLite FTS bookkeeping), the codex_* reasoning carriers,
-            # timestamp (preserved on gateway user replay entries for the
-            # stale-confirmation expiry check — #47868 rejection class),
-            # and every Hermes-internal underscore-prefixed scaffolding key.
-            for schema_foreign in ("tool_name", "codex_reasoning_items", "codex_message_items", "timestamp"):
+            # calls chat.completions.create() directly. Reuse the shared source
+            # metadata copy policy above, then strip remaining schema-foreign
+            # bookkeeping and every Hermes-internal underscore-prefixed key.
+            for schema_foreign in ("tool_name", "codex_reasoning_items", "codex_message_items"):
                 api_msg.pop(schema_foreign, None)
             # api_content (the persist-what-you-send sidecar) carries the
             # exact bytes every main-loop call sent for this message —

agent/conversation_loop.py

@@ -1832,6 +1832,7 @@ def run_conversation(
     persist_user_display_kind: Optional[str] = None,
     persist_user_display_metadata: Optional[Dict[str, Any]] = None,
     moa_config: Optional[dict[str, Any]] = None,
+    persist_user_message_id: Optional[str] = None,
 ) -> Dict[str, Any]:
     """
     Run a complete conversation with tool calling until completion.
@@ -1857,6 +1858,9 @@ def run_conversation(
         persist_user_display_metadata: Optional payload for that event
             (e.g. a delegation's task count).
                 or queuing follow-up prefetch work.
+        moa_config: Optional mixture-of-agents configuration for this turn.
+        persist_user_message_id: Optional stable source identity to retain on
+            the canonical user message and persisted row.
 
     Returns:
         Dict: Complete conversation result with final response and message history
@@ -1908,6 +1912,7 @@ def run_conversation(
         persist_user_timestamp,
         persist_user_display_kind=persist_user_display_kind,
         persist_user_display_metadata=persist_user_display_metadata,
+        persist_user_message_id=persist_user_message_id,
         restore_or_build_system_prompt=_restore_or_build_system_prompt,
         install_safe_stdio=_install_safe_stdio,
         sanitize_surrogates=_sanitize_surrogates,
@@ -2216,13 +2221,15 @@ def run_conversation(
             )
         ]
 
-        # Defensive: repair malformed role-alternation before API call.
-        # Catches cases where the history got wedged into a
-        # ``tool → user`` or ``user → user`` tail (e.g. after empty-
-        # response scaffolding was stripped and a new user message
-        # landed after an orphan tool result). Most providers return
-        # empty content on malformed sequences, which would otherwise
-        # retrigger the empty-retry loop indefinitely.
+        # Defensive: repair malformed assistant/tool structure in canonical
+        # history before the API call. This collapses split assistant turns and
+        # drops orphaned tool results without collapsing adjacent user source
+        # messages. Strict-provider user-role alternation is repaired later by
+        # ``_drop_thinking_only_and_merge_users`` on the per-request copy.
+        # ``repair_message_sequence_with_cursor`` also recomputes the SessionDB
+        # flush cursor (_last_flushed_db_idx) when canonical repair compacts the
+        # list, so turn-end flushing cannot skip shifted assistant/tool rows
+        # (#44837).
         # repair_message_sequence_with_cursor also recomputes the SessionDB
         # flush cursor (_last_flushed_db_idx) when repair compacts the list,
         # so the turn-end flush doesn't skip the assistant/tool chain (#44837).
@@ -2284,6 +2291,17 @@ def run_conversation(
             # Bookkeeping, never a provider field — only the chat-completions
             # transport strips underscore keys, so drop it centrally here.
             api_msg.pop("_row_id", None)
+            # Source ordering/deduplication metadata belongs to the canonical
+            # transcript and SessionDB, never to a provider request. Strip it
+            # at the common API-copy boundary so native Anthropic/Codex paths
+            # do not depend on transport-specific unknown-field filtering.
+            for metadata_key in (
+                "timestamp",
+                "message_id",
+                "platform_message_id",
+                "_source_message_id",
+            ):
+                api_msg.pop(metadata_key, None)
 
             # Inject ephemeral context into the current turn's user message.
             # Sources: memory manager prefetch + plugin pre_llm_call hooks

agent/transports/chat_completions.py

@@ -285,6 +285,9 @@ def convert_messages(
           ``Extra inputs are not permitted, field: 'messages[N].tool_name'``.
           Permissive providers (OpenRouter, MiniMax) silently ignore the
           field, which masked the bug for months.
+        - Transcript metadata: ``timestamp``, ``message_id``, and
+          ``platform_message_id`` are persisted for ordering/deduplication but
+          are not Chat Completions message fields.
         - Hermes-internal scaffolding markers — any top-level message key
           starting with ``_`` (e.g. ``_empty_recovery_synthetic``,
           ``_empty_terminal_sentinel``, ``_thinking_prefill``). These are
@@ -310,6 +313,8 @@ def convert_messages(
                 or "effect_disposition" in msg
                 or "timestamp" in msg  # #47868 — strict providers reject this
                 or "api_content" in msg  # persist-what-you-send sidecar
+                or "message_id" in msg
+                or "platform_message_id" in msg
             ):
                 needs_sanitize = True
                 break
@@ -381,6 +386,8 @@ def mutable_msg() -> dict[str, Any]:
                 or "effect_disposition" in msg
                 or "timestamp" in msg  # #47868 — leak into strict providers
                 or "api_content" in msg  # persist-what-you-send sidecar
+                or "message_id" in msg
+                or "platform_message_id" in msg
             ):
                 out_msg = mutable_msg()
                 out_msg.pop("codex_reasoning_items", None)
@@ -389,6 +396,8 @@ def mutable_msg() -> dict[str, Any]:
                 out_msg.pop("effect_disposition", None)
                 out_msg.pop("timestamp", None)  # #47868 — leak into strict providers
                 out_msg.pop("api_content", None)  # persist-what-you-send sidecar
+                out_msg.pop("message_id", None)
+                out_msg.pop("platform_message_id", None)
 
 
             # Drop all Hermes-internal scaffolding markers (``_``-prefixed).

agent/turn_context.py

@@ -502,6 +502,7 @@ def build_turn_context(
     stream_callback,
     persist_user_message: Optional[Any],
     persist_user_timestamp: Optional[float] = None,
+    persist_user_message_id: Optional[str] = None,
     *,
     persist_user_display_kind: Optional[str] = None,
     persist_user_display_metadata: Optional[Dict[str, Any]] = None,
@@ -721,6 +722,8 @@ def build_turn_context(
     # CLI input is stamped when staged. Gateway input may carry the platform
     # event time. Preserve either value and cover any legacy unstamped handoff.
     stamp_message_timestamp(user_msg, timestamp=persist_user_timestamp)
+    if persist_user_message_id is not None:
+        user_msg["_source_message_id"] = persist_user_message_id
 
     # Hydrate todo store from conversation history.
     if conversation_history and not agent._todo_store.has_items():
@@ -784,6 +787,7 @@ def build_turn_context(
             should_review_memory = True
             agent._turns_since_memory = 0
 
+
     # 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.