feat(api): serve long-lived streams on the event loop

Move GET /api/events, the remote-device command stream and artifact
downloads from Flask to Starlette routes mounted ahead of the Flask
catch-all. On Flask each held an a2wsgi threadpool slot for as long as
its response stayed open, and because uvicorn drops writes after a
client disconnects, a closed tab never released it.

- asgi_auth: one JWT/OIDC gate for Starlette routes; the chat reconnect
  reader uses it too
- ClosingStreamingResponse closes the body iterator and releases the
  SSE slot or file handle even when the client leaves before the first
  frame
- AsyncTopic liveness probe replaces the sync client's socket_timeout
  guard against half-open pub/sub sockets
- ASYNC_REDIS_MAX_CONNECTIONS sizes the async Redis pool; every open
  stream holds a connection and redis-py defaults to 100

docsgpt/api/asgi_auth.py

@@ -0,0 +1,59 @@
+"""Request authentication for the native-async (Starlette) routes.
+
+Mirrors ``authenticate_request`` and the log-context hooks in
+``docsgpt/app.py`` so a route moved off Flask keeps the same gates: the same
+JWT decoder and 401 payloads, and the OIDC revocation check.
+"""
+
+from __future__ import annotations
+
+import uuid
+from contextvars import Token
+from typing import Optional, Tuple
+
+import anyio
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+
+from docsgpt.api.oidc.denylist import is_denied as oidc_session_denied
+from docsgpt.auth import handle_auth
+from docsgpt.core import log_context
+from docsgpt.core.settings import settings
+
+
+async def authenticate(request: Request) -> Tuple[Optional[dict], Optional[JSONResponse]]:
+    """Decode the caller's JWT the way Flask's ``authenticate_request`` does.
+
+    Args:
+        request: The incoming Starlette request.
+
+    Returns:
+        tuple: ``(claims, None)`` for an authenticated caller, ``(None, None)``
+        when no token was sent (the route decides whether that is allowed), or
+        ``(None, response)`` carrying the 401 to return.
+    """
+    decoded = handle_auth(request)
+    if not decoded:
+        return None, None
+    if "error" in decoded:
+        return None, JSONResponse(decoded, status_code=401)
+    # The denylist is a sync Redis read; keep it off the event loop.
+    if settings.AUTH_TYPE == "oidc" and await anyio.to_thread.run_sync(oidc_session_denied, decoded):
+        return None, JSONResponse(
+            {"message": "Authentication error: session revoked", "error": "token_revoked"},
+            status_code=401,
+        )
+    return decoded, None
+
+
+def bind_log_context(endpoint: str, user_id: Optional[str] = None) -> Token:
+    """Stamp this request's log records with a fresh activity id, the endpoint and user.
+
+    Each request runs in its own task, so the binding ends with the request.
+    """
+    return log_context.bind(activity_id=str(uuid.uuid4()), endpoint=endpoint, user_id=user_id)
+
+
+def json_error(message: str, status_code: int) -> JSONResponse:
+    """Return the ``{"success": false, "message": ...}`` body the Flask routes use."""
+    return JSONResponse({"success": False, "message": message}, status_code=status_code)

docsgpt/api/asgi_stream.py

@@ -0,0 +1,44 @@
+"""Streaming response for native-async routes whose streams hold resources."""
+
+from __future__ import annotations
+
+import logging
+from typing import Awaitable, Callable, Optional
+
+import anyio
+from starlette.responses import StreamingResponse
+from starlette.types import Receive, Scope, Send
+
+logger = logging.getLogger(__name__)
+
+
+class ClosingStreamingResponse(StreamingResponse):
+    """``StreamingResponse`` that always finalizes its stream.
+
+    When a client disconnects, Starlette cancels the send loop and drops the
+    body iterator: a generator paused between frames, or one that never
+    started, would leave its cleanup to garbage collection. This closes the
+    iterator and then runs ``on_close`` once the response is over, shielded
+    from the cancellation that ended it.
+    """
+
+    def __init__(self, content, *, on_close: Optional[Callable[[], Awaitable[None]]] = None, **kwargs) -> None:
+        super().__init__(content, **kwargs)
+        self._on_close = on_close
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        try:
+            await super().__call__(scope, receive, send)
+        finally:
+            with anyio.CancelScope(shield=True):
+                aclose = getattr(self.body_iterator, "aclose", None)
+                if aclose is not None:
+                    try:
+                        await aclose()
+                    except Exception:
+                        logger.exception("closing the response stream failed")
+                if self._on_close is not None:
+                    try:
+                        await self._on_close()
+                    except Exception:
+                        logger.exception("response on_close hook failed")

docsgpt/api/async_sse.py

@@ -2,7 +2,7 @@
 
 These Starlette routes serve the chat-stream *reconnect* path on the event
 loop, so a long-lived, mostly-idle tail costs a coroutine instead of one of
-the 32 a2wsgi threadpool slots (see ``docsgpt/asgi.py``). They are the
+the a2wsgi threadpool slots (see ``docsgpt/asgi.py``). They are the
 sole reconnect reader — the old Flask blueprint has been removed. The heavy
 *producer* (``POST /api/answer/stream`` → agent → LLM) stays on the sync
 path untouched.
@@ -24,8 +24,7 @@ from starlette.requests import Request
 from starlette.responses import JSONResponse, StreamingResponse
 from starlette.routing import Route
 
-from docsgpt.api.oidc.denylist import is_denied as oidc_session_denied
-from docsgpt.auth import handle_auth
+from docsgpt.api.asgi_auth import authenticate
 from docsgpt.core.settings import settings
 from docsgpt.events.keys import connection_counter_key
 from docsgpt.storage.db.session import db_readonly
@@ -176,19 +175,14 @@ async def stream_message_events(request: Request) -> JSONResponse | StreamingRes
     cursor → per-user connection cap) then streams snapshot+tail off the
     event loop.
     """
-    # ``handle_auth`` only reads ``request.headers.get("Authorization")``;
-    # Starlette's headers are case-insensitive, so the Flask helper works
-    # verbatim. With AUTH_TYPE unset it returns ``{"sub": "local"}``.
-    decoded = handle_auth(request)
-    if isinstance(decoded, dict) and "error" in decoded:
-        return _json("Authentication error: invalid token", 401)
+    # Same JWT decoder and OIDC revocation check as the Flask routes. With
+    # AUTH_TYPE unset the caller resolves to ``{"sub": "local"}``.
+    decoded, error = await authenticate(request)
+    if error is not None:
+        return error
     user_id = decoded.get("sub") if isinstance(decoded, dict) else None
     if not user_id:
         return _json("Authentication required", 401)
-    if settings.AUTH_TYPE == "oidc" and await anyio.to_thread.run_sync(
-        oidc_session_denied, decoded
-    ):
-        return _json("Authentication error: session revoked", 401)
 
     message_id = request.path_params["message_id"]
     if not _MESSAGE_ID_RE.match(message_id):

docsgpt/api/devices/auth.py

@@ -6,7 +6,7 @@ import base64
 import hashlib
 import logging
 import time
-from typing import Optional, Tuple
+from typing import Callable, Mapping, Optional, Tuple
 
 from flask import jsonify, make_response, request
 
@@ -17,6 +17,9 @@ from docsgpt.storage.db.session import db_readonly, db_session
 
 logger = logging.getLogger(__name__)
 
+#: ``(error_code, http_status)`` for a rejected device request.
+DeviceAuthFailure = Tuple[str, int]
+
 
 def hash_session_token(token: str) -> str:
     """SHA-256 over the opaque session token. Stored in ``devices.token_hash``."""
@@ -45,32 +48,46 @@ def _canonical_payload(method: str, path: str, ts: str, body: bytes) -> str:
     return f"{method} {path} {ts} {body_hash}"
 
 
-def verify_device_session(*, touch: bool = True) -> Tuple[Optional[dict], Optional[tuple]]:
-    """Validate the device session token on the current request.
+def authenticate_device(
+    headers: Mapping[str, str],
+    method: str,
+    path: str,
+    get_body: Callable[[], bytes],
+    *,
+    touch: bool = True,
+) -> Tuple[Optional[dict], Optional[DeviceAuthFailure]]:
+    """Validate a device session token from the raw request parts.
 
-    Returns ``(device_row, None)`` on success or ``(None, response)`` on
-    failure, where ``response`` is a ``(json, status)`` Flask tuple.
+    Framework-neutral core shared by the Flask routes and the native-async
+    SSE stream, which calls it from a worker thread since it reads Postgres.
 
     Args:
+        headers: Request headers (a case-insensitive mapping).
+        method: HTTP method, as the CLI signed it.
+        path: Request path without the query string, as the CLI signed it.
+        get_body: Returns the raw body; only called when signatures are required.
         touch: When True, bump ``last_seen_at`` on the device row.
+
+    Returns:
+        tuple: ``(device_row, None)`` on success or ``(None, (code, status))``.
     """
