fix: resolve remaining P2/P3 issues in WS notification PR

- **Task 1 (auth-disabled installs):** Check  instead
  of  so explicit auth-disabled mode is honored even
  when users exist in the DB. Normalize empty/falsy owner to
  in  note dict and  comparison.
- **Task 2 (closed-WS cleanup):** Replace  in the
  send loop with  so any send failure exits the stream loop to the
   block, which properly unsubscribes the stale queue.
- **Task 3 (token collision):** Extract shared  that
  iterates  prefix-matched rows. Both  and
   now use it, fixing the bug where
  only checked  and could wrongly reject a valid colliding token.

Author: Chukwuebuka-2003

routes/ws_routes.py

@@ -63,6 +63,19 @@ def _validate_api_token(raw_token: str) -> str | None:
         return None
     if len(raw_token) < 12 or len(raw_token) > 100:
         return None
+    match = _find_matching_token(raw_token)
+    if match is None:
+        return None
+    row, _ = match
+    scopes = [s.strip() for s in (row.scopes or "").split(",") if s.strip()]
+    if "notifications:read" not in scopes:
+        return None
+    return row.owner
+
+
+def _find_matching_token(raw_token: str) -> tuple | None:
+    """Find the ApiToken row matching *raw_token* among active rows with the
+    same 8-char prefix.  Returns ``(row, raw_token)`` on match or ``None``."""
     prefix = raw_token[:8]
     try:
         with get_db_session() as db:
@@ -73,35 +86,26 @@ def _validate_api_token(raw_token: str) -> str | None:
             )
             for row in rows:
                 if bcrypt.checkpw(raw_token.encode(), row.token_hash.encode()):
-                    scopes = [s.strip() for s in (row.scopes or "").split(",") if s.strip()]
-                    if "notifications:read" not in scopes:
-                        return None
-                    return row.owner
+                    return row, raw_token
     except Exception:
-        logger.warning("API token validation failed", exc_info=True)
+        logger.warning("API token lookup failed", exc_info=True)
     return None
 
 
 def _revalidate_api_token(raw_token: str, expected_owner: str) -> bool:
-    """Re-validate an API token is still active and maps to *expected_owner*
-    and still has the ``notifications:read`` scope."""
-    prefix = raw_token[:8]
-    try:
-        with get_db_session() as db:
-            row = (
-                db.query(ApiToken)
-                .filter(
-                    ApiToken.token_prefix == prefix,
-                    ApiToken.is_active == True,
-                )
-                .first()
-            )
-            if row and bcrypt.checkpw(raw_token.encode(), row.token_hash.encode()):
-                scopes = [s.strip() for s in (row.scopes or "").split(",") if s.strip()]
-                return row.owner == expected_owner and "notifications:read" in scopes
-    except Exception:
-        logger.warning("API token re-validation failed", exc_info=True)
-    return False
+    """Re-validate an API token is still active, maps to *expected_owner*,
+    and still has the ``notifications:read`` scope.
+
+    Iterates *all* prefix-matched rows (same as ``_validate_api_token``) so
+    that a valid token whose 8-char prefix collides with another active token
+    is not wrongly rejected on the first per-delivery check.
+    """
+    match = _find_matching_token(raw_token)
+    if match is None:
+        return False
+    row, _ = match
+    scopes = [s.strip() for s in (row.scopes or "").split(",") if s.strip()]
+    return row.owner == expected_owner and "notifications:read" in scopes
 
 
 def setup_ws_routes():
@@ -158,13 +162,14 @@ async def ws_notifications(websocket: WebSocket):
                     owner = auth_mgr.get_username_for_token(session_id)
                     used_credential = session_id
 
-        # Fallback: if auth is disabled, allow anonymous
-        if not owner and not (auth_mgr and auth_mgr.is_configured):
-            owner = _ANONYMOUS_OWNER
-
-        if owner is None:
-            await websocket.close(code=4001)
-            return
+        # Fallback: if auth is explicitly disabled at the app level, allow anonymous
+        if not owner:
+            _auth_disabled = not getattr(websocket.app.state, "auth_enabled", True)
+            if _auth_disabled:
+                owner = _ANONYMOUS_OWNER
+            else:
+                await websocket.close(code=4001)
+                return
 
         # ── Subscribe and stream ─────────────────────────────────────────
         q = _subscribe(owner)
@@ -188,7 +193,9 @@ async def ws_notifications(websocket: WebSocket):
                 except WebSocketDisconnect:
                     raise
                 except Exception:
-                    pass
+                    # Any send failure means the client is gone; exit the loop
+                    # so the finally block removes the subscriber queue.
+                    return
         except WebSocketDisconnect:
             pass
         finally:

src/task_scheduler.py

@@ -403,11 +403,13 @@ def add_notification(self, task_name: str, status: str, task_id: str = None, own
         notifications and prevent cross-tenant drain. `body` is the result
         text — populated when output_target='notification' so the client can
         show a rich browser Notification, not just a toast."""
+        from routes.ws_routes import _ANONYMOUS_OWNER
+        note_owner = _ANONYMOUS_OWNER if (owner is not None and not owner) else owner
         note = {
             "task_name": task_name,
             "status": status,
             "task_id": task_id,
-            "owner": owner,
+            "owner": note_owner,
             "body": (body[:500] + "…") if body and len(body) > 500 else body,
             "timestamp": _utcnow().isoformat() + "Z",
         }
@@ -417,13 +419,8 @@ def add_notification(self, task_name: str, status: str, task_id: str = None, own
             self._pending_notifications = self._pending_notifications[-50:]
         # Push to WebSocket subscribers if the channel is available
         try:
-            from routes.ws_routes import push_notification, _ANONYMOUS_OWNER
-            if owner is not None:
-                push_notification(owner, note)
-            else:
-                # Auth-disabled path: push to the anonymous channel so
-                # the WebSocket subscriber receives the notification.
-                push_notification(_ANONYMOUS_OWNER, note)
+            from routes.ws_routes import push_notification
+            push_notification(note_owner, note)
         except Exception:
             pass
 
@@ -436,7 +433,9 @@ def pop_notifications(self, owner: str = None) -> list:
         or when no owner filter is given — preserves backward behaviour
         for the legacy single-user deploy.
         """
-        if owner is None:
+        from routes.ws_routes import _ANONYMOUS_OWNER
+        _owner = _ANONYMOUS_OWNER if (owner is not None and not owner) else owner
+        if _owner is None:
             notes = self._pending_notifications[:]
             self._pending_notifications.clear()
             return notes
@@ -445,7 +444,7 @@ def pop_notifications(self, owner: str = None) -> list:
         # any authenticated user once a second account existed.
         keep, take = [], []
         for n in self._pending_notifications:
-            if n.get("owner") == owner:
+            if n.get("owner") == _owner:
                 take.append(n)
             else:
                 keep.append(n)