Add running class markbook API

Persist the paper's question-layout signals (A-level Section A/B, EITHER/OR choice
groups, marks confidence) onto exam_templates.extraction_meta so the setup UI can
show them. Data was carried by the contract but dropped at persistence.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
Claude-Session: https://claude.ai/code/session_01GruxHXxfdp4kZCgAMVFgvV

docker-compose.dev.yml

@@ -55,6 +55,8 @@ services:
       - CC_COMPOSE_SERVICE=backend-dev
       - RUN_INIT=false
       - INIT_MODE=infra
+      # P2: route exam auto-map through the spike's full recognition pipeline (extraction service)
+      - EXAM_EXTRACT_URL=${EXAM_EXTRACT_URL:-http://192.168.0.203:8899}
     ports:
       - "18000:8000"
     depends_on:

modules/services/exam_extract.py

@@ -0,0 +1,86 @@
+"""Client for the exam extraction SERVICE (docling-exam-spike, P1).
+
+The service runs the spike's FULL recognition pipeline (textlayer → question tree → OMR/figure/table
+sidecars → structure fusion → analyse) and returns the ghost-region contract the app already consumes.
+This module is a thin HTTP client: POST the paper, poll, return the `analyse` suggestions. The app's
+auto-map merges them (coordinate-adapted, id-remapped) in routers/exam/templates.py.
+
+Config: EXAM_EXTRACT_URL (e.g. http://192.168.0.203:8899). If unset, the app keeps its thin first-pass.
+"""
+from __future__ import annotations
+
+import base64
+import os
+import time
+from typing import Any, Dict, Optional
+
+import requests
+
+from modules.logger_tool import initialise_logger
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+
+
+class ExtractError(RuntimeError):
+    pass
+
+
+def service_url() -> Optional[str]:
+    url = os.getenv("EXAM_EXTRACT_URL")
+    return url.rstrip("/") if url else None
+
+
+def is_enabled() -> bool:
+    return bool(service_url())
+
+
+def _get(base: str, slug: str, timeout: int = 30) -> Dict[str, Any]:
+    r = requests.get(f"{base}/api/extract/{slug}", timeout=timeout)
+    r.raise_for_status()
+    return r.json()
+
+
+def get_replica(slug: str, timeout: int = 30) -> Dict[str, Any]:
+    """The digital-replica markdown for a paper (P4): {slug, title, n_questions, total_marks, markdown,
+    questions:[{label, marks, markdown}]}. Raises ExtractError if the paper has no replica yet."""
+    base = service_url()
+    if not base:
+        raise ExtractError("EXAM_EXTRACT_URL not configured")
+    r = requests.get(f"{base}/api/replica/{slug}", timeout=timeout)
+    if r.status_code == 404:
+        raise ExtractError(f"no digital replica for {slug}")
+    r.raise_for_status()
+    return r.json()
+
+
+def extract_suggestions(slug: str, pdf_bytes: bytes, *, force: bool = False,
+                        poll_timeout: int = 1500, poll_interval: int = 5) -> Dict[str, Any]:
+    """POST the paper to the service and poll until the analyse contract is ready.
+
+    Returns the full analyse payload: {status, coordinate_space:'page_fraction', margins,
+    suggestions:{questions[], response_areas[], boundaries[]}, meta}. Raises ExtractError on
+    failure/timeout. A cold paper is ~15 min (Docling per masked page); cached papers return instantly.
+    """
+    base = service_url()
+    if not base:
+        raise ExtractError("EXAM_EXTRACT_URL not configured")
+    payload = {"slug": slug, "pdf_b64": base64.b64encode(pdf_bytes).decode(), "force": force}
+    r = requests.post(f"{base}/api/extract", json=payload, timeout=120)
+    r.raise_for_status()
+    started = r.json()
+    if not started.get("ok", True):
+        raise ExtractError(started.get("error") or "service rejected the request")
+    # cached → fetch the contract straight away; otherwise poll the running job
+    deadline = time.time() + poll_timeout
+    while True:
+        d = _get(base, slug)
+        status = d.get("status")
+        if status == "complete":
+            if not (d.get("suggestions") or {}):
+                raise ExtractError("service returned complete with no suggestions")
+            return d
+        if status == "error":
+            raise ExtractError(f"extraction failed: {d.get('error')}")
+        if time.time() >= deadline:
+            raise ExtractError(f"extraction timed out after {poll_timeout}s (slug={slug})")
+        time.sleep(poll_interval)

routers/exam/__init__.py

@@ -9,10 +9,12 @@ from fastapi import APIRouter
 from routers.exam.templates import router as templates_router
 from routers.exam.batches import router as batches_router
 from routers.exam.bank import router as bank_router
+from routers.exam.corpus import router as corpus_router
 
 router = APIRouter()
 router.include_router(templates_router)
 router.include_router(batches_router)
 router.include_router(bank_router)
+router.include_router(corpus_router)
 
 __all__ = ["router"]

routers/exam/corpus.py

@@ -0,0 +1,98 @@
+"""Exam-bank corpus coverage (/api/exam/corpus) — the state of the collected exam bank.
+
+Read-only view over the seeded exam-board catalogue (eb_specifications + eb_exams): board → subject →
+specification → papers, with per-session QP/MS/ER coverage and rollup counts. Shows what the app has
+COLLECTED (question papers, mark schemes, examiner reports) so a teacher/admin can see the bank's state
+and where science coverage is complete vs thin. Catalogue data is public reference — read as-the-user.
+"""
+from __future__ import annotations
+
+import os
+import re
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends
+
+from modules.logger_tool import initialise_logger
+from routers.exam.dependencies import ExamContext, get_exam_context
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+
+router = APIRouter()
+
+DOC_TYPES = ("QP", "MS", "ER")
+
+
+def _rows(result: Any) -> List[Dict[str, Any]]:
+    data = getattr(result, "data", None)
+    if not data:
+        return []
+    return data if isinstance(data, list) else [data]
+
+
+def _award_level(spec: Dict[str, Any]) -> str:
+    """Best-effort GCSE / AS / A-level from the spec code (AQA GCSE = 8xxx, A-level/AS = 7xxx)."""
+    code = re.sub(r"\D", "", spec.get("award_code") or spec.get("spec_code") or "")
+    if code.startswith("8"):
+        return "GCSE"
+    if code.startswith("7"):
+        return "A-level"
+    return spec.get("award_level") or "Other"
+
+
+@router.get("/corpus")
+async def corpus_coverage(ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]:
+    specs = _rows(
+        ctx.supabase.table("eb_specifications")
+        .select("spec_code, exam_board_code, subject_code, award_code, first_teach").execute()
+    )
+    exams = _rows(
+        ctx.supabase.table("eb_exams")
+        .select("exam_code, spec_code, paper_code, tier, session, type_code, storage_loc").execute()
+    )
+
+    # group exam docs → per spec → per paper (paper_code + session) → which doc types are present
+    by_spec: Dict[str, Dict[str, Dict[str, Any]]] = {}
+    for e in exams:
+        sc = e.get("spec_code")
+        if not sc:
+            continue
+        key = f"{e.get('paper_code') or '?'}|{e.get('session') or '?'}"
+        paper = by_spec.setdefault(sc, {}).setdefault(key, {
+            "paper_code": e.get("paper_code"), "session": e.get("session"),
+            "tier": e.get("tier"), "docs": {}, "exam_codes": {},
+        })
+        dt = (e.get("type_code") or "").upper()
+        if dt in DOC_TYPES:
+            paper["docs"][dt] = bool(e.get("storage_loc"))
+            paper["exam_codes"][dt] = e.get("exam_code")
+
+    totals = {"specs": 0, "papers": 0, "sessions": 0, **{d: 0 for d in DOC_TYPES}}
+    boards: Dict[str, Dict[str, Any]] = {}
+    for s in specs:
+        sc = s["spec_code"]
+        papers_map = by_spec.get(sc, {})
+        if not papers_map:
+            continue
+        totals["specs"] += 1
+        board = s.get("exam_board_code") or "?"
+        level = _award_level(s)
+        papers = sorted(papers_map.values(), key=lambda p: (str(p["session"]), str(p["paper_code"])))
+        counts = {d: sum(1 for p in papers if p["docs"].get(d)) for d in DOC_TYPES}
+        for d in DOC_TYPES:
+            totals[d] += counts[d]
+        totals["papers"] += len(papers)
+        totals["sessions"] += len({p["session"] for p in papers})
+        spec_entry = {
+            "spec_code": sc, "subject": (s.get("subject_code") or "").title(), "level": level,
+            "board": board, "first_teach": s.get("first_teach"),
+            "n_papers": len(papers), "counts": counts, "papers": papers,
+        }
+        boards.setdefault(board, {"board": board, "specs": []})["specs"].append(spec_entry)
+
+    board_list = []
+    for board in sorted(boards):
+        specs_sorted = sorted(boards[board]["specs"], key=lambda x: (x["level"], x["subject"], x["spec_code"]))
+        board_list.append({"board": board, "n_specs": len(specs_sorted), "specs": specs_sorted})
+
+    return {"totals": totals, "boards": board_list}

routers/exam/schemas.py

@@ -80,6 +80,9 @@ class ResponseAreaPayload(BaseModel):
     ] = None
     # Optional Context differentiation (v1 generic; future graph/chart/data_table/diagram/code_block/passage).
     context_type: Optional[str] = None
