fix(api_server): robust Responses turn-start detection

agent/agent_runtime_helpers.py

@@ -519,26 +519,30 @@ def _prune_unanswered_tool_calls(messages: List[Dict]) -> Tuple[List[Dict], int]
     return pruned, repairs
 
 
+def _users_can_merge(prev: Any, msg: Any) -> bool:
+    """Whether ``msg`` can be folded into the preceding user row."""
+    from agent.context_compressor import split_user_originated_turn
+
+    return bool(
+        isinstance(prev, dict)
+        and prev.get("role") == "user"
+        and isinstance(msg, dict)
+        and msg.get("role") == "user"
+        and split_user_originated_turn(prev)[0] is None
+        and prev.get("display_kind") != STEER_DISPLAY_KIND
+        and isinstance(prev.get("content", ""), str)
+        and isinstance(msg.get("content", ""), str)
+    )
+
+
 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 split_user_originated_turn
 
     repairs = 0
     merged: List[Dict] = []
     for msg in messages:
         prev = merged[-1] if merged and isinstance(merged[-1], dict) else None
-        if (
-            prev is not None and prev.get("role") == "user"
-            and isinstance(msg, dict) and msg.get("role") == "user"
-            # 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
-            # A /steer row that ended the previous run is already persisted; merging the next
-            # prompt into it would rewrite it in place and re-break replay parity.
-            and prev.get("display_kind") != STEER_DISPLAY_KIND
-            # Only merge plain-text content; leave multimodal (list) content alone.
-            and isinstance(prev.get("content", ""), str) and isinstance(msg.get("content", ""), str)
-        ):
+        if prev is not None and _users_can_merge(prev, msg):
             prev_content, new_content = prev.get("content", ""), msg.get("content", "")
             prev["content"] = (
                 (prev_content + "\n\n" + new_content) if prev_content and new_content else (prev_content or new_content)
@@ -568,16 +572,57 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
     if not messages:
         return 0
     repairs = 0
+    current_turn_message = None
+    current_turn_owner = None
+    current_turn_idx = getattr(agent, "_persist_user_message_idx", None)
+    if isinstance(current_turn_idx, int) and 0 <= current_turn_idx < len(messages):
+        current_turn_message = messages[current_turn_idx]
     current = messages
     for repair_pass in _SEQUENCE_REPAIR_PASSES:
+        if repair_pass is _merge_consecutive_users and current_turn_message is not None:
+            current_idx = next(
+                (i for i, message in enumerate(current) if message is current_turn_message),
+                -1,
+            )
+            current_turn_owner = current_turn_message
+            while current_idx > 0 and _users_can_merge(
+                current[current_idx - 1], current[current_idx]
+            ):
+                current_idx -= 1
+                current_turn_owner = current[current_idx]
         current, made = repair_pass(current)
         repairs += made
     if repairs > 0:
         # Rewrite in place so persistence/return value/DB flush see the repaired sequence.
         messages[:] = current
+    if current_turn_message is not None:
+        owner = current_turn_message
+        if not any(message is current_turn_message for message in messages):
+            owner = current_turn_owner
+        agent._persist_user_message_idx = next(
+            (index for index, message in enumerate(messages) if message is owner),
+            -1,
+        )
     return repairs
 
 
+def preserve_current_turn_user_idx(
+    agent, previous_messages: List[Dict], messages: List[Dict]
+) -> None:
+    """Carry typed current-turn ownership across an identity-preserving rebuild."""
+    current_turn_idx = getattr(agent, "_persist_user_message_idx", None)
+    if not (
+        isinstance(current_turn_idx, int)
+        and 0 <= current_turn_idx < len(previous_messages)
+    ):
+        return
+    current_turn_message = previous_messages[current_turn_idx]
+    agent._persist_user_message_idx = next(
+        (index for index, message in enumerate(messages) if message is current_turn_message),
+        -1,
+    )
+
+
 def repair_message_sequence_with_cursor(agent, messages: List[Dict]) -> int:
     """Run :func:`repair_message_sequence` and keep ``_last_flushed_db_idx`` consistent. Repair
     shrinks the list in place; counting identity-preserved survivors of the flushed prefix gives
@@ -3252,6 +3297,7 @@ def force_close_tcp_sockets(client: Any) -> int:
 
 __all__ = [
     "convert_to_trajectory_format", "sanitize_tool_call_arguments", "repair_message_sequence",
+    "preserve_current_turn_user_idx",
     "strip_think_blocks", "recover_with_credential_pool", "try_recover_primary_transport",
     "drop_thinking_only_and_merge_users", "restore_primary_runtime", "extract_reasoning",
     "dump_api_request_debug", "prompt_caching_disabled_from_config", "blank_cache_policy_stub",

agent/turn_context.py

@@ -206,12 +206,17 @@ def _maybe_title_session_at_turn_start(agent: Any, messages: List[Any]) -> None:
         logger.debug("Turn-start auto-title dispatch failed", exc_info=True)
 
 
-def reanchor_current_turn_user_idx(messages: List[Any], user_message: Any) -> int:
+def reanchor_current_turn_user_idx(
+    messages: List[Any],
+    user_message: Any,
+    current_turn_user_idx: Optional[int] = None,
+) -> int:
     """Locate this turn's user message after compaction rebuilt ``messages``.
 
     Prefers the LAST user message whose content exactly matches this turn's text, else
     the last user-originated turn; compaction handoffs are never the fallback.
-    Returns -1 when there is no user-originated message.
+    Returns -1 when there is no user-originated message. A valid typed index is
+    authoritative when repair merged the current ask into a rewritten user row.
 
     Compression replaces list entries with fresh copies (and may append a todo-snapshot user message or a
     restored user turn AFTER the surviving copy of the current turn's message), so a pre-compression index
@@ -224,6 +229,14 @@ def reanchor_current_turn_user_idx(messages: List[Any], user_message: Any) -> in
     """
     from agent.context_compressor import user_originated_turn_view
 
+    if (
+        isinstance(current_turn_user_idx, int)
+        and 0 <= current_turn_user_idx < len(messages)
+        and isinstance(messages[current_turn_user_idx], dict)
+        and messages[current_turn_user_idx].get("role") == "user"
+    ):
+        return current_turn_user_idx
+
     fallback = -1
     for i in range(len(messages) - 1, -1, -1):
         msg = messages[i]

agent/turn_iteration_prep.py

@@ -174,32 +174,40 @@ def _is_scaffold_ghost(msg: Dict[str, Any]) -> bool:
             )
         )
 
+    previous_messages = messages
     messages = [msg for msg in messages if not _is_scaffold_ghost(msg)]
+    from agent.agent_runtime_helpers import preserve_current_turn_user_idx
+
+    preserve_current_turn_user_idx(agent, previous_messages, messages)
+    typed_current_turn_idx = getattr(agent, "_persist_user_message_idx", None)
+    if (
+        isinstance(typed_current_turn_idx, int)
+        and 0 <= typed_current_turn_idx < len(messages)
+    ):
+        current_turn_user_idx = typed_current_turn_idx
+    elif user_message is not None:
+        current_turn_user_idx = _reanchor(agent, messages, user_message)
 
     # Repair malformed role alternation (tool→user / user→user tails): providers
     # return empty content on them and the empty-retry loop spins. The _with_cursor
     # variant also recomputes the SessionDB flush cursor after compaction.
     from agent.agent_runtime_helpers import repair_message_sequence_with_cursor
     repaired_seq = repair_message_sequence_with_cursor(agent, messages)
+    typed_current_turn_idx = getattr(agent, "_persist_user_message_idx", None)
+    if (
+        isinstance(typed_current_turn_idx, int)
+        and 0 <= typed_current_turn_idx < len(messages)
+    ):
+        current_turn_user_idx = typed_current_turn_idx
+    elif user_message is not None:
+        current_turn_user_idx = _reanchor(agent, messages, user_message)
     if repaired_seq > 0:
         request_logger.info(
             "Repaired %s message-alternation violations before request (session=%s)",
             repaired_seq,
             agent.session_id or "-",
         )
-        # The merge shrank the list, so the index recorded at turn start can point past this
-        # turn's user row: prefetch would inject into a historical row and index-settling hosts
-        # (hermes-webui) would write the current turn to the FRONT of the context. Re-anchor as
-        # the compression-restart path does (last verbatim row wins, never a historical copy);
-        # without the text the index cannot be re-derived and is left detectably stale.
-        if user_message is not None:
-            _reanchored_idx = _reanchor(agent, messages, user_message)
-            if _reanchored_idx != current_turn_user_idx:
-                request_logger.info(
-                    "Re-anchored current_turn_user_idx %s -> %s after alternation repair (session=%s)",
-                    current_turn_user_idx, _reanchored_idx, agent.session_id or "-",
-                )
-                current_turn_user_idx = _reanchored_idx
+
     return IterationPrep(
         action="fallthrough", messages=messages, request_logger=request_logger,
         current_turn_user_idx=current_turn_user_idx,

contributors/emails/tonyheyifan@gmail.com

@@ -0,0 +1,2 @@
+heyf
+# PR #70695

gateway/platforms/api_server.py

@@ -3622,7 +3622,7 @@ def _turn_runtime_metadata(
     def _finish_turn_result(
         self, agent: Any, result: Any, session_id: Optional[str], *, route, requested_runtime, route_source,
         confirmed_runtime_lock: bool) -> tuple:
-        """Attach usage, effective session id, ``_compressed`` and runtime metadata to a finished turn."""
+        """Attach transcript, usage, session, compression and runtime metadata."""
         usage = {"input_tokens": getattr(agent, "session_prompt_tokens", 0) or 0,
                  "output_tokens": getattr(agent, "session_completion_tokens", 0) or 0,
                  "total_tokens": getattr(agent, "session_total_tokens", 0) or 0}
@@ -3631,6 +3631,13 @@ def _finish_turn_result(
         _eff_sid = getattr(agent, "session_id", session_id)
         if isinstance(_eff_sid, str) and _eff_sid:
             result["session_id"] = _eff_sid
+        if isinstance(result, dict) and isinstance(result.get("messages"), list):
+            from gateway.response_turn_boundary import FULL_TRANSCRIPT_MODE
+
+            result["_transcript_mode"] = FULL_TRANSCRIPT_MODE
+            result["_current_turn_user_idx"] = getattr(
+                agent, "_persist_user_message_idx", None
+            )
         # _compressed tells _build_response_conversation_history to store the compacted
         # transcript as-is (rotation changes session_id; in-place compaction sets a flag).
         _session_rotated = isinstance(_eff_sid, str) and isinstance(session_id, str) and _eff_sid != session_id

gateway/platforms/api_server_openai_routes.py

@@ -947,40 +947,19 @@ async def _handle_delete_response(self, request: "web.Request") -> "web.Response
     def _build_response_conversation_history(
         conversation_history: List[Dict[str, Any]], user_message: Any, result: Dict[str, Any],
         final_response: Any) -> List[Dict[str, Any]]:
-        """Build the stored Responses transcript without duplicating history.
+        from gateway.response_turn_boundary import build_response_conversation_history
 
-        A compressed transcript (``result["_compressed"]``) shares no input-history prefix, so
-        turn-start detection fails; prepending the uncompressed history would bloat the stored
-        context and re-trigger compression every request — it is stored as-is instead.
-        """
-        from gateway.platforms.api_server import APIServerAdapter
-        prior = list(conversation_history)
-        current_user = {"role": "user", "content": user_message}
-        agent_messages = result.get("messages") if isinstance(result, dict) else None
-        if isinstance(agent_messages, list) and agent_messages:
-            turn_start = APIServerAdapter._response_messages_turn_start_index(
-                conversation_history, user_message, result)
-            # turn_start == 0: compression rewrote the transcript or agent_messages is turn-only.
-            if turn_start or result.get("_compressed"):
-                return list(agent_messages)
-            return prior + [current_user] + agent_messages
-        return prior + [current_user, {"role": "assistant", "content": final_response}]
+        return build_response_conversation_history(
+            conversation_history, user_message, result, final_response
+        )
 
     @staticmethod
     def _response_messages_turn_start_index(
         conversation_history: List[Dict[str, Any]], user_message: Any, result: Dict[str, Any],
     ) -> int:
-        """Detect transcript-shaped result["messages"] and return turn start."""
-        agent_messages = result.get("messages") if isinstance(result, dict) else None
-        if not isinstance(agent_messages, list) or not agent_messages:
-            return 0
-        prior = list(conversation_history)
-        expected_prefix = prior + [{"role": "user", "content": user_message}]
-        if agent_messages[:len(expected_prefix)] == expected_prefix:
-            return len(expected_prefix)
-        if prior and agent_messages[:len(prior)] == prior:
-            return len(prior)
-        return 0
+        from gateway.response_turn_boundary import response_messages_turn_start_index
+
+        return response_messages_turn_start_index(conversation_history, user_message, result)
 
     @classmethod
     def _turn_transcript_messages(

gateway/response_turn_boundary.py

@@ -0,0 +1,111 @@
+"""Responses API transcript-boundary policy.
+
+Normal agent runs return an authoritative full transcript. Legacy integrations
+and tests may still return an unmarked current-turn suffix, so that compatibility
+path remains conservative and separate from the explicit full-transcript contract.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+from agent.turn_context import reanchor_current_turn_user_idx
+
+FULL_TRANSCRIPT_MODE = "full"
+
+_SEMANTIC_KEYS = (
+    "role",
+    "content",
+    "tool_calls",
+    "tool_call_id",
+    "name",
+    "function_call",
+)
+_MESSAGE_ROLES = frozenset({"system", "developer", "user", "assistant", "tool"})
+
+
+def semantic_message(message: Any) -> Any:
+    """Project a message to fields that define transcript identity."""
+    if not isinstance(message, dict) or message.get("role") not in _MESSAGE_ROLES:
+        return None
+    return {key: message[key] for key in _SEMANTIC_KEYS if key in message}
+
+
+def semantic_prefix_matches(messages: List[Any], expected: List[Any]) -> bool:
+    """Return whether ``messages`` begins with ``expected`` semantically."""
+    if len(messages) < len(expected):
+        return False
+    for actual, wanted in zip(messages, expected):
+        actual_semantic = semantic_message(actual)
+        wanted_semantic = semantic_message(wanted)
+        if actual_semantic is None or wanted_semantic is None or actual_semantic != wanted_semantic:
+            return False
+    return True
+
+
+def response_messages_turn_start_index(
+    conversation_history: List[Dict[str, Any]],
+    user_message: Any,
+    result: Dict[str, Any],
+) -> int:
+    """Return the first assistant/tool row belonging to the current turn."""
+    agent_messages = result.get("messages") if isinstance(result, dict) else None
+    if not isinstance(agent_messages, list) or not agent_messages:
+        return 0
+
+    if result.get("_transcript_mode") == FULL_TRANSCRIPT_MODE:
+        current_user_idx = reanchor_current_turn_user_idx(
+            agent_messages, user_message, result.get("_current_turn_user_idx")
+        )
+        return current_user_idx + 1 if current_user_idx >= 0 else len(agent_messages)
+
+    prior = list(conversation_history)
+    expected_with_user = prior + [{"role": "user", "content": user_message}]
+    if semantic_prefix_matches(agent_messages, expected_with_user):
+        return len(expected_with_user)
+    if prior and semantic_prefix_matches(agent_messages, prior):
+        return len(prior)
+
+    if result.get("_compressed"):
+        current_user_idx = reanchor_current_turn_user_idx(agent_messages, user_message)
+        return current_user_idx + 1 if current_user_idx >= 0 else len(agent_messages)
+
+    if not any(
+        isinstance(message, dict) and message.get("role") == "user"
+        for message in agent_messages
+    ):
+        return 0
+
+    for index in range(len(agent_messages) - 1, -1, -1):
+        message = agent_messages[index]
+        if isinstance(message, dict) and message.get("role") == "assistant":
+            return index
+    return len(agent_messages)
+
+
+def build_response_conversation_history(
+    conversation_history: List[Dict[str, Any]],
+    user_message: Any,
+    result: Dict[str, Any],
+    final_response: Any,
+) -> List[Dict[str, Any]]:
+    """Build stored history under the explicit full/legacy suffix contract."""
+    agent_messages = result.get("messages") if isinstance(result, dict) else None
+    if isinstance(agent_messages, list):
+        if result.get("_transcript_mode") == FULL_TRANSCRIPT_MODE:
+            return list(agent_messages)
+        if agent_messages:
+            turn_start = response_messages_turn_start_index(conversation_history, user_message, result)
+            if turn_start or result.get("_compressed"):
+                return list(agent_messages)
+            return [
+                *conversation_history,
+                {"role": "user", "content": user_message},
+                *agent_messages,
+            ]
+
+    return [
+        *conversation_history,
+        {"role": "user", "content": user_message},
+        {"role": "assistant", "content": final_response},
+    ]