[fix] - Ponytail-style tests-refactor pass over tests/ and .github/workflows

.github/workflows/ci.yml

@@ -823,9 +823,9 @@ jobs:
         # The wheel is downloaded once per (os, pin) pair and reused on every
         # subsequent run until the pin changes in pyproject.toml. macOS skips
         # this cache entirely -- see the macOS-specific install step below.
-        # The torch==2.13.0+cpu pin (and this cache key) is duplicated in
-        # pip-audit.yml, copilot-setup-steps.yml, and devskim.yml -- grep for
-        # the version string under .github/workflows/ before bumping it here.
+        # The torch==2.13.0+cpu pin (and this cache key) is repeated in other
+        # workflows -- grep for the version string under .github/workflows/
+        # before bumping it here.
         id: torch-wheel-cache
         uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9  # v6.1.0
         with:

.github/workflows/pip-audit.yml

@@ -97,11 +97,9 @@ jobs:
         # requirements.txt/constraints.txt triggers don't force a re-download of
         # the heaviest dependency on every run. macOS skips this cache entirely
         # -- see the macOS-specific install step below (same split ci.yml uses).
-        # The torch==2.13.0+cpu pin (and this cache key) is duplicated in
-        # ci.yml and copilot-setup-steps.yml -- grep for the version string
-        # under .github/workflows/ before bumping it here. (devskim.yml was
-        # listed too until it dropped its install entirely; it now has no torch
-        # pin, so that cross-reference would send a future bump on a dead end.)
+        # The torch==2.13.0+cpu pin (and this cache key) is repeated in other
+        # workflows -- grep for the version string under .github/workflows/
+        # before bumping it here.
         id: torch-wheel-cache
         uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9  # v6.1.0
         with:

tests/README.md

@@ -85,6 +85,7 @@ auto-discovered and need no coverage-source entry.
 | `apipsTest.ps1`, `cmd2index.bat` | Windows-side manual helpers; not collected by pytest. |
 | `nemo_runtime/` | Real NeMo engine, combined NeMo/Numbat CLI/CEL gateway HTTP acceptance, and logging privacy checks with a loopback-only harness (`network_jail.py`, `mock_openai.py`); collected with the main suite and skipped unless `CYCLAW_NEMO_RUNTIME=1`. |
 | `executor_sandbox_double.py`, `spend_live_probe.py` | Helper doubles/probes, not `test_*`-named, so not collected. |
+| `import_graph.py` | The AST import walker shared by the six `test_*_isolation.py` I6 guards; not collected. |
 | `numbat_shaped_events.py` | Drives one representative event per producer family (mainline audit projection, pre-action hook verdicts, CEL monitor, ops/fsconnect/sqlconnect; not the executor or `real_repo_loop`) into a throwaway stream; `python -m tests.numbat_shaped_events --frozen --out ...` regenerates `fixtures/numbat/cyclaw-shaped-*.ndjson`. Not collected; `test_numbat_cyclaw_shaped.py` and `numbat-rules.yml`'s `numbat-stream-contract` job hold its output to the pinned Numbat CLI and schema (issue #1458). |
 
 ## Conventions that bite

tests/import_graph.py

@@ -0,0 +1,21 @@
+"""Shared AST helper for the six ``test_*_isolation.py`` I6 guards.
+
+Each guard used to carry its own byte-identical copy of this walker; one copy
+means a fix to what counts as an import reaches every guard at once.
+"""
+
+from __future__ import annotations
+
+import ast
+
+
+def all_imports(source: str) -> set[str]:
+    """Top-level package names imported anywhere in ``source`` (absolute imports only)."""
+    names: set[str] = set()
+    for node in ast.walk(ast.parse(source)):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                names.add(alias.name.split(".")[0])
+        elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
+            names.add(node.module.split(".")[0])
+    return names

tests/test_agentic_isolation.py

@@ -10,29 +10,16 @@
 
 from __future__ import annotations
 
-import ast
 from pathlib import Path
 
 import pytest
 
+from tests.import_graph import all_imports as _imports
+
 REPO_ROOT = Path(__file__).resolve().parent.parent
 REQUEST_PATH_MODULES = ["gate.py", "gate_ops.py", "gate_auth.py", "gate_memory.py", "graph.py", "mcp_hybrid_server.py"]
 
 
