Optimize dashboard overview loading

api/routers/tasks.py

@@ -9,7 +9,7 @@ from api.database import get_session
 from api.models.task import Task, TaskStatus
 from api.models.token_event import TokenEvent
 from api.models.workstream import Workstream
-from api.schemas.task import TaskCreate, TaskRead, TaskUpdate
+from api.schemas.task import TaskCountRead, TaskCreate, TaskRead, TaskUpdate
 from api.services.lifecycle import status_value, transition_task_status
 from api.task_status import normalize_task_status
 
@@ -24,6 +24,8 @@ async def list_tasks(
     needs_human: bool | None = Query(None),
     priority: str | None = None,
     due_date_before: date | None = None,
+    limit: int | None = Query(None, ge=1, le=5000),
+    offset: int = Query(0, ge=0),
     session: AsyncSession = Depends(get_session),
 ) -> list[Task]:
     q = select(Task)
@@ -40,10 +42,32 @@ async def list_tasks(
     if due_date_before is not None:
         q = q.where(Task.due_date <= due_date_before)
     q = q.order_by(Task.created_at)
+    if offset:
+        q = q.offset(offset)
+    if limit is not None:
+        q = q.limit(limit)
     result = await session.execute(q)
     return list(result.scalars().all())
 
 
+@router.get("/counts", response_model=list[TaskCountRead])
+async def count_tasks(
+    workstream_id: uuid.UUID | None = None,
+    status: str | None = None,
+    session: AsyncSession = Depends(get_session),
+) -> list[TaskCountRead]:
+    q = select(Task.workstream_id, Task.status, func.count()).group_by(Task.workstream_id, Task.status)
+    if workstream_id:
+        q = q.where(Task.workstream_id == workstream_id)
+    if status:
+        q = q.where(Task.status == TaskStatus(normalize_task_status(status)))
+    rows = await session.execute(q)
+    return [
+        TaskCountRead(workstream_id=ws_id, status=task_status, count=count)
+        for ws_id, task_status, count in rows
+    ]
+
+
 @router.post("/", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
 async def create_task(
     body: TaskCreate,