Fix silent-result ownership for repeated prompts

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

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]

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):

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():

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"

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