-    auth = request.headers.get("Authorization", "")
+    auth = headers.get("Authorization", "")
     if not auth.lower().startswith("bearer "):
-        return None, _error("missing_token", 401)
+        return None, ("missing_token", 401)
     token = auth.split(" ", 1)[1].strip()
     if not token:
-        return None, _error("missing_token", 401)
+        return None, ("missing_token", 401)
 
     token_hash = hash_session_token(token)
     with db_readonly() as conn:
         device = DevicesRepository(conn).find_by_token_hash(token_hash)
     if device is None:
-        return None, _error("invalid_token", 401)
+        return None, ("invalid_token", 401)
 
     if settings.REMOTE_DEVICE_REQUIRE_SIGNATURE:
-        sig_error = _verify_signature(device)
-        if sig_error is not None:
-            return None, sig_error
+        failure = _verify_signature(device, headers, method, path, get_body)
+        if failure is not None:
+            return None, failure
 
     if touch:
         try:
@@ -82,23 +99,46 @@ def verify_device_session(*, touch: bool = True) -> Tuple[Optional[dict], Option
     return device, None
 
 
-def _verify_signature(device: dict) -> Optional[tuple]:
+def verify_device_session(*, touch: bool = True) -> Tuple[Optional[dict], Optional[tuple]]:
+    """Validate the device session token on the current Flask request.
+
+    Returns ``(device_row, None)`` on success or ``(None, response)`` on
+    failure, where ``response`` is a ready Flask error response.
+
+    Args:
+        touch: When True, bump ``last_seen_at`` on the device row.
+    """
+    device, failure = authenticate_device(
+        request.headers, request.method, request.path, request.get_data, touch=touch
+    )
+    if failure is not None:
+        return None, _error(*failure)
+    return device, None
+
+
+def _verify_signature(
+    device: dict,
+    headers: Mapping[str, str],
+    method: str,
+    path: str,
+    get_body: Callable[[], bytes],
+) -> Optional[DeviceAuthFailure]:
     """Verify ``X-Device-Signature`` against the stored machine pubkey."""
-    sig_b64 = request.headers.get("X-Device-Signature")
-    ts = request.headers.get("X-Device-Timestamp")
-    fp = request.headers.get("X-Device-Machine-Key")
+    sig_b64 = headers.get("X-Device-Signature")
+    ts = headers.get("X-Device-Timestamp")
+    fp = headers.get("X-Device-Machine-Key")
     if not sig_b64 or not ts or not fp:
-        return _error("missing_signature", 401)
+        return ("missing_signature", 401)
 
     try:
         ts_int = int(ts)
     except (TypeError, ValueError):
-        return _error("invalid_timestamp", 401)
+        return ("invalid_timestamp", 401)
     if abs(time.time() - ts_int) > 300:
-        return _error("timestamp_skew", 401)
+        return ("timestamp_skew", 401)
 
     if fp != device.get("machine_pubkey_fingerprint"):
-        return _error("fingerprint_mismatch", 401)
+        return ("fingerprint_mismatch", 401)
 
     # Defer cryptography import to keep cold-start light when signatures
     # are disabled (the dev default).
@@ -111,26 +151,24 @@ def _verify_signature(device: dict) -> Optional[tuple]:
         logger.error(
             "Signature verification requested but ``cryptography`` is not installed."
         )
-        return _error("signature_unsupported", 500)
+        return ("signature_unsupported", 500)
 
     # The full public key isn't stored — only the fingerprint. For signature
     # verification we accept the pubkey in a header too; we trust it iff
     # its fingerprint matches the stored one. This avoids a separate pubkey
     # column for MVP.
-    pubkey_b64 = request.headers.get("X-Device-Machine-Pubkey")
+    pubkey_b64 = headers.get("X-Device-Machine-Pubkey")
     if not pubkey_b64:
-        return _error("missing_pubkey", 401)
+        return ("missing_pubkey", 401)
     if fingerprint_pubkey(pubkey_b64) != device["machine_pubkey_fingerprint"]:
-        return _error("pubkey_fingerprint_mismatch", 401)
+        return ("pubkey_fingerprint_mismatch", 401)
 
-    payload = _canonical_payload(
-        request.method, request.path, ts, request.get_data()
-    ).encode("utf-8")
+    payload = _canonical_payload(method, path, ts, get_body()).encode("utf-8")
     try:
         pubkey = Ed25519PublicKey.from_public_bytes(base64.b64decode(pubkey_b64))
         pubkey.verify(base64.b64decode(sig_b64), payload)
     except (InvalidSignature, ValueError):
-        return _error("invalid_signature", 401)
+        return ("invalid_signature", 401)
 
     return None
 

docsgpt/api/devices/routes.py

@@ -16,7 +16,6 @@ from docsgpt.api.devices.session import (
     ack_invocation,
     me,
     poll,
-    session_events,
     submit_output,
 )
 from docsgpt.devices.normalizer import normalize_command
@@ -313,10 +312,7 @@ def register(bp: Blueprint) -> None:
     bp.add_url_rule(
         "/api/devices/me", view_func=me, methods=["GET"], endpoint="device_me",
     )
-    bp.add_url_rule(
-        "/api/devices/sessions/<session_id>/events", view_func=session_events,
-        methods=["GET"], endpoint="device_session_events",
-    )
+    # The session's SSE stream is native-async (session_events.py, mounted in asgi.py).
     bp.add_url_rule(
         "/api/devices/sessions/<session_id>/invocations/<invocation_id>/ack",
         view_func=ack_invocation, methods=["POST"], endpoint="device_ack",

docsgpt/api/devices/session.py

@@ -1,4 +1,7 @@
-"""Device session endpoints: poll, SSE, ack, output."""
+"""Device session endpoints: poll, me, ack, output.
+
+The SSE command stream itself is a native-async route (``session_events.py``).
+"""
 
 from __future__ import annotations
 
@@ -6,14 +9,10 @@ import gzip
 import io
 import json
 import logging
-import time
-from typing import Iterator
 
-from flask import Response, jsonify, make_response, request, stream_with_context
+from flask import Response, jsonify, make_response, request
 
 from docsgpt.api.devices.auth import verify_device_session
-from docsgpt.core.settings import settings
-from docsgpt.core.shutdown import is_shutting_down
 from docsgpt.devices.broker import get_broker
 from docsgpt.storage.db.repositories.device_audit_log import (
     DeviceAuditLogRepository,
@@ -82,71 +81,6 @@ def me() -> Response:
     return make_response(jsonify(out), 200)
 
 
-def session_events(session_id: str) -> Response:
-    """``GET /api/devices/sessions/{id}/events`` — SSE invocation stream.
-
-    The ``session_id`` must be the ``session_ticket`` the device's own
-    ``/poll`` just issued (the path it was handed as ``session_url``). A
-    stale, mismatched, or fabricated ticket is rejected with ``410 Gone``
-    before any stream is opened.
-    """
-    device, err = verify_device_session()
-    if err is not None:
-        return err
-    broker = get_broker()
-    if not broker.validate_ticket(device["id"], session_id):
-        return make_response(
-            jsonify({"success": False, "error": "session_ticket_invalid"}), 410
-        )
-    sess = broker.register_session(device["id"], device["user_id"])
-
-    keepalive_interval = float(settings.SSE_KEEPALIVE_SECONDS)
-    idle_seconds = float(settings.REMOTE_DEVICE_SESSION_IDLE_SECONDS)
-
-    @stream_with_context
-    def generate() -> Iterator[str]:
-        try:
-            last_keepalive = time.time()
-            while not sess.closed.is_set():
-                # Break promptly on shutdown (see docsgpt/core/shutdown.py).
-                if is_shutting_down():
-                    break
-                now = time.time()
-                if now - sess.last_activity_at > idle_seconds:
-                    yield _sse_event(
-                        "session_end",
-                        {"reason": "inactivity_timeout"},
-                        sess.last_event_id + 1,
-                    )
-                    sess.last_event_id += 1
-                    broker.close_session(sess.session_id, reason="idle")
-                    return
-                envelope = broker.next_command(sess, timeout=1.0)
-                if envelope is None:
-                    if time.time() - last_keepalive >= keepalive_interval:
-                        last_keepalive = time.time()
-                        yield ": heartbeat\n\n"
-                    continue
-                sess.last_event_id += 1
-                sess.last_activity_at = time.time()
-                yield _sse_event("invocation", envelope, sess.last_event_id)
-                last_keepalive = time.time()
-        except GeneratorExit:
-            logger.debug("device SSE generator exiting for session %s", sess.session_id)
-            raise
-        finally:
-            broker.close_session(sess.session_id, reason="generator_exit")
-
-    return Response(
-        generate(),
-        mimetype="text/event-stream",
-        headers={
-            "Cache-Control": "no-cache",
-            "X-Accel-Buffering": "no",
-        },
-    )
-
-
 def ack_invocation(session_id: str, invocation_id: str) -> Response:
     """CLI acks (accepted / denied / auto_approved) the invocation."""
     device, err = verify_device_session()
@@ -262,11 +196,3 @@ def _as_opt_int(value) -> int | None:
         return int(value)
     except (TypeError, ValueError):
         return None
-
-
-def _sse_event(name: str, payload: dict, event_id: int) -> str:
-    return (
-        f"event: {name}\n"
-        f"id: {event_id}\n"
-        f"data: {json.dumps(payload)}\n\n"
-    )

docsgpt/api/devices/session_events.py

@@ -0,0 +1,133 @@
+"""``GET /api/devices/sessions/{session_id}/events`` — a paired device's command stream.
+
+A native-async Starlette route mounted ahead of Flask in ``docsgpt/asgi.py``.
+The CLI holds this stream open while it waits for commands, so it runs on the
+event loop: the token check and ticket claim hop to a worker thread once, then
+commands arrive through the broker's async ``BLPOP`` and an idle session holds
+no thread.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from functools import partial
+from typing import AsyncIterator
+
+import anyio
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Route
+
+from docsgpt.api.asgi_stream import ClosingStreamingResponse
+from docsgpt.api.devices.auth import authenticate_device
+from docsgpt.core.settings import settings
+from docsgpt.core.shutdown import is_shutting_down
+from docsgpt.devices.broker import DeviceBroker, get_broker
+
+logger = logging.getLogger(__name__)
+
+# Upper bound on one broker poll, so a closed session, idle expiry or a
+# server drain is noticed within this many seconds.
+_POLL_TIMEOUT_SECONDS = 1.0
+
+_SSE_HEADERS = {
+    "Cache-Control": "no-cache",
+    "X-Accel-Buffering": "no",
+}
+
+
+def _error(code: str, status_code: int) -> JSONResponse:
+    return JSONResponse({"success": False, "error": code}, status_code=status_code)
+
+
+def _sse_event(name: str, payload: dict, event_id: int) -> str:
+    return (
+        f"event: {name}\n"
+        f"id: {event_id}\n"
+        f"data: {json.dumps(payload)}\n\n"
+    )
+
+
+async def _session_stream(broker: DeviceBroker, device: dict) -> AsyncIterator[str]:
+    """Deliver queued commands until the session closes, idles out, or the server drains."""
+    # Registration consumes the ticket. Shield it so a disconnect can't leave
+    # a registered session that the ``finally`` below never closes.
+    with anyio.CancelScope(shield=True):
+        sess = await anyio.to_thread.run_sync(
+            broker.register_session, device["id"], device["user_id"]
+        )
+    keepalive_interval = float(settings.SSE_KEEPALIVE_SECONDS)
+    idle_seconds = float(settings.REMOTE_DEVICE_SESSION_IDLE_SECONDS)
+    try:
+        last_keepalive = time.time()
+        # ``closed`` is set when the CLI reconnects and a newer session
+        # replaces this one.
+        while not sess.closed.is_set():
+            # Break promptly on shutdown (see docsgpt/core/shutdown.py).
+            if is_shutting_down():
+                break
+            if time.time() - sess.last_activity_at > idle_seconds:
+                yield _sse_event(
+                    "session_end",
+                    {"reason": "inactivity_timeout"},
+                    sess.last_event_id + 1,
+                )
+                sess.last_event_id += 1
+                broker.close_session(sess.session_id, reason="idle")
+                return
+            envelope = await broker.next_command_async(sess, timeout=_POLL_TIMEOUT_SECONDS)
+            if envelope is None:
+                if time.time() - last_keepalive >= keepalive_interval:
+                    last_keepalive = time.time()
+                    yield ": heartbeat\n\n"
+                continue
+            sess.last_event_id += 1
+            sess.last_activity_at = time.time()
+            yield _sse_event("invocation", envelope, sess.last_event_id)
+            last_keepalive = time.time()
+    finally:
+        broker.close_session(sess.session_id, reason="stream_end")
+
+
+async def device_session_events(request: Request) -> Response:
+    """Stream queued invocations to the device holding the poll-issued ticket.
+
+    The ``session_id`` must be the ``session_ticket`` the device's own
+    ``/poll`` just issued (the path it was handed as ``session_url``). A
+    stale, mismatched, or fabricated ticket is rejected with ``410 Gone``
+    before any stream is opened.
+    """
+    session_id = request.path_params["session_id"]
+    body = await request.body()
+    device, failure = await anyio.to_thread.run_sync(
+        partial(
+            authenticate_device,
+            request.headers,
+            request.method,
+            request.url.path,
+            lambda: body,
+        )
+    )
+    if failure is not None:
+        return _error(*failure)
+    broker = get_broker()
+    if not await anyio.to_thread.run_sync(broker.validate_ticket, device["id"], session_id):
+        return _error("session_ticket_invalid", 410)
+    return ClosingStreamingResponse(
+        _session_stream(broker, device),
+        media_type="text/event-stream",
+        headers=_SSE_HEADERS,
+    )
+
+
+# Mounted in ``docsgpt/asgi.py`` ahead of the Flask catch-all. The device's
+# poll, ack and output endpoints stay on Flask (``session.py``).
+device_session_routes = [
+    Route(
+        "/api/devices/sessions/{session_id}/events",
+        device_session_events,
+        methods=["GET"],
+    ),
+]

docsgpt/api/events/routes.py

@@ -4,6 +4,10 @@ Subscribe-then-snapshot pattern: subscribe to ``user:{user_id}``
 pub/sub, snapshot the Redis Streams backlog past ``Last-Event-ID``
 inside the SUBSCRIBE-ack callback, flush snapshot, then tail live
 events (dedup'd by stream id). See ``docs/runbooks/sse-notifications.md``.
+
+A native-async Starlette route mounted ahead of Flask in ``docsgpt/asgi.py``.
+Every open browser tab holds this stream, so it runs on the event loop with
+the async Redis client instead of pinning a WSGI threadpool slot per tab.
 """
 
 from __future__ import annotations
@@ -12,11 +16,17 @@ import json
 import logging
 import re
 import time
-from typing import Iterator, Optional
+from contextlib import aclosing
+from functools import partial
+from typing import AsyncIterator, Optional
 
-from flask import Blueprint, Response, jsonify, make_response, request, stream_with_context
+from starlette.requests import Request
+from starlette.responses import Response
+from starlette.routing import Route
 
-from docsgpt.cache import get_redis_instance
+from docsgpt.api.asgi_auth import authenticate, bind_log_context, json_error
+from docsgpt.api.asgi_stream import ClosingStreamingResponse
+from docsgpt.cache import PUBSUB_SOCKET_TIMEOUT_SECONDS
 from docsgpt.core.settings import settings
 from docsgpt.core.shutdown import is_shutting_down
 from docsgpt.events.keys import (
@@ -26,14 +36,17 @@ from docsgpt.events.keys import (
     stream_key,
     topic_name,
 )
-from docsgpt.streaming.broadcast_channel import Topic
+from docsgpt.streaming.async_broadcast_channel import AsyncTopic
+from docsgpt.streaming.async_redis import get_async_redis_instance
 
 logger = logging.getLogger(__name__)
 
-events = Blueprint("event_stream", __name__)
-
 SUBSCRIBE_POLL_INTERVAL_SECONDS = 1.0
 
+# Safety TTL on the per-user connection counter: counts orphaned by a hard
+# crash self-heal after this window.
+_COUNTER_TTL_SECONDS = 3600
+
 # WHATWG SSE treats CRLF, CR, and LF equivalently as line terminators.
 _SSE_LINE_SPLIT = re.compile(r"\r\n|\r|\n")
 
@@ -43,6 +56,14 @@ _SSE_LINE_SPLIT = re.compile(r"\r\n|\r|\n")
 # would silently fail.
 _STREAM_ID_RE = re.compile(r"^\d+(-\d+)?$")
 
+_SSE_HEADERS = {
+    "Cache-Control": "no-store",
+    "X-Accel-Buffering": "no",
+    "Connection": "keep-alive",
+    # Marks the response as served by the event loop. Purely diagnostic.
+    "X-SSE-Transport": "async",
+}
+
 # Only emitted at most once per process so a misconfigured deployment
 # doesn't drown the logs.
 _local_user_warned = False
@@ -74,19 +95,19 @@ def _decode(value) -> Optional[str]:
     return str(value)
 
 
-def _oldest_retained_id(redis_client, user_id: str) -> Optional[str]:
+async def _oldest_retained_id(redis_client, user_id: str) -> Optional[str]:
     """Return the id of the oldest entry still in the stream, or ``None``.
 
     Used to detect ``Last-Event-ID`` having slid off the back of the
     MAXLEN'd window.
     """
     try:
-        info = redis_client.xinfo_stream(stream_key(user_id))
+        info = await redis_client.xinfo_stream(stream_key(user_id))
     except Exception:
         return None
     if not isinstance(info, dict):
         return None
-    # redis-py 7.4 returns str-keyed dicts here; the bytes-key probe is
+    # redis-py returns str-keyed dicts here; the bytes-key probe is
     # defence in depth in case ``decode_responses`` is ever flipped.
     first_entry = info.get("first-entry") or info.get(b"first-entry")
     if not first_entry:
@@ -98,7 +119,7 @@ def _oldest_retained_id(redis_client, user_id: str) -> Optional[str]:
         return None
 
 
-def _allow_replay(
+async def _allow_replay(
     redis_client, user_id: str, last_event_id: Optional[str]
 ) -> bool:
     """Per-user sliding-window snapshot-replay budget.
@@ -119,12 +140,12 @@ def _allow_replay(
     window = max(1, int(settings.EVENTS_REPLAY_BUDGET_WINDOW_SECONDS))
     key = replay_budget_key(user_id)
     try:
-        used = int(redis_client.incr(key))
+        used = int(await redis_client.incr(key))
         # Always (re)seed the TTL. Gating on ``used == 1`` would wedge
         # the counter forever if INCR succeeds but EXPIRE raises on
         # the seeding call. EXPIRE on an existing key resets the TTL
         # to ``window`` — within ±1s of the per-window budget semantic.
-        redis_client.expire(key, window)
+        await redis_client.expire(key, window)
     except Exception:
         logger.debug(
             "replay budget probe failed for user=%s; failing open",
@@ -166,10 +187,10 @@ def _replay_floor_id() -> Optional[str]:
     return f"{floor_ms}-0"
 
 
-def _replay_backlog(
+async def _replay_backlog(
     redis_client, user_id: str, last_event_id: Optional[str], max_count: int
-) -> Iterator[tuple[str, str]]:
-    """Yield ``(entry_id, sse_line)`` for backlog entries past ``last_event_id``.
+) -> list[tuple[str, str]]:
+    """Return ``(entry_id, sse_line)`` pairs for backlog entries past ``last_event_id``.
 
     Capped at ``max_count`` rows; clients catch up across reconnects.
     Parse failures are skipped; the Streams id is injected into the
@@ -186,7 +207,7 @@ def _replay_backlog(
         # Exclusive start: '(<id>' skips the already-delivered entry.
         start = f"({last_event_id}"
     try:
-        entries = redis_client.xrange(
+        entries = await redis_client.xrange(
             stream_key(user_id), min=start, max="+", count=max_count
         )
     except Exception as exc:
@@ -196,14 +217,15 @@ def _replay_backlog(
             last_event_id or "-",
             exc,
         )
-        return
+        return []
 
+    lines: list[tuple[str, str]] = []
     for entry_id, fields in entries:
         entry_id_str = _decode(entry_id)
         if not entry_id_str:
             continue
-        # decode_responses=False on the cache client ⇒ field keys/values
-        # are bytes. The string-key fallback covers a future flip of that
+        # decode_responses=False on the client ⇒ field keys/values are
+        # bytes. The string-key fallback covers a future flip of that
         # default without a forced refactor here.
         raw_event = None
         if isinstance(fields, dict):
@@ -223,7 +245,8 @@ def _replay_backlog(
                 "Replay envelope parse failed for entry %s; passing through raw",
                 entry_id_str,
             )
-        yield entry_id_str, _format_sse(event_str, event_id=entry_id_str)
+        lines.append((entry_id_str, _format_sse(event_str, event_id=entry_id_str)))
+    return lines
 
 
 def _truncation_notice_line(oldest_id: str) -> str:
@@ -238,209 +261,102 @@ def _truncation_notice_line(oldest_id: str) -> str:
     )
 
 
-@events.route("/api/events", methods=["GET"])
-def stream_events() -> Response:
-    decoded = getattr(request, "decoded_token", None)
-    user_id = decoded.get("sub") if isinstance(decoded, dict) else None
-    if not user_id:
-        return make_response(
-            jsonify({"success": False, "message": "Authentication required"}),
-            401,
-        )
+async def _release_slot(redis_client, counter_key: str) -> None:
+    try:
+        await redis_client.decr(counter_key)
+    except Exception:
+        logger.debug("SSE connection counter DECR failed for key=%s", counter_key)
 
-    # In dev deployments without AUTH_TYPE configured, every request
-    # resolves to user_id="local" and shares one stream. Surface this so
-    # an accidentally-multi-user dev box doesn't silently cross-stream.
-    global _local_user_warned
-    if user_id == "local" and not _local_user_warned:
-        logger.warning(
-            "SSE serving user_id='local' (AUTH_TYPE not set). "
-            "All clients on this deployment will share one event stream."
-        )
-        _local_user_warned = True
-
-    raw_last_event_id = request.headers.get("Last-Event-ID") or request.args.get(
-        "last_event_id"
-    )
-    last_event_id = _normalize_last_event_id(raw_last_event_id)
-    last_event_id_invalid = raw_last_event_id is not None and last_event_id is None
 
+async def _event_stream(
+    redis_client,
+    user_id: str,
+    last_event_id: Optional[str],
+    last_event_id_invalid: bool,
+    push_enabled: bool,
+) -> AsyncIterator[str]:
+    """Yield the prelude, the replayed backlog, then live events until the stream ends."""
+    connect_ts = time.monotonic()
+    replayed_count = 0
     keepalive_seconds = float(settings.SSE_KEEPALIVE_SECONDS)
-    push_enabled = settings.ENABLE_SSE_PUSH
-    cap = int(settings.SSE_MAX_CONCURRENT_PER_USER)
+    try:
+        # First frame primes intermediaries (Cloudflare, nginx) so they
+        # don't sit on a buffer waiting for body bytes.
+        yield ": connected\n\n"
 
-    redis_client = get_redis_instance()
-    counter_key = connection_counter_key(user_id)
-    counted = False
+        if not push_enabled:
+            yield ": push_disabled\n\n"
+            return
 
-    if push_enabled and redis_client is not None and cap > 0:
-        try:
-            current = int(redis_client.incr(counter_key))
-            counted = True
-        except Exception:
-            current = 0
-            logger.debug(
-                "SSE connection counter INCR failed for user=%s", user_id
-            )
-        if counted:
-            # 1h safety TTL — orphaned counts from hard crashes self-heal.
-            # EXPIRE failure must NOT clobber ``current`` and bypass the cap.
+        replay_lines: list[str] = []
+        max_replayed_id: Optional[str] = None
+        replay_done = False
+
+        # If the client sent a malformed Last-Event-ID, surface the
+        # truncation notice before the subscribe loop. Buffering it into
+        # ``replay_lines`` would lose it when ``subscribe`` returns
+        # immediately (Redis down) and the loop body never runs.
+        if last_event_id_invalid:
+            yield _truncation_notice_line("")
+            replayed_count += 1
+
+        async def _on_subscribe() -> None:
+            # Runs inside AsyncTopic.subscribe after the SUBSCRIBE is
+            # acked. Reading the backlog here means any publisher firing
+            # between SUBSCRIBE-send and SUBSCRIBE-ack has its XADD captured
+            # by XRANGE *and* its PUBLISH buffered at the connection layer
+            # until we read it — closing the replay/subscribe race.
+            #
+            # Truncation contract: ``backlog.truncated`` is emitted ONLY
+            # when the client's ``Last-Event-ID`` has slid off the MAXLEN'd
+            # window or the age floor — the journal is genuinely gone past
+            # the cursor, so the frontend should clear its cursor and
+            # refetch state. Cap-hit skips silently: the cursor advances via
+            # the per-entry ``id:`` headers. Budget exhaustion never reaches
+            # this callback — the route 429s before opening the stream.
+            nonlocal max_replayed_id, replay_done
             try:
-                redis_client.expire(counter_key, 3600)
-            except Exception:
-                logger.debug(
-                    "SSE connection counter EXPIRE failed for user=%s", user_id
-                )
-            if current > cap:
-                try:
-                    redis_client.decr(counter_key)
-                except Exception:
-                    logger.debug(
-                        "SSE connection counter DECR failed for user=%s",
-                        user_id,
-                    )
-                return make_response(
-                    jsonify(
-                        {
-                            "success": False,
-                            "message": "Too many concurrent SSE connections",
-                        }
-                    ),
-                    429,
-                )
+                if redis_client is None or last_event_id is None:
+                    # Fresh session: start live. Replaying the whole
+                    # retained window shipped weeks-old entries on every
+                    # tab-open (MAXLEN caps by count, not age).
+                    return
+                oldest = await _oldest_retained_id(redis_client, user_id)
+                floor = _replay_floor_id()
+                # The snapshot can't reach past whichever is newer: the
+                # MAXLEN'd window edge or the age floor.
+                effective_oldest = oldest
+                if floor and (
+                    effective_oldest is None
+                    or stream_id_compare(floor, effective_oldest) > 0
+                ):
+                    effective_oldest = floor
+                if (
+                    effective_oldest
+                    and stream_id_compare(last_event_id, effective_oldest) < 0
+                ):
+                    replay_lines.append(_truncation_notice_line(effective_oldest))
+                replay_cap = int(settings.EVENTS_REPLAY_MAX_PER_REQUEST)
+                for entry_id, sse_line in await _replay_backlog(
+                    redis_client, user_id, last_event_id, replay_cap
+                ):
+                    replay_lines.append(sse_line)
+                    max_replayed_id = entry_id
+            finally:
+                # Always flip the flag — even on partial-replay failure the
+                # outer loop must reach the flush step so we don't silently
+                # strand whatever entries did land.
+                replay_done = True
 
-    # Replay budget is checked here, before the generator opens the
-    # stream, so a denial can surface as HTTP 429 instead of a silent
-    # snapshot skip. The earlier in-generator skip lost events between
-    # the client's cursor and the first live-tailed entry: the live
-    # tail still carried ``id:`` headers, the frontend advanced
-    # ``lastEventId`` to one of those ids, and the events in between
-    # were never reachable on the next reconnect. 429 keeps the
-    # cursor pinned and lets the frontend back off until the window
-    # slides (eventStreamClient.ts treats 429 as escalated backoff).
-    if push_enabled and redis_client is not None and not _allow_replay(
-        redis_client, user_id, last_event_id
-    ):
-        if counted:
-            try:
-                redis_client.decr(counter_key)
-            except Exception:
-                logger.debug(
-                    "SSE connection counter DECR failed for user=%s",
-                    user_id,
-                )
-        return make_response(
-            jsonify(
-                {
-                    "success": False,
-                    "message": "Replay budget exhausted",
-                }
-            ),
-            429,
+        topic = AsyncTopic(topic_name(user_id))
+        last_keepalive = time.monotonic()
+        subscription = topic.subscribe(
+            on_subscribe=_on_subscribe,
+            poll_timeout=SUBSCRIBE_POLL_INTERVAL_SECONDS,
+            liveness_timeout=PUBSUB_SOCKET_TIMEOUT_SECONDS,
         )
-
-    @stream_with_context
-    def generate() -> Iterator[str]:
-        connect_ts = time.monotonic()
-        replayed_count = 0
-        try:
-            # First frame primes intermediaries (Cloudflare, nginx) so they
-            # don't sit on a buffer waiting for body bytes.
-            yield ": connected\n\n"
-
-            if not push_enabled:
-                yield ": push_disabled\n\n"
-                return
-
-            replay_lines: list[str] = []
-            max_replayed_id: Optional[str] = None
-            replay_done = False
-
-            # If the client sent a malformed Last-Event-ID, surface the
-            # truncation notice synchronously *before* the subscribe
-            # loop. Buffering it into ``replay_lines`` would lose it
-            # when ``Topic.subscribe`` returns immediately (Redis down)
-            # — the loop body never runs, and the flush at line ~335
-            # never fires.
-            if last_event_id_invalid:
-                yield _truncation_notice_line("")
-                replayed_count += 1
-
-            def _on_subscribe_callback() -> None:
-                # Runs synchronously inside Topic.subscribe after the
-                # SUBSCRIBE is acked. By doing XRANGE here, any publisher
-                # firing between SUBSCRIBE-send and SUBSCRIBE-ack has its
-                # XADD captured by XRANGE *and* its PUBLISH buffered at
-                # the connection layer until we read it — closing the
-                # replay/subscribe race the design doc warns about.
-                #
-                # Truncation contract: ``backlog.truncated`` is emitted
-                # ONLY when the client's ``Last-Event-ID`` has slid off
-                # the MAXLEN'd window — that's the case where the
-                # journal is genuinely gone past the cursor and the
-                # frontend should clear its slice cursor and refetch
-                # state. Cap-hit skips the snapshot silently: the
-                # cursor advances via the per-entry ``id:`` headers
-                # and the frontend's slice keeps the latest id so the
-                # next reconnect resumes from there. Budget-exhausted
-                # never reaches this callback — the route 429s before
-                # opening the stream, keeping the cursor pinned.
-                # Conflating these with stale-cursor truncation would
-                # tell the client to clear its cursor and re-receive
-                # the same oldest-N entries on every reconnect —
-                # locking the user out of entries past N.
-                nonlocal max_replayed_id, replay_done
-                try:
-                    if redis_client is None:
-                        return
-                    if last_event_id is None:
-                        # Fresh session: start live. A no-cursor connect has
-                        # no state to catch up on — replaying the whole
-                        # retained window shipped weeks-old entries on every
-                        # tab-open (MAXLEN caps by count, not age). Clients
-                        # that had state present a cursor; the malformed-
-                        # cursor case already got its truncation notice
-                        # before the subscribe loop.
-                        return
-                    oldest = _oldest_retained_id(redis_client, user_id)
-                    floor = _replay_floor_id()
-                    # The snapshot can't reach past whichever is newer:
-                    # the MAXLEN'd window edge or the age floor.
-                    effective_oldest = oldest
-                    if floor and (
-                        effective_oldest is None
-                        or stream_id_compare(floor, effective_oldest) > 0
-                    ):
-                        effective_oldest = floor
-                    if (
-                        effective_oldest
-                        and stream_id_compare(last_event_id, effective_oldest) < 0
-                    ):
-                        # The Last-Event-ID has slid off the replayable
-                        # window. Tell the client so it can fetch full state.
-                        replay_lines.append(
-                            _truncation_notice_line(effective_oldest)
-                        )
-                    replay_cap = int(settings.EVENTS_REPLAY_MAX_PER_REQUEST)
-                    for entry_id, sse_line in _replay_backlog(
-                        redis_client, user_id, last_event_id, replay_cap
-                    ):
-                        replay_lines.append(sse_line)
-                        max_replayed_id = entry_id
-                finally:
-                    # Always flip the flag — even on partial-replay failure
-                    # the outer loop must reach the flush step so we don't
-                    # silently strand whatever entries did land.
-                    replay_done = True
-
-            topic = Topic(topic_name(user_id))
-            last_keepalive = time.monotonic()
-            for payload in topic.subscribe(
-                on_subscribe=_on_subscribe_callback,
-                poll_timeout=SUBSCRIBE_POLL_INTERVAL_SECONDS,
-            ):
-                # Break promptly on shutdown — this a2wsgi thread can't be
-                # cancelled by asyncio (see docsgpt/core/shutdown.py).
+        async with aclosing(subscription):
+            async for payload in subscription:
                 if is_shutting_down():
                     break
 
@@ -466,14 +382,10 @@ def stream_events() -> Response:
                     if isinstance(envelope, dict):
                         candidate = envelope.get("id")
                         # Only trust ids that look like real Redis Streams
-                        # ids (``ms`` or ``ms-seq``). A malformed or
-                        # adversarial publisher could otherwise pin
-                        # dedupe forever — a lex-greater bogus id would
-                        # make every legitimate later id compare ``<=``
-                        # and get dropped silently.
-                        if isinstance(candidate, str) and _STREAM_ID_RE.match(
-                            candidate
-                        ):
+                        # ids. A malformed or adversarial publisher could
+                        # otherwise pin dedupe forever — a lex-greater bogus
+                        # id would make every later id compare ``<=``.
+                        if isinstance(candidate, str) and _STREAM_ID_RE.match(candidate):
                             event_id = candidate
                 except Exception:
                     pass
@@ -489,49 +401,102 @@ def stream_events() -> Response:
                 yield _format_sse(event_str, event_id=event_id)
                 last_keepalive = now
 
-            # Topic.subscribe exited before the first yield (transient
-            # Redis hiccup between SUBSCRIBE-ack and the first poll, or
-            # an immediate Redis-down return). The callback may already
-            # have populated the snapshot — flush it so the client gets
-            # the backlog instead of a silent drop. Safe no-op when the
-            # in-loop flush ran (it clear()'d the buffer) and when the
-            # callback never fired (replay_done stays False).
-            if replay_done and replay_lines:
-                for line in replay_lines:
-                    yield line
-                    replayed_count += 1
-                replay_lines.clear()
-        except GeneratorExit:
-            return
+        # The subscription exited before the first yield (transient Redis
+        # hiccup between SUBSCRIBE-ack and the first poll, or an immediate
+        # Redis-down return). The callback may already have populated the
+        # snapshot — flush it so the client gets the backlog.
+        if replay_done and replay_lines:
+            for line in replay_lines:
+                yield line
+                replayed_count += 1
+            replay_lines.clear()
+    except Exception:
+        # Client disconnect (cancellation / GeneratorExit) is a BaseException
+        # and passes straight through; only genuine bugs land here.
+        logger.exception("SSE event-stream generator crashed for user=%s", user_id)
+    finally:
+        logger.info(
+            "event.disconnect user=%s duration_s=%.1f replayed=%d",
+            user_id,
+            time.monotonic() - connect_ts,
+            replayed_count,
+        )
+
+
+async def stream_events(request: Request) -> Response:
+    """``GET /api/events`` — the caller's live notification stream."""
+    decoded, error = await authenticate(request)
+    if error is not None:
+        return error
+    user_id = decoded.get("sub") if isinstance(decoded, dict) else None
+    if not user_id:
+        return json_error("Authentication required", 401)
+    bind_log_context("event_stream", user_id)
+
+    # In dev deployments without AUTH_TYPE configured, every request
+    # resolves to user_id="local" and shares one stream. Surface this so
+    # an accidentally-multi-user dev box doesn't silently cross-stream.
+    global _local_user_warned
+    if user_id == "local" and not _local_user_warned:
+        logger.warning(
+            "SSE serving user_id='local' (AUTH_TYPE not set). "
+            "All clients on this deployment will share one event stream."
+        )
+        _local_user_warned = True
+
+    raw_last_event_id = request.headers.get("Last-Event-ID") or request.query_params.get(
+        "last_event_id"
+    )
+    last_event_id = _normalize_last_event_id(raw_last_event_id)
+    last_event_id_invalid = raw_last_event_id is not None and last_event_id is None
+
+    push_enabled = settings.ENABLE_SSE_PUSH
+    cap = int(settings.SSE_MAX_CONCURRENT_PER_USER)
+    redis_client = await get_async_redis_instance() if push_enabled else None
+    counter_key = connection_counter_key(user_id)
+    counted = False
+
+    if redis_client is not None and cap > 0:
+        try:
+            current = int(await redis_client.incr(counter_key))
+            counted = True
         except Exception:
-            logger.exception(
-                "SSE event-stream generator crashed for user=%s", user_id
-            )
-        finally:
-            duration_s = time.monotonic() - connect_ts
-            logger.info(
-                "event.disconnect user=%s duration_s=%.1f replayed=%d",
-                user_id,
-                duration_s,
-                replayed_count,
-            )
-            if counted and redis_client is not None:
-                try:
-                    redis_client.decr(counter_key)
-                except Exception:
-                    logger.debug(
-                        "SSE connection counter DECR failed for user=%s on disconnect",
-                        user_id,
-                    )
+            logger.debug("SSE connection counter INCR failed for user=%s", user_id)
+        if counted:
+            # EXPIRE failure must NOT bypass the cap.
+            try:
+                await redis_client.expire(counter_key, _COUNTER_TTL_SECONDS)
+            except Exception:
+                logger.debug("SSE connection counter EXPIRE failed for user=%s", user_id)
+            if current > cap:
+                await _release_slot(redis_client, counter_key)
+                return json_error("Too many concurrent SSE connections", 429)
+
+    # Replay budget is checked before the stream opens so a denial surfaces
+    # as HTTP 429 instead of a silent snapshot skip: a live tail carrying
+    # ``id:`` headers would advance the client's cursor past the entries it
+    # never received. 429 keeps the cursor pinned and the frontend backs off.
+    if redis_client is not None and not await _allow_replay(redis_client, user_id, last_event_id):
+        if counted:
+            await _release_slot(redis_client, counter_key)
+        return json_error("Replay budget exhausted", 429)
 
-    response = Response(generate(), mimetype="text/event-stream")
-    response.headers["Cache-Control"] = "no-store"
-    response.headers["X-Accel-Buffering"] = "no"
-    response.headers["Connection"] = "keep-alive"
     logger.info(
         "event.connect user=%s last_event_id=%s%s",
         user_id,
         last_event_id or "-",
         " (rejected_invalid)" if last_event_id_invalid else "",
     )
-    return response
+    return ClosingStreamingResponse(
+        _event_stream(redis_client, user_id, last_event_id, last_event_id_invalid, push_enabled),
+        media_type="text/event-stream",
+        headers=_SSE_HEADERS,
+        # Released once the response is over, however it ended.
+        on_close=partial(_release_slot, redis_client, counter_key) if counted else None,
+    )
+
+
+# Mounted in ``docsgpt/asgi.py`` ahead of the Flask catch-all.
+event_stream_routes = [
+    Route("/api/events", stream_events, methods=["GET"]),
+]

docsgpt/api/user/artifacts/authz.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Optional
+from typing import Mapping, Optional
 
 from flask import request
 
@@ -40,18 +40,16 @@ class Principal:
         return self.agent_id is not None
 
 
-def resolve_principal() -> Principal:
-    """Resolve the caller to a :class:`Principal`.
+def principal_for(decoded_token: Optional[dict], api_key: Optional[str]) -> Principal:
+    """Resolve a caller from its decoded JWT or ``api_key``, independent of the web framework.
 
-    Priority: a decoded JWT (owner session) first, then an ``api_key`` query/form
-    param (agent-scoped). An unresolvable caller yields an anonymous principal
-    (both fields ``None``) that only a valid ``share_token`` can authorize.
+    Priority: a decoded JWT (owner session) first, then the ``api_key``
+    (agent-scoped). An unresolvable caller yields an anonymous principal (both
+    fields ``None``) that only a valid ``share_token`` can authorize.
     """
-    decoded_token = getattr(request, "decoded_token", None)
     if decoded_token:
         return Principal(user_id=decoded_token.get("sub"))
 
-    api_key = request.args.get("api_key") or request.form.get("api_key")
     if api_key:
         with db_readonly() as conn:
             agent = AgentsRepository(conn).find_by_key(api_key)
@@ -60,6 +58,18 @@ def resolve_principal() -> Principal:
     return Principal()
 
 
+def resolve_principal() -> Principal:
+    """Resolve the current Flask request's caller to a :class:`Principal`.
+
+    Reads the decoded JWT, then an ``api_key`` query/form param; see
+    :func:`principal_for`.
+    """
+    return principal_for(
+        getattr(request, "decoded_token", None),
+        request.args.get("api_key") or request.form.get("api_key"),
+    )
+
+
 def _shared_row_for(conn, conversation_id, share_token):
     """Return the shared_conversations row iff share_token grants this conversation, else None."""
     if not share_token:
@@ -87,7 +97,12 @@ def user_can_access_conversation(
     return _shared_row_for(conn, conversation_id, share_token) is not None
 
 
-def authorize_artifact(conn, artifact: dict, principal: Principal) -> bool:
+def authorize_artifact(
+    conn,
+    artifact: dict,
+    principal: Principal,
+    args: Optional[Mapping[str, str]] = None,
+) -> bool:
     """Authorize a READ of ``artifact`` for ``principal``; missing parent fails closed.
 
     A low-trust agent api_key is confined to a single conversation it proves by
@@ -96,7 +111,13 @@ def authorize_artifact(conn, artifact: dict, principal: Principal) -> bool:
     JWT owner or ``shared_with`` collaborator gets full access to every artifact of
     the parent; a share-token holder is confined to the shared ``first_n_queries``
     snapshot (an artifact whose ``message_id`` is outside it, or NULL, is denied).
+
+    ``args`` are the request's query params; they default to the current Flask
+    request's, and a native-async route passes its own.
     """
+    if args is None:
+        args = request.args
+
     if principal.is_agent_scoped:
         # An agent key is not the owner's session and is embedded in public widget
         # JS, so require the artifact to belong to the owner AND the request to
@@ -104,7 +125,7 @@ def authorize_artifact(conn, artifact: dict, principal: Principal) -> bool:
         # the key alone, without that id, cannot download a known artifact.
         if str(artifact.get("user_id")) != str(principal.user_id):
             return False
-        req_conv = request.args.get("conversation_id")
+        req_conv = args.get("conversation_id")
         parent = artifact.get("conversation_id")
         if not req_conv or parent is None or str(parent) != str(req_conv):
             return False
@@ -114,7 +135,7 @@ def authorize_artifact(conn, artifact: dict, principal: Principal) -> bool:
 
     conversation_id = artifact.get("conversation_id")
     workflow_run_id = artifact.get("workflow_run_id")
-    share_token = request.args.get("share_token")
+    share_token = args.get("share_token")
 
     if conversation_id is not None:
         # Owner or shared_with collaborator: full access to every artifact.

docsgpt/api/user/artifacts/download.py

@@ -0,0 +1,181 @@
+"""``GET /api/artifacts/{artifact_id}/download`` — an artifact version's bytes.
+
+A native-async Starlette route mounted ahead of Flask in ``docsgpt/asgi.py``,
+so a large download to a slow client streams from the event loop instead of
+pinning a WSGI threadpool slot for the whole transfer. Authorization is the
+same parent-derived check the Flask artifact routes use; the database reads
+and storage calls run in worker threads.
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+import re
+from functools import partial
+from typing import AsyncIterator, BinaryIO, Mapping, Optional, Tuple
+
+import anyio
+from starlette.requests import Request
+from starlette.responses import JSONResponse, RedirectResponse, Response
+from starlette.routing import Route
+
+from docsgpt.api.asgi_auth import authenticate, bind_log_context, json_error
+from docsgpt.api.asgi_stream import ClosingStreamingResponse
+from docsgpt.api.user.artifacts.authz import authorize_artifact, principal_for
+from docsgpt.core.settings import settings
+from docsgpt.storage.db.base_repository import looks_like_uuid
+from docsgpt.storage.db.repositories.artifacts import ArtifactsRepository
+from docsgpt.storage.db.session import db_readonly
+from docsgpt.storage.storage_creator import StorageCreator
+
+logger = logging.getLogger(__name__)
+
+# Presigned-URL TTL for private S3 artifact downloads (seconds).
+_PRESIGNED_URL_TTL = 300
+
+_ARTIFACT_URL_ENVELOPE_MIME = "application/vnd.docsgpt.artifact-url+json"
+
+_CHUNK_SIZE = 65536
+
+
+def _sanitize_header_filename(filename: Optional[str], fallback: str) -> str:
+    """Strip CRLF / quotes from a display filename for a Content-Disposition header."""
+    if not filename:
+        return fallback
+    cleaned = re.sub(r'[\r\n"]', "", str(filename)).strip()
+    return cleaned or fallback
+
+
+def _load_version(
+    artifact_id: str, decoded_token: Optional[dict], query: Mapping[str, str]
+) -> Tuple[Optional[dict], Optional[Tuple[str, int]]]:
+    """Authorize the caller and fetch the requested version row; runs in a worker thread.
+
+    Returns:
+        tuple: ``(version_row, None)`` or ``(None, (message, status))``.
+    """
+    principal = principal_for(decoded_token, query.get("api_key"))
+    with db_readonly() as conn:
+        repo = ArtifactsRepository(conn)
+        artifact = repo.get_artifact(artifact_id)
+        if artifact is None:
+            return None, ("Artifact not found", 404)
+        if not authorize_artifact(conn, artifact, principal, args=query):
+            return None, ("Forbidden", 403)
+        version = artifact.get("current_version")
+        version_arg = query.get("version")
+        if version_arg is not None:
+            try:
+                version = int(version_arg)
+            except ValueError:
+                return None, ("Invalid version", 400)
+        version_row = repo.get_version(artifact_id, version)
+    if version_row is None:
+        return None, ("Version not found", 404)
+    return version_row, None
+
+
+async def _read_chunks(file_obj: BinaryIO) -> AsyncIterator[bytes]:
+    """Yield the file in chunks; reads from disk or a socket hop to a worker thread."""
+    in_memory = isinstance(file_obj, io.BytesIO)
+    while True:
+        if in_memory:
+            chunk = file_obj.read(_CHUNK_SIZE)
+        else:
+            chunk = await anyio.to_thread.run_sync(file_obj.read, _CHUNK_SIZE)
+        if not chunk:
+            return
+        yield chunk
+
+
+async def _close_file(file_obj: BinaryIO) -> None:
+    close = getattr(file_obj, "close", None)
+    if callable(close):
+        close()
+
+
+async def download_artifact(request: Request) -> Response:
+    """Serve an artifact version: a 302 (or a JSON URL envelope) on S3, streamed bytes otherwise."""
+    artifact_id = request.path_params["artifact_id"]
+    if not looks_like_uuid(artifact_id):
+        return json_error("Artifact not found", 404)
+    decoded, error = await authenticate(request)
+    if error is not None:
+        return error
+    bind_log_context("artifact_download", decoded.get("sub") if decoded else None)
+    query = request.query_params
+
+    try:
+        version_row, failure = await anyio.to_thread.run_sync(
+            _load_version, artifact_id, decoded, query
+        )
+        if failure is not None:
+            return json_error(*failure)
+
+        # The object key is derived only from the stored path, never client input.
+        storage_path = version_row.get("storage_path")
+        if not storage_path:
+            return json_error("No file for this version", 404)
+
+        filename = _sanitize_header_filename(version_row.get("filename"), f"artifact-{artifact_id}")
+        mime_type = version_row.get("mime_type") or "application/octet-stream"
+        storage = await anyio.to_thread.run_sync(StorageCreator.get_storage)
+
+        # With URL_STRATEGY=="s3" the contract is to hand back a presigned
+        # URL. If the active backend can't mint one, that's a config error:
+        # surface a 500 rather than silently proxying bytes from a backend
+        # the operator expected to be off the hot path.
+        if getattr(settings, "URL_STRATEGY", "backend") == "s3":
+            try:
+                url = await anyio.to_thread.run_sync(
+                    partial(storage.generate_presigned_url, storage_path, expires_in=_PRESIGNED_URL_TTL)
+                )
+            except NotImplementedError:
+                logger.error(
+                    "URL_STRATEGY=s3 but %s cannot mint presigned URLs",
+                    type(storage).__name__,
+                )
+                return json_error("Storage misconfigured", 500)
+            # A 302 to a cross-origin S3 URL can't be read by the app's authed
+            # fetch (the bucket has no CORS grant for the app origin). When the
+            # client opts in via ?disposition=url (or Accept: application/json)
+            # hand the presigned URL back as JSON so it can navigate to it
+            # top-level (no CORS). Default stays a 302 so nothing else breaks.
+            if query.get("disposition") == "url" or (
+                "application/json" in request.headers.get("Accept", "")
+            ):
+                # Tag the envelope with a distinctive media type so the client
+                # keys off a server-set signal, not the JSON body shape. A
+                # ``data`` artifact whose bytes happen to be
+                # ``{"success":true,"url":"..."}`` is streamed under the backend
+                # strategy with its own content-type and can never carry this
+                # vendor type, so it can't be mistaken for a redirect
+                # (open-redirect gadget). Content-Type is CORS-safelisted.
+                return JSONResponse(
+                    {"success": True, "url": url}, media_type=_ARTIFACT_URL_ENVELOPE_MIME
+                )
+            return RedirectResponse(url, status_code=302)
+
+        # Stream the bytes in chunks instead of buffering the whole object in
+        # memory (artifacts can be many MB); the handle closes when the
+        # response ends, including on a client disconnect.
+        file_obj = await anyio.to_thread.run_sync(storage.get_file, storage_path)
+        return ClosingStreamingResponse(
+            _read_chunks(file_obj),
+            media_type=mime_type,
+            headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+            on_close=partial(_close_file, file_obj),
+        )
+    except FileNotFoundError:
+        return json_error("File not found", 404)
+    except Exception:
+        logger.error("Error downloading artifact %s", artifact_id, exc_info=True)
+        return JSONResponse({"success": False}, status_code=400)
+
+
+# Mounted in ``docsgpt/asgi.py`` ahead of the Flask catch-all. The other
+# artifact routes stay on Flask (``routes.py``).
+artifact_download_routes = [
+    Route("/api/artifacts/{artifact_id}/download", download_artifact, methods=["GET"]),
+]

docsgpt/api/user/artifacts/routes.py

@@ -1,18 +1,15 @@
-"""Artifact metadata and download routes (parent-derived authz)."""
+"""Artifact metadata, restore and delete routes (parent-derived authz).
+
+Downloads are served by the native-async route in ``download.py``.
+"""
 
 from __future__ import annotations
 
-import re
-from typing import Optional
-
 from flask import (
-    Response,
     current_app,
     jsonify,
     make_response,
-    redirect,
     request,
-    stream_with_context,
 )
 from flask_restx import Namespace, Resource
 
@@ -24,7 +21,6 @@ from docsgpt.api.user.artifacts.authz import (
     resolve_principal,
     user_can_access_conversation,
 )
-from docsgpt.core.settings import settings
 from docsgpt.storage.db.base_repository import looks_like_uuid
 from docsgpt.storage.db.repositories.artifacts import ArtifactsRepository
 from docsgpt.storage.db.repositories.conversations import ConversationsRepository
@@ -34,20 +30,6 @@ from docsgpt.storage.storage_creator import StorageCreator
 
 artifacts_ns = Namespace("artifacts", description="Artifact operations", path="/api")
 
-# Presigned-URL TTL for private S3 artifact downloads (seconds).
-_PRESIGNED_URL_TTL = 300
-
-
-_ARTIFACT_URL_ENVELOPE_MIME = "application/vnd.docsgpt.artifact-url+json"
-
-
-def _sanitize_header_filename(filename: Optional[str], fallback: str) -> str:
-    """Strip CRLF / quotes from a display filename for a Content-Disposition header."""
-    if not filename:
-        return fallback
-    cleaned = re.sub(r'[\r\n"]', "", str(filename)).strip()
-    return cleaned or fallback
-
 
 def _artifact_summary(artifact: dict) -> dict:
     """Project an artifact identity row to its API metadata shape (owner id withheld)."""
@@ -255,103 +237,6 @@ class GetArtifactVersion(Resource):
             return make_response(jsonify({"success": False}), 400)
 
 
-@artifacts_ns.route("/artifacts/<artifact_id>/download")
-class DownloadArtifact(Resource):
-    @api.doc(description="Download an artifact's bytes (302 to a presigned URL on S3)")
-    def get(self, artifact_id: str):
-        if not looks_like_uuid(artifact_id):
-            return make_response(jsonify({"success": False, "message": "Artifact not found"}), 404)
-        principal = resolve_principal()
-        version_arg = request.args.get("version")
-        try:
-            with db_readonly() as conn:
-                repo = ArtifactsRepository(conn)
-                artifact = repo.get_artifact(artifact_id)
-                if artifact is None:
-                    return make_response(jsonify({"success": False, "message": "Artifact not found"}), 404)
-                if not authorize_artifact(conn, artifact, principal):
-                    return make_response(jsonify({"success": False, "message": "Forbidden"}), 403)
-                version = artifact.get("current_version")
-                if version_arg is not None:
-                    try:
-                        version = int(version_arg)
-                    except ValueError:
-                        return make_response(jsonify({"success": False, "message": "Invalid version"}), 400)
-                version_row = repo.get_version(artifact_id, version)
-
-            if version_row is None:
-                return make_response(jsonify({"success": False, "message": "Version not found"}), 404)
-            # The object key is derived only from the stored path, never client input.
-            storage_path = version_row.get("storage_path")
-            if not storage_path:
-                return make_response(jsonify({"success": False, "message": "No file for this version"}), 404)
-
-            filename = _sanitize_header_filename(version_row.get("filename"), f"artifact-{artifact_id}")
-            mime_type = version_row.get("mime_type") or "application/octet-stream"
-            storage = StorageCreator.get_storage()
-
-            # With URL_STRATEGY=="s3" the contract is to hand back a presigned
-            # URL. If the active backend can't mint one, that's a config error:
-            # surface a 500 rather than silently proxying bytes from a backend
-            # the operator expected to be off the hot path.
-            if getattr(settings, "URL_STRATEGY", "backend") == "s3":
-                try:
-                    url = storage.generate_presigned_url(storage_path, expires_in=_PRESIGNED_URL_TTL)
-                except NotImplementedError:
-                    current_app.logger.error(
-                        "URL_STRATEGY=s3 but %s cannot mint presigned URLs",
-                        type(storage).__name__,
-                    )
-                    return make_response(
-                        jsonify({"success": False, "message": "Storage misconfigured"}),
-                        500,
-                    )
-                # A 302 to a cross-origin S3 URL can't be read by the app's authed
-                # fetch (the bucket has no CORS grant for the app origin). When the
-                # client opts in via ?disposition=url (or Accept: application/json)
-                # hand the presigned URL back as JSON so it can navigate to it
-                # top-level (no CORS). Default stays a 302 so nothing else breaks.
-                if request.args.get("disposition") == "url" or (
-                    "application/json" in request.headers.get("Accept", "")
-                ):
-                    resp = make_response(jsonify({"success": True, "url": url}), 200)
-                    # Tag the envelope with a distinctive media type so the client
-                    # keys off a server-set signal, not the JSON body shape. A
-                    # ``data`` artifact whose bytes happen to be
-                    # ``{"success":true,"url":"..."}`` is streamed under the backend
-                    # strategy with its own (``application/json``) content-type and
-                    # can never carry this vendor type, so it can't be mistaken for a
-                    # redirect (open-redirect gadget). Content-Type is CORS-safelisted,
-                    # so this stays readable cross-origin without expose-headers.
-                    resp.headers["Content-Type"] = _ARTIFACT_URL_ENVELOPE_MIME
-                    return resp
-                return redirect(url, code=302)
-
-            # Stream the bytes in chunks instead of buffering the whole object in
-            # worker memory (artifacts can be many MB); close the handle when done.
-            file_obj = storage.get_file(storage_path)
-
-            def _stream():
-                try:
-                    for chunk in iter(lambda: file_obj.read(65536), b""):
-                        yield chunk
-                finally:
-                    close = getattr(file_obj, "close", None)
-                    if callable(close):
-                        close()
-
-            return Response(
-                stream_with_context(_stream()),
-                mimetype=mime_type,
-                headers={"Content-Disposition": f'attachment; filename="{filename}"'},
-            )
-        except FileNotFoundError:
-            return make_response(jsonify({"success": False, "message": "File not found"}), 404)
-        except Exception as err:
-            current_app.logger.error(f"Error downloading artifact: {err}", exc_info=True)
-            return make_response(jsonify({"success": False}), 400)
-
-
 @artifacts_ns.route("/artifacts/<artifact_id>/restore")
 class RestoreArtifact(Resource):
     @api.doc(description="Restore a prior version by appending it as the new current version")