test(agent,gateway): cover partial-stream recovery and restart helper salvage

tests/gateway/restart_test_helpers.py

@@ -70,6 +70,7 @@ def make_restart_runner(
     runner._restart_task_started = False
     runner._restart_detached = False
     runner._restart_via_service = False
+    runner._detached_restart_helper_started = False
     runner._restart_command_source = None
     runner._restart_drain_timeout = DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
     runner._stop_task = None

tests/gateway/test_restart_drain.py

@@ -180,6 +180,7 @@ def test_load_restart_drain_timeout_prefers_env_then_config_then_default(
 async def test_request_restart_is_idempotent():
     runner, _adapter = make_restart_runner()
     runner.stop = AsyncMock()
+    runner._launch_detached_restart_command = AsyncMock()
 
     # _run_restart is held on self._restart_task and is intentionally NOT in
     # _background_tasks, so _stop_impl's cancel loop can't abort it mid-await
@@ -191,6 +192,7 @@ async def test_request_restart_is_idempotent():
 
     await runner._restart_task
 
+    runner._launch_detached_restart_command.assert_awaited_once_with()
     runner.stop.assert_awaited_once_with(
         restart=True, detached_restart=True, service_restart=False
     )
@@ -263,6 +265,7 @@ async def test_launch_detached_restart_command_uses_setsid(monkeypatch):
     assert cmd[:2] == ["/usr/bin/setsid", "bash"]
     assert "gateway restart" in cmd[-1]
     assert "kill -0 321" in cmd[-1]
+    assert "deadline=$(( $(date +%s) +" in cmd[-1]
     assert kwargs["start_new_session"] is True
     assert kwargs["stdout"] is subprocess.DEVNULL
     assert kwargs["stderr"] is subprocess.DEVNULL
@@ -271,6 +274,22 @@ async def test_launch_detached_restart_command_uses_setsid(monkeypatch):
     assert kwargs["env"].get("_HERMES_GATEWAY") is None
 
 
+@pytest.mark.asyncio
+async def test_detached_restart_helper_is_idempotent(monkeypatch):
+    runner, _adapter = make_restart_runner()
+    popen_calls = []
+
+    monkeypatch.setattr(gateway_run, "_resolve_hermes_bin", lambda: ["/usr/bin/hermes"])
+    monkeypatch.setattr(gateway_run.os, "getpid", lambda: 321)
+    monkeypatch.setattr(shutil, "which", lambda cmd: None)
+    monkeypatch.setattr(subprocess, "Popen", lambda *a, **k: popen_calls.append((a, k)))
+
+    await runner._launch_detached_restart_command()
+    await runner._launch_detached_restart_command()
+
+    assert len(popen_calls) == 1
+
+
 def test_windows_gateway_venv_imports_add_site_packages(monkeypatch, tmp_path):
     venv_dir = tmp_path / "venv"
     site_packages = venv_dir / "Lib" / "site-packages"

tests/run_agent/test_run_agent.py

@@ -4164,7 +4164,9 @@ class TestRunConversation:
             result = agent.run_conversation("ask me")
         # Should recover partial streamed content, not fall through to (empty)
         assert result["completed"] is True
-        assert result["final_response"] == "The answer to your question is that"
+        assert result["final_response"].startswith("The answer to your question is that")
+        assert "No reply:" in result["final_response"]
+        assert result["response_previewed"] is False
         assert result["api_calls"] == 1  # No wasted retries
         # 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()]
@@ -4194,7 +4196,9 @@ class TestRunConversation:
         ):
             result = agent.run_conversation("question")
         # Should use the streamed content, not the old prior-turn fallback
-        assert result["final_response"] == "Fresh partial content from this turn"
+        assert result["final_response"].startswith("Fresh partial content from this turn")
+        assert "No reply:" in result["final_response"]
+        assert result["response_previewed"] is False
         assert result["api_calls"] == 1
 
     def test_interrupt_during_stream_preserves_partial_assistant_text(self, agent):

tests/run_agent/test_turn_completion_explainer.py

@@ -162,6 +162,37 @@ def test_run_conversation_empty_exhausted_surfaces_explanation():
     assert "No reply:" in result["final_response"]
 
 
+def test_run_conversation_partial_stream_recovery_surfaces_explanation():
+    """A long recovered partial stream still needs the visible footer.
+
+    Without this, the gateway marks the turn as previewed and suppresses
+    the final send, leaving messaging users with a fragment and no reason.
+    """
+    agent = _make_agent(max_iterations=10)
+    empty_stub = _mock_response(content=None, finish_reason="stop")
+    recovered = (
+        "I inspected the running gateway and found that the current turn "
+        "stopped after the provider stream timed out."
+    )
+
+    def _fake_api_call(_api_kwargs):
+        agent._current_streamed_assistant_text = recovered
+        return empty_stub
+
+    with (
+        patch.object(agent, "_interruptible_api_call", side_effect=_fake_api_call),
+        patch.object(agent, "_persist_session"),
+        patch.object(agent, "_save_trajectory"),
+        patch.object(agent, "_cleanup_task_resources"),
+    ):
+        result = agent.run_conversation("do something")
+
+    assert result["turn_exit_reason"] == "partial_stream_recovery"
+    assert result["final_response"].startswith(recovered)
+    assert "No reply:" in result["final_response"]
+    assert result["response_previewed"] is False
+
+
 def test_run_conversation_normal_reply_stays_quiet():
     """A normal short reply like 'Done.' must NOT get an explainer footer."""
     agent = _make_agent(max_iterations=10)