From 525edef5f774dbcd92a69bb1b196b9cded894251 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 27 Jul 2026 07:30:53 -0400 Subject: [PATCH] Fix silent-result ownership for repeated prompts --- api/streaming.py | 37 ++++++++----------- .../test_issue4729_reasoning_sse_coalesce.py | 14 +++++-- tests/test_issue765_streaming_persistence.py | 12 +++--- tests/test_run_journal_streaming_static.py | 11 ++++-- tests/test_sprint42.py | 4 +- tests/test_tool_limit_terminal_state.py | 37 ++++++++++++++++++- 6 files changed, 77 insertions(+), 38 deletions(-) diff --git a/api/streaming.py b/api/streaming.py index 1b59a741..4691c2a9 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -1753,22 +1753,19 @@ def _settle_result_messages( return result_messages -def _current_turn_already_has_visible_assistant_answer(messages, *, current_user_text=None): - """Return True when the current real user turn already has visible assistant prose.""" - if current_user_text is not None: - current_user_row = {'role': 'user', 'content': current_user_text} - current_user_tail = _stale_user_tail_candidate(current_user_row) - if current_user_tail: - last_real_user = None - for msg in reversed(list(messages or [])): - if isinstance(msg, dict) and not _is_synthetic_control_message(msg) and msg.get('role') == 'user': - last_real_user = msg - break - if _stale_user_tail_candidate(last_real_user) != current_user_tail: - return False - for msg in reversed(list(messages or [])): +def _current_turn_already_has_visible_assistant_answer(messages, *, active_turn_identity=None): + """Return True only when the token-owned current turn already has visible assistant prose.""" + if not isinstance(active_turn_identity, dict) or not active_turn_identity.get('token'): + return False + current_turn_seen = False + for msg in messages or []: if not isinstance(msg, dict) or _is_synthetic_control_message(msg): continue + if _active_turn_token_matches(msg, active_turn_identity): + current_turn_seen = True + continue + if not current_turn_seen: + continue role = msg.get('role') if role == 'assistant' and _assistant_message_has_final_visible_text(msg) and not msg.get('_error'): return True @@ -9836,15 +9833,13 @@ def _run_agent_streaming( ) if ( not _all_result_messages - and ( - _current_turn_already_has_visible_assistant_answer( + and _current_turn_already_has_visible_assistant_answer( + _align_current_turn_display( _previous_messages, - current_user_text=msg_text, - ) - or _current_turn_already_has_visible_assistant_answer( _previous_context_messages, - current_user_text=msg_text, - ) + _active_turn_identity, + )[0], + active_turn_identity=_active_turn_identity, ) ): _saved_transcript_lacks_final_answer = False diff --git a/tests/test_issue4729_reasoning_sse_coalesce.py b/tests/test_issue4729_reasoning_sse_coalesce.py index 87cabbb6..05e69ce8 100644 --- a/tests/test_issue4729_reasoning_sse_coalesce.py +++ b/tests/test_issue4729_reasoning_sse_coalesce.py @@ -97,10 +97,16 @@ def test_reasoning_buffer_flushed_at_every_boundary(): # terminal catch-all: flush right after the agent run returns, AND a guaranteed-exit # flush in the inner finally so exception / retry paths (which bypass the post-run # flush) still emit the tail before the outer apperror. - assert re.search( - r"agent\.run_conversation\(\*\*_run_conversation_kwargs\)\n\s*# #4729", - STREAMING, - ), "must flush after agent.run_conversation() returns (success path, before done)" + run_idx = STREAMING.index("result = agent.run_conversation(**_run_conversation_kwargs)") + resolve_idx = STREAMING.index( + "_active_turn_identity = _resolve_active_turn_authority(", + run_idx, + ) + flush_idx = STREAMING.index("_flush_reasoning_buffer()", resolve_idx) + assert run_idx < resolve_idx < flush_idx, ( + "must flush after agent.run_conversation() returns, after turn authority is resolved, " + "and before terminal SSE delivery" + ) # the inner finally must also flush (covers exception/retry exits) fin = STREAMING[STREAMING.index(" finally:\n # #4729"):] fin = fin[:700] diff --git a/tests/test_issue765_streaming_persistence.py b/tests/test_issue765_streaming_persistence.py index 36935f0d..c3689b69 100644 --- a/tests/test_issue765_streaming_persistence.py +++ b/tests/test_issue765_streaming_persistence.py @@ -314,10 +314,10 @@ class TestIssue765FollowupHardening: t2.join(timeout=5) assert not errors, f"Concurrent same-session saves should not fail: {errors}" - assert len(replace_sources) == 2, f"Expected 2 replace calls, got {replace_sources}" + assert len(replace_sources) >= 2, f"Expected replace calls, got {replace_sources}" assert len(set(replace_sources)) == 2, ( - "Concurrent same-session saves must use distinct temp files; " - f"got {replace_sources}" + "Concurrent same-session saves must use distinct temp files even if Windows-safe " + f"replace retries one of them; got {replace_sources}" ) data = json.loads(s.path.read_text(encoding="utf-8")) assert data["session_id"] == "same_sid" @@ -337,14 +337,14 @@ class TestIssue765FollowupHardening: "with _agent_lock:\n" " if not ephemeral and not _stream_writeback_is_current(s, stream_id):" ) - save_idx = src.find("_deduplicate_context_messages(_next_context_messages)") + save_idx = src.find("_result_messages = _settle_result_messages(") assert stop_idx != -1, "Success path must stop the checkpoint thread" assert join_idx != -1, "Success path must join the checkpoint thread" assert lock_idx != -1, "Success path must serialize mutation with _agent_lock" - assert save_idx != -1, "Success path restore/mutation block not found" + assert save_idx != -1, "Success path settlement block not found" assert stop_idx < join_idx < lock_idx <= save_idx, ( - "Checkpoint stop/join must happen before the success-path session mutation block" + "Checkpoint stop/join must happen before the success-path settlement block" ) def test_silent_failure_path_does_not_reacquire_agent_lock(self): diff --git a/tests/test_run_journal_streaming_static.py b/tests/test_run_journal_streaming_static.py index c61ba1b8..b2c127bb 100644 --- a/tests/test_run_journal_streaming_static.py +++ b/tests/test_run_journal_streaming_static.py @@ -29,11 +29,14 @@ def test_streaming_journals_sse_events_before_queue_delivery(): def test_streaming_compacts_all_successful_agent_result_writebacks(): src = Path("api/streaming.py").read_text(encoding="utf-8") run_src = src[src.index("def _run_agent_streaming("):] + settle_src = src[src.index("def _settle_result_messages("):] + settle_src = settle_src[: settle_src.index("\n\ndef _current_turn_already_has_visible_assistant_answer(")] - # Normal completion plus both credential self-heal retry-success paths must - # each compact after committing their own result shape. The retries use - # different local variable names, so guard the durable invariant itself. - assert run_src.count("_compact_session_image_parts_for_persistence(s)") == 3 + # Normal completion plus both credential self-heal retry-success paths now + # route through one shared settlement helper, which owns the compaction. + assert "def _settle_result_messages(" in src + assert "_compact_session_image_parts_for_persistence(session)" in settle_src + assert run_src.count("_settle_result_messages(") == 3 def test_visible_process_echo_compare_ignores_all_whitespace(): diff --git a/tests/test_sprint42.py b/tests/test_sprint42.py index a0930603..f7535eee 100644 --- a/tests/test_sprint42.py +++ b/tests/test_sprint42.py @@ -813,9 +813,9 @@ 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 "s.context_messages" in src, \ + assert "next_context_messages" in src and "session.context_messages = _deduplicate_context_messages(next_context_messages)" in src, \ "streaming.py must restore prior reasoning metadata into model context" - assert "s.messages = _merge_display_messages_after_agent_result(" in src, \ + assert "session.messages = _merge_display_messages_after_agent_result(" in src, \ "streaming.py must merge restored result messages into the visible transcript" assert "updated_messages.insert(safe_pos, copy.deepcopy(prev_msg))" in src, \ "streaming.py must reinsert dropped reasoning-only assistant messages" diff --git a/tests/test_tool_limit_terminal_state.py b/tests/test_tool_limit_terminal_state.py index abcab3f2..a54122c6 100644 --- a/tests/test_tool_limit_terminal_state.py +++ b/tests/test_tool_limit_terminal_state.py @@ -569,7 +569,7 @@ def test_tool_limit_exit_uses_shared_settlement_owner(tmp_path, monkeypatch, mar @pytest.mark.parametrize("marker", ["_verification_stop_synthetic", "_pre_verify_synthetic"]) -def test_exception_exit_uses_shared_settlement_owner(tmp_path, monkeypatch, marker): +def test_shared_settlement_owner_handles_exception_style_delta(tmp_path, monkeypatch, marker): prompt = "Fix the failing test." old_turn = [ {"role": "user", "content": prompt, "timestamp": 1.1}, @@ -907,3 +907,38 @@ def test_streaming_empty_result_messages_do_not_treat_prior_assistant_as_current assert apperror_payloads[-1]["type"] == "no_response" assert not [payload for event, payload in events if event == "done"] assert payload["messages"][-1]["_error"] is True + + +def test_streaming_empty_result_same_text_does_not_treat_prior_assistant_as_current_answer( + 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=len(prior), + ) + + 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"][-1]["role"] == "assistant" + assert payload["messages"][-1]["_error"] is True