fix: prevent worker from stopping when another session is still active

Author: maxritter

launcher/tests/unit/test_session_end.py

@@ -1,49 +1,65 @@
 #!/usr/bin/env python3
 """SessionEnd hook - stops worker only when no other sessions are active.
 
-Sends a 'Session Ended' notification. Spec-specific notifications
-(verification_complete, plan_approval) are sent by the spec skills
-via `pilot notify` to avoid duplication.
+Checks session directories directly (no subprocess) for reliability during
+shutdown. Only stops the worker when the current session is the last active one.
 """
 
 from __future__ import annotations
 
-import json
 import os
 import subprocess
 import sys
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).parent))
 
-PILOT_BIN = Path.home() / ".pilot" / "bin" / "pilot"
+SESSIONS_DIR = Path.home() / ".pilot" / "sessions"
+SKIP_NAMES = {"default", "pipes"}
 
 
-def _get_active_session_count() -> int:
-    """Get active session count from the pilot binary."""
+def _has_other_active_sessions() -> bool:
+    """Check if any other Pilot wrapper sessions are still active.
+
+    Iterates session directories directly (no subprocess) to avoid failures
+    during shutdown. Skips the current session via PILOT_SESSION_ID.
+    Returns True on any error (safe default: don't stop the worker).
+    """
     try:
-        result = subprocess.run(
-            [str(PILOT_BIN), "sessions", "--json"],
-            capture_output=True,
-            text=True,
-            check=False,
-            timeout=10,
-        )
-        if result.returncode == 0:
-            data = json.loads(result.stdout)
-            return data.get("count", 0)
-    except (json.JSONDecodeError, OSError, subprocess.TimeoutExpired):
-        pass
-    return 0
+        if not SESSIONS_DIR.exists():
+            return False
+
+        my_session = os.environ.get("PILOT_SESSION_ID", "")
+
+        for entry in SESSIONS_DIR.iterdir():
+            if not entry.is_dir() or entry.name in SKIP_NAMES:
+                continue
+            if entry.name == my_session:
+                continue
+            try:
+                pid = int(entry.name)
+            except ValueError:
+                continue
+            try:
+                os.kill(pid, 0)
+            except OSError:
+                continue
+
+            # Another wrapper PID is alive → another session exists
+            return True
+
+        return False
+    except OSError:
+        # Can't read directory → assume other sessions exist (safe default)
+        return True
 
 
 def main() -> int:
     plugin_root = os.environ.get("CLAUDE_PLUGIN_ROOT", "")
     if not plugin_root:
         return 0
 
-    count = _get_active_session_count()
-    if count > 1:
+    if _has_other_active_sessions():
         return 0
 
     stop_script = Path(plugin_root) / "scripts" / "worker-service.cjs"

pilot/hooks/session_end.py

@@ -1,8 +1,9 @@
-"""Tests for session_end hook — worker stop behavior."""
+"""Tests for session_end hook — worker stop behavior with direct filesystem checks."""
 
 from __future__ import annotations
 
 import os
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import session_end
@@ -17,11 +18,16 @@ def test_returns_early_without_plugin_root():
     assert result == 0
 
 
-def test_skips_stop_when_other_sessions_active():
+def test_skips_stop_when_other_sessions_active(tmp_path: Path):
     """Should skip worker stop when other Pilot sessions are running."""
+    base = tmp_path / "sessions"
+    (base / "1001").mkdir(parents=True)
+    (base / "2002").mkdir(parents=True)
+
     with (
-        patch.dict(os.environ, {"CLAUDE_PLUGIN_ROOT": "/fake/plugin"}),
-        patch("session_end._get_active_session_count", return_value=2),
+        patch.dict(os.environ, {"CLAUDE_PLUGIN_ROOT": "/fake/plugin", "PILOT_SESSION_ID": "1001"}),
+        patch.object(session_end, "SESSIONS_DIR", base),
+        patch("session_end.os.kill", return_value=None),
         patch("session_end.subprocess.run") as mock_run,
     ):
         result = session_end.main()
@@ -30,11 +36,14 @@ def test_skips_stop_when_other_sessions_active():
     mock_run.assert_not_called()
 
 
-def test_stops_worker_when_no_other_sessions():
+def test_stops_worker_when_no_other_sessions(tmp_path: Path):
     """Should stop worker when this is the only active session."""
+    base = tmp_path / "sessions"
+    (base / "1001").mkdir(parents=True)
+
     with (
-        patch.dict(os.environ, {"CLAUDE_PLUGIN_ROOT": "/fake/plugin"}),
-        patch("session_end._get_active_session_count", return_value=1),
+        patch.dict(os.environ, {"CLAUDE_PLUGIN_ROOT": "/fake/plugin", "PILOT_SESSION_ID": "1001"}),
+        patch.object(session_end, "SESSIONS_DIR", base),
         patch("session_end.subprocess.run", return_value=MagicMock(returncode=0)) as mock_run,
     ):
         result = session_end.main()
@@ -44,14 +53,56 @@ def test_stops_worker_when_no_other_sessions():
     assert "stop" in str(mock_run.call_args)
 
 
-def test_stops_worker_when_zero_sessions():
-    """Should stop worker and return 0 when no sessions active."""
+def test_stops_worker_when_zero_sessions(tmp_path: Path):
+    """Should stop worker and return 0 when no sessions exist at all."""
+    base = tmp_path / "sessions"
+    base.mkdir(parents=True)
+
     with (
-        patch.dict(os.environ, {"CLAUDE_PLUGIN_ROOT": "/fake/plugin"}),
-        patch("session_end._get_active_session_count", return_value=0),
+        patch.dict(os.environ, {"CLAUDE_PLUGIN_ROOT": "/fake/plugin", "PILOT_SESSION_ID": "1001"}),
+        patch.object(session_end, "SESSIONS_DIR", base),
         patch("session_end.subprocess.run", return_value=MagicMock(returncode=0)) as mock_run,
     ):
         result = session_end.main()
 
     assert result == 0
     mock_run.assert_called_once()
+
+
+def test_safe_default_on_directory_error():
+    """Should NOT stop worker when directory can't be read (safe default)."""
+    mock_dir = MagicMock()
+    mock_dir.exists.side_effect = OSError("permission denied")
+
+    with (
+        patch.dict(os.environ, {"CLAUDE_PLUGIN_ROOT": "/fake/plugin", "PILOT_SESSION_ID": "1001"}),
+        patch.object(session_end, "SESSIONS_DIR", mock_dir),
+        patch("session_end.subprocess.run") as mock_run,
+    ):
+        result = session_end.main()
+
+    assert result == 0
+    mock_run.assert_not_called()
+
+
+def test_skips_dead_pid_sessions(tmp_path: Path):
+    """Should not count dead PID directories as active sessions."""
+    base = tmp_path / "sessions"
+    (base / "1001").mkdir(parents=True)  # current
+    (base / "9999").mkdir(parents=True)  # dead
+
+    def kill_side_effect(pid: int, _sig: int) -> None:
+        if pid == 9999:
+            raise OSError("No such process")
+
+    with (
+        patch.dict(os.environ, {"CLAUDE_PLUGIN_ROOT": "/fake/plugin", "PILOT_SESSION_ID": "1001"}),
+        patch.object(session_end, "SESSIONS_DIR", base),
+        patch("session_end.os.kill", side_effect=kill_side_effect),
+        patch("session_end.subprocess.run", return_value=MagicMock(returncode=0)) as mock_run,
+    ):
+        result = session_end.main()
+
+    assert result == 0
+    mock_run.assert_called_once()
+    assert "stop" in str(mock_run.call_args)