fix(auth): /ws/events applies the account-state check the other doors apply (trinity-enterprise#858)

src/backend/dependencies.py

@@ -1854,6 +1854,14 @@ def owner_is_workspace_only(owner_row: Optional[dict]) -> bool:
     return is_workspace_only_role((owner_row or {}).get("role"))
 
 
+def owner_is_suspended(owner_row: Optional[dict]) -> bool:
+    """A door that resolves a key's owner ROW itself (`/ws/events`) applies the
+    #995 deactivation check `get_current_user` and `/ws` apply: a suspended
+    owner's still-active key is refused. An unresolvable owner is refused by
+    `owner_is_workspace_only`, so None reads as "not suspended" here."""
+    return bool((owner_row or {}).get("suspended_at"))
+
+
 def require_role(min_role: str):
     """
     Dependency factory that requires the current user to have at least `min_role`.

src/backend/main.py

@@ -26,7 +26,7 @@
 
 from config import CORS_ORIGINS, VOICE_ENABLED
 from models import User
-from dependencies import get_current_user, owner_is_workspace_only, scope_may_open_event_stream, workspace_route
+from dependencies import get_current_user, owner_is_suspended, owner_is_workspace_only, scope_may_open_event_stream, workspace_route
 from services.docker_service import docker_client, list_all_agents_fast
 from utils.helpers import utc_now_iso
 
@@ -1681,6 +1681,12 @@ async def websocket_events_endpoint(
     user_email = key_info.get("user_email")
     # Determine if admin by checking user role
     user_data = db.get_user_by_username(key_info.get("user_id"))  # user_id is actually username
+    # #995 — a suspended owner's key is refused here as on every other door
+    # (`get_current_user`, `/ws`). Same close as an inactive key: the holder
+    # learns nothing more than "this key does not work".
+    if owner_is_suspended(user_data):
+        await websocket.close(code=4001, reason="Invalid or inactive MCP API key")
+        return
     # ent#837 — the event stream is an operator surface, and this handler never
     # runs `get_current_user`, so the operator floor is applied here: a key whose
     # owner is Workspace-only (including one minted before the upgrade) does not

tests/registry.json

@@ -6169,6 +6169,17 @@
         "agent-lifecycle"
       ],
       "description": "recreate_container_with_updated_config re-bakes AGENT_NAME and the trinity.agent-name label from the live name beside TRINITY_AGENT_AUTH_TOKEN (driven through the real recreate, red before the fix on the replayed old name); check_agent_name_env_matches flags a present-and-different env or label, matches absent values and unreadable attrs; start_agent_internal recreates a stuck renamed agent on COLD start only and leaves never-renamed and running agents alone."
+    },
+    {
+      "file": "unit/test_ent858_event_stream_owner_state.py",
+      "feature": "trinity-enterprise#858",
+      "added": "2026-10-20",
+      "categories": [
+        "backend",
+        "auth",
+        "unit"
+      ],
+      "description": "/ws/events applies the #995 account-state check the other doors apply: dependencies.owner_is_suspended truth table (executed); the handler checks it on the owner row it resolves, closes and returns before accept() (shape check, main.py is not importable in the unit island); the close is identical to an inactive key's. Mutation: removing the guard turns both placement tests red."
     }
   ]
 }

tests/unit/test_ent858_event_stream_owner_state.py

@@ -0,0 +1,90 @@
+"""trinity-enterprise#858 — `/ws/events` applies the #995 account-state check
+every other door applies.
+
+`get_current_user` refuses a key whose owner is suspended, and so does `/ws`
+(`resolve_ws_identity`). `/ws/events` authenticates its MCP key itself and never
+runs `get_current_user`, so the same check sits in the handler, on the owner row
+it resolves, before `accept()`.
+
+The owner check is executed here (`dependencies.owner_is_suspended`); the
+placement is a shape check, because `main.py` is not importable in the unit
+island — the same split as the ent#837 owner floor in
+`test_837_workspace_only_floor.py`.
+"""
+from __future__ import annotations
+
+import ast
+import pathlib
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+BACKEND = pathlib.Path(__file__).resolve().parents[2] / "src" / "backend"
+
+
+@pytest.mark.parametrize("row,refused", [
+    ({"suspended_at": "2026-10-20T10:00:00Z", "role": "admin"}, True),
+    ({"suspended_at": "2026-10-20T10:00:00Z", "role": "operator"}, True),
+    ({"suspended_at": None, "role": "admin"}, False),
+    ({"suspended_at": "", "role": "admin"}, False),
+    ({"role": "operator"}, False),
+    (None, False),   # unresolvable owner: owner_is_workspace_only refuses it
+])
+def test_the_owner_check(row, refused):
+    import dependencies as deps
+
+    assert deps.owner_is_suspended(row) is refused
+
+
+def _handler():
+    tree = ast.parse((BACKEND / "main.py").read_text(encoding="utf-8"))
+    return next(n for n in ast.walk(tree)
+                if isinstance(n, ast.AsyncFunctionDef) and n.name == "websocket_events_endpoint")
+
+
+def _closes(node):
+    """(code, reason) of every `websocket.close(...)` inside `node`'s body."""
+    out = []
+    for c in ast.walk(ast.Module(body=node.body, type_ignores=[])):
+        if isinstance(c, ast.Call) and getattr(c.func, "attr", None) == "close":
+            kw = {k.arg: getattr(k.value, "value", None) for k in c.keywords}
+            out.append((kw.get("code"), kw.get("reason")))
+    return out
+
+
+def _guard_calling(fn, name):
+    """The `if <name>(user_data): ...` guard in `fn`, or None."""
+    for node in ast.walk(fn):
+        if isinstance(node, ast.If) and any(
+            isinstance(c, ast.Call)
+            and getattr(c.func, "id", getattr(c.func, "attr", None)) == name
+            and len(c.args) == 1 and isinstance(c.args[0], ast.Name) and c.args[0].id == "user_data"
+            for c in ast.walk(node.test)
+        ):
+            return node
+    return None
+
+
+def test_the_event_stream_refuses_a_suspended_owner_before_accepting():
+    fn = _handler()
+    accept_line = min(n.lineno for n in ast.walk(fn)
+                      if isinstance(n, ast.Call) and getattr(n.func, "attr", None) == "accept")
+    owner_lookup = next(n for n in ast.walk(fn) if isinstance(n, ast.Assign)
+                        and any(isinstance(t, ast.Name) and t.id == "user_data" for t in n.targets))
+    guard = _guard_calling(fn, "owner_is_suspended")
+    assert guard is not None, "websocket_events_endpoint must check owner_is_suspended(user_data)"
+    assert owner_lookup.lineno < guard.lineno < accept_line
+    assert any(isinstance(s, ast.Return) for s in guard.body)
+    assert _closes(guard), "the suspended-owner guard must close the socket"
+
+
+def test_the_refusal_reads_exactly_like_an_inactive_key():
+    """No oracle: a suspended owner's key gets the same close as a dead key."""
+    fn = _handler()
+    inactive = next(
+        node for node in ast.walk(fn)
+        if isinstance(node, ast.If) and isinstance(node.test, ast.UnaryOp)
+        and isinstance(node.test.op, ast.Not) and getattr(node.test.operand, "id", None) == "key_info"
+    )
+    assert _closes(_guard_calling(fn, "owner_is_suspended")) == _closes(inactive)