fix(api): break the import cycle, stop pinning DB connections, revive retries (#66)

The module packages each re-exported their router while core.dependencies
imports identity.models, so importing any submodule dragged the router
graph back through a half-initialised core.dependencies. `import
helprs.core.dependencies` failed on its own; the app booted only because
main.py happens to import admin.views -- which imports a model -- first,
so re-sorting imports would have broken startup. That cycle was also the
reason two services carried function-level imports. The packages are bare
now, both workarounds are gone, and a test imports each module in a fresh
interpreter so the cycle cannot grow back.

FastAPI tears yield-dependencies down only after a streaming body ends, so
the SSE route's DbSession -- and the one behind its auth dependency -- kept
a pooled connection and an open transaction for the life of the stream, up
to CONTAINER_TTL_SECONDS. With 15 connections per worker, 15 viewers
starved the worker, and each idle-in-transaction backend blocked VACUUM.
finalize_session did the same across wait_container. Both now do their
database work in short transactions either side of the wait.

On client disconnect only finalization was detached, not the event drain.
The drain stopped with the generator, so finalize_session then built its
scorecard from a truncated history and posted that to the PR -- defeating
the thing detaching was meant to protect. Both are detached together.

Also:
- container status persisted the enum member NAME ("RUNNING") while the
  column default is a value ("pending"), so a DB-defaulted row raised
  LookupError through the ORM. values_callable plus a data migration.
- per_page=0 divided by zero and per_page=-5 reached Postgres as a
  negative LIMIT; ?status=bogus raised ValueError mid-handler. All three
  were 500s and are now 422s, constrained in the signature.
- a failed container start rolled back the FAILED status *and* the session
  row, so the user got a 502 and a dashboard showing nothing happened.
- mark_failed wrote "failed" and the replay query never selected it, so
  one handler exception dropped a delivery for good and MAX_RETRY_COUNT
  was unreachable. Failed events are replayable and re-claimable, spaced
  by updated_at, and abandonment is now a state that can actually occur.
- core.dependencies went through identity's repository instead of running
  its own SELECT over another module's table.

Author: mariuspruvot

apps/api/alembic/versions/c4d5e6f7a0b1_normalize_container_status_to_values.py

@@ -0,0 +1,34 @@
+"""Normalize container_sessions.status to enum values, not member names.
+
+Without ``values_callable`` on the Enum type, SQLAlchemy persisted the member
+NAME ("RUNNING") while the column's server_default is the value ("pending")
+and the API returns the value. Rows written by the ORM and rows written by the
+database default therefore disagreed, and reading a value-cased row through
+the ORM raised ``LookupError``.
+
+The model now maps by value; this migration brings existing rows in line. Only
+rows written by the ORM need changing, and every enum member is lower-case
+with no internal case distinctions, so ``lower()`` is exactly the mapping.
+
+Revision ID: c4d5e6f7a0b1
+Revises: b3c4d5e6f9a0
+Create Date: 2026-08-01
+
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "c4d5e6f7a0b1"
+down_revision: str | None = "b3c4d5e6f9a0"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.execute("UPDATE container_sessions SET status = lower(status) WHERE status <> lower(status)")
+
+
+def downgrade() -> None:
+    op.execute("UPDATE container_sessions SET status = upper(status) WHERE status <> upper(status)")

apps/api/src/helprs/core/dependencies.py

@@ -6,12 +6,12 @@
 
 from fastapi import Depends, Request
 from jose import JWTError
-from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from helprs.core.config import Settings, get_settings
 from helprs.core.exceptions import UnauthorizedError
 from helprs.core.security import decode_access_token
+from helprs.modules.identity import repository as identity_repository
 from helprs.modules.identity.models import GitHubUser
 
 # Re-export get_settings as a dependency
@@ -33,8 +33,14 @@ async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]:
 DbSession = Annotated[AsyncSession, Depends(get_db)]
 
 
-async def _authenticate(request: Request, session: AsyncSession, settings: Settings, token: str) -> GitHubUser:
-    """Resolve a bearer token to the user it belongs to."""
+async def authenticate_token(session: AsyncSession, settings: Settings, token: str) -> GitHubUser:
+    """Resolve a bearer token to the user it belongs to.
+
+    Public because the SSE endpoint cannot authenticate through the dependency
+    graph: FastAPI tears yield-dependencies down only after the streaming body
+    finishes, so a ``Depends(get_db)`` session would stay checked out for the
+    life of the stream. That route calls this inside its own short session.
+    """
     try:
         payload = decode_access_token(token, settings.SECRET_KEY.get_secret_value())
     except JWTError as e:
@@ -43,17 +49,12 @@ async def _authenticate(request: Request, session: AsyncSession, settings: Setti
     if payload.get("type") == "refresh":
         raise UnauthorizedError("Cannot use refresh token as access token")
 
-    user_id = payload.get("sub")
-    if not user_id:
-        raise UnauthorizedError("Invalid token payload")
-
     try:
-        uuid.UUID(user_id)
-    except (ValueError, AttributeError) as e:
+        user_id = uuid.UUID(payload.get("sub", ""))
+    except (ValueError, AttributeError, TypeError) as e:
         raise UnauthorizedError("Invalid token payload") from e
 
-    result = await session.execute(select(GitHubUser).where(GitHubUser.id == user_id))
-    user = result.scalar_one_or_none()
+    user = await identity_repository.get_by_id(session, user_id)
     if not user:
         raise UnauthorizedError("User not found")
 
@@ -67,28 +68,28 @@ def _bearer_token(request: Request) -> str:
     return auth_header.removeprefix("Bearer ")
 
 
-async def get_current_user(request: Request, session: DbSession, settings: GetSettings) -> GitHubUser:
-    """Authenticate from the ``Authorization`` header."""
-    return await _authenticate(request, session, settings, _bearer_token(request))
-
-
-async def get_current_user_for_stream(request: Request, session: DbSession, settings: GetSettings) -> GitHubUser:
-    """Authenticate an SSE request, allowing the token in the query string.
+def stream_token(request: Request) -> str:
+    """The JWT for an SSE request, which may carry it in the query string.
 
     ``EventSource`` cannot set request headers, so a browser opening a stream
-    has nowhere else to put the JWT. The cost is real — query strings land in
-    proxy logs, browser history and ``Referer`` — so this is granted to the
-    streaming endpoints only, never to the whole API.
+    has nowhere else to put the token. The cost is real -- query strings land
+    in proxy logs, browser history and ``Referer`` -- so this is granted to
+    the streaming endpoints only, never to the whole API. The header form
+    still has to be a well-formed ``Bearer`` value.
     """
-    token = request.headers.get("Authorization", "").removeprefix("Bearer ") or request.query_params.get(
-        "access_token", ""
-    )
+    if request.headers.get("Authorization"):
+        return _bearer_token(request)
+    token = request.query_params.get("access_token", "")
     if not token:
         raise UnauthorizedError("Missing or invalid Authorization header")
-    return await _authenticate(request, session, settings, token)
+    return token
+
+
+async def get_current_user(request: Request, session: DbSession, settings: GetSettings) -> GitHubUser:
+    """Authenticate from the ``Authorization`` header."""
+    return await authenticate_token(session, settings, _bearer_token(request))
 
 
 # Routes take `user: CurrentUser` rather than `user=Depends(get_current_user)`:
 # the Annotated form is a real type for the checker and does not trip B008.
 CurrentUser = Annotated[GitHubUser, Depends(get_current_user)]
-StreamUser = Annotated[GitHubUser, Depends(get_current_user_for_stream)]

apps/api/src/helprs/modules/container/__init__.py

@@ -1,5 +1,8 @@
 """Container module -- ephemeral Docker container lifecycle management."""
 
-from helprs.modules.container.router import router
-
-__all__ = ["router"]
+# Deliberately no re-exports. Importing a router here made every
+# submodule import pull the whole router graph back through
+# core.dependencies, which imports identity.models -- a cycle that left
+# `import helprs.core.dependencies` broken on its own and made startup
+# depend on main.py happening to import admin.views first. Import
+# submodules by their full path instead.

apps/api/src/helprs/modules/container/models.py

@@ -45,7 +45,19 @@ class ContainerSession(Base):
     skill_name: Mapped[str] = mapped_column(String(100), nullable=False)
     container_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
     status: Mapped[ContainerStatus] = mapped_column(
-        Enum(ContainerStatus, name="container_status", native_enum=False, length=20),
+        # values_callable is load-bearing: without it SQLAlchemy persists the
+        # member NAME ("RUNNING"), while this column's server_default is the
+        # value ("pending") and the API returns the value too. Any row created
+        # without an explicit status -- the DB default, raw SQL, a bulk import
+        # -- was then unreadable through the ORM:
+        #   LookupError: 'running' is not among the defined enum values
+        Enum(
+            ContainerStatus,
+            name="container_status",
+            native_enum=False,
+            length=20,
+            values_callable=lambda enum: [member.value for member in enum],
+        ),
         nullable=False,
         default=ContainerStatus.PENDING,
         index=True,

apps/api/src/helprs/modules/container/router.py

@@ -12,7 +12,8 @@
 from fastapi import APIRouter, Request, Response
 from fastapi.responses import StreamingResponse
 
-from helprs.core.dependencies import CurrentUser, DbSession, GetSettings, StreamUser
+from helprs.core.database import get_db_context
+from helprs.core.dependencies import CurrentUser, DbSession, GetSettings, authenticate_token, stream_token
 from helprs.core.exceptions import ConflictError, NotFoundError
 from helprs.core.middleware import limiter
 from helprs.core.security import fernet_decrypt
@@ -101,6 +102,12 @@ async def create_container_session(
         skill_name=body.skill_name,
         user_id=user.id,
     )
+    # Committed before the container is touched. start_container marks the row
+    # FAILED on error and then raises, but that exception unwinds through
+    # get_db, which rolls back -- discarding the FAILED status *and* the row
+    # itself. A failed start left the user with a 502 and a dashboard showing
+    # that nothing had ever happened.
+    await db.commit()
 
     docker = _get_docker_client()
     try:
@@ -151,19 +158,31 @@ def _resume_offset(request: Request, offset: int) -> int:
 async def stream_container_output(
     session_id: UUID,
     request: Request,
-    db: DbSession,
     settings: GetSettings,
-    user: StreamUser,
     offset: int = 0,
 ) -> StreamingResponse:
-    """Relay the container's output as Server-Sent Events."""
-    cs = await get_session_or_404(db, session_id)
-    await verify_session_access(user, cs, db, settings)
+    """Relay the container's output as Server-Sent Events.
+
+    Deliberately takes no ``DbSession`` and no user dependency. FastAPI tears
+    yield-dependencies down only once the streaming body completes, so any
+    ``Depends(get_db)`` here -- including the one behind an authentication
+    dependency -- would keep a pooled connection, and an open transaction,
+    checked out for the life of the stream: up to CONTAINER_TTL_SECONDS. The
+    pool is DB_POOL_SIZE + DB_MAX_OVERFLOW per worker, so a handful of
+    viewers would starve every other request on that worker, and each
+    idle-in-transaction backend blocks VACUUM. Authentication and
+    authorization run in a short session that closes before streaming starts.
+    """
+    async with get_db_context() as db:
+        user = await authenticate_token(db, settings, stream_token(request))
+        cs = await get_session_or_404(db, session_id)
+        await verify_session_access(user, cs, db, settings)
+
+        if cs.status != ContainerStatus.RUNNING or not cs.container_id:
+            raise ConflictError("Container is not running")
 
-    if cs.status != ContainerStatus.RUNNING or not cs.container_id:
-        raise ConflictError("Container is not running")
+        container_id = cs.container_id
 
-    container_id = cs.container_id
     resume_from = _resume_offset(request, offset)
     docker = _get_docker_client()
 
@@ -177,13 +196,23 @@ async def _event_stream():
             status = await finalize_session(session_id, docker)
             finalized = True
             yield _done_event(status)
+        except Exception:
+            # The response has already started, so there is no status code
+            # left to change. Without a frame here the client just sees a
+            # truncated body, and native EventSource silently reconnects
+            # into the same failure.
+            logger.exception("sse_stream_failed", session_id=str(session_id))
+            yield _error_event("Streaming failed. The session will be finalized in the background.")
         finally:
             if not finalized:
-                # The client hung up mid-stream. Finalization must still run,
-                # detached from this request — otherwise the session stays
-                # RUNNING until the reaper mislabels it TIMEOUT, with its
-                # scorecard unextracted and its PR comment never posted.
-                spawn_detached(_finalize_and_close(session_id, docker))
+                # The client hung up, or the stream broke, mid-session. Both
+                # the drain and the finalization have to continue detached
+                # from this request. Detaching only the finalization -- as
+                # this did before -- stops the drain the moment the tab
+                # closes, so finalize_session then builds its scorecard from
+                # a truncated event history and posts that to the PR, quietly
+                # defeating the thing detaching was meant to protect.
+                spawn_detached(_drain_and_finalize(session_id, container_id, docker))
             else:
                 await docker.close()
 
@@ -208,9 +237,24 @@ def _done_event(status: ContainerStatus) -> str:
     return f"event: done\ndata: {json.dumps({'message': message, 'status': status.value})}\n\n"
 
 
-async def _finalize_and_close(session_id: UUID, docker: DockerClient) -> None:
+def _error_event(message: str) -> str:
+    return f"event: error\ndata: {json.dumps({'message': message})}\n\n"
+
+
+async def _drain_and_finalize(session_id: UUID, container_id: str, docker: DockerClient) -> None:
+    """Finish the session with no client attached.
+
+    Keeps consuming the container's output until it exits, then finalizes.
+    Re-reading the log from the start is deliberate and cheap: ``_persist``
+    is idempotent through ``ON CONFLICT DO NOTHING``, and nothing is being
+    yielded to anyone, so the offset does not matter here.
+    """
     try:
+        async for _ in stream_and_persist(docker, container_id, session_id=session_id):
+            pass
         await finalize_session(session_id, docker)
+    except Exception:
+        await logger.aexception("detached_finalize_failed", session_id=str(session_id))
     finally:
         await docker.close()