mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-21 17:25:26 -04:00
fix(code): reap langgraph dev server when startup is cancelled (#4629)
## Problem Pressing `Ctrl+D` shortly after launching `dcode` returns you to the terminal but leaves an orphaned `python -m langgraph_cli dev …` process running: ``` $ ps aux | grep deepagent wbbradley 69643 ... python3 -m langgraph_cli dev --host 127.0.0.1 --port 59934 --no-browser --no-reload --config .../langgraph.json ``` This is not intentional — it's a process leak. ## Root cause To make the splash screen appear instantly, the LangGraph server is started in a background Textual worker (`_start_server_background`) rather than blocking launch. That worker calls `start_server_and_get_agent`, which spawns the `langgraph dev` subprocess early (inside `ServerProcess.start()`) and then spends a few seconds polling `/ok` for health + graph readiness. If the user quits during that window: 1. `super().exit()` tears down Textual, which **cancels all workers**, injecting `asyncio.CancelledError` into the in-flight `server.start()`. 2. `start_server_and_get_agent` guarded cleanup with `except Exception:` — but `CancelledError` is a `BaseException`, **not** an `Exception`, so `server.stop()` was skipped and the exception re-raised. 3. Because the function never returned, `DeepAgentsApp._server_proc` was never assigned, so the belt-and-suspenders `finally` in `run_textual_app` also saw `None` and skipped cleanup. Net result: the `langgraph dev` subprocess is orphaned. When the server has fully finished starting before the quit, `stop()` runs correctly and there is no leak — the leak is specific to quitting *during* startup. ## Fix Catch `BaseException` (which includes `asyncio.CancelledError`) around `server.start()` / `wait_for_graph_ready()` so a cancelled startup still stops the half-started subprocess, then re-raise to preserve cancellation/shutdown semantics. This matches the existing cleanup-on-cancel-and-reraise pattern already used in `skills/trust.py`. --------- Co-authored-by: Mason Daugherty <github@mdrxy.com> Co-authored-by: Mason Daugherty <mason@langchain.dev>
This commit is contained in:
@@ -528,6 +528,7 @@ class ServerProcess:
|
||||
stderr=subprocess.STDOUT,
|
||||
)
|
||||
|
||||
started = False
|
||||
try:
|
||||
await wait_for_server_healthy(
|
||||
self.url,
|
||||
@@ -536,9 +537,24 @@ class ServerProcess:
|
||||
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 @@ class ServerProcess:
|
||||
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)
|
||||
|
||||
|
||||
@@ -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",
|
||||
)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@@ -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 @@ class TestServerProcess:
|
||||
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 @@ class TestServerProcess:
|
||||
):
|
||||
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 @@ class TestServerProcess:
|
||||
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"
|
||||
|
||||
@@ -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 @@ class TestStartServerAndGetAgent:
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user