test(conftest): kill CCB tmux daemons leaked by subprocess ccb calls

test/conftest.py

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
+import glob
+import logging
 import os
 import stat
+import subprocess
 import sys
 from pathlib import Path
 
@@ -81,3 +84,63 @@ def _install_provider_stubs(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) ->
     monkeypatch.setenv("STUB_DELAY", "1.5")
     monkeypatch.setenv("CCB_REPLY_LANG", "en")
     monkeypatch.setenv("CCB_CLAUDE_SKILLS", "0")
+
+
+# ============================================================
+# CCB tmux daemon leak cleanup.
+# Tests in test_v2_phase2_entrypoint.py spawn `ccb` via subprocess.
+# Those subprocesses start the CCB keeper which wraps itself in an
+# independent tmux daemon (socket under tmp_path/.ccb/ccbd/tmux.sock
+# or /run/user/$UID/ccb-runtime/tmux-*.sock). The test's in-process
+# app.shutdown() cannot reach those daemons.
+# ============================================================
+
+_leak_logger = logging.getLogger(__name__)
+
+
+def _safe_kill_tmux_server(sock: str) -> None:
+    try:
+        result = subprocess.run(
+            ['tmux', '-S', sock, 'kill-server'],
+            timeout=5, check=False,
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+        )
+        if result.returncode != 0:
+            _leak_logger.warning(
+                "cleanup: tmux kill-server %s returned rc=%d", sock, result.returncode
+            )
+    except FileNotFoundError:
+        return
+    except subprocess.TimeoutExpired:
+        _leak_logger.warning("cleanup: tmux kill-server %s timed out after 5s", sock)
+
+
+def _runtime_socket_pattern() -> str:
+    return f'/run/user/{os.getuid()}/ccb-runtime/tmux-*.sock'
+
+
+@pytest.fixture(autouse=True)
+def _cleanup_ccb_tmux_per_test(tmp_path):
+    before = set(glob.glob(_runtime_socket_pattern()))
+    try:
+        yield
+    finally:
+        for sock_path in tmp_path.rglob('.ccb/ccbd/tmux.sock'):
+            _safe_kill_tmux_server(str(sock_path))
+        after = set(glob.glob(_runtime_socket_pattern()))
+        for sock in after - before:
+            _safe_kill_tmux_server(sock)
+
+
+@pytest.fixture(autouse=True, scope='session')
+def _cleanup_ccb_tmux_session_end(tmp_path_factory):
+    before = set(glob.glob(_runtime_socket_pattern()))
+    try:
+        yield
+    finally:
+        base_tmp = tmp_path_factory.getbasetemp()
+        for sock_path in base_tmp.rglob('.ccb/ccbd/tmux.sock'):
+            _safe_kill_tmux_server(str(sock_path))
+        after = set(glob.glob(_runtime_socket_pattern()))
+        for sock in after - before:
+            _safe_kill_tmux_server(sock)