feat(comprehension): add SSE question streaming and chat panel (Story 3.3)

Implements the end-to-end Socratic question generation flow: PydanticAI
streaming provider, SSE endpoint with DB-phase / HTTP-phase split,
metadata-only question persistence (hash, never verbatim text), and
a live-updating chat panel with reduced-motion support.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: mariuspruvot

_bmad-output/implementation-artifacts/sprint-status.yaml

@@ -45,7 +45,7 @@
 #   'awaiting-manual-qa', unless invoked with `--skip-manual-qa-gate <reason>`.
 
 generated: 2026-04-09
-last_updated: 2026-04-10 # 3-3 → in-progress (dev-story: question generation + SSE streaming)
+last_updated: 2026-04-11 # 3-3 → review (dev-story complete: question generation + SSE streaming)
 project: helprs
 project_key: NOKEY
 tracking_system: file-system
@@ -114,7 +114,7 @@ development_status:
   # replaced "Story 3.3 backlog" prose with grep-able TODO(story-3.3).
   # make lint && vitest = clean + 24 web tests green (24 = 25 −3 +2).
   3-2-split-view-session-ui-and-diff-viewer: done
-  3-3-question-generation-and-sse-streaming: in-progress
+  3-3-question-generation-and-sse-streaming: review
   3-4-answer-submission-and-feedback-with-code-links: backlog
   3-5-role-adaptation-beyond-diff-and-large-pr-handling: backlog
   epic-3-retrospective: optional

apps/api/alembic/versions/a1b2c3d4e5f6_add_questions_table_and_total_.py