+    # Rich recognition payload (75-exam-marker-region-meta.sql): figure name/description, OMR geometry, unit…
+    # Carried on canvas save so a named context figure survives a round-trip.
+    meta: Optional[Dict[str, Any]] = None
     source: Literal["manual", "ai"] = "manual"
     confirmed: bool = True
     confidence: Optional[float] = Field(default=None, ge=0, le=1)

routers/exam/templates.py

@@ -15,6 +15,7 @@ from __future__ import annotations
 import json
 import math
 import os
+import re
 import tempfile
 import time
 import uuid
@@ -30,6 +31,7 @@ from modules.database.services.exam_projection import project_template, project_
 from modules.database.supabase.utils.client import SupabaseServiceRoleClient
 from modules.database.supabase.utils.storage import StorageAdmin
 from modules.upload_validation import read_pdf_upload_bytes
+from modules.services import exam_extract
 from modules.logger_tool import initialise_logger
 from routers.exam.dependencies import ExamContext, get_exam_context, lookup_exam_code
 from routers.exam.schemas import (
@@ -455,6 +457,32 @@ def _y_to_canvas(y_value: float, page_number: int, pages: List[Dict[str, float]]
     return round(g["page_top"] + (g["page_pt_h"] - (float(y_value) - g["crop_y0"])) / g["page_pt_h"] * g["rendered_h"], 2)
 
 
+def _frac_box_to_canvas(bounds: Optional[Dict[str, Any]], page_number: int,
+                        pages: List[Dict[str, float]]) -> Optional[Dict[str, float]]:
+    """Page-fraction {x,y,w,h} (0..1 per page, from the extraction service) → 780-wide stacked canvas."""
+    if not bounds:
+        return None
+    g = _page_geom(pages, page_number)
+    try:
+        x, y, w, h = (float(bounds["x"]), float(bounds["y"]), float(bounds["w"]), float(bounds["h"]))
+    except (KeyError, TypeError, ValueError):
+        return None
+    return {
+        "x": round(x * g["rendered_w"], 2),
+        "y": round(g["page_top"] + y * g["rendered_h"], 2),
+        "w": round(w * g["rendered_w"], 2),
+        "h": round(h * g["rendered_h"], 2),
+    }
+
+
+def _frac_y_to_canvas(y_frac: Any, page_number: int, pages: List[Dict[str, float]]) -> Optional[float]:
+    g = _page_geom(pages, page_number)
+    try:
+        return round(g["page_top"] + float(y_frac) * g["rendered_h"], 2)
+    except (TypeError, ValueError):
+        return None
+
+
 def _ai_id(template_id: str, *parts: Any) -> str:
     return str(uuid.uuid5(uuid.NAMESPACE_URL, "/".join(["cc-auto-map", template_id, *[str(p) for p in parts]])))
 
@@ -660,6 +688,144 @@ def _run_auto_map_job(job_id: str, ctx: ExamContext, template_id: str, pdf_bytes
         _set_auto_map_status(job_id, {"status": "failed", "template_id": template_id, "error": str(exc)})
 
 
+_ALLOWED_RESPONSE_FORMS = {"lines", "answer-box", "working", "diagram", "tick-boxes", "table", "blanks"}
+_ALLOWED_ANSWER_TYPES = {"written", "mcq", "short", "diagram"}
+_ALLOWED_KINDS = {"response", "context", "question_number", "mark_area", "reference", "furniture"}
+_BOARD_RE = re.compile(r"^(aqa|edexcel|ocr|wjec|eduqas|ccea)-")
+
+
+def _extract_slug(ctx: ExamContext, template: Dict[str, Any]) -> str:
+    """A stable, board-prefixed slug for the extraction-service cache. Prefer the catalogue exam_code
+    (e.g. 'AQA-8463-1H-2022JUN-QP' → 'aqa-8463-1h-2022jun-qp' → board 'aqa' for the right margins);
+    fall back to a template-id slug (structure.py then defaults to AQA content-box margins)."""
+    code = None
+    exam_id = template.get("exam_id")
+    if exam_id:
+        try:
+            row = _first(ctx.supabase.table("eb_exams").select("exam_code").eq("id", exam_id).limit(1).execute())
+            code = (row or {}).get("exam_code")
+        except Exception as exc:
+            logger.info(f"extract slug: eb_exams lookup failed for {exam_id}: {exc}")
+    slug = re.sub(r"[^a-z0-9._-]+", "-", (code or "").lower()).strip("-")
+    if _BOARD_RE.match(slug):
+        return slug
+    return f"aqa-tmpl-{str(template.get('id') or '')[:12]}"
+
+
+def _map_service_contract_to_rows(template_id: str, contract: Dict[str, Any],
+                                  pdf_bytes: bytes) -> Dict[str, List[Dict[str, Any]]]:
+    """Map the extraction service's page-fraction analyse contract onto the app's canvas-space ghost rows.
+
+    Coordinates: page-fraction → the 780-wide stacked canvas. IDs: the service's deterministic uuid5s are
+    re-namespaced per template via _ai_id so two templates of the same paper don't collide and a re-map of
+    the same template re-emits stable ids (so _refresh_ai_rows preserves confirmed ghosts). FK-safe: parts
+    whose parent/owner question is absent are de-parented / dropped rather than crashing the insert.
+    """
+    pages = _pdf_page_geometry(pdf_bytes)
+    sug = contract.get("suggestions") or {}
+
+    def qid(uid: Any) -> str:
+        return _ai_id(template_id, "svc-q", uid)
+
+    questions: List[Dict[str, Any]] = []
+    q_ids: set = set()
+    for q in sug.get("questions") or []:
+        uid = q.get("uid")
+        if not uid:
+            continue
+        rid = qid(uid)
+        q_ids.add(rid)
+        at = q.get("answer_type") if q.get("answer_type") in _ALLOWED_ANSWER_TYPES else None
+        questions.append({
+            "id": rid, "template_id": template_id,
+            "parent_id": qid(q["parent_uid"]) if q.get("parent_uid") else None,
+            "label": q.get("label") or "?", "order": q.get("order", len(questions)),
+            "max_marks": _safe_marks(q.get("max_marks")), "answer_type": at,
+            "is_container": bool(q.get("is_container")),
+            "bounds": _frac_box_to_canvas(q.get("bounds"), q.get("page") or 1, pages),
+            "page": q.get("page"), "source": "ai", "confirmed": False,
+            "confidence": _safe_confidence(q.get("confidence")), "derivation": "extract-service",
+            # analyse contract v2 (migration 78): command verb + stem prose per part
+            "command_word": (q.get("command_word") or None) if not q.get("is_container") else None,
+            "preamble": q.get("preamble") or None,
+        })
+    for q in questions:                                 # FK safety: de-parent a dangling parent_id
+        if q["parent_id"] and q["parent_id"] not in q_ids:
+            q["parent_id"] = None
+
+    response_areas: List[Dict[str, Any]] = []
+    for ra in sug.get("response_areas") or []:
+        uid = ra.get("uid")
+        quid = qid(ra.get("question_uid") or "")
+        if not uid or quid not in q_ids:               # orphan region → drop (FK safety)
+            continue
+        bounds = _frac_box_to_canvas(ra.get("bounds"), ra.get("page") or 1, pages)
+        if not bounds:
+            continue
+        kind = ra.get("kind") if ra.get("kind") in _ALLOWED_KINDS else "response"
+        form = ra.get("response_form") if ra.get("response_form") in _ALLOWED_RESPONSE_FORMS else None
+        response_areas.append({
+            "id": _ai_id(template_id, "svc-ra", uid), "template_id": template_id,
+            "question_id": quid, "page": ra.get("page"), "bounds": bounds,
+            "kind": kind, "response_form": form if kind == "response" else None,
+            "context_type": ra.get("context_type"), "meta": ra.get("meta") or {},
+            "source": "ai", "confirmed": False,
+            "confidence": _safe_confidence(ra.get("confidence")), "derivation": "extract-service",
+        })
+
+    boundaries: List[Dict[str, Any]] = []
+    for i, b in enumerate(sug.get("boundaries") or []):
+        quid = qid(b.get("question_uid") or "")
+        if quid not in q_ids:
+            continue
+        page_index = b.get("page_index")
+        y = _frac_y_to_canvas(b.get("y"), (page_index or 0) + 1, pages)
+        if y is None:
+            continue
+        boundaries.append({
+            "id": _ai_id(template_id, "svc-b", b.get("question_uid") or i), "template_id": template_id,
+            "question_id": quid, "label": b.get("label") or "", "page_index": page_index,
+            "y": y, "bounds": None, "source": "ai", "confirmed": False,
+            "confidence": _safe_confidence(b.get("confidence")), "derivation": "extract-service",
+        })
+
+    return {"questions": questions, "response_areas": response_areas, "boundaries": boundaries}
+
+
+def _run_service_extract_merge(ctx: ExamContext, template_id: str, pdf_bytes: bytes, slug: str) -> Dict[str, List[Dict[str, Any]]]:
+    contract = exam_extract.extract_suggestions(slug, pdf_bytes)
+    rows = _map_service_contract_to_rows(template_id, contract, pdf_bytes)
+    _refresh_ai_rows(ctx, template_id, rows)
+    meta = contract.get("meta") or {}
+    # P3: record provenance + the audit cover-reconciliation gate so the setup UI can flag under-reads,
+    # and store the slug so the digital-text view can resolve this paper's replica.
+    updates: Dict[str, Any] = {"extraction_meta": {
+        "engine": "extract-service", "slug": slug, "audit": meta.get("audit") or {},
+        "counts": {k: len(v) for k, v in rows.items()},
+        # question-layout signals for the setup UI: paper sections (e.g. A-level Section A/B) + EITHER/OR choices
+        "sections": meta.get("sections") or [],
+        "choice_groups": meta.get("choice_groups") or [],
+        "marks_confidence": meta.get("marks_confidence"),
+    }}
+    n_pages = meta.get("n_pages") or meta.get("pages")
+    if n_pages:
+        updates["page_count"] = n_pages
+    ctx.supabase.table("exam_templates").update(updates).eq("id", template_id).execute()
+    return rows
+
+
+def _run_service_extract_job(job_id: str, ctx: ExamContext, template_id: str, pdf_bytes: bytes, slug: str) -> None:
+    _set_auto_map_status(job_id, {"status": "running", "template_id": template_id, "engine": "extract-service", "slug": slug})
+    try:
+        rows = _run_service_extract_merge(ctx, template_id, pdf_bytes, slug)
+        project_template_safe(template_id)
+        _set_auto_map_status(job_id, {"status": "completed", "template_id": template_id,
+                                      "engine": "extract-service", "counts": {k: len(v) for k, v in rows.items()}})
+    except Exception as exc:
+        logger.exception(f"extract-service job failed for template {template_id}: {exc}")
+        _set_auto_map_status(job_id, {"status": "failed", "template_id": template_id, "engine": "extract-service", "error": str(exc)})
+
+
 # ─── templates ───────────────────────────────────────────────────────────────
 
 
@@ -837,6 +1003,14 @@ async def auto_map_template(
         raise HTTPException(status_code=409, detail="Template has recorded marks; auto-map structural refresh is blocked.")
     bucket, path, pdf_bytes = _resolve_template_source(ctx, template)
     source_label = f"{bucket}/{path}"
+    # Extraction-service path (P2): when EXAM_EXTRACT_URL is set, route auto-map through the spike's full
+    # recognition pipeline instead of the thin first-pass. Always async — a cold paper is ~15 min.
+    if exam_extract.is_enabled():
+        slug = _extract_slug(ctx, template)
+        job_id = str(uuid.uuid4())
+        _set_auto_map_status(job_id, {"status": "queued", "template_id": template_id, "engine": "extract-service", "slug": slug})
+        background_tasks.add_task(_run_service_extract_job, job_id, ctx, template_id, pdf_bytes, slug)
+        return JSONResponse(status_code=202, content={"status": "accepted", "job_id": job_id, "engine": "extract-service"})
     try:
         fast_path = _pdf_has_text_layer(pdf_bytes)
     except Exception as exc:
@@ -875,6 +1049,24 @@ async def auto_map_status(
     return body
 
 
+@router.get("/templates/{template_id}/digital-text")
+async def template_digital_text(
+    template_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    """P4: the digital-replica markdown for this template's paper (stem text, parts, marks, answer-space
+    placeholders, spec refs). Resolves the paper via the slug the extraction used."""
+    template = _fetch_template_or_404(ctx, template_id)
+    _require_source_visibility_or_404(ctx, template)
+    if not exam_extract.is_enabled():
+        raise HTTPException(status_code=503, detail="Extraction service not configured")
+    slug = (template.get("extraction_meta") or {}).get("slug") or _extract_slug(ctx, template)
+    try:
+        return exam_extract.get_replica(slug)
+    except exam_extract.ExtractError as exc:
+        raise HTTPException(status_code=404, detail=f"No digital text yet — run auto-map first ({exc})")
+
+
 @router.put("/templates/{template_id}")
 async def replace_template(
     template_id: str,
@@ -955,6 +1147,7 @@ async def replace_template(
                 "kind": ra.kind,
                 "response_form": ra.response_form,
                 "context_type": ra.context_type,   # 73: optional Context differentiation
+                "meta": ra.meta,                   # 75: rich recognition payload (name/description/OMR/…)
                 "source": ra.source,
                 "confirmed": ra.confirmed,
                 "confidence": ra.confidence,

routers/markbook.py

@@ -0,0 +1,256 @@
+"""Running class markbook / gradebook API (/api/markbook).
+
+A markbook is a term-long teacher-editable ledger: roster rows × arbitrary assessment
+columns. It deliberately does not depend on exam-marker batches. All user-facing reads and
+writes use the as-user Supabase client so class/markbook RLS gates are enforced.
+"""
+from __future__ import annotations
+
+import csv
+import io
+import os
+from datetime import date as Date
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import Response
+from pydantic import BaseModel, Field
+
+from modules.logger_tool import initialise_logger
+from routers.exam.dependencies import ExamContext, get_exam_context, resolve_student_names
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+router = APIRouter()
+
+
+class CreateAssessmentRequest(BaseModel):
+    title: str = Field(..., min_length=1, max_length=160)
+    date: Optional[Date] = None
+    max_marks: float = Field(default=100, gt=0)
+
+
+class MarkUpsertRequest(BaseModel):
+    mark: Optional[float] = Field(default=None, ge=0)
+
+
+def _rows(result: Any) -> List[Dict[str, Any]]:
+    data = getattr(result, "data", None)
+    if not data:
+        return []
+    return data if isinstance(data, list) else [data]
+
+
+def _first(result: Any) -> Optional[Dict[str, Any]]:
+    rows = _rows(result)
+    return rows[0] if rows else None
+
+
+def _round(value: Optional[float]) -> Optional[float]:
+    return None if value is None else round(float(value), 1)
+
+
+def _fetch_class_or_404(ctx: ExamContext, class_id: str) -> Dict[str, Any]:
+    row = _first(ctx.supabase.table("classes").select("id, name, institute_id").eq("id", class_id).limit(1).execute())
+    if not row:
+        raise HTTPException(status_code=404, detail="Class not found")
+    return row
+
+
+def _active_roster(ctx: ExamContext, class_id: str) -> List[Dict[str, Any]]:
+    roster = _rows(
+        ctx.supabase.table("class_students")
+        .select("student_id, status, enrolled_at")
+        .eq("class_id", class_id)
+        .eq("status", "active")
+        .execute()
+    )
+    names = resolve_student_names([r["student_id"] for r in roster if r.get("student_id")])
+    return [
+        {
+            "student_id": r["student_id"],
+            "student_name": names.get(r["student_id"]) or r["student_id"],
+            "status": r.get("status"),
+            "enrolled_at": r.get("enrolled_at"),
+        }
+        for r in roster
+        if r.get("student_id")
+    ]
+
+
+def _assessments(ctx: ExamContext, class_id: str) -> List[Dict[str, Any]]:
+    return _rows(
+        ctx.supabase.table("class_assessments")
+        .select("id, class_id, tenant_id, title, date, max_marks, created_at, updated_at")
+        .eq("class_id", class_id)
+        .order("date")
+        .order("created_at")
+        .execute()
+    )
+
+
+def _marks(ctx: ExamContext, assessment_ids: List[str]) -> List[Dict[str, Any]]:
+    if not assessment_ids:
+        return []
+    return _rows(
+        ctx.supabase.table("assessment_marks")
+        .select("assessment_id, student_id, mark, updated_at, updated_by")
+        .in_("assessment_id", assessment_ids)
+        .execute()
+    )
+
+
+def _assemble_grid(ctx: ExamContext, class_id: str) -> Dict[str, Any]:
+    cls = _fetch_class_or_404(ctx, class_id)
+    roster = _active_roster(ctx, class_id)
+    assessments = _assessments(ctx, class_id)
+    assessment_ids = [a["id"] for a in assessments]
+    marks = _marks(ctx, assessment_ids)
+
+    marks_by_student: Dict[str, Dict[str, Optional[float]]] = {r["student_id"]: {} for r in roster}
+    for m in marks:
+        sid = m.get("student_id")
+        aid = m.get("assessment_id")
+        if isinstance(sid, str) and isinstance(aid, str) and sid in marks_by_student and aid in assessment_ids:
+            marks_by_student[sid][aid] = m.get("mark")
+
+    max_total = sum(float(a.get("max_marks") or 0) for a in assessments)
+    students = []
+    all_entered: List[float] = []
+    for idx, student in enumerate(roster, start=1):
+        entered = [
+            float(v)
+            for v in (marks_by_student.get(student["student_id"], {}).get(aid) for aid in assessment_ids)
+            if v is not None
+        ]
+        total = sum(entered) if entered else None
+        students.append(
+            {
+                **student,
+                "row_number": idx,
+                "marks": {aid: marks_by_student.get(student["student_id"], {}).get(aid) for aid in assessment_ids},
+                "total": total,
+                "percentage": _round((total / max_total) * 100) if total is not None and max_total > 0 else None,
+            }
+        )
+        all_entered.extend(entered)
+
+    assessment_summaries = []
+    for a in assessments:
+        vals = []
+        for s in roster:
+            maybe_mark = marks_by_student.get(s["student_id"], {}).get(a["id"])
+            if maybe_mark is not None:
+                vals.append(float(maybe_mark))
+        max_marks = float(a.get("max_marks") or 0)
+        assessment_summaries.append(
+            {
+                "assessment_id": a["id"],
+                "entered_count": len(vals),
+                "average_mark": _round(sum(vals) / len(vals)) if vals else None,
+                "average_percentage": _round((sum(vals) / len(vals) / max_marks) * 100) if vals and max_marks > 0 else None,
+            }
+        )
+
+    return {
+        "class": cls,
+        "students": students,
+        "assessments": assessments,
+        "assessment_summaries": assessment_summaries,
+        "summary": {
+            "student_count": len(roster),
+            "assessment_count": len(assessments),
+            "entered_mark_count": len(all_entered),
+            "class_average_mark": _round(sum(all_entered) / len(all_entered)) if all_entered else None,
+        },
+    }
+
+
+@router.get("/classes/{class_id}/assessments")
+async def list_assessments(class_id: str, ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]:
+    _fetch_class_or_404(ctx, class_id)
+    return {"assessments": _assessments(ctx, class_id)}
+
+
+@router.post("/classes/{class_id}/assessments")
+async def create_assessment(class_id: str, body: CreateAssessmentRequest, ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]:
+    cls = _fetch_class_or_404(ctx, class_id)
+    row = {
+        "class_id": class_id,
+        "tenant_id": cls["institute_id"],
+        "title": body.title.strip(),
+        "date": body.date.isoformat() if body.date else None,
+        "max_marks": body.max_marks,
+    }
+    created = _first(ctx.supabase.table("class_assessments").insert(row).execute())
+    if not created:
+        raise HTTPException(status_code=500, detail="Failed to create assessment")
+    logger.info(f"Markbook assessment {created.get('id')} created for class {class_id} by {ctx.user_id}")
+    return created
+
+
+@router.get("/classes/{class_id}/grid")
+async def get_grid(class_id: str, ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]:
+    return _assemble_grid(ctx, class_id)
+
+
+@router.put("/classes/{class_id}/assessments/{assessment_id}/marks/{student_id}")
+async def upsert_mark(
+    class_id: str,
+    assessment_id: str,
+    student_id: str,
+    body: MarkUpsertRequest,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    cls = _fetch_class_or_404(ctx, class_id)
+    assessment = _first(
+        ctx.supabase.table("class_assessments")
+        .select("id, class_id, tenant_id, max_marks")
+        .eq("id", assessment_id)
+        .eq("class_id", class_id)
+        .limit(1)
+        .execute()
+    )
+    if not assessment:
+        raise HTTPException(status_code=404, detail="Assessment not found")
+    roster_row = _first(
+        ctx.supabase.table("class_students")
+        .select("student_id")
+        .eq("class_id", class_id)
+        .eq("student_id", student_id)
+        .eq("status", "active")
+        .limit(1)
+        .execute()
+    )
+    if not roster_row:
+        raise HTTPException(status_code=404, detail="Student is not active in this class")
+    if body.mark is not None and body.mark > float(assessment.get("max_marks") or 0):
+        raise HTTPException(status_code=422, detail="mark exceeds assessment max_marks")
+
+    row = {
+        "assessment_id": assessment_id,
+        "student_id": student_id,
+        "tenant_id": assessment.get("tenant_id") or cls["institute_id"],
+        "mark": body.mark,
+        "updated_by": ctx.user_id,
+    }
+    upserted = _first(ctx.supabase.table("assessment_marks").upsert(row, on_conflict="assessment_id,student_id").execute())
+    if not upserted:
+        raise HTTPException(status_code=500, detail="Failed to upsert mark")
+    return {"status": "ok", "mark": upserted}
+
+
+@router.get("/classes/{class_id}/csv")
+async def export_csv(class_id: str, ctx: ExamContext = Depends(get_exam_context)) -> Response:
+    data = _assemble_grid(ctx, class_id)
+    assessments = data["assessments"]
+    buf = io.StringIO()
+    writer = csv.writer(buf)
+    writer.writerow(["row", "student_name", "student_id"] + [a["title"] for a in assessments] + ["total", "percentage"])
+    for student in data["students"]:
+        writer.writerow(
+            [student["row_number"], student.get("student_name") or "", student.get("student_id") or ""]
+            + ["" if student["marks"].get(a["id"]) is None else student["marks"].get(a["id"]) for a in assessments]
+            + ["" if student["total"] is None else student["total"], "" if student["percentage"] is None else student["percentage"]]
+        )
+    filename = f"markbook-{class_id}.csv"
+    return Response(content=buf.getvalue(), media_type="text/csv", headers={"Content-Disposition": f'attachment; filename="{filename}"'})

run/routers.py

@@ -41,6 +41,7 @@ from routers.transcribe.keywords import router as keywords_router
 from routers.me.bootstrap_router import router as me_bootstrap_router
 from routers import tlsync_token as tlsync_token_router
 from routers.exam import router as exam_router
+from routers.markbook import router as markbook_router
 
 def register_routes(app: FastAPI):
     logger.info("Starting to register routes...")
@@ -138,6 +139,9 @@ def register_routes(app: FastAPI):
     # Exam-marker Routes (as-user Supabase, RLS-enforced; spec §4)
     app.include_router(exam_router, prefix="/api/exam", tags=["Exam"])
 
+    # Running markbook / gradebook Routes (as-user Supabase, RLS-enforced)
+    app.include_router(markbook_router, prefix="/api/markbook", tags=["Markbook"])
+
     # Transcription Routes (CIS Phase 1)
     app.include_router(sessions_router, prefix="/transcribe", tags=["Transcription Sessions"])
     app.include_router(canvas_events_router, prefix="/transcribe", tags=["Transcription Canvas Events"])

tests/test_exam_corpus.py

@@ -0,0 +1,48 @@
+"""Test the exam-bank corpus coverage grouping (/api/exam/corpus)."""
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from routers.exam.corpus import router
+from routers.exam.dependencies import ExamContext, get_exam_context
+
+TEACHER = "00000000-0000-0000-0000-000000000001"
+INST_A = "10000000-0000-0000-0000-000000000001"
+
+
+class FakeResult:
+    def __init__(self, data): self.data = data
+
+
+class FakeQuery:
+    def __init__(self, store, table): self.store, self.table = store, table
+    def select(self, *_a, **_k): return self
+    def execute(self): return FakeResult(list(self.store.get(self.table, [])))
+
+
+class FakeSupabase:
+    def __init__(self, store): self.store = store
+    def table(self, name): return FakeQuery(self.store, name)
+
+
+def _client(store):
+    app = FastAPI(); app.include_router(router, prefix="/api/exam")
+    app.dependency_overrides[get_exam_context] = lambda: ExamContext(TEACHER, "tok", FakeSupabase(store), [INST_A])
+    return TestClient(app)
+
+
+def test_corpus_groups_papers_by_session_with_qp_ms_er_coverage():
+    store = {
+        "eb_specifications": [{"spec_code": "AQA-PHYS-8463", "exam_board_code": "AQA", "subject_code": "PHYSICS", "award_code": "8463", "first_teach": "2016"}],
+        "eb_exams": [
+            {"exam_code": "a", "spec_code": "AQA-PHYS-8463", "paper_code": "8463/1H", "session": "2022-Jun", "type_code": "QP", "storage_loc": "cc.examboards/a.pdf"},
+            {"exam_code": "b", "spec_code": "AQA-PHYS-8463", "paper_code": "8463/1H", "session": "2022-Jun", "type_code": "MS", "storage_loc": "cc.examboards/b.pdf"},
+            {"exam_code": "c", "spec_code": "AQA-PHYS-8463", "paper_code": "8463/1H", "session": "2022-Jun", "type_code": "ER", "storage_loc": None},
+            {"exam_code": "d", "spec_code": "AQA-PHYS-8463", "paper_code": "8463/2H", "session": "2023-Jun", "type_code": "QP", "storage_loc": "cc.examboards/d.pdf"},
+        ],
+    }
+    body = _client(store).get("/api/exam/corpus").json()
+    assert body["totals"]["specs"] == 1 and body["totals"]["papers"] == 2
+    spec = body["boards"][0]["specs"][0]
+    assert spec["level"] == "GCSE" and spec["subject"] == "Physics"
+    assert spec["counts"] == {"QP": 2, "MS": 1, "ER": 0}  # ER present but not stored → not counted
+    p = next(p for p in spec["papers"] if p["paper_code"] == "8463/1H")
+    assert p["docs"] == {"QP": True, "MS": True, "ER": False}

tests/test_exam_extract_mapping.py

@@ -0,0 +1,75 @@
+"""Test the extraction-service contract → app ghost-row mapping (P2).
+
+Mocks the PDF geometry so the mapper is tested in isolation: page-fraction → 780-canvas conversion,
+per-template uuid remap, FK-safety (orphan regions / dangling parents), and rich meta passthrough.
+"""
+import routers.exam.templates as T
+
+TID = "11111111-1111-1111-1111-111111111111"
+
+# two pages, each rendered 780 wide × 1000 tall; page 2 stacked below page 1
+_GEOM = [
+    {"rendered_w": 780.0, "rendered_h": 1000.0, "page_top": 0.0, "page_pt_w": 595.0, "page_pt_h": 842.0, "crop_x0": 0.0, "crop_y0": 0.0},
+    {"rendered_w": 780.0, "rendered_h": 1000.0, "page_top": 1000.0, "page_pt_w": 595.0, "page_pt_h": 842.0, "crop_x0": 0.0, "crop_y0": 0.0},
+]
+
+_CONTRACT = {
+    "coordinate_space": "page_fraction",
+    "suggestions": {
+        "questions": [
+            {"uid": "Q1", "label": "1", "order": 0, "max_marks": 5, "is_container": True, "page": 1},
+            {"uid": "Q1a", "parent_uid": "Q1", "label": "1(a)", "order": 1, "max_marks": 3,
+             "answer_type": "short", "is_container": False, "page": 1},
+            {"uid": "Q1b", "parent_uid": "Q1", "label": "1(b)", "order": 2, "max_marks": 2,
+             "answer_type": "mcq", "is_container": False, "page": 2},
+        ],
+        "response_areas": [
+            {"uid": "RA1", "question_uid": "Q1a", "page": 1, "kind": "response", "response_form": "answer-box",
+             "confidence": 0.9, "bounds": {"x": 0.1, "y": 0.2, "w": 0.5, "h": 0.05},
+             "meta": {"unit": "m/s", "quantity": "v", "n_lines": 1}},
+            {"uid": "RA2", "question_uid": "Q1b", "page": 2, "kind": "response", "response_form": "tick-boxes",
+             "bounds": {"x": 0.2, "y": 0.3, "w": 0.4, "h": 0.2},
+             "meta": {"select_n": 2, "n_options": 5, "boxes": [{"x0": 0.2, "y0": 0.3, "fill": 0.8}]}},
+            {"uid": "CTX1", "question_uid": "Q1b", "page": 2, "kind": "context", "response_form": None,
+             "bounds": {"x": 0.1, "y": 0.1, "w": 0.6, "h": 0.15}},
+            {"uid": "ORPH", "question_uid": "GHOST", "page": 1, "kind": "response", "response_form": "lines",
+             "bounds": {"x": 0.1, "y": 0.5, "w": 0.5, "h": 0.1}},   # orphan owner → must be dropped
+        ],
+        "boundaries": [
+            {"question_uid": "Q1", "label": "1", "page_index": 0, "y": 0.18, "confidence": 0.85},
+        ],
+    },
+    "meta": {"n_pages": 2},
+}
+
+
+def test_contract_maps_to_canvas_rows(monkeypatch):
+    monkeypatch.setattr(T, "_pdf_page_geometry", lambda _b: _GEOM)
+    rows = T._map_service_contract_to_rows(TID, _CONTRACT, b"%PDF-1.4")
+
+    q = rows["questions"]
+    assert len(q) == 3
+    container = next(x for x in q if x["label"] == "1")
+    a = next(x for x in q if x["label"] == "1(a)")
+    assert container["is_container"] is True and container["parent_id"] is None
+    assert a["parent_id"] == container["id"]          # parent remapped to the container's per-template id
+    assert a["answer_type"] == "short" and a["max_marks"] == 3
+    assert all(x["source"] == "ai" and x["confirmed"] is False for x in q)
+
+    ra = rows["response_areas"]
+    assert len(ra) == 3                                # orphan (owner GHOST) dropped
+    ids = {r["id"] for r in ra}
+    assert len(ids) == 3                               # ids unique
+    box = next(r for r in ra if r["response_form"] == "answer-box")
+    # page-fraction → 780-canvas: x=0.1*780=78, y=0.2*1000=200, w=0.5*780=390, h=0.05*1000=50
+    assert box["bounds"] == {"x": 78.0, "y": 200.0, "w": 390.0, "h": 50.0}
+    assert box["meta"]["unit"] == "m/s"               # rich meta preserved
+    tick = next(r for r in ra if r["response_form"] == "tick-boxes")
+    # page 2 y stacked: 0.3*1000 + page_top(1000) = 1300
+    assert tick["bounds"]["y"] == 1300.0 and tick["meta"]["select_n"] == 2
+    ctx = next(r for r in ra if r["kind"] == "context")
+    assert ctx["response_form"] is None               # context carries no answer form
+    assert all(r["question_id"] in {x["id"] for x in q} for r in ra)   # FK-safe
+
+    b = rows["boundaries"]
+    assert len(b) == 1 and b[0]["y"] == 180.0 and b[0]["question_id"] == container["id"]

tests/test_markbook.py

@@ -0,0 +1,167 @@
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+import routers.markbook as markbook_mod
+from routers.exam.dependencies import ExamContext
+from routers.markbook import router
+
+TEACHER = "00000000-0000-0000-0000-000000000001"
+INST = "10000000-0000-0000-0000-000000000001"
+CLASS = "c-1"
+
+
+class FakeResult:
+    def __init__(self, data):
+        self.data = data
+
+
+class FakeQuery:
+    def __init__(self, store, table):
+        self.store = store
+        self.table = table
+        self.rows = list(store.get(table, []))
+        self._filters = []
+        self._op = None
+        self._payload = None
+        self._limit = None
+
+    def select(self, *_a, **_k):
+        self._op = "select"; return self
+
+    def insert(self, payload):
+        self._op = "insert"; self._payload = payload; return self
+
+    def upsert(self, payload, **_kwargs):
+        self._op = "upsert"; self._payload = payload; return self
+
+    def eq(self, k, v):
+        self._filters.append(("eq", k, v)); self.rows = [r for r in self.rows if r.get(k) == v]; return self
+
+    def in_(self, k, vals):
+        vals = set(vals); self._filters.append(("in", k, vals)); self.rows = [r for r in self.rows if r.get(k) in vals]; return self
+
+    def order(self, *_a, **_k):
+        return self
+
+    def limit(self, n):
+        self._limit = n; return self
+
+    def _match(self, row):
+        for op, k, v in self._filters:
+            if op == "eq" and row.get(k) != v:
+                return False
+            if op == "in" and row.get(k) not in v:
+                return False
+        return True
+
+    def execute(self):
+        backing = self.store.setdefault(self.table, [])
+        if self._op in ("insert", "upsert"):
+            payloads = self._payload if isinstance(self._payload, list) else [self._payload]
+            out = []
+            for p in payloads:
+                row = dict(p)
+                if self._op == "upsert" and self.table == "assessment_marks":
+                    existing = next((r for r in backing if r.get("assessment_id") == row.get("assessment_id") and r.get("student_id") == row.get("student_id")), None)
+                    if existing:
+                        existing.update(row); out.append(existing); continue
+                if self._op == "upsert" and row.get("id") is not None:
+                    existing = next((r for r in backing if r.get("id") == row["id"]), None)
+                    if existing:
+                        existing.update(row); out.append(existing); continue
+                row.setdefault("id", f"gen-{self.table}-{len(backing)}")
+                backing.append(row); out.append(row)
+            return FakeResult(out)
+        rows = self.rows[: self._limit] if self._limit is not None else self.rows
+        return FakeResult(rows)
+
+
+class FakeSupabase:
+    def __init__(self, store):
+        self.store = store
+
+    def table(self, name):
+        return FakeQuery(self.store, name)
+
+
+def make_client(store):
+    app = FastAPI()
+    app.include_router(router, prefix="/api/markbook")
+    from routers.exam.dependencies import get_exam_context
+    app.dependency_overrides[get_exam_context] = lambda: ExamContext(TEACHER, "tok", FakeSupabase(store), [INST])
+    return TestClient(app)
+
+
+def base_store(**extra):
+    store = {
+        "classes": [{"id": CLASS, "name": "10A Maths", "institute_id": INST}],
+        "class_students": [
+            {"class_id": CLASS, "student_id": "s1", "status": "active"},
+            {"class_id": CLASS, "student_id": "s2", "status": "active"},
+            {"class_id": CLASS, "student_id": "s3", "status": "inactive"},
+        ],
+    }
+    store.update(extra)
+    return store
+
+
+@pytest.fixture(autouse=True)
+def names(monkeypatch):
+    monkeypatch.setattr(markbook_mod, "resolve_student_names", lambda ids: {sid: {"s1": "Alice", "s2": "Bob"}.get(sid, sid) for sid in ids})
+
+
+def test_create_assessment_sets_class_and_tenant():
+    store = base_store()
+    c = make_client(store)
+    r = c.post(f"/api/markbook/classes/{CLASS}/assessments", json={"title": "Homework 1", "date": "2026-09-10", "max_marks": 20})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["class_id"] == CLASS
+    assert body["tenant_id"] == INST
+    assert body["title"] == "Homework 1"
+    assert body["max_marks"] == 20
+
+
+def test_grid_reuses_active_roster_and_computes_summaries():
+    store = base_store(
+        class_assessments=[
+            {"id": "a1", "class_id": CLASS, "tenant_id": INST, "title": "HW", "date": "2026-09-01", "max_marks": 10},
+            {"id": "a2", "class_id": CLASS, "tenant_id": INST, "title": "Quiz", "date": "2026-09-08", "max_marks": 20},
+        ],
+        assessment_marks=[
+            {"assessment_id": "a1", "student_id": "s1", "mark": 8},
+            {"assessment_id": "a2", "student_id": "s1", "mark": 18},
+            {"assessment_id": "a1", "student_id": "s2", "mark": 6},
+        ],
+    )
+    body = make_client(store).get(f"/api/markbook/classes/{CLASS}/grid").json()
+    assert [s["student_name"] for s in body["students"]] == ["Alice", "Bob"]
+    alice = body["students"][0]
+    assert alice["marks"] == {"a1": 8, "a2": 18}
+    assert alice["total"] == 26
+    assert alice["percentage"] == 86.7
+    assert body["assessment_summaries"][0]["average_mark"] == 7.0
+    assert body["summary"]["entered_mark_count"] == 3
+
+
+def test_mark_upsert_rejects_over_max_and_inactive_students():
+    store = base_store(class_assessments=[{"id": "a1", "class_id": CLASS, "tenant_id": INST, "title": "HW", "max_marks": 10}])
+    c = make_client(store)
+    assert c.put(f"/api/markbook/classes/{CLASS}/assessments/a1/marks/s1", json={"mark": 11}).status_code == 422
+    assert c.put(f"/api/markbook/classes/{CLASS}/assessments/a1/marks/s3", json={"mark": 5}).status_code == 404
+    ok = c.put(f"/api/markbook/classes/{CLASS}/assessments/a1/marks/s1", json={"mark": 9})
+    assert ok.status_code == 200
+    assert ok.json()["mark"]["mark"] == 9
+
+
+def test_csv_export_has_roster_rows_and_assessment_columns():
+    store = base_store(
+        class_assessments=[{"id": "a1", "class_id": CLASS, "tenant_id": INST, "title": "HW", "max_marks": 10}],
+        assessment_marks=[{"assessment_id": "a1", "student_id": "s1", "mark": 8}],
+    )
+    text = make_client(store).get(f"/api/markbook/classes/{CLASS}/csv").text
+    lines = text.strip().splitlines()
+    assert lines[0] == "row,student_name,student_id,HW,total,percentage"
+    assert lines[1].startswith("1,Alice,s1,8,8")
+    assert lines[2].startswith("2,Bob,s2,,,")