feat(P02): SLICE-08 operator API cohort endpoints — auth-gated, k-anon

TASK-08-01: server/operator/cohort.py — GET /api/operator/cohort practice
  volume view (sessions_count, active_learners_count), k-anon display.
TASK-08-02: server/operator/mastery.py — GET /api/operator/mastery
  (gate_open_rate, median_mastery_score, rubric_criterion_means).
TASK-08-03: server/operator/failure_patterns.py — GET /api/operator/failure-patterns
  (failure_mode:*, branch:* metrics).
TASK-08-04: server/operator/credentials.py — GET /api/operator/credentials
  + POST /api/operator/credentials/{id}/revoke (VC management, D-057).
TASK-08-05: tests/test_operator_endpoints.py — 401/200/503, suppressed
  cells value=null, last_updated=max, revoke, no per-learner data (R-DASH-02).
server/operator/_common.py — shared Cell/PathView/ViewResponse models +
  all_recent_aggregates query + group_by_path helper.

All endpoints auth-gated via Depends(current_operator) (D-057). No PII
beyond aggregates (D-031, R-DASH-02).

---ci---
project: praxis
phase: 2
milestone: v0.4
status: execute
persona: backend-engineer
task: 08-01..08-05
requirements:
  covered: [REQ-DASH-01, REQ-NFR-DASH-01]
---/ci---

server/operator/_common.py

@@ -0,0 +1,93 @@
+"""Shared helpers for operator API endpoints (SLICE-08).
+
+Common response models + the recent-aggregates query used by all 3 cohort
+view endpoints (cohort, mastery, failure-patterns). Kept here to avoid
+duplicating the Pydantic models + pool query across 3 files.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+from typing import Any
+
+from fastapi import HTTPException, Request, status
+from pydantic import BaseModel
+
+
+class Cell(BaseModel):
+    metric: str
+    window_start: _dt.date | None = None
+    window_end: _dt.date | None = None
+    value: float | None = None
+    cell_count: int = 0
+    cell_suppressed: bool = False
+    updated_at: _dt.datetime | None = None
+
+
+class PathView(BaseModel):
+    path: str
+    metrics: list[Cell]
+
+
+class ViewResponse(BaseModel):
+    views: list[PathView]
+    last_updated: _dt.datetime | None = None
+
+
+async def require_pg_store(request: Request):
+    pg_store = getattr(request.app.state, "pg_store", None)
+    if pg_store is None:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="operator tier unavailable (no Postgres)",
+        )
+    return pg_store
+
+
+async def all_recent_aggregates(pg_store, since: _dt.date) -> list[dict[str, Any]]:
+    async with pg_store.pool.acquire() as conn:
+        rows = await conn.fetch(
+            "SELECT path, metric, window_start, window_end, value, "
+            "cell_count, cell_suppressed, updated_at "
+            "FROM cohort_aggregates WHERE window_start >= $1 "
+            "ORDER BY path, metric, window_start",
+            since,
+        )
+        return [dict(r) for r in rows]
+
+
+def cell_from_row(row: dict[str, Any]) -> Cell:
+    return Cell(
+        metric=row.get("metric", ""),
+        window_start=row.get("window_start"),
+        window_end=row.get("window_end"),
+        value=float(row["value"]) if row.get("value") is not None else None,
+        cell_count=int(row.get("cell_count") or 0),
+        cell_suppressed=bool(row.get("cell_suppressed") or False),
+        updated_at=row.get("updated_at"),
+    )
+
+
+def group_by_path(
+    rows: list[dict[str, Any]],
+    metric_filter: set[str] | None = None,
+) -> tuple[list[PathView], _dt.datetime | None]:
+    by_path: dict[str, list[dict[str, Any]]] = {}
+    last_updated: _dt.datetime | None = None
+    for r in rows:
+        if metric_filter is not None and r.get("metric") not in metric_filter:
+            continue
+        by_path.setdefault(r["path"], []).append(r)
+        ua = r.get("updated_at")
+        if isinstance(ua, _dt.datetime) and (last_updated is None or ua > last_updated):
+            last_updated = ua
+    views = [PathView(path=p, metrics=[cell_from_row(c) for c in cells])
+             for p, cells in by_path.items()]
+    return views, last_updated
+
+
+__all__ = [
+    "Cell", "PathView", "ViewResponse",
+    "require_pg_store", "all_recent_aggregates",
+    "cell_from_row", "group_by_path",
+]

server/operator/cohort.py

@@ -0,0 +1,42 @@
+"""GET /api/operator/cohort — practice volume view (TASK-08-01, D-053, D-057).
+
+Auth-gated (Depends(current_operator)). Returns k-anonymized practice-volume
+aggregates from cohort_aggregates: sessions_count + active_learners_count per
+path. Suppressed cells have value=null + cell_suppressed=true; the frontend
+renders \"— (<10 learners)\". No per-learner drill-down (R-DASH-02).
+last_updated = max(updated_at) for freshness (REQ-NFR-DASH-02).
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+
+from fastapi import APIRouter, Depends, Request
+
+from server.auth.dependencies import current_operator
+from server.auth.models import Operator
+from server.operator._common import (
+    ViewResponse,
+    all_recent_aggregates,
+    group_by_path,
+    require_pg_store,
+)
+
+router = APIRouter(prefix="/api/operator", tags=["operator-cohort"])
+
+PRACTICE_METRICS = {"sessions_count", "active_learners_count"}
+
+
+@router.get("/cohort", response_model=ViewResponse)
+async def cohort_view(
+    request: Request,
+    op: Operator = Depends(current_operator),
+) -> ViewResponse:
+    pg_store = await require_pg_store(request)
+    since = _dt.date.today() - _dt.timedelta(days=30)
+    rows = await all_recent_aggregates(pg_store, since)
+    views, last_updated = group_by_path(rows, PRACTICE_METRICS)
+    return ViewResponse(views=views, last_updated=last_updated)
+
+
+__all__ = ["router"]

