fix(projects): batch the summary queries — the fan-out was exhausting the pool

Reported live: Projects and Snippets showed skeletons that never resolved,
/knowledge worked intermittently. The logs named it exactly:

    QueuePool limit of size 5 overflow 10 reached, connection timed out, 30.00
    GET /api/settings  500  30584.0ms
    GET /api/projects  200  30882.9ms

/api/projects was not hanging — it was waiting out the 30-second checkout
timeout and then returning 200 with summaries silently missing, because
_attach swallowed the TimeoutError. Nobody waits 31 seconds, so it read as a
hang.

THE SHAPE: routes/projects.py ran asyncio.gather over every project. Each
_attach called get_project_summary, which opened its own session for three
queries and then called get_project_milestone_summary — which opened one more
session PER MILESTONE. So 25 projects asked for roughly 250 concurrent
checkouts against a pool of 15 (SQLAlchemy's default 5 + 10 overflow).

That is why unrelated routes failed too. Snippets and /knowledge were never
broken; they queued behind the burst and inherited its timeout. /api/settings
returning 500 while /api/projects returned 200 is the same cause wearing two
faces.

The comment above the gather said "one backend pass instead of N+1 frontend
calls". It did remove the N+1 from the network — and recreated it against the
connection pool, where it is worse, because the browser had at least been
serialising those calls.

Now: get_project_summaries() does all projects in four queries and one session,
and get_project_milestone_summaries() does all milestones in two. Two sessions
total for the whole page, independent of how many projects exist.

The progress calculation is extracted to _progress_from_counts and shared by
both the batch and single paths, so the cancelled-exclusion rule cannot drift
into two versions that disagree about whether a milestone is finished.

Tests assert the SESSION COUNT, not just the values. An implementation that
returned identical output while opening a session per project would pass a
correctness test and reproduce the outage.

Deliberately NOT done: raising pool_size. It would move the cliff rather than
remove it, and this endpoint now needs two connections regardless of scale.

Closes #2384.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UaYUaouG9jjhATyuxCKrQs

src/scribe/services/milestones.py

@@ -156,26 +156,82 @@ async def get_milestone_progress(milestone_id: int) -> dict:
         for status, count in rows.fetchall():
             status_counts[status] = count
 
-        total = sum(status_counts.values())
-        cancelled = status_counts.get("cancelled", 0)
-        completed = status_counts.get("done", 0)
-        # Cancelled tasks are resolved work, not pending — exclude them from the
-        # percent-complete denominator so a milestone whose only open task was
-        # cancelled still reaches 100% (and auto-collapses) instead of stalling.
-        active_total = total - cancelled
-        pct = round(completed / active_total * 100, 1) if active_total > 0 else 0.0
+    # Same rule as the batch path, computed in one place so the two cannot
+    # drift on the cancelled-exclusion.
+    return _progress_from_counts(status_counts)
 
-        return {
-            "total": total,
-            "completed": completed,
-            "pct": pct,
-            "status_counts": {
-                "todo": status_counts.get("todo", 0),
-                "in_progress": status_counts.get("in_progress", 0),
-                "done": status_counts.get("done", 0),
-                "cancelled": cancelled,
-            },
-        }
+
+def _progress_from_counts(status_counts: dict[str, int]) -> dict:
+    """The progress shape, computed from already-fetched counts.
+
+    Split out of get_milestone_progress so the batch path can reuse the rule
+    rather than restate it — the cancelled-exclusion below is easy to get
+    subtly different in a second copy, and then two screens disagree about
+    whether a milestone is finished.
+    """
+    total = sum(status_counts.values())
+    cancelled = status_counts.get("cancelled", 0)
+    completed = status_counts.get("done", 0)
+    # Cancelled tasks are resolved work, not pending — excluded from the
+    # denominator so a milestone whose only open task was cancelled reaches
+    # 100% instead of stalling.
+    active_total = total - cancelled
+    return {
+        "total": total,
+        "completed": completed,
+        "pct": round(completed / active_total * 100, 1) if active_total > 0 else 0.0,
+        "status_counts": {
+            "todo": status_counts.get("todo", 0),
+            "in_progress": status_counts.get("in_progress", 0),
+            "done": status_counts.get("done", 0),
+            "cancelled": cancelled,
+        },
+    }
+
+
+async def get_project_milestone_summaries(
+    user_id: int, project_ids: list[int]
+) -> dict[int, list[dict]]:
+    """Milestone summaries for MANY projects in two queries total.
+
+    The per-project version below is a nested fan-out: one query to list a
+    project's milestones, then one more per milestone for its progress. Called
+    for 25 projects concurrently it asked for ~250 pooled connections against a
+    pool of 15, and every one of them waited out the 30-second checkout timeout
+    (#2384). This does the same work in two queries and one session.
+    """
+    if not project_ids:
+        return {}
+
+    async with async_session() as session:
+        milestones = list((await session.execute(
+            select(Milestone).where(
+                Milestone.user_id == user_id,
+                Milestone.project_id.in_(project_ids),
+                Milestone.deleted_at.is_(None),
+            ).order_by(Milestone.order_index.asc(), Milestone.created_at.asc())
+        )).scalars().all())
+
+        counts: dict[int, dict[str, int]] = {}
+        if milestones:
+            rows = await session.execute(
+                select(Note.milestone_id, Note.status, func.count(Note.id))
+                .where(
+                    Note.milestone_id.in_([m.id for m in milestones]),
+                    Note.status.isnot(None),
+                    Note.deleted_at.is_(None),
+                )
+                .group_by(Note.milestone_id, Note.status)
+            )
+            for milestone_id, status, count in rows.fetchall():
+                counts.setdefault(milestone_id, {})[status] = count
+
+    out: dict[int, list[dict]] = {pid: [] for pid in project_ids}
+    for m in milestones:
+        entry = m.to_dict()
+        entry.update(_progress_from_counts(counts.get(m.id, {})))
+        out.setdefault(m.project_id, []).append(entry)
+    return out
 
 
 async def get_project_milestone_summary(user_id: int, project_id: int) -> list[dict]: