fix(gateway): stream synthesized recovery finals

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 ───────────

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"