feat(resources): add EvictionExecutor with SIGTERM/grace/SIGKILL sequence

circuitforge_core/resources/agent/eviction_executor.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import logging
+import os
+import signal
+import time
+from dataclasses import dataclass
+
+import psutil
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_GRACE_S = 5.0
+
+
+@dataclass(frozen=True)
+class EvictionResult:
+    success: bool
+    method: str   # "sigterm", "sigkill", "already_gone", "not_found", "error"
+    message: str
+
+
+class EvictionExecutor:
+    def __init__(self, grace_period_s: float = _DEFAULT_GRACE_S) -> None:
+        self._default_grace = grace_period_s
+
+    def evict_pid(
+        self,
+        pid: int,
+        grace_period_s: float | None = None,
+    ) -> EvictionResult:
+        grace = grace_period_s if grace_period_s is not None else self._default_grace
+
+        if not psutil.pid_exists(pid):
+            return EvictionResult(
+                success=False, method="not_found",
+                message=f"PID {pid} not found"
+            )
+
+        try:
+            os.kill(pid, signal.SIGTERM)
+        except ProcessLookupError:
+            return EvictionResult(
+                success=True, method="already_gone",
+                message=f"PID {pid} vanished before SIGTERM"
+            )
+        except PermissionError as exc:
+            return EvictionResult(
+                success=False, method="error",
+                message=f"Permission denied terminating PID {pid}: {exc}"
+            )
+
+        # Wait for grace period
+        deadline = time.monotonic() + grace
+        while time.monotonic() < deadline:
+            if not psutil.pid_exists(pid):
+                logger.info("PID %d exited cleanly after SIGTERM", pid)
+                return EvictionResult(
+                    success=True, method="sigterm",
+                    message=f"PID {pid} exited after SIGTERM"
+                )
+            time.sleep(0.05)
+
+        # Escalate to SIGKILL
+        if psutil.pid_exists(pid):
+            try:
+                os.kill(pid, signal.SIGKILL)
+                logger.warning("PID %d required SIGKILL", pid)
+                return EvictionResult(
+                    success=True, method="sigkill",
+                    message=f"PID {pid} killed with SIGKILL"
+                )
+            except ProcessLookupError:
+                pass
+
+        return EvictionResult(
+            success=True, method="sigkill",
+            message=f"PID {pid} is gone"
+        )

tests/test_resources/test_eviction_executor.py

@@ -0,0 +1,43 @@
+import signal
+from unittest.mock import patch, call
+import pytest
+from circuitforge_core.resources.agent.eviction_executor import EvictionExecutor, EvictionResult
+
+
+def test_evict_by_pid_sends_sigterm_then_sigkill():
+    executor = EvictionExecutor(grace_period_s=0.01)
+    # pid_exists always True → grace period expires → SIGKILL fires
+    with patch("os.kill") as mock_kill, \
+         patch("circuitforge_core.resources.agent.eviction_executor.psutil") as mock_psutil:
+        mock_psutil.pid_exists.return_value = True
+        result = executor.evict_pid(pid=1234, grace_period_s=0.01)
+
+    assert result.success is True
+    calls = mock_kill.call_args_list
+    assert call(1234, signal.SIGTERM) in calls
+    assert call(1234, signal.SIGKILL) in calls
+
+
+def test_evict_pid_succeeds_on_sigterm_alone():
+    executor = EvictionExecutor(grace_period_s=0.1)
+    with patch("os.kill"), \
+         patch("circuitforge_core.resources.agent.eviction_executor.psutil") as mock_psutil:
+        mock_psutil.pid_exists.side_effect = [True, False]  # gone after SIGTERM
+        result = executor.evict_pid(pid=5678, grace_period_s=0.01)
+    assert result.success is True
+    assert result.method == "sigterm"
+
+
+def test_evict_pid_not_found_returns_failure():
+    executor = EvictionExecutor()
+    with patch("circuitforge_core.resources.agent.eviction_executor.psutil") as mock_psutil:
+        mock_psutil.pid_exists.return_value = False
+        result = executor.evict_pid(pid=9999)
+    assert result.success is False
+    assert "not found" in result.message.lower()
+
+
+def test_eviction_result_is_immutable():
+    result = EvictionResult(success=True, method="sigterm", message="ok")
+    with pytest.raises((AttributeError, TypeError)):
+        result.success = False  # type: ignore