-def _imports(source: str) -> set[str]:
-    """Top-level module names imported by ``source`` (import X / from X import ...)."""
-    names: set[str] = set()
-    tree = ast.parse(source)
-    for node in ast.walk(tree):
-        if isinstance(node, ast.Import):
-            for alias in node.names:
-                names.add(alias.name.split(".")[0])
-        elif isinstance(node, ast.ImportFrom):
-            if node.module and node.level == 0:
-                names.add(node.module.split(".")[0])
-    return names
-
-
 @pytest.mark.parametrize("module_file", REQUEST_PATH_MODULES)
 def test_request_path_does_not_import_agentic(module_file):
     source = (REPO_ROOT / module_file).read_text(encoding="utf-8")

tests/test_bounded_executor.py

@@ -0,0 +1,101 @@
+"""BoundedExecutor: the real capacity bound behind /query and the auth workers.
+
+The gate tests replace it with an AsyncMock, so these exercise the semaphore
+itself. Threads are coordinated with Events, never sleeps.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextvars
+import threading
+
+import pytest
+
+from utils.bounded_executor import BoundedExecutor, WorkCapacityExceeded
+
+
+@pytest.mark.parametrize("workers", [0, -1, True, 2.0, "2"])
+def test_capacity_must_be_a_positive_int(workers):
+    """Zero, negatives, bools and non-ints are refused at construction."""
+    with pytest.raises(ValueError, match="positive integer"):
+        BoundedExecutor(workers, name="bad")
+
+
+def test_full_capacity_rejects_then_frees_after_completion():
+    """A call beyond capacity is refused immediately; capacity returns once the worker finishes."""
+    ex = BoundedExecutor(1, name="bound")
+    started, release = threading.Event(), threading.Event()
+
+    def blocking() -> str:
+        started.set()
+        release.wait()
+        return "first"
+
+    async def scenario() -> list[str]:
+        first = asyncio.create_task(ex.run(blocking))
+        await asyncio.to_thread(started.wait)
+        with pytest.raises(WorkCapacityExceeded):
+            await ex.run(lambda: "refused")
+        release.set()
+        return [await first, await ex.run(lambda: "second")]
+
+    try:
+        assert asyncio.run(scenario()) == ["first", "second"]
+    finally:
+        release.set()
+        ex.close()
+
+
+def test_cancelled_caller_keeps_capacity_until_worker_finishes():
+    """Cancelling the awaiting request does not free the slot while its worker still runs."""
+    ex = BoundedExecutor(1, name="cancel")
+    started, release = threading.Event(), threading.Event()
+
+    def blocking() -> None:
+        started.set()
+        release.wait()
+
+    async def scenario() -> None:
+        task = asyncio.create_task(ex.run(blocking))
+        await asyncio.to_thread(started.wait)
+        task.cancel()
+        (outcome,) = await asyncio.gather(task, return_exceptions=True)
+        assert isinstance(outcome, asyncio.CancelledError)
+        with pytest.raises(WorkCapacityExceeded):
+            await ex.run(lambda: None)
+
+    try:
+        asyncio.run(scenario())
+    finally:
+        release.set()
+        ex.close()
+
+
+def test_failed_submit_releases_its_slot():
+    """A submit that raises (executor closed) gives the slot back instead of leaking it."""
+    ex = BoundedExecutor(1, name="closed")
+    ex.close()
+
+    async def scenario() -> None:
+        for _ in range(2):
+            # A leaked slot would turn the second attempt into WorkCapacityExceeded.
+            with pytest.raises(RuntimeError):
+                await ex.run(lambda: None)
+
+    asyncio.run(scenario())
+
+
+def test_worker_sees_the_callers_context():
+    """Context variables set by the caller are visible inside the worker thread."""
+    request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="unset")
+    ex = BoundedExecutor(1, name="ctx")
+
+    async def scenario() -> str:
+        request_id.set("req-42")
+        return await ex.run(request_id.get)
+
+    try:
+        assert asyncio.run(scenario()) == "req-42"
+    finally:
+        ex.close()

tests/test_gate_auth_admin.py

@@ -26,11 +26,12 @@ async def _allow(_request: Request) -> None:
     return None
 
 
-def _client(manager):
+def _client(manager, events=None):
     app = FastAPI()
 
-    async def audit(_event):
-        return None
+    async def audit(event):
+        if events is not None:
+            events.append(event)
 
     register_auth_routes(
         app, _cfg(), audit=audit, enforce_rate_limit=_allow, auth_manager=manager,
@@ -174,3 +175,101 @@ def test_operator_cannot_open_audit_summary(self, manager):
         client = _client(manager)
         _login(client, "alice", _GOOD)
         assert client.get("/auth/audit/summary").status_code == 403
+
+
+class TestAdminLifecycle:
+    """Success and refusal paths of role/disable/enable/delete, per the CLAUDE.md route table."""
+
+    def test_last_admin_cannot_be_demoted_disabled_or_deleted(self, manager):
+        manager.create_user("root", _GOOD, role="admin")
+        client = _client(manager)
+        h = {"x-cyclaw-csrf": _login(client, "root", _GOOD)}
+        attempts = {
+            "set_role": client.post("/auth/users/root/role", json={"role": "operator"}, headers=h),
+            "disable_user": client.post("/auth/users/root/disable", headers=h),
+            "delete_user": client.delete("/auth/users/root", headers=h),
+        }
+        for action, r in attempts.items():
+            assert r.status_code == 403, action
+            assert r.json()["detail"]["code"] == "AUTH_LAST_ADMIN"
+            assert r.json()["detail"]["details"]["action"] == action
+        assert manager.get_user("root").role == "admin"
+
+    def test_admin_sets_role_and_it_is_audited(self, manager):
+        manager.create_user("root", _GOOD, role="admin")
+        manager.create_user("bob", _GOOD, role="operator")
+        events: list[dict] = []
+        client = _client(manager, events)
+        h = {"x-cyclaw-csrf": _login(client, "root", _GOOD)}
+        bad = client.post("/auth/users/bob/role", json={"role": "god"}, headers=h)
+        assert bad.status_code == 422
+        assert bad.json()["detail"]["code"] == "AUTH_POLICY"
+        r = client.post("/auth/users/bob/role", json={"role": "admin"}, headers=h)
+        assert r.status_code == 200
+        assert manager.get_user("bob").role == "admin"
+        assert {"event": "auth_role_set", "username": "root", "target": "bob", "role": "admin"} in events
+
+    def test_disable_blocks_login_until_enable(self, manager):
+        manager.create_user("root", _GOOD, role="admin")
+        manager.create_user("bob", _GOOD, role="operator")
+        events: list[dict] = []
+        client = _client(manager, events)
+        h = {"x-cyclaw-csrf": _login(client, "root", _GOOD)}
+        other = _client(manager)
+        r = client.post("/auth/users/bob/disable", headers=h)
+        assert r.status_code == 200
+        r = other.post("/auth/login", json={"username": "bob", "password": _GOOD})
+        assert r.status_code == 401
+        r = client.post("/auth/users/bob/enable", headers=h)
+        assert r.status_code == 200
+        r = other.post("/auth/login", json={"username": "bob", "password": _GOOD})
+        assert r.status_code == 200
+        kinds = [e["event"] for e in events if e.get("target") == "bob"]
+        assert kinds == ["auth_user_disabled", "auth_user_enabled"]
+
+    def test_operator_toggles_operators_but_not_admins(self, manager):
+        manager.create_user("root", _GOOD, role="admin")
+        manager.create_user("alice", _GOOD, role="operator")
+        manager.create_user("carol", _GOOD, role="operator")
+        client = _client(manager)
+        h = {"x-cyclaw-csrf": _login(client, "alice", _GOOD)}
+        r = client.post("/auth/users/carol/disable", headers=h)
+        assert r.status_code == 200
+        r = client.post("/auth/users/carol/enable", headers=h)
+        assert r.status_code == 200
+        r = client.post("/auth/users/root/disable", headers=h)
+        assert r.status_code == 403
+        assert r.json()["detail"]["code"] == "AUTH_PERMISSION_DENIED"
+        assert not manager.get_user("root").disabled
+
+    def test_admin_hard_deletes_and_unknown_users_are_404(self, manager):
+        manager.create_user("root", _GOOD, role="admin")
+        manager.create_user("carol", _GOOD, role="operator")
+        events: list[dict] = []
+        client = _client(manager, events)
+        h = {"x-cyclaw-csrf": _login(client, "root", _GOOD)}
+        r = client.delete("/auth/users/carol", headers=h)
+        assert r.status_code == 200
+        assert manager.get_user("carol") is None
+        assert {"event": "auth_user_deleted", "username": "root", "target": "carol"} in events
+        for r in (
+            client.delete("/auth/users/ghost", headers=h),
+            client.post("/auth/users/ghost/disable", headers=h),
+            client.post("/auth/users/ghost/enable", headers=h),
+        ):
+            assert r.status_code == 404
+            assert r.json()["detail"]["code"] == "AUTH_USER_NOT_FOUND"
+
+    def test_lifecycle_writes_need_csrf(self, manager):
+        manager.create_user("root", _GOOD, role="admin")
+        manager.create_user("bob", _GOOD, role="operator")
+        client = _client(manager)
+        _login(client, "root", _GOOD)
+        for r in (
+            client.post("/auth/users/bob/disable"),
+            client.post("/auth/users/bob/role", json={"role": "audit"}),
+            client.delete("/auth/users/bob"),
+        ):
+            assert r.status_code == 403
+            assert r.json()["detail"]["code"] == "CSRF_TOKEN_INVALID"
+        assert not manager.get_user("bob").disabled

tests/test_gate_query_auth.py

@@ -167,6 +167,9 @@ def test_session_cookie_works(self, manager, user):
         assert r.json()["username"] == username
 
     def test_device_token_works(self, manager, user):
+        """The realistic device-token case: no Sec-Fetch-Site/Origin at all
+        (curl, PowerShell, Telegram's client) -- _looks_cross_site's absent-
+        header allowance passes it straight through."""
         username, _password = user
         token = manager.create_device_token(username, "telegram")
         r = _client(manager).post(
@@ -257,18 +260,6 @@ def test_a_forged_cross_site_header_is_rejected_even_with_a_valid_device_token(s
         assert r.status_code == 403
         assert r.json()["detail"]["code"] == "CROSS_SITE_BLOCKED"
 
-    def test_device_token_caller_with_no_browser_headers_still_works(self, manager, user):
-        """The realistic device-token case: no Sec-Fetch-Site/Origin at all
-        (curl, PowerShell, Telegram's client) -- _looks_cross_site's absent-
-        header allowance passes it straight through."""
-        username, _password = user
-        token = manager.create_device_token(username, "telegram")
-        r = _client(manager).post(
-            "/query", json={}, headers={"Authorization": f"Bearer {token}"},
-        )
-        assert r.status_code == 200
-        assert r.json()["username"] == username
-
 
 class TestAttachFlipsProbe:
     def test_attach_makes_enforcement_probe_true(self, manager):

tests/test_guardrails_isolation.py

@@ -10,28 +10,16 @@
 
 from __future__ import annotations
 
-import ast
 from pathlib import Path
 
 import pytest
 
+from tests.import_graph import all_imports as _imports
+
 REPO_ROOT = Path(__file__).resolve().parent.parent
 REQUEST_PATH_MODULES = ["gate.py", "gate_ops.py", "gate_auth.py", "gate_memory.py", "graph.py", "mcp_hybrid_server.py"]
 
 
-def _imports(source: str) -> set[str]:
-    names: set[str] = set()
-    tree = ast.parse(source)
-    for node in ast.walk(tree):
-        if isinstance(node, ast.Import):
-            for alias in node.names:
-                names.add(alias.name.split(".")[0])
-        elif isinstance(node, ast.ImportFrom):
-            if node.module and node.level == 0:
-                names.add(node.module.split(".")[0])
-    return names
-
-
 @pytest.mark.parametrize("module_file", REQUEST_PATH_MODULES)
 def test_request_path_does_not_import_guardrails(module_file):
     source = (REPO_ROOT / module_file).read_text(encoding="utf-8")

tests/test_memory_isolation.py

@@ -12,6 +12,8 @@
 
 import pytest
 
+from tests.import_graph import all_imports as _all_imports
+
 REPO_ROOT = Path(__file__).resolve().parent.parent
 
 # Modules that must not top-level-import the memory package.
@@ -55,19 +57,6 @@ def _toplevel_imports(source: str) -> set[str]:
     return names
 
 
-def _all_imports(source: str) -> set[str]:
-    names: set[str] = set()
-    tree = ast.parse(source)
-    for node in ast.walk(tree):
-        if isinstance(node, ast.Import):
-            for alias in node.names:
-                names.add(alias.name.split(".")[0])
-        elif isinstance(node, ast.ImportFrom):
-            if node.module and node.level == 0:
-                names.add(node.module.split(".")[0])
-    return names
-
-
 @pytest.mark.parametrize("module_file", NO_TOPLEVEL_MEMORY)
 def test_no_toplevel_import_of_memory(module_file: str):
     path = REPO_ROOT / module_file