@@ -0,0 +1,81 @@
+"""add questions table and total_questions
+
+Revision ID: a1b2c3d4e5f6
+Revises: 9036c7377667
+Create Date: 2026-04-10 14:00:00.000000
+
+Story 3.3: introduces the ``questions`` table (metadata-only — no
+``text`` column; see FR35/NFR14) and adds ``sessions.total_questions``
+so the session detail endpoint + SSE stream know up-front how many
+questions to ask. Existing rows get ``total_questions = 0`` via the
+``server_default``.
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "a1b2c3d4e5f6"
+down_revision: str | None = "9036c7377667"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ---- sessions.total_questions (additive, nullable=False + default) ----
+    op.add_column(
+        "sessions",
+        sa.Column(
+            "total_questions",
+            sa.Integer(),
+            nullable=False,
+            server_default=sa.text("0"),
+        ),
+    )
+
+    # ---- questions table -------------------------------------------------
+    op.create_table(
+        "questions",
+        sa.Column("session_id", sa.Uuid(), nullable=False),
+        sa.Column("number", sa.Integer(), nullable=False),
+        sa.Column("topic", sa.String(length=32), nullable=False),
+        sa.Column("text_hash", sa.String(length=64), nullable=False),
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(
+            ["session_id"],
+            ["sessions.id"],
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "session_id",
+            "number",
+            name="uq_questions_session_number",
+        ),
+    )
+    op.create_index(
+        "ix_questions_session_id",
+        "questions",
+        ["session_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_questions_session_id", table_name="questions")
+    op.drop_table("questions")
+    op.drop_column("sessions", "total_questions")

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

@@ -1,7 +1,8 @@
 """Async SQLAlchemy engine and session factory."""
 
 import uuid
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, AsyncIterator
+from contextlib import asynccontextmanager
 from datetime import datetime
 
 from sqlalchemy import DateTime, func
@@ -10,6 +11,64 @@
 
 from helprs.core.config import get_settings
 
+# Module-level slot holding the active session factory. Populated by
+# ``helprs.main`` during lifespan startup and read by ``get_db_context``
+# (a plain async context manager — not a FastAPI dependency — used by
+# streaming endpoints that need to open a fresh short-lived session
+# inside a stream generator AFTER the request-scoped ``get_db`` session
+# has already been released).
+#
+# The FastAPI dep graph keeps using ``app.state.session_factory`` via
+# ``core.dependencies.get_db`` — ``get_db_context`` does NOT replace it;
+# it is additive, for code paths that live outside the request lifecycle.
+_session_factory_holder: async_sessionmaker[AsyncSession] | None = None
+
+
+def set_session_factory(factory: async_sessionmaker[AsyncSession]) -> None:
+    """Register the session factory for ``get_db_context`` to use.
+
+    Called once by the lifespan startup in ``helprs.main`` after the
+    engine is created. Tests that need ``get_db_context`` outside the
+    app lifespan can call this directly in a fixture.
+    """
+    global _session_factory_holder
+    _session_factory_holder = factory
+
+
+def clear_session_factory() -> None:
+    """Reset the module-level slot (test teardown)."""
+    global _session_factory_holder
+    _session_factory_holder = None
+
+
+@asynccontextmanager
+async def get_db_context() -> AsyncIterator[AsyncSession]:
+    """Open a short-lived AsyncSession outside the FastAPI dep graph.
+
+    Used by SSE streaming code that needs to commit small writes
+    (per-question persistence) AFTER the request-scoped
+    ``get_db`` dependency has released its session. Holding the
+    request-scoped session open across 10s+ LLM calls would exhaust
+    the connection pool under load — this helper is the escape hatch.
+
+    Commits on clean exit and rolls back on exception. Never use
+    ``get_db_context`` for request-scoped DB work — the FastAPI
+    ``get_db`` dependency is the right tool there.
+    """
+    if _session_factory_holder is None:
+        raise RuntimeError(
+            "get_db_context called before set_session_factory — "
+            "this is a lifespan bug in helprs.main or a missing "
+            "test fixture setup."
+        )
+    async with _session_factory_holder() as session:
+        try:
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise
+
 
 class Base(AsyncAttrs, DeclarativeBase):
     """Base model with common columns for all entities."""

apps/api/src/helprs/main.py

@@ -8,7 +8,12 @@
 from fastapi import APIRouter, FastAPI
 
 from helprs.core.config import get_settings
-from helprs.core.database import create_engine, create_session_factory
+from helprs.core.database import (
+    clear_session_factory,
+    create_engine,
+    create_session_factory,
+    set_session_factory,
+)
 from helprs.core.exceptions import DomainError, domain_exception_handler
 from helprs.core.middleware import configure_logging, configure_sentry, setup_middleware
 
@@ -108,8 +113,13 @@ async def lifespan(app: FastAPI):
         engine = create_engine()
         reaper_task: asyncio.Task | None = None
         try:
+            session_factory = create_session_factory(engine)
             app.state.engine = engine
-            app.state.session_factory = create_session_factory(engine)
+            app.state.session_factory = session_factory
+            # Register the factory for ``get_db_context`` (used by the
+            # SSE stream generator for per-question writes outside the
+            # request-scoped dep graph — Story 3.3).
+            set_session_factory(session_factory)
             app.state.replay_semaphore = asyncio.Semaphore(_REPLAY_CONCURRENCY)
             app.state.replay_tasks = set()
 
@@ -141,6 +151,7 @@ async def lifespan(app: FastAPI):
             if tracked:
                 await asyncio.gather(*tracked, return_exceptions=True)
 
+            clear_session_factory()
             await engine.dispose()
 
     app = FastAPI(
@@ -160,6 +171,7 @@ async def lifespan(app: FastAPI):
     api_router = APIRouter(prefix="/api/v1")
 
     from helprs.modules.comprehension.presentation.routers import router as comprehension_router
+    from helprs.modules.comprehension.presentation.sse import sse_router as comprehension_sse_router
     from helprs.modules.identity.router import router as identity_router
     from helprs.modules.installation.router import router as installation_router
     from helprs.modules.webhook.router import router as webhook_router
@@ -168,6 +180,11 @@ async def lifespan(app: FastAPI):
     api_router.include_router(installation_router)
     api_router.include_router(webhook_router)
     api_router.include_router(comprehension_router)
+    # Story 3.3: SSE streaming for Socratic question generation. Mounted
+    # AFTER the detail router so FastAPI's route matcher considers the
+    # more specific ``/{session_id}/stream`` path alongside the existing
+    # ``/{session_id}`` detail route.
+    api_router.include_router(comprehension_sse_router)
 
     app.include_router(api_router)
 

apps/api/src/helprs/modules/comprehension/application/commands.py

@@ -24,3 +24,9 @@ class StartSessionCommand:
     # field in a frozen dataclass still lets callers mutate labels after
     # construction. Immutable by construction keeps the command safe.
     pr_labels: tuple[str, ...]
+    # Story 3.3: optional PR diff line count, used by
+    # ``StartSessionHandler`` to size the session via
+    # ``estimate_question_count``. ``None`` for legacy callers (existing
+    # webhook-handler tests without an updated fixture) — the handler
+    # falls back to ``5`` questions in that case.
+    pr_diff_line_count: int | None = None

apps/api/src/helprs/modules/comprehension/application/handlers.py

@@ -29,6 +29,7 @@
 from helprs.modules.comprehension.application.commands import StartSessionCommand
 from helprs.modules.comprehension.application.queries import GetSessionQuery, GetSessionResult
 from helprs.modules.comprehension.domain.entities import PRContext, Session
+from helprs.modules.comprehension.domain.services import estimate_question_count
 from helprs.modules.comprehension.infrastructure.repositories import SqlAlchemySessionRepository
 from helprs.modules.installation.service import (
     get_default_suppression_labels,
@@ -126,8 +127,16 @@ async def handle(self, cmd: StartSessionCommand) -> StartSessionResult:
             pr_diff_url=cmd.pr_diff_url,
         )
 
+        # TODO(story-3.5): replace with role-adaptive + large-PR sizing
+        # (see FR39–FR41). Story 3.3 uses the minimal line-count
+        # heuristic so the end-to-end flow ships first.
+        total_questions = estimate_question_count(cmd.pr_diff_line_count) if cmd.pr_diff_line_count is not None else 5
+
         if len(existing) == 0:
-            author, reviewer = await repo.add_pair(pr_ctx=pr_ctx)
+            author, reviewer = await repo.add_pair(
+                pr_ctx=pr_ctx,
+                total_questions=total_questions,
+            )
             pair: tuple[Session, Session] = (author, reviewer)
             created = True
             comment_needed = True
@@ -156,7 +165,10 @@ async def handle(self, cmd: StartSessionCommand) -> StartSessionResult:
                 orphan_session_id=str(orphan.id),
             )
             await repo.delete_one(session_id=orphan.id)
-            author, reviewer = await repo.add_pair(pr_ctx=pr_ctx)
+            author, reviewer = await repo.add_pair(
+                pr_ctx=pr_ctx,
+                total_questions=total_questions,
+            )
             pair = (author, reviewer)
             created = True
             comment_needed = True
@@ -259,8 +271,11 @@ async def handle(self, query: GetSessionQuery) -> GetSessionResult:
         #    exceptions".
         installation_token = await mint_installation_token(domain_session.github_installation_id, self._settings)
 
+        # Story 3.3: count persisted questions for the detail response.
+        question_count = await repo.count_questions(session_id=query.session_id)
+
         return GetSessionResult(
             session=domain_session,
             installation_token=installation_token,
-            question_count=0,  # Story 3.3 will count QuestionModel rows here
+            question_count=question_count,
         )

apps/api/src/helprs/modules/comprehension/domain/entities.py

@@ -75,6 +75,11 @@ class Session:
     pr_diff_url: str
     role: SessionRole
     status: SessionStatus
+    # Story 3.3: number of questions this session plans to ask, set at
+    # creation time by ``estimate_question_count``. Zero on legacy rows
+    # created before the Story 3.3 migration — the SSE endpoint falls
+    # back to ``5`` in that case.
+    total_questions: int
     created_at: datetime
     updated_at: datetime