fix(streaming): preserve terminal failure precedence

api/streaming.py

@@ -2597,8 +2597,10 @@ def _assistant_content_part_is_tool_use(part) -> bool:
         return True
     if part_type:
         return False
-    return any(part.get(key) for key in ('tool_use_id', 'tool_call_id', 'call_id')) and bool(
-        part.get('name') or part.get('tool_name') or part.get('input') or part.get('args')
+    if _message_content_part_text(part).strip():
+        return False
+    return any(key in part for key in ('tool_use_id', 'tool_call_id', 'call_id')) and any(
+        key in part for key in ('name', 'tool_name', 'input', 'args')
     )
 
 
@@ -2612,8 +2614,12 @@ def _assistant_message_has_final_visible_text(message) -> bool:
         for idx, part in enumerate(content):
             if _assistant_content_part_is_tool_use(part):
                 last_tool_idx = idx
-        tail_parts = content[last_tool_idx + 1:] if last_tool_idx >= 0 else content
-        return bool(_message_text(tail_parts).strip())
+        if last_tool_idx >= 0:
+            tail_parts = content[last_tool_idx + 1:]
+            return bool(_message_text(tail_parts).strip())
+        if message.get('tool_calls'):
+            return False
+        return bool(_message_text(content).strip())
     if message.get('tool_calls'):
         return False
     return bool(_message_text(content).strip())
@@ -8794,7 +8800,8 @@ def _run_agent_streaming(
                 # the result/agent, use the captured message so the classifier can
                 # surface the real cause (model_not_found / auth) instead of the
                 # misleading no_response "silent rate limit, try again" fallback.
-                if not _last_err and _captured_terminal_error[0]:
+                _captured_terminal_failure = bool(_captured_terminal_error[0])
+                if not _last_err and _captured_terminal_failure:
                     _last_err = _captured_terminal_error[0]
                 _classification = _classify_provider_error(
                     str(_last_err) if _last_err else '',
@@ -8804,7 +8811,8 @@ def _run_agent_streaming(
                 _is_quota = _classification['type'] == 'quota_exhausted'
                 _is_auth = _classification['type'] == 'auth_mismatch'
                 _drop_replayed_assistant = (
-                    _agent_result_terminal_failure(result)
+                    _captured_terminal_failure
+                    or _agent_result_terminal_failure(result)
                     or bool(getattr(agent, '_last_error', None))
                     or ('error' in result and result.get('error') is not None)
                 )
@@ -8818,7 +8826,8 @@ def _run_agent_streaming(
                 )
                 _is_agent_result_terminal = _agent_result_terminal_failure(result)
                 _terminal_failure = (
-                    _is_agent_result_terminal
+                    _captured_terminal_failure
+                    or _is_agent_result_terminal
                     or (
                         _saved_transcript_lacks_final_answer
                         and _classification['type'] not in {'cancelled', 'interrupted'}

tests/test_issue5121_provider_auth_terminal_error.py

@@ -329,6 +329,53 @@ def test_auth_401_seeded_replayed_assistant_does_not_satisfy_current_turn(tmp_pa
     assert not any(event == "done" for event, _ in events)
 
 
+def test_captured_terminal_http_400_beats_structured_final_answer(tmp_path, monkeypatch):
+    session = _prepare_session("captured_terminal_http_400", "stream_captured_terminal_http_400", pending_user_message="Please use the tool")
+
+    class CapturedTerminalHttp400Agent(MockAgent):
+        def __init__(self, status_callback=None, **kwargs):
+            super().__init__(**kwargs)
+            self.status_callback = status_callback
+
+        def run_conversation(self, **kwargs):
+            history = list(kwargs.get("conversation_history") or [])
+            status_cb = getattr(self, "status_callback", None)
+            if status_cb is not None:
+                status_cb("lifecycle", "❌ Non-retryable error (HTTP 400): invalid model")
+            if self.stream_delta_callback is not None:
+                self.stream_delta_callback("Partial text before failure")
+            return {
+                "messages": history + [
+                    {
+                        "role": "assistant",
+                        "content": [
+                            {"type": "tool_use", "name": "weather.lookup", "input": {"city": "Leeds"}},
+                            {"type": "output_text", "output_text": "It is 18C and sunny."},
+                        ],
+                        "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "weather.lookup", "arguments": "{}"}}],
+                    }
+                ],
+                "error": "",
+            }
+
+    fake_queue = _run_stream(
+        monkeypatch,
+        session,
+        "stream_captured_terminal_http_400",
+        CapturedTerminalHttp400Agent,
+        workspace=str(tmp_path),
+    )
+    saved = Session.load("captured_terminal_http_400")
+    assert saved is not None
+
+    events = _queue_events(fake_queue)
+    apperrors = [data for event, data in events if event == "apperror"]
+    assert apperrors, "expected apperror for captured terminal HTTP 400"
+    assert apperrors[-1]["type"] == "model_not_found"
+    assert not any(event == "done" for event, _ in events)
+    assert saved.messages[-1]["_error"] is True
+
+
 def test_auth_retry_success_does_not_append_error_turn(tmp_path, monkeypatch):
     session = _prepare_session("auth_retry", "stream_auth_retry", pending_user_message="Please retry")
     agent_cls = _build_auth_failure_agent(token_text="")

tests/test_issue5686_completed_output_text.py

@@ -55,6 +55,43 @@ class TestIssue5686CompletedOutputText:
         ]
         assert streaming._session_lacks_final_assistant_answer(messages) is True
 
+    def test_session_lacks_final_answer_false_for_untyped_visible_text_with_call_metadata(self):
+        messages = list(PRIOR_DISPLAY) + [
+            {"role": "user", "content": "What is the weather?"},
+            {
+                "role": "assistant",
+                "content": [
+                    {"call_id": "call_1", "name": "weather.lookup", "text": "It is 18C and sunny."},
+                ],
+            },
+        ]
+        assert streaming._session_lacks_final_assistant_answer(messages) is False
+
+    def test_session_lacks_final_answer_true_for_untyped_tool_part_with_empty_args(self):
+        messages = list(PRIOR_DISPLAY) + [
+            {"role": "user", "content": "What is the weather?"},
+            {
+                "role": "assistant",
+                "content": [
+                    {"call_id": "call_1", "args": {}},
+                ],
+            },
+        ]
+        assert streaming._session_lacks_final_assistant_answer(messages) is True
+
+    def test_session_lacks_final_answer_true_for_top_level_tool_calls_without_in_content_boundary(self):
+        messages = list(PRIOR_DISPLAY) + [
+            {"role": "user", "content": "What is the weather?"},
+            {
+                "role": "assistant",
+                "content": [
+                    {"type": "text", "text": "Checking the tool result"},
+                ],
+                "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "weather.lookup", "arguments": "{}"}}],
+            },
+        ]
+        assert streaming._session_lacks_final_assistant_answer(messages) is True
+
     def test_session_lacks_final_answer_false_for_plain_output_text_parts(self):
         messages = list(PRIOR_DISPLAY) + [
             {"role": "user", "content": "Summarize"},