From c845e140875f22dfee1fc0c9619806a9c8fd1f6d Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Fri, 19 Jun 2026 05:06:33 +0000 Subject: [PATCH] fix(#4283/#4393): pass-4 recovered-user keep decision uses actual kept neighbours Codex+Opus gate caught that the forward-scan-only keep produced [user, _recovered user, assistant] -> adjacent users -> strict-provider 400 once the anchoring assistant's prev kept neighbour was a user. Keep a recovered user ONLY when it separates two assistants (prev_kept==assistant AND next==assistant); mirror in _api_safe_message_positions. + 3 regression tests. --- api/streaming.py | 81 +++- ...test_issue4283_recovered_context_replay.py | 368 ++++++++++++++++++ 2 files changed, 446 insertions(+), 3 deletions(-) create mode 100644 tests/test_issue4283_recovered_context_replay.py diff --git a/api/streaming.py b/api/streaming.py index 8e4cef70..8d0493ab 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -3448,7 +3448,7 @@ def _sanitize_messages_for_api(messages, *, cfg: dict = None): # Second pass: build the sanitized list, dropping orphaned tool messages. clean = [] - for msg in messages: + for idx, msg in enumerate(messages): if not isinstance(msg, dict): continue # Skip display-only Thinking entries. They are visible transcript @@ -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 diff --git a/tests/test_issue4283_recovered_context_replay.py b/tests/test_issue4283_recovered_context_replay.py new file mode 100644 index 00000000..f9fd0f80 --- /dev/null +++ b/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}"