Merge pull request #4475 from nesquena/release/stage-4393

Release v0.51.512 — RW (interrupted-turn user prompt no longer replays, #4393)

CHANGELOG.md

@@ -3,6 +3,12 @@
 
 ## [Unreleased]
 
+## [v0.51.512] — 2026-06-19 — Release RW (interrupted-turn user prompt no longer replays)
+
+### Fixed
+
+- **A user prompt from an interrupted turn no longer replays in every subsequent request (#4283).** When a turn was interrupted, its recovered user message could be re-sent on later requests (and, in some orphaned-tool shapes, produce two adjacent same-role messages that strict providers reject with a 400). The API-message sanitizer now decides whether to keep a recovered user message based on the actual post-sanitization neighbours — it is kept only when it genuinely separates two assistant turns, and dropped otherwise — with the same logic applied in both the sanitizer and the safe-position mirror. Thanks @kaishi00.
+
 ## [v0.51.511] — 2026-06-19 — Release RV (virtual transcript renders during programmatic scrolls)
 
 ### Fixed

api/streaming.py

@@ -3466,6 +3466,12 @@ def _sanitize_messages_for_api(messages, *, cfg: dict = None):
         # API 400 errors on strict providers (empty assistant content).
         if msg.get('_partial') and not str(msg.get('content') or '').strip():
             continue
+        # Note: _recovered user messages are NOT skipped here — they may need
+        # to be retained to preserve role alternation when a kept assistant
+        # follows.  The _recovered skip happens in a final pass after orphaned
+        # tool_calls are stripped, so the anchor check is exact (#4283).
+        # Temporarily mark _recovered users so the final pass can find them.
+        is_recovered = msg.get('_recovered') and msg.get('role') == 'user'
         role = msg.get('role')
         if role == 'tool':
             tid = msg.get('tool_call_id') or ''
@@ -3473,6 +3479,8 @@ def _sanitize_messages_for_api(messages, *, cfg: dict = None):
                 # Orphaned tool result — skip to avoid 400 from strict providers.
                 continue
         sanitized = {k: v for k, v in msg.items() if k in _API_SAFE_MSG_KEYS}
+        if is_recovered:
+            sanitized['_recovered'] = True  # temporary marker — stripped before return
         if strip_native_images and 'content' in sanitized:
             sanitized['content'] = _strip_native_image_parts_from_content(sanitized.get('content'))
         if sanitized.get('role'):
@@ -3505,7 +3513,35 @@ def _sanitize_messages_for_api(messages, *, cfg: dict = None):
             else:
                 msg = dict(msg, tool_calls=kept)
         filtered_clean.append(msg)
-    return filtered_clean
+
+    # Fourth pass: drop _recovered user messages unless removing one would fuse
+    # two same-role neighbours.  Operating on filtered_clean (post orphaned-tool/
+    # tool_calls stripping) means the neighbour check is exact (#4283).  The
+    # decision uses the ACTUAL kept sequence: the previously-kept message's role
+    # (`final[-1]`) and the next surviving message's role.  A _recovered user is
+    # kept ONLY when it separates two assistants (prev kept == assistant AND next
+    # surviving == assistant) — i.e. it is an answered turn whose removal would
+    # leave `assistant, assistant` adjacency.  In every other case dropping it is
+    # safe and correct: it would either leave a clean `user, assistant` pair, or
+    # (if next is a user) it is a stale unanswered prompt that must not replay.
+    # Deciding only on "an assistant follows" (ignoring the prev kept role) is the
+    # bug that re-introduced `user, _recovered user, assistant` → adjacent users
+    # → strict-provider 400 once the anchoring assistant's predecessor was a user.
+    final = []
+    for i, msg in enumerate(filtered_clean):
+        if msg.get('_recovered') and msg.get('role') == 'user':
+            prev_role = final[-1].get('role') if final else None
+            next_role = None
+            for j in range(i + 1, len(filtered_clean)):
+                next_role = filtered_clean[j].get('role')
+                break
+            # Keep only if this recovered user actually separates two assistants.
+            if not (prev_role == 'assistant' and next_role == 'assistant'):
+                continue  # drop — fusing the neighbours is clean, or it's a stale prompt
+            # Keep but strip the temporary marker
+            msg = {k: v for k, v in msg.items() if k != '_recovered'}
+        final.append(msg)
+    return final
 
 
 def _api_safe_message_positions(messages):
@@ -3531,12 +3567,17 @@ def _api_safe_message_positions(messages):
             continue
         if msg.get('_partial') and not str(msg.get('content') or '').strip():
             continue
+        # Note: _recovered user messages are NOT skipped here — deferred to
+        # a final pass after orphaned tool_calls stripping (#4283).
+        is_recovered = msg.get('_recovered') and msg.get('role') == 'user'
         role = msg.get('role')
         if role == 'tool':
             tid = msg.get('tool_call_id') or ''
             if not tid or tid not in valid_tool_call_ids:
                 continue
         sanitized = {k: v for k, v in msg.items() if k in _API_SAFE_MSG_KEYS}
+        if is_recovered:
+            sanitized['_recovered'] = True  # temporary marker — stripped before return
         if sanitized.get('role'):
             out.append((idx, sanitized))
 
@@ -3564,7 +3605,24 @@ def _api_safe_message_positions(messages):
             else:
                 msg = dict(msg, tool_calls=kept)
         filtered_out.append((idx, msg))
-    return filtered_out
+
+    # Fourth pass: drop _recovered user messages unless removing one would fuse
+    # two same-role neighbours — mirrors _sanitize_messages_for_api pass 4 (#4283).
+    # Decide on the ACTUAL kept sequence: prev kept role (final_out[-1]) + next
+    # surviving role. Keep ONLY when it separates two assistants; otherwise drop.
+    final_out = []
+    for i, (idx, msg) in enumerate(filtered_out):
+        if msg.get('_recovered') and msg.get('role') == 'user':
+            prev_role = final_out[-1][1].get('role') if final_out else None
+            next_role = None
+            for j in range(i + 1, len(filtered_out)):
+                next_role = filtered_out[j][1].get('role')
+                break
+            if not (prev_role == 'assistant' and next_role == 'assistant'):
+                continue
+            msg = {k: v for k, v in msg.items() if k != '_recovered'}
+        final_out.append((idx, msg))
+    return final_out
 
 
 def _deduplicate_context_messages(messages):
@@ -5016,6 +5074,23 @@ def _materialize_pending_user_turn_before_error(session) -> bool:
     if pending_attachments:
         recovered['attachments'] = list(pending_attachments)
     session.messages.append(recovered)
+    # Mirror to context_messages so the _recovered flag survives the state.db
+    # round-trip (#4283).  state.db has no _recovered column, so without this
+    # mirror the next turn's reconciled_state_db_messages_for_session(
+    # prefer_context=True) finds the recovered user as a flagless state.db
+    # delta and _sanitize_messages_for_api cannot filter it — causing the
+    # interrupted turn's prompt to be prepended to every subsequent turn.
+    # Placing the mirror here (rather than in _persist_cancelled_turn) covers
+    # all three callers: cancel, provider-error, and exception paths.
+    ctx = getattr(session, 'context_messages', None)
+    if isinstance(ctx, list) and ctx:
+        rec_text = " ".join(str(recovered.get('content') or '').split())
+        if not any(
+            isinstance(e, dict) and e.get('role') == 'user'
+            and " ".join(str(e.get('content') or '').split()) == rec_text
+            for e in ctx[-8:]
+        ):
+            ctx.append({k: v for k, v in recovered.items() if k != 'timestamp'})
     # The new user turn is now committed to messages (#3831): retire a positive
     # truncation watermark left over from a prior retry/undo/edit so it cannot
     # freeze at the old edit boundary and later drop these post-edit turns on an

tests/test_issue4283_recovered_context_replay.py

@@ -0,0 +1,368 @@
+"""Regression tests for #4283: interrupted-turn user message replay.
+
+Three shapes must be covered:
+
+1. **Pure cancel** — recovered user followed only by an _error marker.
+   The recovered user must be stripped (no assistant to anchor).
+
+2. **Interrupted-with-saved-partial** — recovered user followed by a kept
+   _partial assistant.  The user must be retained to preserve role
+   alternation (adjacent assistants → 400 on strict providers).
+
+3. **Orphaned-tool-calls divergence** — the forward-scan anchor check must
+   operate on the post-sanitized list, not the raw input, so it doesn't
+   keep a recovered user based on an assistant that pass 3 will drop
+   (Nesquena review on PR #4393).
+"""
+from __future__ import annotations
+
+from api.streaming import (
+    _sanitize_messages_for_api,
+    _api_safe_message_positions,
+    _materialize_pending_user_turn_before_error,
+)
+
+
+# ── Shape 1: pure cancel — recovered user stripped ─────────────────────────
+
+def test_recovered_user_stripped_when_no_kept_assistant_follows():
+    """Pure cancel shape: recovered user followed only by an _error marker.
+    The user should be stripped — no adjacent-assistant risk.
+    """
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "stale prompt", "_recovered": True},
+        {"role": "assistant", "content": "Task cancelled.", "_error": True},
+        {"role": "user", "content": "Q3"},
+    ]
+    result = _sanitize_messages_for_api(messages)
+    roles = [m["role"] for m in result]
+    assert roles == ["user", "assistant", "user"]
+    assert not any("stale prompt" in m.get("content", "") for m in result)
+
+
+def test_recovered_user_stripped_when_next_is_user():
+    """Recovered user followed by another user (no assistant between).
+    The recovered user should be stripped — the next user replaces it.
+    """
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "stale", "_recovered": True},
+        {"role": "user", "content": "Q3"},
+    ]
+    result = _sanitize_messages_for_api(messages)
+    roles = [m["role"] for m in result]
+    assert roles == ["user", "assistant", "user"]
+    assert not any("stale" in m.get("content", "") for m in result)
+
+
+def test_recovered_user_stripped_at_end_of_list():
+    """Recovered user at the very end — nothing follows, no anchor.
+    Should be stripped.
+    """
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "stale", "_recovered": True},
+    ]
+    result = _sanitize_messages_for_api(messages)
+    roles = [m["role"] for m in result]
+    assert roles == ["user", "assistant"]
+
+
+# ── Shape 2: interrupted-with-saved-partial — recovered user kept ──────────
+
+def test_recovered_user_kept_when_anchoring_partial_assistant():
+    """Dropping a _recovered user that precedes a kept _partial assistant
+    would produce adjacent assistant messages → 400 on strict providers.
+    The user must be retained.
+    """
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "Q2", "_recovered": True},
+        {"role": "assistant", "content": "Partial answer…", "_partial": True},
+        {"role": "user", "content": "Q3"},
+    ]
+    result = _sanitize_messages_for_api(messages)
+    roles = [m["role"] for m in result]
+    assert roles == ["user", "assistant", "user", "assistant", "user"]
+    # The recovered user is retained because it anchors the partial assistant.
+    assert any(
+        m.get("role") == "user" and "Q2" in m.get("content", "")
+        for m in result
+    )
+
+
+def test_recovered_user_kept_when_anchoring_assistant_with_tool_calls():
+    """Recovered user followed by an assistant with tool_calls (no _partial).
+    The user must be retained to preserve role alternation.
+    """
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "Q2", "_recovered": True},
+        {"role": "assistant", "content": "", "tool_calls": [{"id": "tc1", "type": "function", "function": {"name": "search", "arguments": "{}"}}]},
+        {"role": "tool", "content": "result", "tool_call_id": "tc1"},
+        {"role": "user", "content": "Q3"},
+    ]
+    result = _sanitize_messages_for_api(messages)
+    roles = [m["role"] for m in result]
+    # The recovered user is retained because it anchors the assistant with tool_calls.
+    assert "user" in roles
+    assert roles.count("user") >= 2  # Q1 + Q2 (recovered, kept) or Q1 + Q3
+
+
+# ── Shape 3: orphaned-tool-calls divergence (Nesquena review) ──────────────
+
+def test_stale_user_dropped_when_anchor_is_orphaned_empty_assistant():
+    """Shape (a) from Nesquena's review: an empty assistant with orphaned
+    tool_calls looks like a valid anchor to a forward scan, but pass 3
+    strips it (all tool_calls orphaned, no content → dropped).  The
+    recovered user must then also be dropped — operating on the
+    post-sanitized list catches this.
+    """
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "stale", "_recovered": True},
+        # Assistant with tool_calls but no matching tool response → orphaned
+        {"role": "assistant", "content": "", "tool_calls": [{"id": "tc_orphan", "type": "function", "function": {"name": "search", "arguments": "{}"}}]},
+        {"role": "user", "content": "Q3"},
+    ]
+    result = _sanitize_messages_for_api(messages)
+    roles = [m["role"] for m in result]
+    # The orphaned assistant is dropped → recovered user has no anchor → dropped
+    assert roles == ["user", "assistant", "user"]
+    assert not any("stale" in m.get("content", "") for m in result)
+
+
+def test_no_adjacent_assistants_when_orphan_tool_between():
+    """Shape (b) from Nesquena's review: recovered user, orphan tool row,
+    then a real assistant.  The orphan tool is dropped by pass 2, which
+    would leave adjacent [recovered_user, assistant] — but since the
+    recovered user DOES anchor the assistant, it's kept, and the roles
+    are correct: [user, assistant, user, assistant, user].
+    """
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "Q2", "_recovered": True},
+        # Orphan tool (no matching assistant tool_call in history)
+        {"role": "tool", "content": "orphan result", "tool_call_id": "tc_missing"},
+        {"role": "assistant", "content": "A2"},
+        {"role": "user", "content": "Q3"},
+    ]
+    result = _sanitize_messages_for_api(messages)
+    roles = [m["role"] for m in result]
+    # Orphan tool dropped, recovered user anchors A2 → kept
+    assert roles == ["user", "assistant", "user", "assistant", "user"]
+
+
+def _assert_no_adjacent_same_role(roles):
+    pairs = [(roles[i], roles[i + 1]) for i in range(len(roles) - 1) if roles[i] == roles[i + 1]]
+    assert not pairs, f"adjacent same-role messages would 400 on strict providers: {pairs} in {roles}"
+
+
+def test_recovered_user_dropped_when_prev_kept_is_user_orphan_tool_calls_assistant():
+    """Regression (Codex/Opus gate on #4393): the pass-4 keep decision must use
+    the ACTUAL kept neighbours, not just 'an assistant follows'.
+
+    Shape: U_prev, A1, recovered U, assistant(orphaned tool_calls, empty content),
+    A2.  Pass 3 drops the orphaned-tool-calls assistant.  The recovered user is
+    correctly KEPT here because its kept neighbours are A1 (prev) and A2 (next) —
+    it separates two assistants.  Result must alternate with no adjacent users.
+    """
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "Q2", "_recovered": True},
+        {"role": "assistant", "content": "", "tool_calls": [
+            {"id": "tc_orphan", "type": "function", "function": {"name": "f", "arguments": "{}"}}]},
+        {"role": "assistant", "content": "A2"},
+    ]
+    result = _sanitize_messages_for_api(messages)
+    roles = [m["role"] for m in result]
+    _assert_no_adjacent_same_role(roles)
+    assert roles == ["user", "assistant", "user", "assistant"]
+    assert not any("_recovered" in m for m in result)
+
+
+def test_recovered_user_dropped_when_orphan_assistant_leaves_user_prev():
+    """The bug case: dropping the orphaned-tool assistant makes the recovered
+    user's previous kept neighbour a USER.  Keeping the recovered user would then
+    produce [user, user, assistant] → strict-provider 400.  It must be dropped.
+
+    Shape: U_prev, assistant(orphaned tool_calls, empty), recovered U, A_next.
+    """
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "", "tool_calls": [
+            {"id": "tc_orphan", "type": "function", "function": {"name": "f", "arguments": "{}"}}]},
+        {"role": "user", "content": "Q2", "_recovered": True},
+        {"role": "assistant", "content": "A2"},
+    ]
+    result = _sanitize_messages_for_api(messages)
+    roles = [m["role"] for m in result]
+    _assert_no_adjacent_same_role(roles)
+    # orphan assistant dropped (pass 3); recovered user would fuse to [user, assistant]
+    # cleanly if kept, but prev-kept is a user → it's a stale unanswered prompt → drop.
+    assert roles == ["user", "assistant"]
+    assert not any(m.get("role") == "user" and m.get("content") == "Q2" for m in result)
+
+
+def test_api_safe_positions_no_adjacent_users_with_orphan_tool_calls():
+    """The _api_safe_message_positions mirror must make the same neighbour-aware
+    decision (Codex/Opus said mirror the fix in both functions)."""
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "", "tool_calls": [
+            {"id": "tc_orphan", "type": "function", "function": {"name": "f", "arguments": "{}"}}]},
+        {"role": "user", "content": "Q2", "_recovered": True},
+        {"role": "assistant", "content": "A2"},
+    ]
+    positions = _api_safe_message_positions(messages)
+    kept_roles = [messages[idx]["role"] for idx, _ in positions]
+    _assert_no_adjacent_same_role(kept_roles)
+
+
+# ── _api_safe_message_positions parity ─────────────────────────────────────
+
+def test_api_safe_positions_strips_recovered_user():
+    """Same as sanitize but via _api_safe_message_positions."""
+    ctx = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "stale", "_recovered": True, "timestamp": 123},
+    ]
+    positions = _api_safe_message_positions(ctx)
+    roles = [msg["role"] for _, msg in positions]
+    assert roles == ["user", "assistant"]
+
+
+def test_api_safe_positions_keeps_recovered_user_with_partial():
+    """_api_safe_message_positions mirrors sanitize for the anchor case."""
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "Q2", "_recovered": True},
+        {"role": "assistant", "content": "Partial…", "_partial": True},
+    ]
+    positions = _api_safe_message_positions(messages)
+    roles = [msg["role"] for _, msg in positions]
+    assert roles == ["user", "assistant", "user", "assistant"]
+
+
+# ── context_messages mirror (Fix 2) ────────────────────────────────────────
+
+class _DummySession:
+    """Minimal session for _materialize_pending_user_turn_before_error tests."""
+    def __init__(self, messages=None, context_messages=None, pending_msg=""):
+        self.messages = messages or []
+        self.context_messages = context_messages
+        self.pending_user_message = pending_msg
+        self.pending_attachments = []
+        self.pending_started_at = None
+        self.active_stream_id = "stream-test"
+        self.truncation_watermark = None
+        self.path = ""
+        self.session_id = "test-4283"
+
+    def save(self, *args, **kwargs):
+        pass
+
+
+def test_materialize_mirrors_recovered_user_to_context_messages():
+    """_materialize_pending_user_turn_before_error must mirror the recovered
+    user to context_messages so the _recovered flag survives the state.db
+    round-trip (#4283).
+    """
+    s = _DummySession(
+        messages=[{"role": "assistant", "content": "A1"}],
+        context_messages=[
+            {"role": "user", "content": "Q1"},
+            {"role": "assistant", "content": "A1"},
+        ],
+        pending_msg="restart the gateway",
+    )
+    appended = _materialize_pending_user_turn_before_error(s)
+
+    assert appended is True
+    ctx_users = [m for m in s.context_messages if m.get("role") == "user"]
+    assert any(m.get("_recovered") for m in ctx_users)
+    assert any("restart the gateway" in m.get("content", "") for m in ctx_users)
+
+
+def test_materialize_does_not_duplicate_context_messages():
+    """Repeated calls must not grow context_messages unboundedly."""
+    s = _DummySession(
+        messages=[{"role": "assistant", "content": "A1"}],
+        context_messages=[
+            {"role": "user", "content": "Q1"},
+            {"role": "assistant", "content": "A1"},
+        ],
+        pending_msg="restart the gateway",
+    )
+    _materialize_pending_user_turn_before_error(s)
+    ctx_len_after_first = len(s.context_messages)
+
+    s.pending_user_message = "restart the gateway"
+    _materialize_pending_user_turn_before_error(s)
+    assert len(s.context_messages) == ctx_len_after_first
+
+
+def test_materialize_skips_mirror_when_context_messages_empty():
+    """When context_messages is empty/None (first-turn error), the mirror
+    should be skipped — prefer_context falls back to session.messages.
+    """
+    s = _DummySession(
+        messages=[],
+        context_messages=None,
+        pending_msg="first turn error",
+    )
+    appended = _materialize_pending_user_turn_before_error(s)
+
+    assert appended is True
+    assert s.context_messages is None
+    assert s.messages[-1].get("_recovered") is True
+
+
+def test_sanitize_strips_recovered_user_from_context_messages():
+    """End-to-end: recovered user mirrored to context_messages with
+    _recovered flag → _sanitize_messages_for_api strips it (pure cancel).
+    """
+    ctx = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "stale prompt", "_recovered": True, "timestamp": 123},
+    ]
+    result = _sanitize_messages_for_api(ctx)
+    roles = [m["role"] for m in result]
+    assert roles == ["user", "assistant"]
+    assert not any("stale prompt" in m.get("content", "") for m in result)
+
+
+# ── No _recovered marker leaks into API output ─────────────────────────────
+
+def test_no_recovered_marker_in_output():
+    """The temporary _recovered marker must be stripped from all messages
+    in the sanitized output — it must never reach the API.
+    """
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "A1"},
+        {"role": "user", "content": "Q2", "_recovered": True},
+        {"role": "assistant", "content": "Partial…", "_partial": True},
+        {"role": "user", "content": "Q3"},
+    ]
+    result = _sanitize_messages_for_api(messages)
+    for msg in result:
+        assert "_recovered" not in msg, f"_recovered leaked into sanitized output: {msg}"
+
+    # Also verify the positions path strips the marker (#4393 review note).
+    from api.streaming import _api_safe_message_positions
+    pos_result = _api_safe_message_positions(messages)
+    for idx, msg in pos_result:
+        assert "_recovered" not in msg, f"_recovered leaked into positions output: {msg}"