diff --git a/libs/code/deepagents_code/client/launch/server.py b/libs/code/deepagents_code/client/launch/server.py index dfea201c53..ea9bef8665 100644 --- a/libs/code/deepagents_code/client/launch/server.py +++ b/libs/code/deepagents_code/client/launch/server.py @@ -528,6 +528,7 @@ async def start( stderr=subprocess.STDOUT, ) + started = False try: await wait_for_server_healthy( self.url, @@ -536,9 +537,24 @@ async def start( read_log=self._read_log_file, local=True, ) - except Exception: - self.stop() - raise + started = True + finally: + if not started: + # Reap the subprocess we just spawned if startup did not + # complete — including cancellation (e.g. Ctrl+D / SIGINT before + # the health check returns). A `finally` rather than `except + # Exception` is deliberate: `asyncio.CancelledError` is a + # `BaseException`, so an `except Exception` guard would skip this + # and orphan the process. The inner guard stops a `stop()` error + # from masking the exception already propagating; `stop()` is + # effectively non-raising today, so if it does fire it signals an + # unexpected leak — hence `error`, not `warning`. + try: + self.stop() + except Exception: + logger.exception( + "Error stopping server during startup cleanup", + ) async def wait_for_graph_ready( self, @@ -634,14 +650,29 @@ def _stop_process(self) -> None: self._process.wait(timeout=_SHUTDOWN_TIMEOUT) except subprocess.TimeoutExpired: logger.warning("Server did not stop gracefully, killing") - self._process.kill() + # `kill()` lives in this handler, so a raise here would escape + # the sibling `except OSError` below. Guard it explicitly: + # `ProcessLookupError` just means the process already exited + # (benign — nothing left to reap), while any other `OSError` + # means SIGKILL failed and the process is likely orphaned. try: + self._process.kill() self._process.wait(timeout=2) except subprocess.TimeoutExpired: logger.warning( "Server process pid=%d did not exit after SIGKILL", self._process.pid, ) + except ProcessLookupError: + logger.debug( + "Server process pid=%d already exited before SIGKILL", + self._process.pid, + ) + except OSError: + logger.exception( + "Failed to SIGKILL server process pid=%d; it may be orphaned", + self._process.pid, + ) except OSError: logger.warning("Error stopping server", exc_info=True) diff --git a/libs/code/deepagents_code/client/launch/server_manager.py b/libs/code/deepagents_code/client/launch/server_manager.py index 136ddea279..cfc79a65c6 100644 --- a/libs/code/deepagents_code/client/launch/server_manager.py +++ b/libs/code/deepagents_code/client/launch/server_manager.py @@ -401,19 +401,34 @@ async def start_server_and_get_agent( owns_config_dir=True, scaffold=_scaffold_workspace, ) + started = False try: await server.start() await server.wait_for_graph_ready("agent") - except Exception: - server.stop() - raise - - agent = RemoteAgent( - url=server.url, - graph_name="agent", - ) - - return agent, server, None + agent = RemoteAgent( + url=server.url, + graph_name="agent", + ) + started = True + return agent, server, None + finally: + if not started: + # Startup failed or was cancelled before the server was handed off + # to the caller (which records the reference only on success). If + # `start()` itself failed it already reaped its own subprocess, so + # this `stop()` is then an idempotent no-op; this cleanup is the sole + # reaper only when `start()` succeeded but `wait_for_graph_ready()` + # (or `RemoteAgent()`) failed afterward. A `finally` rather than + # `except Exception` is deliberate: `asyncio.CancelledError` is a + # `BaseException`, so an `except Exception` guard would skip cleanup + # and orphan the process. The inner guard stops a `stop()` error + # from masking the exception already propagating. + try: + server.stop() + except Exception: + logger.exception( + "Error stopping server during startup cleanup", + ) # ------------------------------------------------------------------ diff --git a/libs/code/tests/unit_tests/test_server.py b/libs/code/tests/unit_tests/test_server.py index 68dbd3f457..93e9b9eede 100644 --- a/libs/code/tests/unit_tests/test_server.py +++ b/libs/code/tests/unit_tests/test_server.py @@ -2,7 +2,10 @@ from __future__ import annotations +import asyncio +import logging import os +import signal import socket import threading from types import SimpleNamespace @@ -416,9 +419,14 @@ async def test_wait_for_graph_ready_checks_logs_after_transport_error( await server.wait_for_graph_ready("agent") async def test_start_cleans_up_partial_state_on_health_failure( - self, tmp_path: Path + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Failed startup should stop the process and remove owned resources.""" + # `_stop_process` preserves the log file when debug mode is on, which + # would defeat the `not log_path.exists()` assertion below; pin it off + # so an ambient `DEEPAGENTS_CODE_DEBUG` in the environment can't flake. + monkeypatch.delenv("DEEPAGENTS_CODE_DEBUG", raising=False) + config_dir = tmp_path / "runtime" config_dir.mkdir() (config_dir / "langgraph.json").write_text("{}") @@ -456,7 +464,7 @@ async def test_start_cleans_up_partial_state_on_health_failure( ): await server.start() - process.send_signal.assert_called_once() + process.send_signal.assert_called_once_with(signal.SIGTERM) process.wait.assert_called_once() log_file.close.assert_called_once() assert server._process is None @@ -464,6 +472,120 @@ async def test_start_cleans_up_partial_state_on_health_failure( assert not config_dir.exists() assert not log_path.exists() + async def test_start_cleans_up_partial_state_on_cancellation( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A cancelled startup must reap the subprocess it already spawned. + + `asyncio.CancelledError` is a `BaseException`, not an `Exception`, so + `start()` must clean up in a `finally` — otherwise the `langgraph dev` + subprocess spawned before the health check is orphaned when the caller + is cancelled mid-startup (e.g. Ctrl+D). Regression: PR #4629. + """ + # See sibling health-failure test: pin debug off so the log file is + # unlinked and the `not log_path.exists()` assertion can't flake. + monkeypatch.delenv("DEEPAGENTS_CODE_DEBUG", raising=False) + + config_dir = tmp_path / "runtime" + config_dir.mkdir() + (config_dir / "langgraph.json").write_text("{}") + + log_path = tmp_path / "server.log" + log_path.write_text("booting") + + process = MagicMock() + process.pid = 1234 + process.poll.return_value = None + + log_file = MagicMock() + log_file.name = str(log_path) + + server = ServerProcess(config_dir=config_dir, owns_config_dir=True) + + with ( + patch( + "deepagents_code.client.launch.server._find_free_port", + return_value=12345, + ), + patch( + "deepagents_code.client.launch.server.tempfile.NamedTemporaryFile", + return_value=log_file, + ), + patch( + "deepagents_code.client.launch.server.subprocess.Popen", + return_value=process, + ), + patch( + "deepagents_code.client.launch.server.wait_for_server_healthy", + new=AsyncMock(side_effect=asyncio.CancelledError), + ), + pytest.raises(asyncio.CancelledError), + ): + await server.start() + + process.send_signal.assert_called_once_with(signal.SIGTERM) + process.wait.assert_called_once() + log_file.close.assert_called_once() + assert server._process is None + assert server._log_file is None + assert not config_dir.exists() + assert not log_path.exists() + + async def test_start_cleanup_error_does_not_mask_startup_error( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """A failing `stop()` during cleanup must not mask the startup error. + + `start()`'s `finally` guards `stop()` so that if reaping the subprocess + itself raises, the in-flight startup exception still propagates + unchanged (rather than being replaced by the cleanup error) and the + failure is logged at `error` level. + """ + config_dir = tmp_path / "runtime" + config_dir.mkdir() + (config_dir / "langgraph.json").write_text("{}") + + log_path = tmp_path / "server.log" + log_path.write_text("booting") + + process = MagicMock() + process.pid = 1234 + process.poll.return_value = None + + log_file = MagicMock() + log_file.name = str(log_path) + + server = ServerProcess(config_dir=config_dir, owns_config_dir=True) + + with ( + patch( + "deepagents_code.client.launch.server._find_free_port", + return_value=12345, + ), + patch( + "deepagents_code.client.launch.server.tempfile.NamedTemporaryFile", + return_value=log_file, + ), + patch( + "deepagents_code.client.launch.server.subprocess.Popen", + return_value=process, + ), + patch( + "deepagents_code.client.launch.server.wait_for_server_healthy", + new=AsyncMock(side_effect=RuntimeError("startup boom")), + ), + patch.object( + server, "stop", side_effect=RuntimeError("cleanup boom") + ) as mock_stop, + caplog.at_level(logging.ERROR), + # The original startup error propagates, not the cleanup error. + pytest.raises(RuntimeError, match="startup boom"), + ): + await server.start() + + mock_stop.assert_called_once() + assert "Error stopping server during startup cleanup" in caplog.text + async def test_start_rescaffolds_when_config_missing(self, tmp_path: Path) -> None: """A missing langgraph.json should be rebuilt via the scaffold hook.""" config_dir = tmp_path / "runtime" diff --git a/libs/code/tests/unit_tests/test_server_manager.py b/libs/code/tests/unit_tests/test_server_manager.py index 53815588a7..06fae5ca98 100644 --- a/libs/code/tests/unit_tests/test_server_manager.py +++ b/libs/code/tests/unit_tests/test_server_manager.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import os from typing import TYPE_CHECKING from unittest.mock import AsyncMock, MagicMock, patch @@ -279,6 +280,111 @@ async def test_stops_server_when_graph_readiness_fails( mock_server.stop.assert_called_once() mock_agent.assert_not_called() + @pytest.mark.parametrize( + "interrupt", + [asyncio.CancelledError, KeyboardInterrupt, SystemExit], + ) + async def test_stops_server_when_start_interrupted( + self, interrupt: type[BaseException], tmp_path: Path, monkeypatch + ) -> None: + """A quit during startup must still reap the half-started server. + + The langgraph subprocess is spawned inside `ServerProcess.start()` + before this function returns, so the caller has not yet stored a + reference to it (`DeepAgentsApp._server_proc` is assigned only on + successful return). When the background startup worker is interrupted + mid-`start()` — e.g. the user presses Ctrl+D before the health check + completes — this `except` clause is the only thing that can stop the + orphaned subprocess. The interrupts covered here are all `BaseException` + subclasses rather than `Exception`, so an `except Exception` guard would + leak the process (regression: PR #4629). + """ + project_root = tmp_path / "project" + project_root.mkdir() + monkeypatch.chdir(project_root) + + work_dir = tmp_path / "runtime" + work_dir.mkdir() + + mock_server = MagicMock() + mock_server.start = AsyncMock(side_effect=interrupt) + mock_server.wait_for_graph_ready = AsyncMock() + mock_server.stop = MagicMock() + mock_server.url = "http://127.0.0.1:2024" + + with ( + patch.dict(os.environ, {}, clear=False), + patch( + "deepagents_code.client.launch.server_manager.tempfile.mkdtemp", + return_value=str(work_dir), + ), + patch("deepagents_code.client.launch.server_manager._write_checkpointer"), + patch("deepagents_code.client.launch.server_manager._write_pyproject"), + patch( + "deepagents_code.client.launch.server.ServerProcess", + return_value=mock_server, + ), + patch("deepagents_code.client.remote_client.RemoteAgent") as mock_agent, + pytest.raises(interrupt), + ): + await start_server_and_get_agent( + assistant_id="agent", + mcp_config_path=None, + ) + + mock_server.start.assert_awaited_once() + mock_server.stop.assert_called_once() + # The interrupt must propagate: graph readiness is never reached, and + # no client is handed back to a caller that is being torn down. + mock_server.wait_for_graph_ready.assert_not_awaited() + mock_agent.assert_not_called() + + async def test_start_cleanup_error_does_not_mask_interrupt( + self, tmp_path: Path, monkeypatch + ) -> None: + """A failure inside `stop()` must not replace the in-flight interrupt. + + Cleanup runs while a `BaseException` (here `CancelledError`) is + propagating. If `stop()` itself raises, that error is swallowed and + logged so the original cancellation stays the propagated exception, + preserving cancellation semantics instead of surfacing the teardown + error (regression: PR #4629). + """ + project_root = tmp_path / "project" + project_root.mkdir() + monkeypatch.chdir(project_root) + + work_dir = tmp_path / "runtime" + work_dir.mkdir() + + mock_server = MagicMock() + mock_server.start = AsyncMock(side_effect=asyncio.CancelledError) + mock_server.wait_for_graph_ready = AsyncMock() + mock_server.stop = MagicMock(side_effect=RuntimeError("kill failed")) + mock_server.url = "http://127.0.0.1:2024" + + with ( + patch.dict(os.environ, {}, clear=False), + patch( + "deepagents_code.client.launch.server_manager.tempfile.mkdtemp", + return_value=str(work_dir), + ), + patch("deepagents_code.client.launch.server_manager._write_checkpointer"), + patch("deepagents_code.client.launch.server_manager._write_pyproject"), + patch( + "deepagents_code.client.launch.server.ServerProcess", + return_value=mock_server, + ), + patch("deepagents_code.client.remote_client.RemoteAgent"), + pytest.raises(asyncio.CancelledError), + ): + await start_server_and_get_agent( + assistant_id="agent", + mcp_config_path=None, + ) + + mock_server.stop.assert_called_once() + def test_relative_paths_written_verbatim_to_langgraph_json( self, tmp_path: Path ) -> None: