From 4a7b851c8cf2500f21956c0dd8aaffa441891ae6 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 27 Jul 2026 17:01:34 -0400 Subject: [PATCH] fix(#6481): materialize pending turns without borrowed display indexes --- api/streaming.py | 166 +++++++++++------- ...1_terminal_failure_transcript_evaluator.py | 11 +- ...issue6481_verification_evidence_phantom.py | 43 ++++- tests/test_sprint42.py | 2 +- tests/test_stale_stream_writeback.py | 4 +- tests/test_tool_limit_terminal_state.py | 95 +++++++++- 6 files changed, 251 insertions(+), 70 deletions(-) diff --git a/api/streaming.py b/api/streaming.py index 3fd6ec50..f9dc14eb 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -1550,29 +1550,14 @@ def _mark_active_turn_checkpoint_in_history(messages, identity, msg_text, *, all return messages, True -def _display_row_can_accept_context_index(display, context, identity, msg_text): - if not isinstance(identity, dict) or not _active_turn_boundary_is_valid(identity): - return False - idx = identity['current_turn_user_idx'] - if idx < 0 or idx >= len(display) or idx >= len(context): - return False - expected_text = identity.get('text') if identity.get('text') is not None else msg_text - display_row = display[idx] - context_row = context[idx] - if ( - not isinstance(display_row, dict) - or not isinstance(context_row, dict) - or display_row.get('role') != 'user' - or context_row.get('role') != 'user' - or _normalize_user_text(display_row.get('content')) != _normalize_user_text(expected_text) - or _normalize_user_text(context_row.get('content')) != _normalize_user_text(expected_text) - ): - return False - return ( - display_row.get('timestamp') == context_row.get('timestamp') - and (display_row.get('_source') or 'webui') == (context_row.get('_source') or 'webui') - and copy.deepcopy(display_row.get('attachments') or []) == copy.deepcopy(context_row.get('attachments') or []) - ) +def _owner_projection_current_turn_row(messages, identity): + messages = list(messages or []) + if not isinstance(identity, dict): + return None + for message in messages: + if _active_turn_token_matches(message, identity): + return copy.deepcopy(message) + return None def _find_active_turn_checkpoint_index(result_messages, previous_context, identity, msg_text): @@ -1685,18 +1670,20 @@ def _align_current_turn_display(previous_display, previous_context, identity): identity.get('text'), allow_index_fallback=False, ) - if ( - not _active_turn_has_checkpoint(display, identity) - and _display_row_can_accept_context_index(display, context, identity, identity.get('text')) - ): - _mark_active_turn_checkpoint(display[identity['current_turn_user_idx']], identity) if _active_turn_has_checkpoint(display, identity): return display, context - checkpoint = next( - (message for message in context if _active_turn_token_matches(message, identity)), - None, + checkpoint = _owner_projection_current_turn_row( + context, + identity, ) - if checkpoint is not None: + if checkpoint is None and identity.get('token'): + checkpoint = _materialize_active_turn_user( + identity, + identity.get('text'), + identity.get('source') or 'webui', + ) + context.append(copy.deepcopy(checkpoint)) + if checkpoint is not None and not _active_turn_has_checkpoint(display, identity): display.append(copy.deepcopy(checkpoint)) return display, context @@ -1714,7 +1701,13 @@ def _prepare_marker_clean_writeback( 'verification_nudge_seen': has_verification_nudge, 'active_turn_identity': copy.deepcopy(active_turn_identity), } - if cleaned or not result_messages: + if not result_messages: + return ( + cleaned, + list(previous_context_messages or []), + provenance, + ) + if cleaned: return ( cleaned, _restore_reasoning_metadata(previous_context_messages, cleaned), @@ -1759,7 +1752,11 @@ def _settle_result_messages( msg_text, source, ) - session.context_messages = _deduplicate_context_messages(next_context_messages) + session.context_messages = ( + _deduplicate_context_messages(next_context_messages) + if result_messages + else list(next_context_messages or []) + ) if result_messages: session.context_messages = _settle_current_turn_boundary( previous_context_messages, @@ -5031,6 +5028,7 @@ def _deduplicate_context_messages(messages): return messages seen = set() deduped = [] + user_exact_index = {} for msg in messages: if _is_context_compression_marker(msg): marker_key = ( @@ -5049,6 +5047,26 @@ def _deduplicate_context_messages(messages): deduped.append(msg) continue key = _message_identity(msg) + if isinstance(msg, dict) and msg.get('role') == 'user' and key is not None: + user_exact_key = ( + key, + msg.get('timestamp'), + msg.get('_ts'), + msg.get('_source') or 'webui', + json.dumps(msg.get('attachments') or [], sort_keys=True, ensure_ascii=False), + ) + prior_exact_idx = user_exact_index.get(user_exact_key) + if msg.get('_active_turn_token'): + if prior_exact_idx is not None: + deduped[prior_exact_idx] = msg + continue + if key in seen: + deduped.append(msg) + user_exact_index[user_exact_key] = len(deduped) - 1 + continue + elif prior_exact_idx is not None: + continue + user_exact_index[user_exact_key] = len(deduped) if key is not None and key in seen: continue if key is not None: @@ -6142,25 +6160,11 @@ def _merge_display_messages_after_agent_result( for message in result_messages ) _active_turn_identity = _writeback_provenance.get('active_turn_identity') - previous_display, _ = _mark_active_turn_checkpoint_in_history( + previous_display, _ = _align_current_turn_display( previous_display, + previous_context, _active_turn_identity, - msg_text, - allow_index_fallback=False, ) - if ( - not _active_turn_has_checkpoint(previous_display, _active_turn_identity) - and _display_row_can_accept_context_index( - previous_display, - previous_context, - _active_turn_identity, - msg_text, - ) - ): - _mark_active_turn_checkpoint( - previous_display[_active_turn_identity['current_turn_user_idx']], - _active_turn_identity, - ) # Same marker filter for the model-history inputs: the synthetic verify-loop # answer/nudge live in the agent's returned messages and prior context, and # would otherwise slip into the merged transcript as a real delta. (#5334) @@ -6508,12 +6512,25 @@ def _turn_transcript_lacks_final_assistant_answer( msg_text, source: str = "webui", drop_replayed_assistant: bool = False, + active_turn_identity=None, ) -> bool: """Return True when an already-merged transcript still lacks a final assistant answer.""" merged_messages = list(merged_messages or []) previous_display = list(previous_display or []) - current_user_idx = _find_current_user_turn(merged_messages, msg_text) - if current_user_idx is None or current_user_idx < len(previous_display): + current_user_token_idx = next( + ( + idx for idx, message in enumerate(merged_messages) + if _active_turn_token_matches(message, active_turn_identity) + ), + None, + ) + current_user_idx = current_user_token_idx + if current_user_idx is None: + current_user_idx = _find_current_user_turn(merged_messages, msg_text) + if current_user_idx is None or ( + current_user_token_idx is None + and current_user_idx < len(previous_display) + ): # The active turn lives after the durable transcript boundary. If the # merged display only exposes an older user row, materialize the pending # prompt so a replayed assistant row cannot satisfy the wrong turn. @@ -6560,15 +6577,26 @@ def _merged_transcript_lacks_final_assistant_answer( msg_text, source: str = "webui", drop_replayed_assistant: bool = False, + active_turn_identity=None, ) -> bool: """Return True when the current turn still lacks a final assistant answer.""" previous_display = list(previous_display or []) + _verification_nudge_seen = any( + isinstance(message, dict) + and message.get('role') == 'user' + and _is_synthetic_control_message(message) + for message in result_messages or [] + ) merged_messages = _merge_display_messages_after_agent_result( previous_display, previous_context, _restore_reasoning_metadata(previous_display, result_messages), msg_text, source=source, + verification_nudge_provenance={ + 'verification_nudge_seen': _verification_nudge_seen, + 'active_turn_identity': active_turn_identity, + }, ) return _turn_transcript_lacks_final_assistant_answer( merged_messages, @@ -6576,6 +6604,7 @@ def _merged_transcript_lacks_final_assistant_answer( msg_text, source=source, drop_replayed_assistant=drop_replayed_assistant, + active_turn_identity=active_turn_identity, ) @@ -9440,12 +9469,18 @@ def _run_agent_streaming( state_messages=_external_state_messages, ) or [] ) - _previous_context_messages = _new_turn_context_from_messages( + _previous_owner_context_messages = list( reconciled_state_db_messages_for_session( s, prefer_context=True, state_messages=_external_state_messages, - ), + ) or [] + ) + _previous_owner_context_messages = _deduplicate_context_messages( + _previous_owner_context_messages + ) + _previous_context_messages = _new_turn_context_from_messages( + _previous_owner_context_messages, msg_text, ) # Dedup before feeding to agent — merge_session_messages_append_only @@ -9650,7 +9685,9 @@ def _run_agent_streaming( return with _stream_writeback_stage(_writeback_timings, "merge_result"): _tool_limit_reached = _agent_result_tool_limit_reached(result) - _result_messages = result.get('messages') or _previous_context_messages + _result_messages = result.get('messages') + if _result_messages is None: + _result_messages = _previous_context_messages _result_messages = _drop_synthetic_max_iteration_summary_requests( _result_messages, enabled=_tool_limit_reached, @@ -9692,7 +9729,7 @@ def _run_agent_streaming( _result_messages = _settle_result_messages( s, _previous_messages, - _previous_context_messages, + _previous_owner_context_messages, _result_messages, msg_text, _turn_pending_source, @@ -9872,18 +9909,19 @@ def _run_agent_streaming( ) _saved_transcript_lacks_final_answer = _merged_transcript_lacks_final_assistant_answer( _previous_messages, - _previous_context_messages, + _previous_owner_context_messages, _all_result_messages, msg_text, source=getattr(s, 'pending_user_source', None) or 'webui', drop_replayed_assistant=_drop_replayed_assistant, + active_turn_identity=_active_turn_identity, ) if ( not _all_result_messages and _current_turn_already_has_visible_assistant_answer( _align_current_turn_display( _previous_messages, - _previous_context_messages, + _previous_owner_context_messages, _active_turn_identity, )[0], active_turn_identity=_active_turn_identity, @@ -10029,7 +10067,9 @@ def _run_agent_streaming( # evaluates False on next conceptual pass. # Since we're in a flat block, directly run the # post-result merge logic here. - _result_messages = result.get('messages') or _previous_context_messages + _result_messages = result.get('messages') + if _result_messages is None: + _result_messages = _previous_context_messages _result_messages = _drop_synthetic_max_iteration_summary_requests( _result_messages, enabled=_agent_result_tool_limit_reached(result), @@ -10037,7 +10077,7 @@ def _run_agent_streaming( _result_messages = _settle_result_messages( s, _previous_messages, - _previous_context_messages, + _previous_owner_context_messages, _result_messages, msg_text, _turn_pending_source, @@ -11247,11 +11287,13 @@ def _run_agent_streaming( getattr(s, 'active_stream_id', None), ) return - _result_messages = _heal_result.get('messages') or _previous_context_messages + _result_messages = _heal_result.get('messages') + if _result_messages is None: + _result_messages = _previous_context_messages _result_messages = _settle_result_messages( s, _previous_messages, - _previous_context_messages, + _previous_owner_context_messages, _result_messages, msg_text, _turn_pending_source, diff --git a/tests/test_issue5141_terminal_failure_transcript_evaluator.py b/tests/test_issue5141_terminal_failure_transcript_evaluator.py index 48da1e72..d326f8c7 100644 --- a/tests/test_issue5141_terminal_failure_transcript_evaluator.py +++ b/tests/test_issue5141_terminal_failure_transcript_evaluator.py @@ -93,7 +93,14 @@ def test_turn_evaluator_materializes_pending_user_after_display_boundary(): def test_merged_wrapper_delegates_to_turn_evaluator(): calls = [] - def _fake_evaluator(merged_messages, previous_display, msg_text, source="webui", drop_replayed_assistant=False): + def _fake_evaluator( + merged_messages, + previous_display, + msg_text, + source="webui", + drop_replayed_assistant=False, + active_turn_identity=None, + ): calls.append( { "merged_len": len(list(merged_messages or [])), @@ -101,6 +108,7 @@ def test_merged_wrapper_delegates_to_turn_evaluator(): "msg_text": msg_text, "source": source, "drop_replayed_assistant": drop_replayed_assistant, + "active_turn_identity": active_turn_identity, } ) return True @@ -125,4 +133,5 @@ def test_merged_wrapper_delegates_to_turn_evaluator(): assert calls[0]["msg_text"] == "hello" assert calls[0]["source"] == "cli" assert calls[0]["drop_replayed_assistant"] is True + assert calls[0]["active_turn_identity"] is None assert calls[0]["merged_len"] >= 1 diff --git a/tests/test_issue6481_verification_evidence_phantom.py b/tests/test_issue6481_verification_evidence_phantom.py index 0820f28d..679bb2fa 100644 --- a/tests/test_issue6481_verification_evidence_phantom.py +++ b/tests/test_issue6481_verification_evidence_phantom.py @@ -17,7 +17,11 @@ VERIFICATION_EVIDENCE = { def _current_turn_prefix(*, include_attempted_answer): messages = [ - {"role": "user", "content": "Fix the failing test."}, + { + "role": "user", + "content": "Fix the failing test.", + "_active_turn_token": "stream:turn", + }, { "role": "assistant", "content": [{"type": "tool_use", "name": "terminal", "input": {"cmd": "pytest"}}], @@ -279,6 +283,7 @@ def test_divergent_context_index_does_not_relabel_same_text_display_row(): {"role": "assistant", "content": "The earlier attempt is complete."}, ] previous_context = [ + {"role": "user", "content": prompt, "timestamp": 1.0, "attachments": []}, { "role": "user", "content": prompt, @@ -308,6 +313,42 @@ def test_divergent_context_index_does_not_relabel_same_text_display_row(): assert aligned_display[-1]["_active_turn_token"] == token +def test_missing_exact_checkpoint_materializes_current_turn_without_retokening_history(): + prompt = "Fix the failing test." + token = "direct-stream:1.9" + previous_display = [ + {"role": "user", "content": prompt, "timestamp": 1.1}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + previous_context = [ + {"role": "user", "content": prompt, "timestamp": 1.1, "attachments": []}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + + aligned_display, aligned_context = streaming._align_current_turn_display( + previous_display, + previous_context, + _writeback_provenance( + prompt=prompt, + timestamp=1.9, + current_turn_user_idx=0, + token=token, + )["active_turn_identity"], + ) + + expected = [ + ("user", prompt), + ("assistant", "The earlier attempt is complete."), + ("user", prompt), + ] + assert _role_content_sequence(aligned_display) == expected + assert _role_content_sequence(aligned_context) == expected + assert aligned_display[0].get("_active_turn_token") is None + assert aligned_context[0].get("_active_turn_token") is None + assert aligned_display[-1]["_active_turn_token"] == token + assert aligned_context[-1]["_active_turn_token"] == token + + def test_shared_settlement_owner_contract(): prompt = "Fix the failing test." previous = [ diff --git a/tests/test_sprint42.py b/tests/test_sprint42.py index f7535eee..c629e0f6 100644 --- a/tests/test_sprint42.py +++ b/tests/test_sprint42.py @@ -813,7 +813,7 @@ def test_streaming_restores_prior_reasoning_metadata_after_followup(): src = (REPO / 'api' / 'streaming.py').read_text(encoding="utf-8") assert "def _restore_reasoning_metadata(" in src, \ "streaming.py must define a helper to restore prior reasoning metadata" - assert "next_context_messages" in src and "session.context_messages = _deduplicate_context_messages(next_context_messages)" in src, \ + assert "next_context_messages" in src and "_deduplicate_context_messages(next_context_messages)" in src, \ "streaming.py must restore prior reasoning metadata into model context" assert "session.messages = _merge_display_messages_after_agent_result(" in src, \ "streaming.py must merge restored result messages into the visible transcript" diff --git a/tests/test_stale_stream_writeback.py b/tests/test_stale_stream_writeback.py index d071dc28..786b895c 100644 --- a/tests/test_stale_stream_writeback.py +++ b/tests/test_stale_stream_writeback.py @@ -174,7 +174,7 @@ def test_success_path_checks_stream_ownership_before_persisting_result(): src = Path("api/streaming.py").read_text(encoding="utf-8") guard = "if not ephemeral and not _stream_writeback_is_current(s, stream_id):" guard_pos = src.find(guard) - result_merge_pos = src.find("_result_messages = result.get('messages') or _previous_context_messages") + result_merge_pos = src.find("_result_messages = result.get('messages')") compression_pos = src.find("Handle context compression side effects") assert guard_pos != -1 @@ -192,7 +192,7 @@ def test_self_heal_retry_success_checks_stream_ownership_before_writeback(): guard = "if not ephemeral and not _stream_writeback_is_current(s, stream_id):" assert guard in block - assert block.index(guard) < block.index("_result_messages = _heal_result.get('messages') or _previous_context_messages") + assert block.index(guard) < block.index("_result_messages = _heal_result.get('messages')") assert block.index(guard) < block.index("s.save()") diff --git a/tests/test_tool_limit_terminal_state.py b/tests/test_tool_limit_terminal_state.py index 1c03289c..b5b10d05 100644 --- a/tests/test_tool_limit_terminal_state.py +++ b/tests/test_tool_limit_terminal_state.py @@ -645,7 +645,11 @@ def test_missing_agent_authority_fails_closed(tmp_path, monkeypatch): @pytest.mark.parametrize("marker", ["_verification_stop_synthetic", "_pre_verify_synthetic"]) def test_synthetic_only_verification_delta_preserves_prior_histories(tmp_path, monkeypatch, marker): prior_turn = [ - {"role": "user", "content": "Fix the failing test."}, + { + "role": "user", + "content": "Fix the failing test.", + "_active_turn_token": build_active_turn_token("stream-tool-limit", 1.0), + }, {"role": "assistant", "content": [{"type": "tool_use", "name": "terminal"}]}, { "role": "tool", @@ -945,6 +949,46 @@ def test_streaming_empty_result_same_text_does_not_treat_prior_assistant_as_curr assert payload["messages"][-1]["_error"] is True +def test_streaming_empty_result_without_exact_checkpoint_materializes_current_user( + tmp_path, + monkeypatch, +): + prompt = "Fix the failing test." + prior = [ + {"role": "user", "content": prompt, "timestamp": 1.1}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + result = {"messages": []} + + events, payload = _run_streaming_with_fake_agent( + tmp_path, + monkeypatch, + result, + prior_messages=prior, + prior_context_messages=prior, + msg_text=prompt, + pending_started_at=1.9, + current_turn_user_idx=0, + ) + + apperror_payloads = [event_payload for event, event_payload in events if event == "apperror"] + assert apperror_payloads, "expected silent-failure apperror" + assert apperror_payloads[-1]["type"] == "no_response" + assert not [event_payload for event, event_payload in events if event == "done"] + assert [(message["role"], message.get("content")) for message in payload["messages"][:-1]] == [ + ("user", prompt), + ("assistant", "The earlier attempt is complete."), + ("user", prompt), + ] + assert payload["messages"][0].get("_active_turn_token") is None + assert payload["messages"][2]["_active_turn_token"] == build_active_turn_token( + "stream-tool-limit", + 1.9, + ) + assert payload["messages"][-1]["role"] == "assistant" + assert payload["messages"][-1]["_error"] is True + + def test_streaming_empty_result_divergent_context_index_does_not_reuse_historical_assistant( tmp_path, monkeypatch, @@ -955,7 +999,8 @@ def test_streaming_empty_result_divergent_context_index_does_not_reuse_historica {"role": "user", "content": prompt, "timestamp": 1.0}, {"role": "assistant", "content": "The earlier attempt is complete."}, ] - context_only_current = [ + prior_context = [ + {"role": "user", "content": prompt, "timestamp": 1.0, "attachments": []}, { "role": "user", "content": prompt, @@ -971,7 +1016,7 @@ def test_streaming_empty_result_divergent_context_index_does_not_reuse_historica monkeypatch, result, prior_messages=prior, - prior_context_messages=context_only_current, + prior_context_messages=prior_context, msg_text=prompt, pending_started_at=1.9, current_turn_user_idx=0, @@ -989,3 +1034,47 @@ def test_streaming_empty_result_divergent_context_index_does_not_reuse_historica assert payload["messages"][0].get("_active_turn_token") is None assert payload["messages"][-1]["role"] == "assistant" assert payload["messages"][-1]["_error"] is True + + +def test_streaming_tool_limit_empty_result_without_exact_checkpoint_materializes_current_user( + tmp_path, + monkeypatch, +): + prompt = "Fix the failing test." + prior = [ + {"role": "user", "content": prompt, "timestamp": 1.1}, + {"role": "assistant", "content": "The earlier attempt is complete."}, + ] + result = { + "turn_exit_reason": "max_iterations_reached(30/30)", + "messages": [], + } + + events, payload = _run_streaming_with_fake_agent( + tmp_path, + monkeypatch, + result, + prior_messages=prior, + prior_context_messages=prior, + msg_text=prompt, + pending_started_at=1.9, + current_turn_user_idx=0, + ) + + apperror_payloads = [event_payload for event, event_payload in events if event == "apperror"] + assert apperror_payloads, "expected tool-limit apperror" + assert apperror_payloads[-1]["type"] == "tool_limit_reached" + assert apperror_payloads[-1]["terminal_state"] == "tool_limit_reached" + assert not [event_payload for event, event_payload in events if event == "done"] + assert [(message["role"], message.get("content")) for message in payload["messages"][:-1]] == [ + ("user", prompt), + ("assistant", "The earlier attempt is complete."), + ("user", prompt), + ] + assert payload["messages"][0].get("_active_turn_token") is None + assert payload["messages"][2]["_active_turn_token"] == build_active_turn_token( + "stream-tool-limit", + 1.9, + ) + assert payload["messages"][-1]["role"] == "assistant" + assert payload["messages"][-1]["_error"] is True