Merge cb527188c9 into 9e145d9ceb

coworker/server/manager.py

@@ -4091,6 +4091,21 @@ class SessionManager:
             if not clients:
                 self._session_clients.pop(session_id, None)
 
+    @staticmethod
+    def _shutdown_engine(engine: TurnEngine) -> None:
+        """Stop all processes owned by an engine before dropping its last reference."""
+        try:
+            engine.request_interrupt()
+        except Exception:
+            pass
+        executor = getattr(engine, "executor", None)
+        shutdown = getattr(executor, "shutdown", None)
+        if callable(shutdown):
+            try:
+                shutdown()
+            except Exception:
+                pass
+
     async def broadcast_session(self, session_id: str, message: dict) -> None:
         """Fan a turn event out to every socket viewing this session. Best-effort: a dead socket
         is dropped, never fatal to the turn (delivery is socket-independent)."""
@@ -4104,7 +4119,19 @@ class SessionManager:
         await self.scheduler.stop()
         await self.stop_gateway()
         await self.mcp.aclose()
-        self.audit_store.close()
+        for engine in list(self._engines.values()):
+            self._shutdown_engine(engine)
+        self._engines.clear()
+        for store in (
+            self.audit_store,
+            self.session_store,
+            self.memory_store,
+            self.task_store,
+        ):
+            try:
+                store.close()
+            except Exception:
+                pass
 
     # -- automation (scheduled tasks) -------------------------------------------
     def approval_prompt_data(self, session_id: str, request) -> dict[str, Any]:
@@ -5498,12 +5525,7 @@ class SessionManager:
             return {"ok": False, "error": "internal sessions cannot be deleted here"}
         engine = self._engines.pop(session_id, None)
         if engine is not None:
-            try:
-                # (was engine.interrupt() — a method that never existed; the AttributeError
-                # was silently swallowed, so deleting a running session never stopped it.)
-                engine.request_interrupt()
-            except Exception:
-                pass
+            self._shutdown_engine(engine)
         record = self.session_store.load(session_id)
         ok = self.session_store.delete(session_id)
         # Deleting a session is the one implicit unsubscribe (otherwise subscriptions are permanent).

coworker/tools/shell.py

@@ -71,6 +71,15 @@ class Executor(ABC):
     def close(self) -> None:  # pragma: no cover - default no-op
         pass
 
+    def shutdown(self) -> None:
+        """Release every process owned by this executor.
+
+        ``close`` deliberately leaves detached background work alone so a transient shell
+        recovery does not kill it. Session deletion and server shutdown need the opposite
+        guarantee: no command started by the session may outlive that session.
+        """
+        self.close()
+
 
 class _BackgroundTask:
     """One detached background command: its own process (not the persistent shell), a
@@ -424,6 +433,17 @@ class LocalExecutor(Executor):
             self._proc.terminate()
         except (ProcessLookupError, OSError):
             pass
+        try:
+            self._proc.wait(timeout=5)
+        except (subprocess.TimeoutExpired, OSError):
+            pass
+
+    def shutdown(self) -> None:
+        """Stop the persistent shell and all detached background tasks for this session."""
+        self.interrupt_now()
+        self.close()
+        for task_id in list(self._bg_tasks):
+            self.background_kill(task_id)
 
     def _result(
         self, command, exit_code, output, *, timed_out, truncated=False, error=None

tests/test_server.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import asyncio
+
 import pytest
 from fastapi.testclient import TestClient
 
@@ -759,6 +761,8 @@ def test_workspace_command_trust_controls_live_engine(tmp_path):
         )
         assert not after.allowed and after.needs_user
 
+    manager.delete_session("trust")
+    assert "trust" not in manager._engines
     manager.workspace_trust.set_trusted(proj, True)
     proj.rename(tmp_path / "moved-project")
     assert client.post(
@@ -768,6 +772,39 @@ def test_workspace_command_trust_controls_live_engine(tmp_path):
     assert manager.trusted_workspaces() == []
 
 
+def test_delete_session_stops_its_shell_and_releases_workspace(tmp_path):
+    project = tmp_path / "project"
+    project.mkdir()
+    manager = SessionManager(
+        data_dir=tmp_path / "data", provider=ScriptedProvider([])
+    )
+    engine = manager.get_engine("session", workspace=str(project))
+    assert engine is not None
+    shell = engine.executor._proc
+    manager.save("session", engine)
+
+    assert manager.delete_session("session")["ok"] is True
+    assert shell.poll() is not None
+
+    project.rename(tmp_path / "renamed-project")
+    asyncio.run(manager.aclose())
+
+
+def test_manager_aclose_stops_live_session_shells(tmp_path):
+    project = tmp_path / "project"
+    project.mkdir()
+    manager = SessionManager(
+        data_dir=tmp_path / "data", provider=ScriptedProvider([])
+    )
+    engine = manager.get_engine("session", workspace=str(project))
+    assert engine is not None
+    shell = engine.executor._proc
+
+    asyncio.run(manager.aclose())
+
+    assert shell.poll() is not None
+
+
 def test_recent_workspaces_exclude_scratch_dirs(tmp_path):
     # Scratch dirs get touched like any workspace, but must never show up as
     # "recent projects" in the folder gate (owner call, 2026-07-03).

tests/test_shell.py

@@ -208,6 +208,21 @@ def test_background_task_kill(executor):
     assert res["status"] == "exited"
 
 
+def test_shutdown_stops_shell_and_background_tasks(tmp_path):
+    ex = LocalExecutor(cwd=tmp_path, default_timeout=10)
+    try:
+        shell = ex._proc
+        started = ex.run_background(ECHO_THEN_SLEEP)
+        task = ex._bg_tasks[started["task_id"]]
+
+        ex.shutdown()
+
+        assert shell.poll() is not None
+        assert task.proc.poll() is not None
+    finally:
+        ex.shutdown()
+
+
 def test_background_unknown_task_errors(executor):
     reg = ToolRegistry()
     reg.register_all(shell_tools(executor))