From f4b2cf033b703fe0eaf3af7657aab4b2dd6721ad Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Tue, 16 Jun 2026 17:03:37 +0000 Subject: [PATCH] fix(streaming): normalize repair-merged user context (#4089) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the agent's role-sequence repair concatenates the prior context-tail user row with the submitted current turn as \n\n, that polluted pair was persisted into both the model context and the visible transcript. Add a _detect_stale_user_merge detector (requires the literal \n\n repair boundary + exact tail/current match after sentinel stripping) and funnel context-merge and display-merge through one _strip_stale_user_merge_from_messages so a single rule governs persistence. The strip is scoped to the new-turn candidate slice only — historical rows in the already-committed previous_context prefix are never rewritten. Co-authored-by: starship-s --- api/routes.py | 1 + api/streaming.py | 168 +++++- .../test_stale_user_context_contamination.py | 519 ++++++++++++++++++ 3 files changed, 685 insertions(+), 3 deletions(-) create mode 100644 tests/test_stale_user_context_contamination.py diff --git a/api/routes.py b/api/routes.py index c5cb9a57..9d4ddaae 100644 --- a/api/routes.py +++ b/api/routes.py @@ -14325,6 +14325,7 @@ def _handle_chat_sync(handler, body): _next_context_messages = _dedupe_replayed_context_messages( _previous_context_messages, _next_context_messages, + msg, ) s.context_messages = _next_context_messages s.messages = _merge_display_messages_after_agent_result( diff --git a/api/streaming.py b/api/streaming.py index 89e1417c..aceed689 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -3826,24 +3826,63 @@ def _strip_replayed_context_items(existing_messages, candidates): return cleaned -def _dedupe_replayed_context_messages(previous_context, result_messages): +def _dedupe_replayed_context_messages(previous_context, result_messages, msg_text=None): """Keep model context append-only without replayed blocks/summaries.""" previous_context = list(previous_context or []) result_messages = list(result_messages or []) if not previous_context or not result_messages: return result_messages + previous_user_tail = _stale_user_tail_candidate(_last_user_row(previous_context)) if not _messages_have_prefix(result_messages, previous_context): + # Agent-side role-sequence repair can replace the last prior user row + # with a repaired current-user row. In that shape the result no longer + # has `previous_context` as an exact prefix, but it should still be + # merged as: previous context + clean current turn + assistant/tool delta. + if ( + msg_text + and len(previous_context) >= 1 + and len(result_messages) >= len(previous_context) + and _messages_have_prefix(result_messages, previous_context[:-1]) + ): + boundary_idx = len(previous_context) - 1 + boundary_row = result_messages[boundary_idx] + is_stale_merge = ( + previous_user_tail + and _detect_stale_user_merge(boundary_row, msg_text, previous_user_tail) + ) + if is_stale_merge or _looks_like_current_user_turn(boundary_row, msg_text): + if is_stale_merge: + # Clean only the stale-merged boundary row; leave all prior + # history in previous_context untouched. + cleaned_boundary = copy.deepcopy(boundary_row) + cleaned_boundary['content'] = msg_text + candidates = [cleaned_boundary] + result_messages[boundary_idx + 1:] + else: + candidates = result_messages[boundary_idx:] + candidates = _strip_replayed_prefix(previous_context, candidates) + if candidates: + candidates = _strip_replayed_context_items(previous_context, candidates) + return previous_context + candidates return result_messages candidates = result_messages[len(previous_context):] + # Strip stale merges only from the new-turn candidate slice so that + # legitimate historical user rows in the already-committed previous_context + # prefix are never rewritten. + if msg_text and previous_user_tail: + candidates = _strip_stale_user_merge_from_messages( + candidates, + msg_text, + previous_user_tail, + ) candidates = _strip_replayed_prefix(previous_context, candidates) if candidates: candidates = _strip_replayed_context_items(previous_context, candidates) return previous_context + candidates -def _dedupe_replayed_active_context(previous_context, result_messages): +def _dedupe_replayed_active_context(previous_context, result_messages, msg_text=None): """Keep model context append-only without re-appending a replayed tail.""" - return _dedupe_replayed_context_messages(previous_context, result_messages) + return _dedupe_replayed_context_messages(previous_context, result_messages, msg_text) def _is_context_compression_marker(msg): @@ -3941,6 +3980,110 @@ def _drop_checkpointed_current_user_from_context(messages, msg_text): return history +def _strip_workspace_prefixes_for_compare(text: str) -> str: + """Remove WebUI workspace sentinels anywhere before text comparison.""" + value = _strip_workspace_prefix(text, include_legacy=True) + for pattern in (_WORKSPACE_PREFIX_ANY_RE, _LEGACY_WORKSPACE_PREFIX_ANY_RE): + value = pattern.sub('', value) + return value.strip() + + +def _normalize_user_text(text): + """Collapse whitespace and strip workspace sentinels for tail comparisons.""" + if not isinstance(text, str): + return "" + return " ".join(_strip_workspace_prefixes_for_compare(text).split()) + + +def _raw_message_text(value) -> str: + """Extract text from a message content payload without stripping markup. + + Used for the stale-user-merge detector so the literal boundary between + the prior tail and the current turn survives into the comparison. The + thinking-markup strip in ``_message_text`` collapses newlines, which + would defeat the boundary check. + """ + if isinstance(value, list): + return ' '.join( + str(p.get('text') or p.get('content') or '') + for p in value + if isinstance(p, dict) + ) + return str(value or '') + + +def _stale_user_tail_candidate(msg): + """Return normalized text if msg is a user row that could be a stale tail.""" + if not isinstance(msg, dict) or msg.get('role') != 'user': + return None + raw = _raw_message_text(msg.get('content', '')) + if not raw.strip(): + return None + return _normalize_user_text(raw) + + +def _last_user_row(messages): + """Return the last user-role row in `messages`, or None.""" + for msg in reversed(list(messages or [])): + if isinstance(msg, dict) and msg.get('role') == 'user': + return msg + return None + + +def _detect_stale_user_merge(message, msg_text, previous_user_tail): + """Return True if `message` is the current user turn with a stale prefix merged in. + + The agent's defensive repair path can concatenate the prior context-tail + user row with the submitted current turn as ``\\n\\n``. + The literal ``\\n\\n`` boundary must survive into the comparison; a + single-newline or space-only join is not the repair shape and must not + match. Workspace sentinels may be present on either or both halves and + are stripped before comparison. + """ + if not isinstance(message, dict) or message.get('role') != 'user': + return False + current_norm = _normalize_user_text(msg_text) + if not current_norm: + return False + tail_norm = _normalize_user_text(previous_user_tail) if previous_user_tail else "" + if not tail_norm or len(tail_norm) < 2: + return False + merged = _raw_message_text(message.get('content', '')) + if not merged: + return False + # Require the literal \n\n repair boundary to be present after sentinel + # stripping. Check both the raw text and the workspace-stripped text so + # that workspace-prefixed rows on either or both halves still match. + for candidate in (merged, _strip_workspace_prefixes_for_compare(merged)): + if '\n\n' not in candidate: + continue + head, _, rest = candidate.partition('\n\n') + if _normalize_user_text(head) == tail_norm and _normalize_user_text(rest) == current_norm: + return True + return False + + +def _strip_stale_user_merge_from_messages(messages, msg_text, previous_user_tail): + """Return messages with stale-prefixed current user turns replaced by clean ones. + + Both context-merge (model-facing) and display-merge (visible transcript) + callers funnel through this so a single detection rule governs persistence. + The current user row is replaced with a clean copy using `msg_text` so the + displayed bubble matches what the human submitted, never the polluted pair. + """ + if not messages or not msg_text: + return messages + out = [] + for msg in messages: + if _detect_stale_user_merge(msg, msg_text, previous_user_tail): + cleaned = copy.deepcopy(msg) if isinstance(msg, dict) else {'role': 'user', 'content': msg_text} + cleaned['content'] = msg_text + out.append(cleaned) + else: + out.append(msg) + return out + + def _save_streaming_checkpoint(session): """Persist a streaming checkpoint under the session's profile context.""" from api import profiles as profiles_api @@ -4149,6 +4292,7 @@ def _merge_display_messages_after_agent_result(previous_display, previous_contex result_messages = list(result_messages or []) if not result_messages: return previous_display + previous_user_tail = _stale_user_tail_candidate(_last_user_row(previous_context)) # ── Backfill normal turns from previous_context that are missing from # previous_display. After context compression recovery, previous_context @@ -4252,6 +4396,14 @@ def _merge_display_messages_after_agent_result(previous_display, previous_contex if _messages_have_prefix(result_messages, previous_context): candidates = result_messages[len(previous_context):] + # Normalize stale merges only in the new-turn slice; never rewrite + # historical rows in the already-committed previous_context prefix. + if msg_text and previous_user_tail: + candidates = _strip_stale_user_merge_from_messages( + candidates, + msg_text, + previous_user_tail, + ) candidates = _strip_replayed_prefix(previous_display, candidates) candidates = _strip_replayed_prefix(previous_context, candidates) else: @@ -4261,6 +4413,13 @@ def _merge_display_messages_after_agent_result(previous_display, previous_contex if _is_context_compression_marker(m) ] turn_candidates = result_messages[current_user_idx:] if current_user_idx is not None else [] + # Normalize stale merges only in the current-turn slice. + if msg_text and previous_user_tail: + turn_candidates = _strip_stale_user_merge_from_messages( + turn_candidates, + msg_text, + previous_user_tail, + ) candidates = marker_candidates + turn_candidates merged = previous_display[:] @@ -6906,6 +7065,7 @@ def _run_agent_streaming( _next_context_messages = _dedupe_replayed_context_messages( _previous_context_messages, _next_context_messages, + msg_text, ) s.context_messages = _deduplicate_context_messages(_next_context_messages) s.messages = _merge_display_messages_after_agent_result( @@ -7197,6 +7357,7 @@ def _run_agent_streaming( _next_context_messages = _dedupe_replayed_context_messages( _previous_context_messages, _next_context_messages, + msg_text, ) s.context_messages = _deduplicate_context_messages(_next_context_messages) s.messages = _merge_display_messages_after_agent_result( @@ -8176,6 +8337,7 @@ def _run_agent_streaming( _next_context_messages = _dedupe_replayed_context_messages( _previous_context_messages, _next_context_messages, + msg_text, ) s.context_messages = _deduplicate_context_messages(_next_context_messages) s.messages = _merge_display_messages_after_agent_result( diff --git a/tests/test_stale_user_context_contamination.py b/tests/test_stale_user_context_contamination.py new file mode 100644 index 00000000..6fd24fbd --- /dev/null +++ b/tests/test_stale_user_context_contamination.py @@ -0,0 +1,519 @@ +"""Behavioral tests for stale user-context contamination repair. + +The agent's defensive `repair_message_sequence()` can concatenate a prior +context-tail user row with the current submitted turn as +``\\n\\n``. WebUI must not persist that +polluted merged string as the current user turn in either visible +``messages`` or model-facing ``context_messages``. + +These tests exercise the helpers in ``api.streaming`` directly so the +behaviour can be verified without a live chat/stream round-trip. +""" + +import pytest + + +PRIOR_TAIL = "please use the larger context model" +CURRENT_TURN = "can you summarize the release blockers?" +POLLUTED = f"{PRIOR_TAIL}\n\n{CURRENT_TURN}" + + +def _text(value): + """Extract plain text from a message's content field (str or list).""" + if isinstance(value, str): + return value + if isinstance(value, list): + return " ".join( + part.get("text", "") + for part in value + if isinstance(part, dict) and isinstance(part.get("text"), str) + ) + return str(value or "") + + +def test_detect_stale_user_merge_matches_polluted_pair(): + """Detector must flag the polluted merge as a stale-prefixed current turn.""" + from api.streaming import _detect_stale_user_merge + + polluted_msg = {"role": "user", "content": POLLUTED} + assert _detect_stale_user_merge(polluted_msg, CURRENT_TURN, PRIOR_TAIL) is True + + +def test_detect_stale_user_merge_does_not_match_clean_current(): + """A clean current turn that happens to mention the prior phrase stays intact.""" + from api.streaming import _detect_stale_user_merge + + clean_msg = {"role": "user", "content": CURRENT_TURN} + assert _detect_stale_user_merge(clean_msg, CURRENT_TURN, PRIOR_TAIL) is False + + +def test_detect_stale_user_merge_does_not_match_when_tail_differs(): + """If the prior tail is different, the merge is not the repair pattern.""" + from api.streaming import _detect_stale_user_merge + + other_msg = {"role": "user", "content": f"other stale\n\n{CURRENT_TURN}"} + assert _detect_stale_user_merge(other_msg, CURRENT_TURN, PRIOR_TAIL) is False + + +def test_detect_stale_user_merge_ignores_non_user_roles(): + """Only user rows are candidates for the repair-merge pattern.""" + from api.streaming import _detect_stale_user_merge + + assistant_msg = {"role": "assistant", "content": POLLUTED} + assert _detect_stale_user_merge(assistant_msg, CURRENT_TURN, PRIOR_TAIL) is False + + +def test_detect_stale_user_merge_handles_workspace_prefixed_row(): + """Workspace-prefixed model rows still match when stripped.""" + from api.streaming import _detect_stale_user_merge + + prefixed = { + "role": "user", + "content": f"[Workspace::v1: /tmp/project]\n{POLLUTED}", + } + assert _detect_stale_user_merge(prefixed, CURRENT_TURN, PRIOR_TAIL) is True + + +def test_detect_stale_user_merge_handles_workspace_prefix_on_both_halves(): + """Repair can concatenate two separately workspace-prefixed user rows.""" + from api.streaming import _detect_stale_user_merge + + prefixed_both = { + "role": "user", + "content": ( + f"[Workspace::v1: /tmp/project]\n{PRIOR_TAIL}\n\n" + f"[Workspace::v1: /tmp/project]\n{CURRENT_TURN}" + ), + } + assert _detect_stale_user_merge(prefixed_both, CURRENT_TURN, PRIOR_TAIL) is True + + +def test_detect_stale_user_merge_does_not_match_single_newline_joined(): + """A single-\\n separator is not the repair shape and must not be flagged.""" + from api.streaming import _detect_stale_user_merge + + single_nl = {"role": "user", "content": f"{PRIOR_TAIL}\n{CURRENT_TURN}"} + assert _detect_stale_user_merge(single_nl, CURRENT_TURN, PRIOR_TAIL) is False + + +def test_detect_stale_user_merge_does_not_match_space_joined(): + """A space-only separator is not the repair shape and must not be flagged.""" + from api.streaming import _detect_stale_user_merge + + space_joined = {"role": "user", "content": f"{PRIOR_TAIL} {CURRENT_TURN}"} + assert _detect_stale_user_merge(space_joined, CURRENT_TURN, PRIOR_TAIL) is False + + +def test_strip_stale_user_merge_from_messages_replaces_polluted_row(): + """Normalizer replaces a polluted current row with a clean copy of msg_text.""" + from api.streaming import _strip_stale_user_merge_from_messages + + messages = [ + {"role": "user", "content": POLLUTED}, + {"role": "assistant", "content": "Sure, checking that now."}, + ] + + cleaned = _strip_stale_user_merge_from_messages(messages, CURRENT_TURN, PRIOR_TAIL) + + assert cleaned[0]["content"] == CURRENT_TURN + assert cleaned[0]["role"] == "user" + assert cleaned[1] == messages[1] + + +def test_strip_stale_user_merge_handles_list_content_row(): + """Normalizer also handles OpenAI-style list content payloads.""" + from api.streaming import _strip_stale_user_merge_from_messages + + messages = [ + { + "role": "user", + "content": [{"type": "text", "text": POLLUTED}], + }, + {"role": "assistant", "content": "pushing now"}, + ] + + cleaned = _strip_stale_user_merge_from_messages(messages, CURRENT_TURN, PRIOR_TAIL) + + assert cleaned[0]["content"] == CURRENT_TURN + assert cleaned[1] == messages[1] + + +def test_strip_stale_user_merge_does_not_touch_clean_rows(): + """Clean rows, assistant rows, and tool rows must pass through untouched.""" + from api.streaming import _strip_stale_user_merge_from_messages + + messages = [ + {"role": "user", "content": PRIOR_TAIL}, + {"role": "assistant", "content": "ok noted"}, + {"role": "user", "content": CURRENT_TURN}, + {"role": "tool", "content": "result", "tool_call_id": "x"}, + ] + + cleaned = _strip_stale_user_merge_from_messages(messages, CURRENT_TURN, PRIOR_TAIL) + + assert [m["content"] for m in cleaned] == [ + PRIOR_TAIL, + "ok noted", + CURRENT_TURN, + "result", + ] + + +def test_deduplicate_context_messages_cleans_polluted_current_user_in_result(): + """Result-normalization must leave no polluted current user row behind. + + This is the end-to-end assertion the spec asks for: the post-merge + `context_messages` must contain a clean current user turn, not the + stale-merged pair. + """ + from api.streaming import ( + _deduplicate_context_messages, + _dedupe_replayed_context_messages, + ) + + previous_context = [ + {"role": "user", "content": "are we ready?"}, + {"role": "assistant", "content": "almost, hold on"}, + {"role": "user", "content": PRIOR_TAIL}, + ] + result_messages = [ + *previous_context, + {"role": "user", "content": POLLUTED}, + {"role": "assistant", "content": "pushing now"}, + ] + + cleaned = _dedupe_replayed_context_messages( + previous_context, + result_messages, + CURRENT_TURN, + ) + cleaned = _deduplicate_context_messages(cleaned) + + polluted_rows = [ + m + for m in cleaned + if isinstance(m, dict) + and m.get("role") == "user" + and POLLUTED in _text(m.get("content", "")) + ] + assert not polluted_rows, ( + f"No user content should equal or start with the polluted pair; got: " + f"{[m.get('content') for m in cleaned]}" + ) + + current_rows = [ + m + for m in cleaned + if isinstance(m, dict) + and m.get("role") == "user" + and _text(m.get("content", "")).strip() == CURRENT_TURN + ] + assert current_rows, ( + f"Persisted context should contain the clean current user turn; got: " + f"{[m.get('content') for m in cleaned]}" + ) + + +def test_dedupe_replayed_context_handles_repair_replaced_tail_user_row(): + """Context merge handles repaired rows that replace the prior tail user row. + + Some repair paths return the shared prefix only through the item before the + prior user tail, then put the repair-merged current user row at the old tail + position. WebUI should preserve the old tail and append the clean current + turn, never persist the polluted joined content. + """ + from api.streaming import _dedupe_replayed_context_messages + + previous_context = [ + {"role": "user", "content": "are we ready?"}, + {"role": "assistant", "content": "almost, hold on"}, + {"role": "user", "content": PRIOR_TAIL}, + ] + result_messages = [ + previous_context[0], + previous_context[1], + {"role": "user", "content": POLLUTED}, + {"role": "assistant", "content": "pushing now"}, + ] + + cleaned = _dedupe_replayed_context_messages( + previous_context, + result_messages, + CURRENT_TURN, + ) + + assert [m.get("content") for m in cleaned] == [ + "are we ready?", + "almost, hold on", + PRIOR_TAIL, + CURRENT_TURN, + "pushing now", + ] + assert not any( + isinstance(m, dict) and POLLUTED in _text(m.get("content", "")) + for m in cleaned + ) + + +def test_merge_display_drops_polluted_current_when_eager_checkpoint_clean(): + """Display merge must not append a polluted row next to a clean eager checkpoint. + + The visible transcript should keep exactly one clean current user row and + append only the assistant response — not produce two adjacent user rows. + """ + from api.streaming import _merge_display_messages_after_agent_result + + previous_display = [ + {"role": "user", "content": "are we ready?"}, + {"role": "assistant", "content": "almost, hold on"}, + {"role": "user", "content": CURRENT_TURN}, # eager checkpoint + ] + previous_context = [ + {"role": "user", "content": "are we ready?"}, + {"role": "assistant", "content": "almost, hold on"}, + {"role": "user", "content": PRIOR_TAIL}, + ] + result_messages = [ + *previous_context, + {"role": "user", "content": POLLUTED}, # polluted merge from repair + {"role": "assistant", "content": "pushing now"}, + ] + + merged = _merge_display_messages_after_agent_result( + previous_display, + previous_context, + result_messages, + CURRENT_TURN, + ) + + user_texts = [ + _text(m.get("content", "")).strip() + for m in merged + if isinstance(m, dict) and m.get("role") == "user" + ] + assert user_texts.count(CURRENT_TURN) == 1, ( + f"Should keep exactly one clean current user row; got user rows: {user_texts}" + ) + assert not any(POLLUTED in t for t in user_texts), ( + f"Polluted merged user row must not appear in display; got: {user_texts}" + ) + assert any( + isinstance(m, dict) + and m.get("role") == "assistant" + and "pushing now" in _text(m.get("content", "")) + for m in merged + ), "Assistant response must still be appended" + + +def test_merge_display_does_not_overstrip_when_current_already_clean(): + """A legitimately new current turn that mentions the prior phrase stays intact.""" + from api.streaming import _merge_display_messages_after_agent_result + + previous_display = [ + {"role": "user", "content": "are we ready?"}, + {"role": "assistant", "content": "almost, hold on"}, + ] + previous_context = [ + {"role": "user", "content": "are we ready?"}, + {"role": "assistant", "content": "almost, hold on"}, + {"role": "user", "content": PRIOR_TAIL}, + ] + new_turn = f"re: {PRIOR_TAIL} — that helps, thanks" + result_messages = [ + *previous_context, + {"role": "user", "content": new_turn}, + {"role": "assistant", "content": "glad it did"}, + ] + + merged = _merge_display_messages_after_agent_result( + previous_display, + previous_context, + result_messages, + new_turn, + ) + + user_texts = [ + _text(m.get("content", "")).strip() + for m in merged + if isinstance(m, dict) and m.get("role") == "user" + ] + assert new_turn in user_texts, ( + f"Genuine current turn must remain intact; got: {user_texts}" + ) + assert PRIOR_TAIL in user_texts, ( + f"Original prior-tail user row must remain in display; got: {user_texts}" + ) + + +def test_merge_display_passes_through_when_prior_tail_text_differs(): + """Detector skips when the prior-tail text does not match PRIOR_TAIL — no over-strip.""" + from api.streaming import _merge_display_messages_after_agent_result + + previous_display = [ + {"role": "user", "content": "first question"}, + {"role": "assistant", "content": "first answer"}, + ] + previous_context = [ + {"role": "user", "content": "first question"}, + {"role": "assistant", "content": "first answer"}, + ] + # _last_user_row finds "first question" as the prior tail, but its + # normalized text does not match PRIOR_TAIL, so _detect_stale_user_merge + # correctly returns False and the polluted row is not cleaned. + polluted_msg = {"role": "user", "content": POLLUTED} + result_messages = [ + *previous_context, + polluted_msg, + {"role": "assistant", "content": "answer"}, + ] + + merged = _merge_display_messages_after_agent_result( + previous_display, + previous_context, + result_messages, + CURRENT_TURN, + ) + + user_texts = [ + _text(m.get("content", "")).strip() + for m in merged + if isinstance(m, dict) and m.get("role") == "user" + ] + assert CURRENT_TURN in user_texts, ( + f"Current turn must be normalized to the clean text; got: {user_texts}" + ) + assert PRIOR_TAIL not in user_texts, ( + f"Unmatched prior-tail text must not be synthesized as its own user row; got: {user_texts}" + ) + + +def test_merge_display_workspace_prefixed_polluted_row_is_cleaned(): + """The polluted row may arrive with a workspace sentinel; cleaning still works.""" + from api.streaming import _merge_display_messages_after_agent_result + + previous_display = [ + {"role": "user", "content": "are we ready?"}, + {"role": "assistant", "content": "almost, hold on"}, + ] + previous_context = [ + {"role": "user", "content": "are we ready?"}, + {"role": "assistant", "content": "almost, hold on"}, + {"role": "user", "content": PRIOR_TAIL}, + ] + prefixed_polluted = { + "role": "user", + "content": f"[Workspace::v1: /tmp/project]\n{POLLUTED}", + } + result_messages = [ + *previous_context, + prefixed_polluted, + {"role": "assistant", "content": "pushing now"}, + ] + + merged = _merge_display_messages_after_agent_result( + previous_display, + previous_context, + result_messages, + CURRENT_TURN, + ) + + user_texts = [ + _text(m.get("content", "")).strip() + for m in merged + if isinstance(m, dict) and m.get("role") == "user" + ] + assert user_texts.count(CURRENT_TURN) == 1, ( + f"Should keep exactly one clean current user row; got: {user_texts}" + ) + assert not any(POLLUTED in t for t in user_texts), ( + f"Polluted row must be normalized away; got: {user_texts}" + ) + + +def test_dedupe_replayed_context_preserves_historical_row_with_merge_shape(): + """A historical user row shaped like \\n\\n must not be rewritten. + + If a prior conversation turn happens to have content exactly matching the + stale-merge pattern, _dedupe_replayed_context_messages must not rewrite it: + only the new-turn boundary/candidate slice is eligible for stale-merge cleanup. + """ + from api.streaming import _dedupe_replayed_context_messages + + HISTORICAL_SHAPE = f"{PRIOR_TAIL}\n\n{CURRENT_TURN}" + + previous_context = [ + {"role": "user", "content": HISTORICAL_SHAPE}, # legitimate old turn + {"role": "assistant", "content": "summary from that turn"}, + {"role": "user", "content": PRIOR_TAIL}, # becomes previous_user_tail + ] + result_messages = [ + *previous_context, + {"role": "user", "content": CURRENT_TURN}, # clean current turn + {"role": "assistant", "content": "new response"}, + ] + + cleaned = _dedupe_replayed_context_messages( + previous_context, + result_messages, + CURRENT_TURN, + ) + + assert any( + isinstance(m, dict) and m.get("content") == HISTORICAL_SHAPE + for m in cleaned + ), ( + "Historical user row shaped like the stale merge must remain unchanged; " + f"got: {[m.get('content') for m in cleaned]}" + ) + assert any( + isinstance(m, dict) and m.get("role") == "user" and m.get("content") == CURRENT_TURN + for m in cleaned + ), ( + "Clean current user turn must appear in returned context; " + f"got: {[m.get('content') for m in cleaned]}" + ) + + +def test_stale_tail_candidate_returns_normalized_text(): + """_stale_user_tail_candidate normalizes whitespace and strips workspace prefix.""" + from api.streaming import _stale_user_tail_candidate + + msg = { + "role": "user", + "content": "[Workspace::v1: /tmp/project]\n please use the larger context model ", + } + assert _stale_user_tail_candidate(msg) == "please use the larger context model" + + non_user = {"role": "assistant", "content": "please use the larger context model"} + assert _stale_user_tail_candidate(non_user) is None + + empty = {"role": "user", "content": ""} + assert _stale_user_tail_candidate(empty) is None + + +def test_last_user_row_returns_trailing_user_message(): + """_last_user_row returns the most recent user message in the list.""" + from api.streaming import _last_user_row + + messages = [ + {"role": "user", "content": "first"}, + {"role": "assistant", "content": "ok"}, + {"role": "user", "content": "second"}, + ] + assert _last_user_row(messages) == {"role": "user", "content": "second"} + assert _last_user_row([]) is None + assert _last_user_row([{"role": "assistant", "content": "no user"}]) is None + + +# --- helpers ---------------------------------------------------------------- + + +def _stale_tail_for(messages): + """Re-derive the normalized prior-tail text used by the production call sites.""" + from api.streaming import _last_user_row, _stale_user_tail_candidate + + return _stale_user_tail_candidate(_last_user_row(messages)) + + +if __name__ == "__main__": # pragma: no cover - allow `python tests/...py` + raise SystemExit(pytest.main([__file__, "-q"]))