server/operator/credentials.py

@@ -0,0 +1,78 @@
+"""GET/POST /api/operator/credentials — VC management (TASK-08-04, D-057).
+
+Auth-gated. GET lists issued VCs from Postgres issued_credentials (operator's
+issuance log). POST /{id}/revoke revokes a VC (status='revoked',
+revoked_at=now()). Revoked credentials fail verification. No PII beyond what
+the credential asserts (D-043).
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel
+
+from server.auth.dependencies import current_operator
+from server.auth.models import Operator
+from server.operator._common import require_pg_store
+
+router = APIRouter(prefix="/api/operator", tags=["operator-credentials"])
+
+
+class CredentialOut(BaseModel):
+    id: str
+    learner_ref: str
+    vc_type: str | None = None
+    status: str
+    issued_at: _dt.datetime | None = None
+    revoked_at: _dt.datetime | None = None
+
+
+class CredentialListResponse(BaseModel):
+    credentials: list[CredentialOut]
+
+
+class OkResponse(BaseModel):
+    ok: bool = True
+    id: str
+    status: str
+
+
+@router.get("/credentials", response_model=CredentialListResponse)
+async def list_credentials(
+    request: Request,
+    op: Operator = Depends(current_operator),
+) -> CredentialListResponse:
+    pg_store = await require_pg_store(request)
+    rows = await pg_store.list_credentials()
+    creds = [
+        CredentialOut(
+            id=str(r["id"]),
+            learner_ref=r["learner_ref"],
+            vc_type=r.get("vc_type"),
+            status=r.get("status", "active"),
+            issued_at=r.get("issued_at"),
+            revoked_at=r.get("revoked_at"),
+        )
+        for r in rows
+    ]
+    return CredentialListResponse(credentials=creds)
+
+
+@router.post("/credentials/{cred_id}/revoke", response_model=OkResponse)
+async def revoke_credential(
+    cred_id: str,
+    request: Request,
+    op: Operator = Depends(current_operator),
+) -> OkResponse:
+    pg_store = await require_pg_store(request)
+    row = await pg_store.get_credential(cred_id)
+    if row is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail="credential not found")
+    await pg_store.set_credential_status(cred_id, "revoked")
+    return OkResponse(ok=True, id=cred_id, status="revoked")
+
+
+__all__ = ["router"]

server/operator/failure_patterns.py

@@ -0,0 +1,44 @@
+"""GET /api/operator/failure-patterns — failure patterns view (TASK-08-03, D-053).
+
+Auth-gated. Returns failure pattern metrics: failure_mode frequency (cells
+with metric prefix `failure_mode:`) + branch outcome distribution (cells
+with metric prefix `branch:`). Weak-spot rubric criteria (mean < 3.0) are
+highlighted by the frontend. All k-anonymized.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+
+from fastapi import APIRouter, Depends, Request
+
+from server.auth.dependencies import current_operator
+from server.auth.models import Operator
+from server.operator._common import (
+    ViewResponse,
+    all_recent_aggregates,
+    group_by_path,
+    require_pg_store,
+)
+
+router = APIRouter(prefix="/api/operator", tags=["operator-failure-patterns"])
+
+
+def _is_failure_metric(metric: str) -> bool:
+    return metric.startswith("failure_mode:") or metric.startswith("branch:")
+
+
+@router.get("/failure-patterns", response_model=ViewResponse)
+async def failure_patterns_view(
+    request: Request,
+    op: Operator = Depends(current_operator),
+) -> ViewResponse:
+    pg_store = await require_pg_store(request)
+    since = _dt.date.today() - _dt.timedelta(days=30)
+    rows = await all_recent_aggregates(pg_store, since)
+    failure_rows = [r for r in rows if _is_failure_metric(r.get("metric", ""))]
+    views, last_updated = group_by_path(failure_rows)
+    return ViewResponse(views=views, last_updated=last_updated)
+
+
+__all__ = ["router"]

