fix(mcp): per-process PID isolation prevents cross-session crash on restart

- _stdio_pids: set → Dict[int,str] tracks pid→server_name
- SIGTERM-first with 2s grace before SIGKILL escalation
- hasattr guard for SIGKILL on platforms without it
- Updated tests for dict-based tracking and 3-phase kill sequence
This commit is contained in:
Jefferson
2026-04-23 20:01:17 +02:00
committed by Teknium
parent c7d023937c
commit 67c8f837fc
2 changed files with 51 additions and 23 deletions

View File

@@ -77,7 +77,7 @@ class TestStdioPidTracking:
from tools.mcp_tool import _stdio_pids, _lock
with _lock:
# Might have residual state from other tests, just check type
assert isinstance(_stdio_pids, set)
assert isinstance(_stdio_pids, dict)
def test_kill_orphaned_noop_when_empty(self):
"""_kill_orphaned_mcp_children does nothing when no PIDs tracked."""
@@ -96,7 +96,7 @@ class TestStdioPidTracking:
# Use a PID that definitely doesn't exist
fake_pid = 999999999
with _lock:
_stdio_pids.add(fake_pid)
_stdio_pids[fake_pid] = "test"
# Should not raise (ProcessLookupError is caught)
_kill_orphaned_mcp_children()
@@ -105,40 +105,49 @@ class TestStdioPidTracking:
assert fake_pid not in _stdio_pids
def test_kill_orphaned_uses_sigkill_when_available(self, monkeypatch):
"""Unix-like platforms should keep using SIGKILL for orphan cleanup."""
"""SIGTERM-first then SIGKILL after 2s for orphan cleanup."""
from tools.mcp_tool import _kill_orphaned_mcp_children, _stdio_pids, _lock
fake_pid = 424242
with _lock:
_stdio_pids.clear()
_stdio_pids.add(fake_pid)
_stdio_pids[fake_pid] = "test"
fake_sigkill = 9
monkeypatch.setattr(signal, "SIGKILL", fake_sigkill, raising=False)
with patch("tools.mcp_tool.os.kill") as mock_kill:
with patch("tools.mcp_tool.os.kill") as mock_kill, \
patch("time.sleep") as mock_sleep:
_kill_orphaned_mcp_children()
mock_kill.assert_called_once_with(fake_pid, fake_sigkill)
# SIGTERM, then alive-check (signal 0), then SIGKILL
mock_kill.assert_any_call(fake_pid, signal.SIGTERM)
mock_kill.assert_any_call(fake_pid, 0) # alive check
mock_kill.assert_any_call(fake_pid, fake_sigkill)
assert mock_kill.call_count == 3
mock_sleep.assert_called_once_with(2)
with _lock:
assert fake_pid not in _stdio_pids
def test_kill_orphaned_falls_back_without_sigkill(self, monkeypatch):
"""Windows-like signal modules without SIGKILL should fall back to SIGTERM."""
"""Without SIGKILL, SIGTERM is used for both phases."""
from tools.mcp_tool import _kill_orphaned_mcp_children, _stdio_pids, _lock
fake_pid = 434343
with _lock:
_stdio_pids.clear()
_stdio_pids.add(fake_pid)
_stdio_pids[fake_pid] = "test"
monkeypatch.delattr(signal, "SIGKILL", raising=False)
with patch("tools.mcp_tool.os.kill") as mock_kill:
with patch("tools.mcp_tool.os.kill") as mock_kill, \
patch("time.sleep") as mock_sleep:
_kill_orphaned_mcp_children()
mock_kill.assert_called_once_with(fake_pid, signal.SIGTERM)
# SIGTERM phase, alive check raises (process gone), no escalation
mock_kill.assert_any_call(fake_pid, signal.SIGTERM)
assert mock_sleep.called
with _lock:
assert fake_pid not in _stdio_pids