scripts/agent_loop.py

@@ -333,6 +333,15 @@ def _agent_alive(state: dict) -> bool:
         return True
 
 
+def _is_claude_process(pid: int) -> bool:
+    """Return True if pid's comm name indicates it is a claude/node process."""
+    try:
+        comm = Path(f"/proc/{pid}/comm").read_text().strip()
+        return comm in ("claude", "node")
+    except OSError:
+        return False
+
+
 def _agent_age_seconds(state: dict) -> float:
     """Seconds elapsed since the agent was launched, from the state file timestamp."""
     try:
@@ -367,11 +376,13 @@ def _git_summary() -> str:
 def _kill_agent(state: dict) -> None:
     """Forcefully stop the running agent."""
     pid = state.get("pid")
-    if pid:
+    if pid and _is_claude_process(pid):
         try:
             os.kill(pid, 9)
         except ProcessLookupError:
             pass
+    elif pid:
+        print(f"WARNING: pid {pid} is not a claude process — skipping kill to avoid hitting recycled PID")
 
 
 # ── subcommands ───────────────────────────────────────────────────────────────

scripts/test_agent_loop.py

@@ -88,21 +88,47 @@ class TestAgentAlive(unittest.TestCase):
         self.assertFalse(agent_loop._agent_alive({"pid": None}))
 
 
+class TestIsClaudeProcess(unittest.TestCase):
+    def test_returns_true_for_claude_comm(self):
+        with patch.object(agent_loop.Path, "read_text", return_value="claude\n"):
+            self.assertTrue(agent_loop._is_claude_process(1234))
+
+    def test_returns_true_for_node_comm(self):
+        with patch.object(agent_loop.Path, "read_text", return_value="node\n"):
+            self.assertTrue(agent_loop._is_claude_process(1234))
+
+    def test_returns_false_for_other_process(self):
+        with patch.object(agent_loop.Path, "read_text", return_value="bash\n"):
+            self.assertFalse(agent_loop._is_claude_process(1234))
+
+    def test_returns_false_when_proc_missing(self):
+        with patch.object(agent_loop.Path, "read_text", side_effect=OSError):
+            self.assertFalse(agent_loop._is_claude_process(1234))
+
+
 class TestKillAgent(unittest.TestCase):
     def test_kill_sends_sigkill(self):
-        with patch("agent_loop.os.kill") as mock_kill:
-            agent_loop._kill_agent({"pid": 1234})
-            mock_kill.assert_called_once_with(1234, 9)
+        with patch("agent_loop._is_claude_process", return_value=True):
+            with patch("agent_loop.os.kill") as mock_kill:
+                agent_loop._kill_agent({"pid": 1234})
+                mock_kill.assert_called_once_with(1234, 9)
 
     def test_kill_ignores_missing_process(self):
-        with patch("agent_loop.os.kill", side_effect=ProcessLookupError):
-            agent_loop._kill_agent({"pid": 1234})  # Should not raise.
+        with patch("agent_loop._is_claude_process", return_value=True):
+            with patch("agent_loop.os.kill", side_effect=ProcessLookupError):
+                agent_loop._kill_agent({"pid": 1234})  # Should not raise.
 
     def test_kill_noop_when_no_pid(self):
         with patch("agent_loop.os.kill") as mock_kill:
             agent_loop._kill_agent({})
             mock_kill.assert_not_called()
 
+    def test_kill_skips_recycled_pid(self):
+        with patch("agent_loop._is_claude_process", return_value=False):
+            with patch("agent_loop.os.kill") as mock_kill:
+                agent_loop._kill_agent({"pid": 1234})
+                mock_kill.assert_not_called()
+
 
 class TestStartAgent(unittest.TestCase):
     def _make_mock_proc(self, pid=42):