diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 212f7cc09ac4..e26a6c514232 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -127,6 +127,22 @@ def _ra(): return run_agent +def _emit_synthesized_final_delta(agent: Any, final_response: str) -> None: + """Stream final text synthesized outside the normal model delta path. + + A few recovery paths assign ``final_response`` from already-available + text and then break out of the loop without another model stream event. + Gateway/SSE clients only see streamed deltas, so emit that synthesized + final answer before the loop closes. + """ + if not final_response or not getattr(agent, "stream_delta_callback", None): + return + try: + agent.stream_delta_callback(final_response) + except Exception: + pass + + def _restore_or_build_system_prompt(agent, system_message, conversation_history): """Restore the cached system prompt from the session DB or build it fresh. @@ -3470,6 +3486,7 @@ def _stop_spinner(): f"⚠️ Tool guardrail halted {decision.tool_name}: {decision.code}" ) messages.append({"role": "assistant", "content": final_response}) + _emit_synthesized_final_delta(agent, final_response) break # Reset per-turn retry counters after successful tool @@ -3576,6 +3593,7 @@ def _stop_spinner(): ) final_response = _recovered agent._response_was_previewed = True + _emit_synthesized_final_delta(agent, final_response) break # If the previous turn already delivered real content alongside @@ -3602,6 +3620,7 @@ def _stop_spinner(): # fallback text as the final response and break. final_response = agent._strip_think_blocks(fallback).strip() agent._response_was_previewed = True + _emit_synthesized_final_delta(agent, final_response) break # ── Post-tool-call empty response nudge ─────────── diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 3d0dcedddd05..468b9ed14d82 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -3033,11 +3033,14 @@ def _fake_api_call(api_kwargs): return empty_stub status_messages = [] + stream_deltas = [] + agent.stream_delta_callback = stream_deltas.append def _capture_status(msg): status_messages.append(msg) with ( + patch.object(agent, "_has_stream_consumers", return_value=False), patch.object(agent, "_interruptible_api_call", side_effect=_fake_api_call), patch.object(agent, "_persist_session"), patch.object(agent, "_save_trajectory"), @@ -3049,6 +3052,7 @@ def _capture_status(msg): assert result["completed"] is True assert result["final_response"] == "The answer to your question is that" assert result["api_calls"] == 1 # No wasted retries + assert "The answer to your question is that" in stream_deltas # Should emit the stream-interrupted status, NOT the empty-retry status recovery_msgs = [m for m in status_messages if "stream interrupted" in m.lower()] assert len(recovery_msgs) >= 1, f"Expected stream recovery status, got: {status_messages}" @@ -3080,6 +3084,33 @@ def _fake_api_call(api_kwargs): assert result["final_response"] == "Fresh partial content from this turn" assert result["api_calls"] == 1 + def test_prior_turn_fallback_emits_stream_delta(self, agent): + """SSE clients receive fallback text synthesized from the previous turn.""" + self._setup_agent(agent) + tool_call = _mock_tool_call(name="memory", arguments='{"action":"add","text":"ok"}', call_id="mem1") + tool_resp = _mock_response( + content="Earlier housekeeping answer", + finish_reason="tool_calls", + tool_calls=[tool_call], + ) + empty_stub = _mock_response(content="", finish_reason="stop") + agent.client.chat.completions.create.side_effect = [tool_resp, empty_stub] + agent.valid_tool_names.add("memory") + stream_deltas = [] + agent.stream_delta_callback = stream_deltas.append + + with ( + patch.object(agent, "_has_stream_consumers", return_value=False), + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + ): + result = agent.run_conversation("thanks") + + assert result["final_response"] == "Earlier housekeeping answer" + assert result["api_calls"] == 2 + assert "Earlier housekeeping answer" in stream_deltas + def test_nous_401_refreshes_after_remint_and_retries(self, agent): self._setup_agent(agent) agent.provider = "nous"