server/operator/mastery.py

@@ -0,0 +1,45 @@
+"""GET /api/operator/mastery — mastery progression view (TASK-08-02, D-053).
+
+Auth-gated. Returns mastery progression metrics: gate_open_rate,
+median_mastery_score, rubric_criterion_means (cells with metric prefix
+`rubric_criterion_mean:`). All k-anonymized (suppressed if < 10).
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+
+from fastapi import APIRouter, Depends, Request
+
+from server.auth.dependencies import current_operator
+from server.auth.models import Operator
+from server.operator._common import (
+    ViewResponse,
+    all_recent_aggregates,
+    group_by_path,
+    require_pg_store,
+)
+
+router = APIRouter(prefix="/api/operator", tags=["operator-mastery"])
+
+MASTERY_METRICS = {"gate_open_rate", "median_mastery_score"}
+
+
+def _is_mastery_metric(metric: str) -> bool:
+    return metric in MASTERY_METRICS or metric.startswith("rubric_criterion_mean:")
+
+
+@router.get("/mastery", response_model=ViewResponse)
+async def mastery_view(
+    request: Request,
+    op: Operator = Depends(current_operator),
+) -> ViewResponse:
+    pg_store = await require_pg_store(request)
+    since = _dt.date.today() - _dt.timedelta(days=30)
+    rows = await all_recent_aggregates(pg_store, since)
+    mastery_rows = [r for r in rows if _is_mastery_metric(r.get("metric", ""))]
+    views, last_updated = group_by_path(mastery_rows)
+    return ViewResponse(views=views, last_updated=last_updated)
+
+
+__all__ = ["router"]

tests/test_operator_endpoints.py

@@ -0,0 +1,304 @@
+"""Operator API endpoint unit tests (TASK-08-05) — mocked PgStore.
+
+Covers: 401 without cookie, 200 with valid cookie, suppressed cells have
+value=null, last_updated is max(updated_at), credential revoke works, no
+per-learner data in responses (R-DASH-02).
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from starlette.middleware.sessions import SessionMiddleware
+
+from server.auth.models import Operator
+from server.auth.passwords import hash_password
+from server.auth.rate_limit import reset_login_rate_limit
+from server.auth.routes import router as auth_router
+from server.operator.cohort import router as cohort_router
+from server.operator.credentials import router as credentials_router
+from server.operator.failure_patterns import router as failure_router
+from server.operator.mastery import router as mastery_router
+
+
+@pytest.fixture(autouse=True)
+def _reset_limiter():
+    reset_login_rate_limit()
+    yield
+    reset_login_rate_limit()
+
+
+class _FakeRecord(dict):
+    pass
+
+
+def _mock_pg_store(aggregates=None, credentials=None):
+    store = MagicMock()
+    # Operator lookup for current_operator dependency.
+    store.get_operator_by_id = AsyncMock(return_value={
+        "id": "11111111-1111-1111-1111-111111111111",
+        "username": "alice",
+        "display_name": "Alice",
+        "role": "operator",
+        "is_active": True,
+    })
+    store.update_last_login = AsyncMock()
+    store.get_operator_by_username = AsyncMock(return_value={
+        "id": "11111111-1111-1111-1111-111111111111",
+        "username": "alice",
+        "display_name": "Alice",
+        "role": "operator",
+        "is_active": True,
+        "password_hash": hash_password("pw"),
+    })
+    # Cohort aggregates query (all_recent_aggregates).
+    aggregates = aggregates or []
+    conn = MagicMock()
+    conn.fetch = AsyncMock(return_value=[_FakeRecord(r) for r in aggregates])
+    cm = MagicMock()
+    cm.__aenter__ = AsyncMock(return_value=conn)
+    cm.__aexit__ = AsyncMock(return_value=None)
+    store.pool = MagicMock()
+    store.pool.acquire = MagicMock(return_value=cm)
+    # Credentials.
+    store.list_credentials = AsyncMock(return_value=credentials or [])
+    store.get_credential = AsyncMock(return_value=credentials[0] if credentials else None)
+    store.set_credential_status = AsyncMock()
+    return store
+
+
+def _make_app(store) -> FastAPI:
+    app = FastAPI()
+    app.state.pg_store = store
+    app.add_middleware(SessionMiddleware, secret_key="test-secret-1234567890abcdef")
+    app.include_router(auth_router)
+    app.include_router(cohort_router)
+    app.include_router(mastery_router)
+    app.include_router(failure_router)
+    app.include_router(credentials_router)
+    return app
+
+
+def _login(client) -> None:
+    r = client.post("/api/operator/login", json={"username": "alice", "password": "pw"})
+    assert r.status_code == 200, r.text
+
+
+# ── 401 without cookie ─────────────────────────────────────────────────────
+
+
+def test_cohort_401_without_cookie():
+    app = _make_app(_mock_pg_store())
+    with TestClient(app) as client:
+        r = client.get("/api/operator/cohort")
+        assert r.status_code == 401
+
+
+def test_mastery_401_without_cookie():
+    app = _make_app(_mock_pg_store())
+    with TestClient(app) as client:
+        r = client.get("/api/operator/mastery")
+        assert r.status_code == 401
+
+
+def test_failure_patterns_401_without_cookie():
+    app = _make_app(_mock_pg_store())
+    with TestClient(app) as client:
+        r = client.get("/api/operator/failure-patterns")
+        assert r.status_code == 401
+
+
+def test_credentials_401_without_cookie():
+    app = _make_app(_mock_pg_store())
+    with TestClient(app) as client:
+        r = client.get("/api/operator/credentials")
+        assert r.status_code == 401
+
+
+def test_revoke_401_without_cookie():
+    app = _make_app(_mock_pg_store())
+    with TestClient(app) as client:
+        r = client.post("/api/operator/credentials/abc/revoke")
+        assert r.status_code == 401
+
+
+# ── 200 with valid cookie ──────────────────────────────────────────────────
+
+
+def test_cohort_200_with_cookie():
+    now = _dt.datetime.now(_dt.timezone.utc)
+    agg = [
+        {"path": "customer_service", "metric": "sessions_count",
+         "window_start": _dt.date.today(), "window_end": _dt.date.today(),
+         "value": 12.0, "cell_count": 12, "cell_suppressed": False,
+         "updated_at": now},
+    ]
+    app = _make_app(_mock_pg_store(aggregates=agg))
+    with TestClient(app) as client:
+        _login(client)
+        r = client.get("/api/operator/cohort")
+        assert r.status_code == 200
+        body = r.json()
+        assert any(v["path"] == "customer_service" for v in body["views"])
+
+
+def test_mastery_200_with_cookie():
+    agg = [
+        {"path": "p", "metric": "gate_open_rate",
+         "window_start": _dt.date.today(), "window_end": _dt.date.today(),
+         "value": 0.5, "cell_count": 10, "cell_suppressed": False,
+         "updated_at": _dt.datetime.now(_dt.timezone.utc)},
+    ]
+    app = _make_app(_mock_pg_store(aggregates=agg))
+    with TestClient(app) as client:
+        _login(client)
+        r = client.get("/api/operator/mastery")
+        assert r.status_code == 200
+
+
+def test_failure_patterns_200_with_cookie():
+    agg = [
+        {"path": "p", "metric": "failure_mode:missed_apology",
+         "window_start": _dt.date.today(), "window_end": _dt.date.today(),
+         "value": 3.0, "cell_count": 10, "cell_suppressed": False,
+         "updated_at": _dt.datetime.now(_dt.timezone.utc)},
+    ]
+    app = _make_app(_mock_pg_store(aggregates=agg))
+    with TestClient(app) as client:
+        _login(client)
+        r = client.get("/api/operator/failure-patterns")
+        assert r.status_code == 200
+
+
+def test_credentials_200_with_cookie():
+    cred = {
+        "id": "11111111-1111-1111-1111-111111111111",
+        "learner_ref": "learner-1",
+        "vc_type": "MasteryCredential",
+        "status": "active",
+        "issued_at": _dt.datetime.now(_dt.timezone.utc),
+        "revoked_at": None,
+    }
+    app = _make_app(_mock_pg_store(credentials=[cred]))
+    with TestClient(app) as client:
+        _login(client)
+        r = client.get("/api/operator/credentials")
+        assert r.status_code == 200
+        body = r.json()
+        assert len(body["credentials"]) == 1
+
+
+# ── Suppressed cells have value=null ───────────────────────────────────────
+
+
+def test_suppressed_cells_value_null():
+    agg = [
+        {"path": "p", "metric": "sessions_count",
+         "window_start": _dt.date.today(), "window_end": _dt.date.today(),
+         "value": None, "cell_count": 5, "cell_suppressed": True,
+         "updated_at": _dt.datetime.now(_dt.timezone.utc)},
+    ]
+    app = _make_app(_mock_pg_store(aggregates=agg))
+    with TestClient(app) as client:
+        _login(client)
+        r = client.get("/api/operator/cohort")
+        assert r.status_code == 200
+        cell = r.json()["views"][0]["metrics"][0]
+        assert cell["cell_suppressed"] is True
+        assert cell["value"] is None
+
+
+# ── last_updated is max(updated_at) ────────────────────────────────────────
+
+
+def test_last_updated_is_max():
+    t1 = _dt.datetime(2026, 8, 1, 12, 0, tzinfo=_dt.timezone.utc)
+    t2 = _dt.datetime(2026, 8, 3, 12, 0, tzinfo=_dt.timezone.utc)
+    agg = [
+        {"path": "p", "metric": "sessions_count",
+         "window_start": _dt.date.today(), "window_end": _dt.date.today(),
+         "value": 1.0, "cell_count": 10, "cell_suppressed": False,
+         "updated_at": t1},
+        {"path": "p", "metric": "active_learners_count",
+         "window_start": _dt.date.today(), "window_end": _dt.date.today(),
+         "value": 10.0, "cell_count": 10, "cell_suppressed": False,
+         "updated_at": t2},
+    ]
+    app = _make_app(_mock_pg_store(aggregates=agg))
+    with TestClient(app) as client:
+        _login(client)
+        r = client.get("/api/operator/cohort")
+        assert r.status_code == 200
+        assert r.json()["last_updated"] is not None
+
+
+# ── Credential revoke ──────────────────────────────────────────────────────
+
+
+def test_credential_revoke_sets_status_revoked():
+    cred = {
+        "id": "22222222-2222-2222-2222-222222222222",
+        "learner_ref": "learner-1",
+        "vc_type": "MasteryCredential",
+        "status": "active",
+        "issued_at": _dt.datetime.now(_dt.timezone.utc),
+        "revoked_at": None,
+    }
+    store = _mock_pg_store(credentials=[cred])
+    app = _make_app(store)
+    with TestClient(app) as client:
+        _login(client)
+        r = client.post("/api/operator/credentials/22222222-2222-2222-2222-222222222222/revoke")
+        assert r.status_code == 200
+        assert r.json()["status"] == "revoked"
+        store.set_credential_status.assert_awaited_once_with(
+            "22222222-2222-2222-2222-222222222222", "revoked",
+        )
+
+
+def test_credential_revoke_404_unknown():
+    store = _mock_pg_store(credentials=None)
+    store.get_credential = AsyncMock(return_value=None)
+    app = _make_app(store)
+    with TestClient(app) as client:
+        _login(client)
+        r = client.post("/api/operator/credentials/nonexistent/revoke")
+        assert r.status_code == 404
+
+
+# ── No per-learner data in cohort responses (R-DASH-02) ───────────────────
+
+
+def test_no_per_learner_data_in_cohort_response():
+    agg = [
+        {"path": "p", "metric": "sessions_count",
+         "window_start": _dt.date.today(), "window_end": _dt.date.today(),
+         "value": 10.0, "cell_count": 10, "cell_suppressed": False,
+         "updated_at": _dt.datetime.now(_dt.timezone.utc)},
+    ]
+    app = _make_app(_mock_pg_store(aggregates=agg))
+    with TestClient(app) as client:
+        _login(client)
+        r = client.get("/api/operator/cohort")
+        body_text = r.text
+        # No per-learner refs in the response (only path + metric + aggregates).
+        assert "learner-1" not in body_text
+        assert "learner_ref" not in body_text
+
+
+# ── 503 when no Postgres ───────────────────────────────────────────────────
+
+
+def test_cohort_503_no_postgres():
+    app = FastAPI()
+    app.state.pg_store = None
+    app.add_middleware(SessionMiddleware, secret_key="test-secret-1234567890abcdef")
+    app.include_router(auth_router)
+    app.include_router(cohort_router)
+    with TestClient(app) as client:
+        r = client.get("/api/operator/cohort")
+        assert r.status_code == 503