feat(seed): expand corpus to 1178 papers + download-only/unseed/granular reset

PRIMARY — corpus breadth (505->1178 papers, 18->60 specs, all URLs HEAD-verified):
- AQA (enumerated): Maths, English Lang/Lit, Geography, Computer Science, Business,
  Psychology, MFL (French/Spanish/German), GCSE + A-level, on top of round-1 sciences.
- Edexcel + OCR (confirmed direct URLs via research): Maths, English, Geography, History,
  Business, Computer Science, GCSE + A-level.
- generate_corpus_manifest.py: _subj/_mfl AQA builders, Edexcel/OCR spec+URL tables,
  derived exam_code (_mk_exam_code) matching the locked convention, concurrent re-verify.
Verified on dev .94: eb_specifications=60, eb_exams=1178, QP=469, doc_type all 'pdf',
seed idempotent (uploaded=673 new, skipped=505), failed=0.

SECONDARY:
- --download-only + persistent bucket-shaped local store (manifests/_corpus_store/, gitignored):
  download-once, seed-many, offline-repeatable; --store-dir/--no-store. (_store_path/_item_bytes/
  download_corpus). Verified: store populated, seed reads offline (download_cached).
- --unseed [--board/--spec]: inverse loader — storage objects (Storage API; protect_delete blocks
  raw SQL), first-sweep seed templates, eb_exams, eb_specifications. Verified reversible on .94.
- Granular admin reset: POST /admin/reset?scope=all|exam-corpus|timetable. reset_environment.reset(scope)
  adds EXAM_CORPUS_TABLES (10) + cc.examboards storage cleanup + TIMETABLE_TABLES (13); 'all' now also
  clears the exam subsystem the legacy reset missed. No schema migration required.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

api/services/docling/template.py

@@ -44,6 +44,34 @@ def _furn_kind(it):
     return "chrome_text"
 
 
+def synthesize_part_box(part_band, content_x_band):
+    """Return the one authoritative S5 part-box projection.
+
+    Parts remain boxes in S5, but the box is a projection rather than intrinsic
+    geometry: document content margins provide the x-extent and the part band
+    provides y. The band end is already bounded by the next part in bands.py;
+    the original label box remains a separate anchor for rendering/review.
+
+    Coordinates stay in the first-pass PDF-point BOTTOMLEFT bbox shape.
+    """
+    if not content_x_band:
+        return None
+    try:
+        x_left = content_x_band["x_left"]
+        x_right = content_x_band["x_right"]
+        y_start = part_band["y_start"]
+        y_end = part_band["y_end"]
+    except KeyError:
+        return None
+    return {
+        "l": round(x_left, 1),
+        "t": round(y_start, 1),
+        "r": round(x_right, 1),
+        "b": round(y_end, 1),
+        "coord_origin": "BOTTOMLEFT",
+    }
+
+
 def build(structured, bands, furniture, pdf=None, page_roles=None):
     page_roles = page_roles or {}
     part_bbox = {p["label"]: p.get("bbox")
@@ -124,10 +152,15 @@ def build(structured, bands, furniture, pdf=None, page_roles=None):
         main = [{"question": m["question"], "y_start": m["y_start"], "y_end": m["y_end"],
                  "is_start": m.get("is_start", True),
                  "source": "auto", "confirmed": False} for m in pb["main"]]
-        part = [{"label": p["label"], "question": p["question"],
-                 "y_start": p["y_start"], "y_end": p["y_end"],
-                 "label_box": part_bbox.get(p["label"]),     # app may render a box instead of lines
-                 "source": "auto", "confirmed": False} for p in pb["part"]]
+        part = []
+        for p in pb["part"]:
+            part.append({
+                "label": p["label"], "question": p["question"],
+                "y_start": p["y_start"], "y_end": p["y_end"],
+                "label_box": part_bbox.get(p["label"]),  # anchor, not the part extent
+                "box": synthesize_part_box(p, xband),
+                "source": "auto", "confirmed": False,
+            })
         pr = page_roles.get(pgs) or page_roles.get(pg) or {}
         pages[pgs] = {
             "role": pr.get("role", "question"),

docker-entrypoint.sh

@@ -75,6 +75,14 @@ if [ "$RUN_INIT" = "true" ]; then
                 }
                 print_success "GAIS data import completed"
                 ;;
+            "exam-corpus")
+                print_status "Seeding exam-paper corpus (manifest-gated; skips if none configured)..."
+                python3 main.py --mode exam-corpus || {
+                    print_error "Exam corpus seed failed!"
+                    exit 1
+                }
+                print_success "Exam corpus seed completed"
+                ;;
             "full")
                 print_status "Running full initialization..."
                 python3 main.py --mode infra || exit 1

main.py

@@ -323,6 +323,52 @@ def run_gais_data_mode():
 
 # Old clear_dev_redis_queue function removed - now handled by Redis Manager
 
+def run_exam_corpus_mode():
+    """Seed the public exam-paper corpus from a manifest (optional, gated).
+
+    Env controls:
+      EXAM_CORPUS_MANIFEST    - path to the corpus manifest (required to do anything)
+      EXAM_CORPUS_DRY_RUN     - 'true' to validate + report only
+      EXAM_CORPUS_FORCE       - 'true' to re-upload/overwrite existing objects
+      EXAM_CORPUS_BOARD/_SPEC - filter to one exam_board_code / spec_code
+      EXAM_CORPUS_USER_SUBSET - 'true' to also seed a user-side test subset
+      EXAM_CORPUS_FIRST_SWEEP - 'true' to run the docling/auto-map first pass
+
+    Skips gracefully (success) when no manifest is configured/present, so it is safe
+    in a comma-mode list (e.g. INIT_MODE=infra,seed,exam-corpus) before papers exist.
+    Buckets are NOT created here — infra mode (buckets.py) owns provisioning.
+    """
+    logger.info("Running in exam-corpus seed mode")
+    manifest = os.getenv("EXAM_CORPUS_MANIFEST")
+    if not manifest or not os.path.exists(manifest):
+        logger.warning(
+            f"exam-corpus: no manifest at EXAM_CORPUS_MANIFEST={manifest!r}; skipping (nothing to seed yet)"
+        )
+        return True
+    try:
+        from run.initialization.seed_exam_corpus import load
+        rep = load(
+            manifest,
+            dry_run=_truthy_env("EXAM_CORPUS_DRY_RUN"),
+            force=_truthy_env("EXAM_CORPUS_FORCE"),
+            board_filter=os.getenv("EXAM_CORPUS_BOARD") or None,
+            spec_filter=os.getenv("EXAM_CORPUS_SPEC") or None,
+            user_subset=_truthy_env("EXAM_CORPUS_USER_SUBSET"),
+            do_first_sweep=_truthy_env("EXAM_CORPUS_FIRST_SWEEP"),
+        )
+        if rep.errors:
+            logger.error(f"exam-corpus seed completed with {len(rep.errors)} error(s)")
+            return False
+        logger.info(
+            f"exam-corpus seed ok: specs={rep.specs_upserted} papers={rep.papers_upserted} "
+            f"uploaded={rep.files_uploaded}"
+        )
+        return True
+    except Exception as e:
+        logger.error(f"exam-corpus seed failed: {e}")
+        return False
+
+
 def run_development_mode():
     """Run the server in development mode with auto-reload"""
     logger.info("Running in development mode")
@@ -411,7 +457,7 @@ Startup modes:
     
     parser.add_argument(
         '--mode', '-m',
-        choices=['infra', 'seed', 'seed-test', 'gais-data', 'dev', 'prod'],
+        choices=['infra', 'seed', 'seed-test', 'gais-data', 'exam-corpus', 'dev', 'prod'],
         default='dev',
         help='Startup mode (default: dev)'
     )
@@ -446,7 +492,12 @@ if __name__ == "__main__":
         # Run GAIS data import
         success = run_gais_data_mode()
         sys.exit(0 if success else 1)
-    
+
+    elif args.mode == 'exam-corpus':
+        # Seed the public exam-paper corpus from a manifest (gated; skips if none configured)
+        success = run_exam_corpus_mode()
+        sys.exit(0 if success else 1)
+
     elif args.mode == 'dev':
         # Run development server
         run_development_mode()

routers/database/tools/classes_router.py

@@ -323,7 +323,7 @@ async def get_class(
     # Enrollment requests (pending)
     reqs = (
         sb.supabase.table("enrollment_requests")
-        .select("id, student_id, status, created_at")
+        .select("id, student_id, status, requested_at")
         .eq("class_id", class_id)
         .eq("status", "pending")
         .execute()

routers/database/tools/platform_admin_router.py

@@ -126,13 +126,24 @@ async def platform_stats(
 
 @router.post("/reset")
 async def reset_environment(
+    scope: str = "all",
     _: dict = Depends(require_platform_admin),
 ) -> Dict[str, Any]:
-    """DESTRUCTIVE: wipe all test data. Neo4j + Supabase. Platform admin only."""
+    """DESTRUCTIVE: wipe test data. Platform admin only.
+
+    scope (query param):
+      - all          : full wipe (Neo4j + Supabase data + auth users) AND exam subsystem + storage.
+      - exam-corpus  : ONLY the exam corpus — eb_*/exam_* tables + cc.examboards storage objects
+                       (load/unload the public corpus without touching schools/users).
+      - timetable    : ONLY timetable/calendar materialization tables.
+    """
+    if scope not in ("all", "exam-corpus", "timetable"):
+        raise HTTPException(status_code=400, detail="scope must be one of: all, exam-corpus, timetable")
     import asyncio
+    import functools
     from run.initialization.reset_environment import reset as _reset
     loop = asyncio.get_event_loop()
-    result = await loop.run_in_executor(None, _reset)
+    result = await loop.run_in_executor(None, functools.partial(_reset, scope))
     return {"status": "ok", **result}
 
 

routers/exam/templates.py

@@ -12,13 +12,19 @@ join keys (spec §2).
 """
 from __future__ import annotations
 
+import json
 import os
+import tempfile
+import time
 import uuid
 from typing import Any, Dict, List, Optional, Tuple
 
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, UploadFile
-from fastapi.responses import Response
+from fastapi.responses import JSONResponse, Response
 
+from api.services.docling import AutoMapError, auto_map
+from api.services.docling import extract as docling_extract
+from api.services.docling.regions import detect_response_regions_from_pdf
 from modules.database.services.exam_projection import project_template, project_template_safe
 from modules.database.supabase.utils.client import SupabaseServiceRoleClient
 from modules.database.supabase.utils.storage import StorageAdmin
@@ -37,6 +43,8 @@ router = APIRouter()
 
 SOURCE_CABINET_NAME = "Exam Marker Template Sources"
 SOURCE_BUCKET_FALLBACK = "cc.users"
+AUTO_MAP_JOB_PREFIX = "exam:auto-map"
+_AUTO_MAP_JOB_STATUS: Dict[str, Dict[str, Any]] = {}
 
 
 # ─── helpers ─────────────────────────────────────────────────────────────────
@@ -224,6 +232,341 @@ async def _upload_template_source_file(
     return file_id
 
 
+def _job_key(job_id: str) -> str:
+    return f"{AUTO_MAP_JOB_PREFIX}:{job_id}"
+
+
+def _redis_client() -> Any:
+    try:
+        import redis
+    except Exception:
+        return None
+    try:
+        url = os.getenv("LOCAL_REDIS_URL") or os.getenv("REDIS_URL")
+        if url:
+            client = redis.Redis.from_url(url, decode_responses=True, socket_timeout=2)
+        else:
+            client = redis.Redis(
+                host=os.getenv("REDIS_HOST", "localhost"),
+                port=int(os.getenv("REDIS_PORT", "6379")),
+                db=int(os.getenv("REDIS_DB_DEV", os.getenv("REDIS_DB", "0"))),
+                password=os.getenv("REDIS_PASSWORD") or None,
+                decode_responses=True,
+                socket_timeout=2,
+            )
+        client.ping()
+        return client
+    except Exception:
+        return None
+
+
+def _set_auto_map_status(job_id: str, payload: Dict[str, Any]) -> None:
+    status = {"job_id": job_id, "updated_at": int(time.time()), **payload}
+    _AUTO_MAP_JOB_STATUS[job_id] = status
+    client = _redis_client()
+    if client is not None:
+        try:
+            client.setex(_job_key(job_id), int(os.getenv("EXAM_AUTO_MAP_JOB_TTL", "3600")), json.dumps(status))
+        except Exception as exc:
+            logger.warning(f"auto-map redis status write failed for {job_id}: {exc}")
+
+
+def _get_auto_map_status(job_id: str) -> Optional[Dict[str, Any]]:
+    client = _redis_client()
+    if client is not None:
+        try:
+            raw = client.get(_job_key(job_id))
+            if raw:
+                return json.loads(raw)
+        except Exception as exc:
+            logger.warning(f"auto-map redis status read failed for {job_id}: {exc}")
+    return _AUTO_MAP_JOB_STATUS.get(job_id)
+
+
+def _resolve_template_source(ctx: ExamContext, template: Dict[str, Any]) -> Tuple[str, str, bytes]:
+    bucket: Optional[str] = None
+    path: Optional[str] = None
+    if template.get("exam_id"):
+        storage_loc = _lookup_exam_storage_loc(template["exam_id"])
+        if not storage_loc:
+            raise HTTPException(status_code=404, detail="Template source not found")
+        try:
+            bucket, path = _parse_storage_loc(storage_loc)
+        except ValueError:
+            raise HTTPException(status_code=404, detail="Template source not found")
+    elif template.get("source_file_id"):
+        # Same scoped service-role exception as source-pdf: owner gate has already passed.
+        file_row = _first(
+            SupabaseServiceRoleClient().supabase.table("files")
+            .select("bucket, path, mime_type, name")
+            .eq("id", template["source_file_id"])
+            .limit(1)
+            .execute()
+        )
+        if not file_row or not file_row.get("bucket") or not file_row.get("path"):
+            raise HTTPException(status_code=404, detail="Template source not found")
+        bucket = file_row["bucket"]
+        path = file_row["path"]
+    else:
+        raise HTTPException(status_code=404, detail="Template source not found")
+    try:
+        return bucket, path, StorageAdmin().download_file(bucket, path)
+    except Exception as exc:
+        logger.warning(f"Template source download failed for template {template.get('id')}: {exc}")
+        raise HTTPException(status_code=404, detail="Template source not found")
+
+
+def _pdf_has_text_layer(pdf_bytes: bytes) -> bool:
+    with tempfile.NamedTemporaryFile(prefix="cc-auto-map-detect-", suffix=".pdf", delete=False) as fh:
+        fh.write(pdf_bytes)
+        tmp = fh.name
+    try:
+        return bool(docling_extract.has_text_layer(tmp))
+    finally:
+        try:
+            os.unlink(tmp)
+        except OSError:
+            pass
+
+
+def _pdf_page_geometry(pdf_bytes: bytes) -> List[Dict[str, float]]:
+    with tempfile.NamedTemporaryFile(prefix="cc-auto-map-geom-", suffix=".pdf", delete=False) as fh:
+        fh.write(pdf_bytes)
+        tmp = fh.name
+    try:
+        import fitz
+        doc = fitz.open(tmp)
+        pages: List[Dict[str, float]] = []
+        page_top = 0.0
+        try:
+            for page in doc:
+                media = page.mediabox
+                crop = page.cropbox
+                rendered_w = float(crop.width or page.rect.width or 595.0)
+                rendered_h = float(crop.height or page.rect.height or 842.0)
+                pages.append({
+                    "media_x0": float(media.x0),
+                    "crop_x0": float(crop.x0),
+                    "crop_y0": float(crop.y0),
+                    "page_pt_w": float(crop.width or page.rect.width or 1),
+                    "page_pt_h": float(crop.height or page.rect.height or 1),
+                    "rendered_w": rendered_w,
+                    "rendered_h": rendered_h,
+                    "page_top": page_top,
+                })
+                page_top += rendered_h
+        finally:
+            doc.close()
+        return pages
+    except Exception as exc:
+        logger.warning(f"PDF geometry read failed; falling back to A4 page geometry: {exc}")
+        return []
+    finally:
+        try:
+            os.unlink(tmp)
+        except OSError:
+            pass
+
+
+def _page_geom(pages: List[Dict[str, float]], page_number: int) -> Dict[str, float]:
+    if 1 <= page_number <= len(pages):
+        return pages[page_number - 1]
+    return {
+        "media_x0": 0.0, "crop_x0": 0.0, "crop_y0": 0.0,
+        "page_pt_w": 595.0, "page_pt_h": 842.0,
+        "rendered_w": 595.0, "rendered_h": 842.0,
+        "page_top": (page_number - 1) * 842.0,
+    }
+
+
+def _box_to_canvas(box: Optional[Dict[str, Any]], page_number: int, pages: List[Dict[str, float]]) -> Optional[Dict[str, float]]:
+    if not box:
+        return None
+    g = _page_geom(pages, page_number)
+    if box.get("coord_origin") == "TOPLEFT" and {"x", "y", "w", "h"}.issubset(box):
+        scale = 0.5 if box.get("unit") == "px" else 1.0
+        return {
+            "x": round(float(box["x"]) * scale, 2),
+            "y": round(g["page_top"] + float(box["y"]) * scale, 2),
+            "w": round(float(box["w"]) * scale, 2),
+            "h": round(float(box["h"]) * scale, 2),
+        }
+    if not {"l", "t", "r", "b"}.issubset(box):
+        return None
+    l, t, r, b = (float(box[k]) for k in ("l", "t", "r", "b"))
+    # Canvas pages are rendered from the PDF CropBox with page_left fixed at 0.
+    # Docling boxes are in PDF user-space coordinates, so subtract the CropBox
+    # origin instead of adding it; otherwise cropped PDFs shift right/down.
+    x = (l - g["crop_x0"]) / g["page_pt_w"] * g["rendered_w"]
+    y = g["page_top"] + (g["page_pt_h"] - (t - g["crop_y0"])) / g["page_pt_h"] * g["rendered_h"]
+    w = (r - l) / g["page_pt_w"] * g["rendered_w"]
+    h = (t - b) / g["page_pt_h"] * g["rendered_h"]
+    return {"x": round(x, 2), "y": round(y, 2), "w": round(w, 2), "h": round(h, 2)}
+
+
+def _response_form_from_region_type(region_type: Any) -> Optional[str]:
+    return {
+        "answer_lines": "lines",
+        "answer_box": "answer-box",
+        "working_space": "working",
+        "lines": "lines",
+        "answer-box": "answer-box",
+        "working": "working",
+    }.get(str(region_type or ""))
+
+
+def _y_to_canvas(y_value: float, page_number: int, pages: List[Dict[str, float]]) -> float:
+    g = _page_geom(pages, page_number)
+    return round(g["page_top"] + (g["page_pt_h"] - (float(y_value) - g["crop_y0"])) / g["page_pt_h"] * g["rendered_h"], 2)
+
+
+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]])))
+
+
+def _safe_confidence(value: Any = None) -> float:
+    if isinstance(value, (int, float)):
+        return max(0.0, min(1.0, float(value)))
+    return 0.75
+
+
+def _margin_values(first_pass: Dict[str, Any], page_number: int) -> Dict[str, Optional[float]]:
+    vals: Dict[str, Optional[float]] = {"left": None, "right": None, "top": None, "bottom": None}
+    for m in first_pass.get("margins") or []:
+        edge = m.get("edge")
+        if edge not in vals:
+            continue
+        if m.get("scope") == "document" and edge in {"left", "right"}:
+            vals[edge] = m.get("value")
+        elif m.get("scope") == "page" and int(m.get("page") or -1) == page_number:
+            vals[edge] = m.get("value")
+    return vals
+
+
+def _map_first_pass_to_rows(template_id: str, first_pass: Dict[str, Any], pdf_bytes: bytes, extra_regions: Optional[List[Dict[str, Any]]] = None) -> Dict[str, List[Dict[str, Any]]]:
+    pages_geom = _pdf_page_geometry(pdf_bytes)
+    questions: List[Dict[str, Any]] = []
+    response_areas: List[Dict[str, Any]] = []
+    boundaries: List[Dict[str, Any]] = []
+    layout: List[Dict[str, Any]] = []
+    q_ids: Dict[str, str] = {}
+    first_part_by_page: Dict[int, str] = {}
+    pages_obj = first_pass.get("pages") or {}
+
+    for page_key in sorted(pages_obj, key=lambda k: int(k)):
+        page_number = int(page_key)
+        page_index = page_number - 1
+        page = pages_obj[page_key]
+        margins = _margin_values(first_pass, page_number)
+        layout.append({
+            "id": _ai_id(template_id, "layout", page_number),
+            "template_id": template_id,
+            "page_index": page_index,
+            "role": page.get("role"),
+            "margin_left": margins["left"],
+            "margin_right": margins["right"],
+            "margin_top": margins["top"],
+            "margin_bottom": margins["bottom"],
+            "margins_enabled": bool(page.get("margins_enabled", True)),
+            "source": "ai",
+            "confirmed": False,
+            "confidence": 0.8,
+            "derivation": "docling-page-layout",
+            "meta": {"role_source": page.get("role_source"), "schema": first_pass.get("meta", {}).get("schema")},
+        })
+        for band in page.get("main_bands") or []:
+            label = str(band.get("question") or "").strip()
+            if not label:
+                continue
+            qid = q_ids.setdefault(label, _ai_id(template_id, "question", label))
+            if not any(q["id"] == qid for q in questions):
+                questions.append({"id": qid, "template_id": template_id, "label": label, "order": len(q_ids) - 1, "max_marks": 0, "is_container": True, "source": "ai", "confirmed": False, "confidence": _safe_confidence(band.get("confidence")), "derivation": "docling-main-band"})
+            for edge, yv in (("start", band.get("y_start")), ("end", band.get("y_end"))):
+                if yv is not None:
+                    boundaries.append({"id": _ai_id(template_id, "boundary", label, edge, page_number), "template_id": template_id, "question_id": qid, "label": f"{label}:{edge}", "page_index": page_index, "y": _y_to_canvas(float(yv), page_number, pages_geom), "bounds": None, "source": "ai", "confirmed": False, "confidence": _safe_confidence(band.get("confidence")), "derivation": "docling-main-band"})
+        for band in page.get("part_bands") or []:
+            label = str(band.get("label") or "").strip()
+            parent_label = str(band.get("question") or "").strip()
+            if not label:
+                continue
+            parent_id = q_ids.setdefault(parent_label, _ai_id(template_id, "question", parent_label or label.split(".")[0]))
+            if parent_label and not any(q["id"] == parent_id for q in questions):
+                questions.append({"id": parent_id, "template_id": template_id, "label": parent_label, "order": len(q_ids) - 1, "max_marks": 0, "is_container": True, "source": "ai", "confirmed": False, "confidence": 0.7, "derivation": "docling-inferred-main-question"})
+            pid = _ai_id(template_id, "part", label)
+            first_part_by_page.setdefault(page_index, pid)
+            bounds = None
+            y1, y2 = band.get("y_start"), band.get("y_end")
+            if margins["left"] is not None and margins["right"] is not None and y1 is not None and y2 is not None:
+                top = max(float(y1), float(y2)); bottom = min(float(y1), float(y2))
+                bounds = _box_to_canvas({"l": margins["left"], "r": margins["right"], "t": top, "b": bottom, "coord_origin": "BOTTOMLEFT"}, page_number, pages_geom)
+            bounds = bounds or _box_to_canvas(band.get("label_box"), page_number, pages_geom)
+            questions.append({"id": pid, "template_id": template_id, "parent_id": parent_id, "label": label, "order": len(questions), "max_marks": 0, "is_container": False, "bounds": bounds, "page": page_number, "source": "ai", "confirmed": False, "confidence": _safe_confidence(band.get("confidence")), "derivation": "docling-part-band-x-margins"})
+
+    default_qid = questions[0]["id"] if questions else _ai_id(template_id, "question", "auto")
+    for page_key in sorted(pages_obj, key=lambda k: int(k)):
+        page_number = int(page_key); page_index = page_number - 1; page = pages_obj[page_key]
+        owner_qid = first_part_by_page.get(page_index, default_qid)
+        for collection, kind, context_type, derivation in (("furniture", "furniture", None, "docling-furniture"), ("figures", "context", "figure", "docling-context-figure"), ("tables", "context", "data_table", "docling-table")):
+            for idx, item in enumerate(page.get(collection) or []):
+                bounds = _box_to_canvas(item.get("box"), page_number, pages_geom)
+                if bounds:
+                    row = {"id": _ai_id(template_id, collection, page_number, idx), "template_id": template_id, "question_id": owner_qid, "page": page_number, "bounds": bounds, "kind": kind, "source": "ai", "confirmed": False, "confidence": 0.65, "derivation": derivation}
+                    if context_type:
+                        row["context_type"] = context_type
+                    response_areas.append(row)
+    for idx, region in enumerate(extra_regions or []):
+        page_index = int(region.get("page_index", 0))
+        bounds = _box_to_canvas(region.get("bbox") or {}, page_index + 1, pages_geom)
+        if bounds:
+            response_form = _response_form_from_region_type(region.get("region_type"))
+            if response_form:
+                response_areas.append({"id": _ai_id(template_id, "region", page_index, idx), "template_id": template_id, "question_id": first_part_by_page.get(page_index, default_qid), "page": page_index + 1, "bounds": bounds, "kind": "response", "response_form": response_form, "source": "ai", "confirmed": False, "confidence": _safe_confidence(region.get("confidence")), "derivation": region.get("detection_method") or "opencv-response-region"})
+    return {"questions": questions, "response_areas": response_areas, "boundaries": boundaries, "layout": layout}
+
+
+def _refresh_ai_rows(ctx: ExamContext, template_id: str, rows: Dict[str, List[Dict[str, Any]]]) -> None:
+    sb = ctx.supabase
+    for table in ("exam_response_areas", "exam_boundaries", "exam_template_layout", "exam_questions"):
+        sb.table(table).delete().eq("template_id", template_id).eq("source", "ai").eq("confirmed", False).execute()
+    for table, key in (("exam_questions", "questions"), ("exam_response_areas", "response_areas"), ("exam_boundaries", "boundaries"), ("exam_template_layout", "layout")):
+        payload = rows.get(key) or []
+        if payload:
+            sb.table(table).insert(payload).execute()
+
+
+def _run_auto_map_merge(ctx: ExamContext, template_id: str, pdf_bytes: bytes, source_label: str) -> Dict[str, List[Dict[str, Any]]]:
+    first_pass = auto_map(pdf_bytes, source_pdf=source_label)
+    extra_regions: List[Dict[str, Any]] = []
+    try:
+        with tempfile.NamedTemporaryFile(prefix="cc-auto-map-regions-", suffix=".pdf", delete=False) as fh:
+            fh.write(pdf_bytes)
+            tmp = fh.name
+        try:
+            extra_regions = detect_response_regions_from_pdf(tmp)
+        finally:
+            try:
+                os.unlink(tmp)
+            except OSError:
+                pass
+    except Exception as exc:
+        logger.info(f"auto-map response-region detection skipped for template {template_id}: {exc}")
+    rows = _map_first_pass_to_rows(template_id, first_pass, pdf_bytes, extra_regions)
+    _refresh_ai_rows(ctx, template_id, rows)
+    updates = {"exam_code": first_pass.get("meta", {}).get("paper_code"), "page_count": first_pass.get("meta", {}).get("n_pages")}
+    ctx.supabase.table("exam_templates").update({k: v for k, v in updates.items() if v is not None}).eq("id", template_id).execute()
+    return rows
+
+
+def _run_auto_map_job(job_id: str, ctx: ExamContext, template_id: str, pdf_bytes: bytes, source_label: str) -> None:
+    _set_auto_map_status(job_id, {"status": "running", "template_id": template_id})
+    try:
+        rows = _run_auto_map_merge(ctx, template_id, pdf_bytes, source_label)
+        _set_auto_map_status(job_id, {"status": "completed", "template_id": template_id, "counts": {k: len(v) for k, v in rows.items()}})
+    except Exception as exc:
+        logger.exception(f"auto-map job failed for template {template_id}: {exc}")
+        _set_auto_map_status(job_id, {"status": "failed", "template_id": template_id, "error": str(exc)})
+
+
 # ─── templates ───────────────────────────────────────────────────────────────
 
 
@@ -339,48 +682,61 @@ async def get_template_source_pdf(
     template = _fetch_template_or_404(ctx, template_id)
     _require_source_visibility_or_404(ctx, template)
 
-    bucket: Optional[str] = None
-    path: Optional[str] = None
-
-    if template.get("exam_id"):
-        storage_loc = _lookup_exam_storage_loc(template["exam_id"])
-        if not storage_loc:
-            raise HTTPException(status_code=404, detail="Template source not found")
-        try:
-            bucket, path = _parse_storage_loc(storage_loc)
-        except ValueError:
-            raise HTTPException(status_code=404, detail="Template source not found")
-    elif template.get("source_file_id"):
-        # Resolve the file row via service role (authz already done above: the caller proved they
-        # can see this template, and source_file_id is the template's own file). Reading `files`
-        # as-the-user trips a pre-existing broken RLS policy on cabinet_memberships
-        # (42P17 infinite recursion) — documented service-role exception, like the catalogue lookup.
-        file_row = _first(
-            SupabaseServiceRoleClient().supabase.table("files")
-            .select("bucket, path, mime_type, name")
-            .eq("id", template["source_file_id"])
-            .limit(1)
-            .execute()
-        )
-        if not file_row or not file_row.get("bucket") or not file_row.get("path"):
-            raise HTTPException(status_code=404, detail="Template source not found")
-        bucket = file_row["bucket"]
-        path = file_row["path"]
-    else:
-        raise HTTPException(status_code=404, detail="Template source not found")
-
-    if not bucket or not path:
-        raise HTTPException(status_code=404, detail="Template source not found")
-
-    try:
-        pdf_bytes = StorageAdmin().download_file(bucket, path)
-    except Exception as exc:
-        logger.warning(f"Template source download failed for template {template_id}: {exc}")
-        raise HTTPException(status_code=404, detail="Template source not found")
-
+    _, _, pdf_bytes = _resolve_template_source(ctx, template)
     return Response(content=pdf_bytes, media_type="application/pdf")
 
 
+@router.post("/templates/{template_id}/auto-map", response_model=None)
+async def auto_map_template(
+    template_id: str,
+    background_tasks: BackgroundTasks,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any] | JSONResponse:
+    template = _fetch_template_or_404(ctx, template_id)
+    _require_owner(ctx, template)
+    _require_source_visibility_or_404(ctx, template)
+    if _template_has_recorded_marks(ctx, template_id):
+        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}"
+    try:
+        fast_path = _pdf_has_text_layer(pdf_bytes)
+    except Exception as exc:
+        logger.warning(f"auto-map text-layer detection failed for template {template_id}; falling back to OCR queue: {exc}")
+        fast_path = False
+    if not fast_path:
+        job_id = str(uuid.uuid4())
+        _set_auto_map_status(job_id, {"status": "queued", "template_id": template_id})
+        background_tasks.add_task(_run_auto_map_job, job_id, ctx, template_id, pdf_bytes, source_label)
+        return JSONResponse(status_code=202, content={"status": "accepted", "job_id": job_id})
+    try:
+        _run_auto_map_merge(ctx, template_id, pdf_bytes, source_label)
+    except (AutoMapError, ValueError) as exc:
+        raise HTTPException(status_code=422, detail=f"Auto-map failed: {exc}")
+    except Exception as exc:
+        logger.exception(f"auto-map failed for template {template_id}: {exc}")
+        raise HTTPException(status_code=502, detail=f"Auto-map failed: {exc}")
+    background_tasks.add_task(project_template_safe, template_id)
+    return await get_template(template_id, ctx)
+
+
+@router.get("/templates/{template_id}/auto-map/{job_id}/status")
+async def auto_map_status(
+    template_id: str,
+    job_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    template = _fetch_template_or_404(ctx, template_id)
+    _require_owner(ctx, template)
+    status = _get_auto_map_status(job_id)
+    if not status or status.get("template_id") != template_id:
+        raise HTTPException(status_code=404, detail="Auto-map job not found")
+    body = dict(status)
+    if body.get("status") == "completed":
+        body["template"] = await get_template(template_id, ctx)
+    return body
+
+
 @router.put("/templates/{template_id}")
 async def replace_template(
     template_id: str,

run/initialization/buckets.py

@@ -46,7 +46,7 @@ def initialize_buckets() -> dict:
                     file_size_limit=1000 * 1024 * 1024,  # 1GB
                 )
             },
-            # Exam Board files
+            # Exam Board files (admin-curated public exam corpus: QP/MS/insert/ER + specs)
             {
                 "id": "cc.examboards",
                 "options": CreateBucketOptions(
@@ -55,6 +55,34 @@ def initialize_buckets() -> dict:
                     file_size_limit=1000 * 1024 * 1024,  # 1GB
                 )
             },
+            # ── Storage taxonomy bins (access scoped by RLS on bucket + leading path segment; RLS = D1) ──
+            # Platform-managed public/shared assets (readable by all authenticated users).
+            {
+                "id": "cc.public",
+                "options": CreateBucketOptions(
+                    name="Classroom Copilot Public",
+                    public=False,
+                    file_size_limit=1000 * 1024 * 1024,  # 1GB
+                )
+            },
+            # Institute-scoped operational assets: cc.institutes/{institute_id}/...
+            {
+                "id": "cc.institutes",
+                "options": CreateBucketOptions(
+                    name="Classroom Copilot Institutes",
+                    public=False,
+                    file_size_limit=1000 * 1024 * 1024,  # 1GB
+                )
+            },
+            # Platform-admin-only assets, seeds, intake/staging for unidentified papers.
+            {
+                "id": "cc.admin",
+                "options": CreateBucketOptions(
+                    name="Classroom Copilot Admin",
+                    public=False,
+                    file_size_limit=1000 * 1024 * 1024,  # 1GB
+                )
+            },
         ]
         
         results = {}
@@ -81,11 +109,17 @@ def initialize_buckets() -> dict:
                     logger.error(f"Failed to create bucket: {bucket['id']}")
                     
             except Exception as e:
-                results[bucket["id"]] = {
-                    "status": "error",
-                    "error": str(e)
-                }
-                logger.error(f"Error creating bucket {bucket['id']}: {str(e)}")
+                # Idempotent: an already-existing bucket is not a failure on re-run.
+                if any(s in str(e).lower() for s in ("already exists", "duplicate", "resource already")):
+                    results[bucket["id"]] = {"status": "exists", "result": str(e)}
+                    success_count += 1
+                    logger.info(f"Bucket already exists (ok): {bucket['id']}")
+                else:
+                    results[bucket["id"]] = {
+                        "status": "error",
+                        "error": str(e)
+                    }
+                    logger.error(f"Error creating bucket {bucket['id']}: {str(e)}")
         
         # Determine overall success
         if success_count == total_count:

run/initialization/manifests/_corpus_store/.gitignore

@@ -0,0 +1,3 @@
+# Persistent local corpus store — PDFs are NOT committed (re-downloadable from manifest).
+*
+!.gitignore

run/initialization/manifests/generate_corpus_manifest.py

@@ -0,0 +1,501 @@
+#!/usr/bin/env python3
+"""
+generate_corpus_manifest.py — build the public exam-corpus manifest from OFFICIAL sources,
+verifying every source URL is live before it is written.
+
+Output: exam-corpus.yaml (consumed by run/initialization/seed_exam_corpus.py).
+
+Sources (all official exam-board hosts; public past-paper PDFs):
+  AQA      filestore.aqa.org.uk  — fully templatable; enumerated + HEAD-verified here.
+  Edexcel  qualifications.pearson.com — date suffix non-derivable; confirmed URLs embedded.
+  OCR      www.ocr.org.uk/Images — opaque doc-id; confirmed URLs embedded.
+
+Every URL is HEAD/GET-checked (200 + application/pdf) before inclusion, so the committed
+manifest never carries a dead or wrong-cased link. Re-run to refresh as more sessions go public.
+
+Conventions (locked — see ~/cc/ideas/2026-06-07-exam-paper-ingestion.md):
+  session   = "YYYY-Mon"        e.g. 2022-Jun
+  exam_code = BOARD-award-PAPER-SESSIONCOMPACT-ROLE   e.g. AQA-8463-1H-2022JUN-QP
+"""
+from __future__ import annotations
+import concurrent.futures as cf
+import os
+import sys
+import urllib.error
+import urllib.request
+from typing import Any, Dict, List, Optional, Tuple
+
+import yaml
+
+AQA_BASE = "https://filestore.aqa.org.uk/sample-papers-and-mark-schemes"
+ROLE_TOKEN = {"QP": "QP", "MS": "MS", "ER": "WRE"}       # AQA filestore role tokens
+MONTHS = {"JUN": ("june", "Jun"), "NOV": ("november", "Nov")}
+FETCHED = "2026-06-07"
+
+
+def head_ok(url: str, timeout: int = 20) -> bool:
+    """True iff the URL resolves to a real PDF (200 + application/pdf), following redirects.
+    AQA soft-404s redirect to www.aqa.org.uk/req_path=... (text/html), so we check content-type.
+    Uses a tiny Range GET (stdlib urllib) so we never pull the whole PDF just to verify it."""
+    req = urllib.request.Request(url, headers={"Range": "bytes=0-3", "User-Agent": "cc-corpus/1.0"})
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as r:
+            ctype = (r.headers.get("content-type") or "").lower()
+            return r.status in (200, 206) and "pdf" in ctype
+    except urllib.error.HTTPError as e:
+        # A 206/200 PDF never lands here; 404/redirect-to-html will.
+        ctype = (e.headers.get("content-type") or "").lower() if e.headers else ""
+        return e.code in (200, 206) and "pdf" in ctype
+    except Exception:
+        return False
+
+
+# ─────────────────────────── AQA catalogue ───────────────────────────
+# spec_code, subject, award, award_level, first_teach, [(filestore_papercode, paper_code, tier), ...]
+def _gcse_single(award: str) -> List[Tuple[str, str, Optional[str]]]:
+    out = []
+    for paper in ("1", "2"):
+        for tier in ("F", "H"):
+            out.append((f"{award}{paper}{tier}", f"{award}/{paper}{tier}", tier))
+    return out
+
+def _trilogy(award: str) -> List[Tuple[str, str, Optional[str]]]:
+    out = []
+    for subj in ("B", "C", "P"):
+        for paper in ("1", "2"):
+            for tier in ("F", "H"):
+                out.append((f"{award}{subj}{paper}{tier}", f"{award}/{subj}/{paper}{tier}", tier))
+    return out
+
+def _alevel(award: str, papers=("1", "2", "3")) -> List[Tuple[str, str, Optional[str]]]:
+    return [(f"{award}{p}", f"{award}/{p}", None) for p in papers]
+
+def _subj(award: str, papers, tiers=(None,)) -> List[Tuple[str, str, Optional[str]]]:
+    """Generic GCSE/A-level builder. tiers=('F','H') for tiered subjects (Maths/Science),
+    tiers=(None,) for untiered (English/Geography/CS/Business/Psychology)."""
+    out = []
+    for p in papers:
+        for t in tiers:
+            tl = t or ""
+            out.append((f"{award}{p}{tl}", f"{award}/{p}{tl}", t))
+    return out
+
+def _mfl(award: str) -> List[Tuple[str, str, Optional[str]]]:
+    """AQA MFL: Listening/Reading/Writing papers, each Foundation/Higher (Speaking is teacher-conducted,
+    no public QP). Filestore code encodes skill+tier, e.g. 8658LH = French Listening Higher."""
+    out = []
+    for skill in ("L", "R", "W"):
+        for t in ("F", "H"):
+            out.append((f"{award}{skill}{t}", f"{award}/{skill}{t}", t))
+    return out
+
+AQA_SPECS = [
+    # ── Sciences (round 1 — kept at full depth) ──────────────────────────────────────
+    ("AQA-BIOL-8461", "BIOLOGY", "8461", "GCSE", "2016", _gcse_single("8461")),
+    ("AQA-CHEM-8462", "CHEMISTRY", "8462", "GCSE", "2016", _gcse_single("8462")),
+    ("AQA-PHYS-8463", "PHYSICS", "8463", "GCSE", "2016", _gcse_single("8463")),
+    ("AQA-COMB-8464", "COMBINED SCIENCE TRILOGY", "8464", "GCSE", "2016", _trilogy("8464")),
+    ("AQA-BIOL-7401", "BIOLOGY", "7401", "AS", "2015", _alevel("7401", ("1", "2"))),
+    ("AQA-BIOL-7402", "BIOLOGY", "7402", "A-level", "2015", _alevel("7402")),
+    ("AQA-CHEM-7404", "CHEMISTRY", "7404", "AS", "2015", _alevel("7404", ("1", "2"))),
+    ("AQA-CHEM-7405", "CHEMISTRY", "7405", "A-level", "2015", _alevel("7405")),
+    ("AQA-PHYS-7407", "PHYSICS", "7407", "AS", "2015", _alevel("7407", ("1", "2"))),
+    ("AQA-PHYS-7408", "PHYSICS", "7408", "A-level", "2015", _alevel("7408")),
+    # ── Round 2 breadth — high-volume core (Maths, English) ───────────────────────────
+    ("AQA-MATH-8300", "MATHEMATICS", "8300", "GCSE", "2015", _subj("8300", ("1", "2", "3"), ("F", "H"))),
+    ("AQA-MATH-7357", "MATHEMATICS", "7357", "A-level", "2017", _alevel("7357", ("1", "2", "3"))),
+    ("AQA-MATH-7356", "MATHEMATICS", "7356", "AS", "2017", _alevel("7356", ("1", "2"))),
+    ("AQA-ENGL-8700", "ENGLISH LANGUAGE", "8700", "GCSE", "2015", _subj("8700", ("1", "2"))),
+    ("AQA-ENGLIT-8702", "ENGLISH LITERATURE", "8702", "GCSE", "2015", _subj("8702", ("1", "2"))),
+    ("AQA-ENGL-7702", "ENGLISH LANGUAGE", "7702", "A-level", "2015", _alevel("7702", ("1", "2"))),
+    ("AQA-ENGLIT-7712", "ENGLISH LITERATURE A", "7712", "A-level", "2015", _alevel("7712", ("1", "2"))),
+    # ── Round 2 breadth — humanities / others ─────────────────────────────────────────
+    ("AQA-GEOG-8035", "GEOGRAPHY", "8035", "GCSE", "2016", _subj("8035", ("1", "2", "3"))),
+    ("AQA-GEOG-7037", "GEOGRAPHY", "7037", "A-level", "2016", _alevel("7037", ("1", "2"))),
+    ("AQA-COMP-8525", "COMPUTER SCIENCE", "8525", "GCSE", "2020", _subj("8525", ("1", "2"))),
+    ("AQA-COMP-7517", "COMPUTER SCIENCE", "7517", "A-level", "2015", _alevel("7517", ("1", "2"))),
+    ("AQA-BUS-8132", "BUSINESS", "8132", "GCSE", "2017", _subj("8132", ("1", "2"))),
+    ("AQA-BUS-7132", "BUSINESS", "7132", "A-level", "2015", _alevel("7132", ("1", "2", "3"))),
+    ("AQA-PSYC-8182", "PSYCHOLOGY", "8182", "GCSE", "2017", _subj("8182", ("1", "2"))),
+    ("AQA-PSYC-7182", "PSYCHOLOGY", "7182", "A-level", "2015", _alevel("7182", ("1", "2", "3"))),
+    # ── Round 2 breadth — modern foreign languages (Listening/Reading/Writing, F+H) ───
+    ("AQA-FREN-8658", "FRENCH", "8658", "GCSE", "2016", _mfl("8658")),
+    ("AQA-SPAN-8698", "SPANISH", "8698", "GCSE", "2016", _mfl("8698")),
+    ("AQA-GERM-8668", "GERMAN", "8668", "GCSE", "2016", _mfl("8668")),
+    ("AQA-FREN-7652", "FRENCH", "7652", "A-level", "2016", _alevel("7652", ("1", "2"))),
+    ("AQA-SPAN-7692", "SPANISH", "7692", "A-level", "2016", _alevel("7692", ("1", "2"))),
+    ("AQA-GERM-7662", "GERMAN", "7662", "A-level", "2016", _alevel("7662", ("1", "2"))),
+]
+AQA_SESSIONS = ["JUN18", "JUN19", "NOV20", "NOV21", "JUN22", "JUN23", "JUN24"]
+AQA_ROLES = ["QP", "MS", "ER"]
+
+
+def aqa_url(papercode: str, role: str, session: str) -> Tuple[str, str]:
+    mon = session[:3]
+    yy = session[3:]
+    folder, _ = MONTHS[mon]
+    year = "20" + yy
+    fname = f"AQA-{papercode}-{ROLE_TOKEN[role]}-{session}.PDF"
+    return f"{AQA_BASE}/{year}/{folder}/{fname}", fname
+
+
+def session_pretty(session: str) -> Tuple[str, str]:
+    mon = session[:3]       # "JUN" | "NOV"
+    yy = session[3:]        # "22"
+    _, pretty = MONTHS[mon]
+    # ("2022-Jun" display session, "2022JUN" compact for exam_code — year-first, matches the
+    # locked exam_code convention and the Edexcel/OCR entries).
+    return f"20{yy}-{pretty}", f"20{yy}{mon}"
+
+
+def build_aqa() -> Dict[str, Any]:
+    candidates: List[Tuple[str, str, str, str, str, str, Optional[str], str, str, str]] = []
+    # (spec_code, subject, award, paper_fc, paper_code, tier, role, session, url, fname)
+    spec_meta = {}
+    for spec_code, subject, award, level, first_teach, papers in AQA_SPECS:
+        spec_meta[spec_code] = (subject, award, level, first_teach)
+        for paper_fc, paper_code, tier in papers:
+            for session in AQA_SESSIONS:
+                for role in AQA_ROLES:
+                    url, fname = aqa_url(paper_fc, role, session)
+                    candidates.append((spec_code, subject, award, paper_fc, paper_code, tier,
+                                       role, session, url, fname))
+
+    print(f"[AQA] HEAD-verifying {len(candidates)} candidate URLs...", file=sys.stderr)
+    live: Dict[int, bool] = {}
+    with cf.ThreadPoolExecutor(max_workers=24) as ex:
+        futs = {ex.submit(head_ok, c[8]): i for i, c in enumerate(candidates)}
+        done = 0
+        for fut in cf.as_completed(futs):
+            i = futs[fut]
+            live[i] = fut.result()
+            done += 1
+            if done % 60 == 0:
+                print(f"  ...{done}/{len(candidates)} ({sum(live.values())} live)", file=sys.stderr)
+
+    specs: Dict[str, Dict[str, Any]] = {}
+    for i, c in enumerate(candidates):
+        if not live.get(i):
+            continue
+        spec_code, subject, award, paper_fc, paper_code, tier, role, session, url, fname = c
+        sess_pretty, sess_compact = session_pretty(session)
+        token = paper_fc[len(award):]  # "1H" / "P1H" / "1"
+        exam_code = f"AQA-{award}-{token}-{sess_compact}-{role}"
+        spec = specs.setdefault(spec_code, {"papers": []})
+        spec["papers"].append({
+            "exam_code": exam_code,
+            "paper_code": paper_code,
+            "tier": tier,
+            "session": sess_pretty,
+            "doc_type": role,
+            "file": {
+                "source": f"url:{url}",
+                "original_name": fname,
+                "provenance": {"source_url": url, "fetched": FETCHED,
+                               "license": "AQA public past paper"},
+            },
+        })
+
+    spec_list = []
+    for spec_code, subject, award, level, first_teach, _papers in AQA_SPECS:
+        if spec_code not in specs:
+            continue
+        papers = sorted(specs[spec_code]["papers"], key=lambda p: p["exam_code"])
+        spec_list.append({
+            "spec_code": spec_code, "exam_board_code": "AQA", "subject_code": subject,
+            "award_code": award, "award_level": level, "first_teach": first_teach,
+            "papers": papers,
+        })
+        print(f"[AQA] {spec_code}: {len(papers)} live papers", file=sys.stderr)
+    return {"exam_board_code": "AQA", "specifications": spec_list}
+
+
+# ─────────────── Edexcel / OCR — confirmed direct URLs (re-verified at build) ───────────────
+# These boards aren't templatable (Edexcel has a non-derivable date suffix; OCR uses opaque
+# doc-ids), so confirmed URLs are listed as 6-tuples: (spec_code, paper_code, tier, session, role,
+# url). exam_code is DERIVED (see _mk_exam_code) so it always matches the locked convention.
+EXAM_CODE_PREFIX = {"EDEXCEL": "EDX", "OCR": "OCR"}
+
+def _ec_token(paper_code: str) -> str:
+    t = paper_code.split("/")[-1]
+    return str(int(t)) if t.isdigit() else t   # "01"->"1", "1H"->"1H", "1CH"->"1CH", "11"->"11"
+
+def _mk_exam_code(prefix: str, award: str, paper_code: str, session: str, role: str) -> str:
+    y, m = session.split("-")
+    return f"{prefix}-{award}-{_ec_token(paper_code)}-{y}{m.upper()}-{role}"
+
+_PE = "https://qualifications.pearson.com/content/dam/pdf"
+_EDX = f"{_PE}/GCSE/Science/2016"
+_OCR = "https://www.ocr.org.uk/Images"
+
+EDEXCEL_SPECS = {
+    "EDX-BIOL-1BI0": ("BIOLOGY", "1BI0", "GCSE", "2016"),
+    "EDX-CHEM-1CH0": ("CHEMISTRY", "1CH0", "GCSE", "2016"),
+    "EDX-PHYS-1PH0": ("PHYSICS", "1PH0", "GCSE", "2016"),
+    "EDX-COMB-1SC0": ("COMBINED SCIENCE", "1SC0", "GCSE", "2016"),
+    "EDX-MATH-1MA1": ("MATHEMATICS", "1MA1", "GCSE", "2015"),
+    "EDX-ENGL-1EN0": ("ENGLISH LANGUAGE", "1EN0", "GCSE", "2015"),
+    "EDX-ENGLIT-1ET0": ("ENGLISH LITERATURE", "1ET0", "GCSE", "2015"),
+    "EDX-GEOG-1GA0": ("GEOGRAPHY A", "1GA0", "GCSE", "2016"),
+    "EDX-HIST-1HI0": ("HISTORY", "1HI0", "GCSE", "2016"),
+    "EDX-BUS-1BS0": ("BUSINESS", "1BS0", "GCSE", "2017"),
+    "EDX-COMP-1CP2": ("COMPUTER SCIENCE", "1CP2", "GCSE", "2020"),
+    "EDX-MATH-9MA0": ("MATHEMATICS", "9MA0", "A-level", "2017"),
+    "EDX-ENGL-9EN0": ("ENGLISH LANGUAGE", "9EN0", "A-level", "2015"),
+    "EDX-ENGLIT-9ET0": ("ENGLISH LITERATURE", "9ET0", "A-level", "2015"),
+    "EDX-GEOG-9GE0": ("GEOGRAPHY", "9GE0", "A-level", "2016"),
+}
+EDEXCEL_PAPERS = [
+    # ── Sciences (round 1) ──
+    ("EDX-BIOL-1BI0", "1BI0/1H", "H", "2024-Jun", "QP", f"{_EDX}/Exam-materials/1bi0-1h-que-20240511.pdf"),
+    ("EDX-BIOL-1BI0", "1BI0/2F", "F", "2023-Jun", "QP", f"{_EDX}/Exam-materials/1bi0-2f-que-20230610.pdf"),
+    ("EDX-BIOL-1BI0", "1BI0/2H", "H", "2023-Jun", "QP", f"{_EDX}/Exam-materials/1bi0-2h-que-20230610.pdf"),
+    ("EDX-BIOL-1BI0", "1BI0/1F", "F", "2023-Jun", "MS", f"{_EDX}/Exam-materials/1bi0-1f-rms-20230824.pdf"),
+    ("EDX-BIOL-1BI0", "1BI0/1H", "H", "2024-Jun", "MS", f"{_EDX}/Exam-materials/1bi0-1h-rms-20240822.pdf"),
+    ("EDX-BIOL-1BI0", "1BI0/1H", "H", "2022-Jun", "MS", f"{_EDX}/exam-materials/1bi0-1h-rms-20220825.pdf"),
+    ("EDX-CHEM-1CH0", "1CH0/1F", "F", "2023-Jun", "QP", f"{_EDX}/Exam-materials/1ch0-1f-que-20230523.pdf"),
+    ("EDX-CHEM-1CH0", "1CH0/1H", "H", "2024-Jun", "QP", f"{_EDX}/Exam-materials/1ch0-1h-que-20240518.pdf"),
+    ("EDX-CHEM-1CH0", "1CH0/2H", "H", "2024-Jun", "MS", f"{_EDX}/Exam-materials/1ch0-2h-rms-20240822.pdf"),
+    ("EDX-PHYS-1PH0", "1PH0/1H", "H", "2023-Jun", "QP", f"{_EDX}/Exam-materials/1ph0-1h-que-20230526.pdf"),
+    ("EDX-PHYS-1PH0", "1PH0/2F", "F", "2023-Jun", "QP", f"{_EDX}/Exam-materials/1ph0-2f-que-20230617.pdf"),
+    ("EDX-PHYS-1PH0", "1PH0/1H", "H", "2024-Jun", "QP", f"{_EDX}/Exam-materials/1ph0-1h-que-20240523.pdf"),
+    ("EDX-PHYS-1PH0", "1PH0/2H", "H", "2023-Jun", "MS", f"{_EDX}/Exam-materials/1ph0-2h-rms-20230824.pdf"),
+    ("EDX-PHYS-1PH0", "1PH0/2H", "H", "2022-Jun", "MS", f"{_EDX}/exam-materials/1ph0-2h-rms-20220825.pdf"),
+    ("EDX-COMB-1SC0", "1SC0/1CH", None, "2023-Jun", "MS", f"{_EDX}/Exam-materials/1sc0-1ch-rms-20230824.pdf"),
+    # ── Maths 1MA1 (round 2) ──
+    ("EDX-MATH-1MA1", "1MA1/1H", "H", "2023-Jun", "QP", f"{_PE}/GCSE/Mathematics/2015/Exam-materials/1ma1-1h-que-20230520.pdf"),
+    ("EDX-MATH-1MA1", "1MA1/1H", "H", "2023-Jun", "MS", f"{_PE}/GCSE/Mathematics/2015/Exam-materials/1ma1-1h-rms-20230824.pdf"),
+    ("EDX-MATH-1MA1", "1MA1/1F", "F", "2023-Jun", "MS", f"{_PE}/GCSE/Mathematics/2015/Exam-materials/1ma1-1f-rms-20230824.pdf"),
+    ("EDX-MATH-1MA1", "1MA1/1F", "F", "2024-Jun", "QP", f"{_PE}/GCSE/Mathematics/2015/Exam-materials/1ma1-1f-que-20240517.pdf"),
+    ("EDX-MATH-1MA1", "1MA1/1H", "H", "2024-Jun", "QP", f"{_PE}/GCSE/Mathematics/2015/Exam-materials/1ma1-1h-que-20240517.pdf"),
+    ("EDX-MATH-1MA1", "1MA1/1F", "F", "2024-Jun", "MS", f"{_PE}/GCSE/Mathematics/2015/Exam-materials/1ma1-1f-rms-20240822.pdf"),
+    ("EDX-MATH-1MA1", "1MA1/1H", "H", "2023-Nov", "MS", f"{_PE}/GCSE/Mathematics/2015/Exam-materials/1ma1-1h-rms-20240111.pdf"),
+    ("EDX-MATH-1MA1", "1MA1/1H", "H", "2022-Jun", "MS", f"{_PE}/GCSE/mathematics/2015/exam-materials/1ma1-1h-rms-20220825.pdf"),
+    ("EDX-MATH-1MA1", "1MA1/3H", "H", "2022-Jun", "MS", f"{_PE}/GCSE/mathematics/2015/exam-materials/1ma1-3h-rms-20220825.pdf"),
+    # ── English Language 1EN0 / Literature 1ET0 (round 2) ──
+    ("EDX-ENGL-1EN0", "1EN0/01", None, "2024-Jun", "QP", f"{_PE}/GCSE/English-Language/2015/Exam-materials/1en0-01-que-20240524.pdf"),
+    ("EDX-ENGL-1EN0", "1EN0/01", None, "2023-Nov", "QP", f"{_PE}/GCSE/English-Language/2015/Exam-materials/1en0-01-que-20231108.pdf"),
+    ("EDX-ENGL-1EN0", "1EN0/01", None, "2024-Jun", "MS", f"{_PE}/GCSE/English-Language/2015/Exam-materials/1en0-01-rms-20240822.pdf"),
+    ("EDX-ENGL-1EN0", "1EN0/02", None, "2024-Jun", "MS", f"{_PE}/GCSE/English-Language/2015/Exam-materials/1en0-02-rms-20240822.pdf"),
+    ("EDX-ENGL-1EN0", "1EN0/01", None, "2023-Jun", "MS", f"{_PE}/GCSE/English-Language/2015/Exam-materials/1en0-01-rms-20230824.pdf"),
+    ("EDX-ENGL-1EN0", "1EN0/02", None, "2023-Jun", "MS", f"{_PE}/GCSE/English-Language/2015/Exam-materials/1en0-02-rms-20230824.pdf"),
+    ("EDX-ENGLIT-1ET0", "1ET0/01", None, "2023-Jun", "QP", f"{_PE}/GCSE/English-Literature/2015/Exam-materials/1et0-01-que-20230518.pdf"),
+    ("EDX-ENGLIT-1ET0", "1ET0/02", None, "2023-Jun", "QP", f"{_PE}/GCSE/English-Literature/2015/Exam-materials/1et0-02-que-20230525.pdf"),
+    ("EDX-ENGLIT-1ET0", "1ET0/02", None, "2024-Jun", "QP", f"{_PE}/GCSE/English-Literature/2015/Exam-materials/1et0-02-que-20240521.pdf"),
+    ("EDX-ENGLIT-1ET0", "1ET0/01", None, "2023-Jun", "MS", f"{_PE}/GCSE/English-Literature/2015/Exam-materials/1et0-01-rms-20230824.pdf"),
+    ("EDX-ENGLIT-1ET0", "1ET0/01", None, "2024-Jun", "MS", f"{_PE}/GCSE/English-Literature/2015/Exam-materials/1et0-01-rms-20240822.pdf"),
+    # ── A-level Maths 9MA0 / English 9EN0 / 9ET0 (round 2) ──
+    ("EDX-MATH-9MA0", "9MA0/01", None, "2023-Jun", "QP", f"{_PE}/A-Level/Mathematics/2017/Exam-materials/9ma0-01-que-20230607.pdf"),
+    ("EDX-MATH-9MA0", "9MA0/31", None, "2023-Jun", "QP", f"{_PE}/A-Level/Mathematics/2017/Exam-materials/9ma0-31-que-20230621.pdf"),
+    ("EDX-MATH-9MA0", "9MA0/02", None, "2024-Jun", "QP", f"{_PE}/A-Level/Mathematics/2017/Exam-materials/9ma0-02-que-20240612.pdf"),
+    ("EDX-MATH-9MA0", "9MA0/31", None, "2023-Jun", "MS", f"{_PE}/A-Level/Mathematics/2017/Exam-materials/9ma0-31-rms-20230817.pdf"),
+    ("EDX-MATH-9MA0", "9MA0/01", None, "2024-Jun", "MS", f"{_PE}/A-Level/Mathematics/2017/Exam-materials/9ma0-01-rms-20240815.pdf"),
+    ("EDX-ENGL-9EN0", "9EN0/01", None, "2024-Jun", "MS", f"{_PE}/A-Level/English-Language/2015/Exam-materials/9en0-01-rms-20240815.pdf"),
+    ("EDX-ENGL-9EN0", "9EN0/02", None, "2024-Jun", "MS", f"{_PE}/A-Level/English-Language/2015/Exam-materials/9en0-02-rms-20240815.pdf"),
+    ("EDX-ENGLIT-9ET0", "9ET0/01", None, "2024-Jun", "QP", f"{_PE}/A-Level/English-Literature/2015/Exam-materials/9et0-01-que-20240525.pdf"),
+    ("EDX-ENGLIT-9ET0", "9ET0/01", None, "2023-Jun", "MS", f"{_PE}/A-Level/English-Literature/2015/Exam-materials/9et0-01-rms-20230817.pdf"),
+    ("EDX-ENGLIT-9ET0", "9ET0/03", None, "2023-Jun", "MS", f"{_PE}/A-Level/English-Literature/2015/Exam-materials/9et0-03-rms-20230817.pdf"),
+    # ── Humanities (round 2) ──
+    ("EDX-GEOG-1GA0", "1GA0/01", None, "2023-Jun", "QP", f"{_PE}/GCSE/Geography-A/2016/Exam-materials/1ga0-01-que-20230523.pdf"),
+    ("EDX-GEOG-1GA0", "1GA0/01", None, "2023-Jun", "MS", f"{_PE}/GCSE/Geography-A/2016/Exam-materials/1ga0-01-rms-20230824.pdf"),
+    ("EDX-GEOG-1GA0", "1GA0/02", None, "2023-Jun", "QP", f"{_PE}/GCSE/Geography-A/2016/Exam-materials/1ga0-02-que-20230610.pdf"),
+    ("EDX-GEOG-1GA0", "1GA0/02", None, "2023-Jun", "MS", f"{_PE}/GCSE/Geography-A/2016/Exam-materials/1ga0-02-rms-20230824.pdf"),
+    ("EDX-GEOG-1GA0", "1GA0/01", None, "2024-Jun", "MS", f"{_PE}/GCSE/Geography-A/2016/Exam-materials/1ga0-01-rms-20240822.pdf"),
+    ("EDX-GEOG-1GA0", "1GA0/03", None, "2024-Jun", "QP", f"{_PE}/GCSE/Geography-A/2016/Exam-materials/1ga0-03-que-20240615.pdf"),
+    ("EDX-HIST-1HI0", "1HI0/10", None, "2023-Jun", "QP", f"{_PE}/GCSE/History/2016/Exam-materials/1hi0-10-que-20230519.pdf"),
+    ("EDX-HIST-1HI0", "1HI0/10", None, "2023-Jun", "MS", f"{_PE}/GCSE/History/2016/Exam-materials/1hi0-10-rms-20230824.pdf"),
+    ("EDX-HIST-1HI0", "1HI0/12", None, "2023-Jun", "MS", f"{_PE}/GCSE/History/2016/Exam-materials/1hi0-12-rms-20230824.pdf"),
+    ("EDX-HIST-1HI0", "1HI0/13", None, "2024-Jun", "MS", f"{_PE}/GCSE/History/2016/Exam-materials/1hi0-13-rms-20240822.pdf"),
+    ("EDX-HIST-1HI0", "1HI0/33", None, "2023-Jun", "MS", f"{_PE}/GCSE/History/2016/Exam-materials/1hi0-33-rms-20230824.pdf"),
+    ("EDX-BUS-1BS0", "1BS0/01", None, "2023-Jun", "QP", f"{_PE}/GCSE/Business/2017/Exam-materials/1bs0-01-que-20230519.pdf"),
+    ("EDX-BUS-1BS0", "1BS0/02", None, "2023-Jun", "QP", f"{_PE}/GCSE/Business/2017/Exam-materials/1bs0-02-que-20230613.pdf"),
+    ("EDX-BUS-1BS0", "1BS0/02", None, "2023-Jun", "MS", f"{_PE}/GCSE/Business/2017/Exam-materials/1bs0-02-rms-20230824.pdf"),
+    ("EDX-BUS-1BS0", "1BS0/02", None, "2024-Jun", "QP", f"{_PE}/GCSE/Business/2017/Exam-materials/1bs0-02-que-20240606.pdf"),
+    ("EDX-BUS-1BS0", "1BS0/01", None, "2024-Jun", "MS", f"{_PE}/GCSE/Business/2017/Exam-materials/1bs0-01-rms-20240822.pdf"),
+    ("EDX-COMP-1CP2", "1CP2/01", None, "2023-Jun", "QP", f"{_PE}/GCSE/Computer-science/2020/Exam-materials/1cp2-01-que-20230520.pdf"),
+    ("EDX-COMP-1CP2", "1CP2/01", None, "2023-Jun", "MS", f"{_PE}/GCSE/Computer-science/2020/Exam-materials/1cp2-01-rms-20230824.pdf"),
+    ("EDX-COMP-1CP2", "1CP2/02", None, "2023-Jun", "QP", f"{_PE}/GCSE/Computer-science/2020/Exam-materials/1cp2-02-que-20230526.pdf"),
+    ("EDX-COMP-1CP2", "1CP2/01", None, "2024-Jun", "QP", f"{_PE}/GCSE/Computer-Science/2020/Exam-materials/1cp2-01-que-20240702.pdf"),
+    ("EDX-COMP-1CP2", "1CP2/01", None, "2024-Jun", "MS", f"{_PE}/GCSE/Computer-science/2020/Exam-materials/1cp2-01-rms-20240822.pdf"),
+    ("EDX-GEOG-9GE0", "9GE0/01", None, "2023-Jun", "QP", f"{_PE}/A-Level/Geography/2016/Exam-materials/9ge0-01-que-20230518.pdf"),
+]
+
+OCR_SPECS = {
+    "OCR-BIOL-J247": ("BIOLOGY", "J247", "GCSE", "2016"),
+    "OCR-CHEM-J248": ("CHEMISTRY", "J248", "GCSE", "2016"),
+    "OCR-PHYS-J249": ("PHYSICS", "J249", "GCSE", "2016"),
+    "OCR-COMB-J250": ("COMBINED SCIENCE", "J250", "GCSE", "2016"),
+    "OCR-MATH-J560": ("MATHEMATICS", "J560", "GCSE", "2015"),
+    "OCR-ENGL-J351": ("ENGLISH LANGUAGE", "J351", "GCSE", "2015"),
+    "OCR-ENGLIT-J352": ("ENGLISH LITERATURE", "J352", "GCSE", "2015"),
+    "OCR-COMP-J277": ("COMPUTER SCIENCE", "J277", "GCSE", "2020"),
+    "OCR-GEOG-J383": ("GEOGRAPHY A", "J383", "GCSE", "2016"),
+    "OCR-BUS-J204": ("BUSINESS", "J204", "GCSE", "2017"),
+    "OCR-HIST-J411": ("HISTORY B (SHP)", "J411", "GCSE", "2016"),
+    "OCR-MATH-H240": ("MATHEMATICS A", "H240", "A-level", "2017"),
+    "OCR-ENGLIT-H472": ("ENGLISH LITERATURE", "H472", "A-level", "2015"),
+    "OCR-ENGL-H470": ("ENGLISH LANGUAGE", "H470", "A-level", "2015"),
+}
+OCR_PAPERS = [
+    # ── Sciences (round 1) ──
+    ("OCR-BIOL-J247", "J247/01", "F", "2024-Jun", "QP", f"{_OCR}/727713-question-paper-paper-1.pdf"),
+    ("OCR-BIOL-J247", "J247/01", "F", "2024-Jun", "MS", f"{_OCR}/727745-mark-scheme-paper-1.pdf"),
+    ("OCR-BIOL-J247", "J247/03", "H", "2024-Jun", "QP", f"{_OCR}/727715-question-paper-paper-3.pdf"),
+    ("OCR-BIOL-J247", "J247/03", "H", "2024-Jun", "MS", f"{_OCR}/727747-mark-scheme-paper-3.pdf"),
+    ("OCR-BIOL-J247", "J247/01", "F", "2023-Jun", "QP", f"{_OCR}/704945-question-paper-paper-1.pdf"),
+    ("OCR-BIOL-J247", "J247/03", "H", "2023-Jun", "MS", f"{_OCR}/704979-mark-scheme-paper-3.pdf"),
+    ("OCR-BIOL-J247", "J247/03", "H", "2022-Jun", "QP", f"{_OCR}/678031-question-paper-paper-3.pdf"),
+    ("OCR-BIOL-J247", "J247/01", "F", "2022-Jun", "MS", f"{_OCR}/678076-mark-scheme-paper-1.pdf"),
+    ("OCR-CHEM-J248", "J248/01", "F", "2024-Jun", "QP", f"{_OCR}/727718-question-paper-paper-1.pdf"),
+    ("OCR-CHEM-J248", "J248/03", "H", "2024-Jun", "MS", f"{_OCR}/727751-mark-scheme-paper-3.pdf"),
+    ("OCR-CHEM-J248", "J248/01", "F", "2023-Jun", "QP", f"{_OCR}/704950-question-paper-paper-1.pdf"),
+    ("OCR-CHEM-J248", "J248/03", "H", "2022-Jun", "QP", f"{_OCR}/678036-question-paper-paper-3.pdf"),
+    ("OCR-PHYS-J249", "J249/01", "F", "2024-Jun", "QP", f"{_OCR}/727724-question-paper-paper-1.pdf"),
+    ("OCR-PHYS-J249", "J249/03", "H", "2024-Jun", "MS", f"{_OCR}/727755-mark-scheme-paper-3.pdf"),
+    ("OCR-PHYS-J249", "J249/01", "F", "2023-Jun", "QP", f"{_OCR}/704956-question-paper-paper-1.pdf"),
+    ("OCR-PHYS-J249", "J249/03", "H", "2022-Jun", "MS", f"{_OCR}/678086-mark-scheme-paper-3.pdf"),
+    ("OCR-COMB-J250", "J250/01", "F", "2024-Jun", "QP", f"{_OCR}/727730-question-paper-paper-1.pdf"),
+    ("OCR-COMB-J250", "J250/07", "H", "2024-Jun", "MS", f"{_OCR}/727763-mark-scheme-paper-7.pdf"),
+    # ── Maths J560 (round 2) ──
+    ("OCR-MATH-J560", "J560/01", "F", "2024-Jun", "QP", f"{_OCR}/727817-question-paper-paper-1.pdf"),
+    ("OCR-MATH-J560", "J560/01", "F", "2024-Jun", "MS", f"{_OCR}/727824-mark-scheme-paper-1.pdf"),
+    ("OCR-MATH-J560", "J560/04", "H", "2024-Jun", "QP", f"{_OCR}/727820-question-paper-paper-4.pdf"),
+    ("OCR-MATH-J560", "J560/04", "H", "2024-Jun", "MS", f"{_OCR}/727827-mark-scheme-paper-4.pdf"),
+    ("OCR-MATH-J560", "J560/01", "F", "2023-Jun", "QP", f"{_OCR}/705050-question-paper-paper-1.pdf"),
+    ("OCR-MATH-J560", "J560/01", "F", "2023-Jun", "MS", f"{_OCR}/705057-mark-scheme-paper-1.pdf"),
+    ("OCR-MATH-J560", "J560/04", "H", "2023-Jun", "QP", f"{_OCR}/705053-question-paper-paper-4.pdf"),
+    ("OCR-MATH-J560", "J560/04", "H", "2023-Jun", "MS", f"{_OCR}/705060-mark-scheme-paper-4.pdf"),
+    ("OCR-MATH-J560", "J560/01", "F", "2022-Jun", "QP", f"{_OCR}/678149-question-paper-paper-1.pdf"),
+    ("OCR-MATH-J560", "J560/01", "F", "2022-Jun", "MS", f"{_OCR}/678156-mark-scheme-paper-1.pdf"),
+    ("OCR-MATH-J560", "J560/04", "H", "2022-Jun", "QP", f"{_OCR}/678152-question-paper-paper-4.pdf"),
+    ("OCR-MATH-J560", "J560/04", "H", "2022-Jun", "MS", f"{_OCR}/678159-mark-scheme-paper-4.pdf"),
+    # ── English Language J351 / Literature J352 (round 2) ──
+    ("OCR-ENGL-J351", "J351/01", None, "2024-Jun", "QP", f"{_OCR}/727556-question-paper-communicating-information-and-ideas.pdf"),
+    ("OCR-ENGL-J351", "J351/01", None, "2024-Jun", "MS", f"{_OCR}/727658-mark-scheme-communication-information-and-ideas.pdf"),
+    ("OCR-ENGL-J351", "J351/02", None, "2024-Jun", "QP", f"{_OCR}/727558-question-paper-exploring-effects-and-impact.pdf"),
+    ("OCR-ENGL-J351", "J351/02", None, "2024-Jun", "MS", f"{_OCR}/727659-mark-scheme-exploring-effects-and-impact.pdf"),
+    ("OCR-ENGL-J351", "J351/01", None, "2023-Jun", "QP", f"{_OCR}/704782-question-paper-communicating-information-and-ideas.pdf"),
+    ("OCR-ENGL-J351", "J351/01", None, "2023-Jun", "MS", f"{_OCR}/704888-mark-scheme-communication-information-and-ideas.pdf"),
+    ("OCR-ENGL-J351", "J351/01", None, "2022-Jun", "QP", f"{_OCR}/677852-question-paper-communicating-information-and-ideas.pdf"),
+    ("OCR-ENGL-J351", "J351/01", None, "2022-Jun", "MS", f"{_OCR}/677967-mark-scheme-communication-information-and-ideas.pdf"),
+    ("OCR-ENGLIT-J352", "J352/01", None, "2024-Jun", "QP", f"{_OCR}/727830-question-paper-exploring-modern-and-literary-heritage-texts.pdf"),
+    ("OCR-ENGLIT-J352", "J352/01", None, "2024-Jun", "MS", f"{_OCR}/727832-mark-scheme-exploring-modern-and-literary-heritage-texts.pdf"),
+    ("OCR-ENGLIT-J352", "J352/02", None, "2024-Jun", "QP", f"{_OCR}/727831-question-paper-exploring-poetry-and-shakespeare.pdf"),
+    ("OCR-ENGLIT-J352", "J352/02", None, "2024-Jun", "MS", f"{_OCR}/727833-mark-scheme-exploring-poetry-and-shakespeare.pdf"),
+    ("OCR-ENGLIT-J352", "J352/01", None, "2023-Jun", "QP", f"{_OCR}/705069-question-paper-exploring-modern-and-literary-heritage-texts.pdf"),
+    ("OCR-ENGLIT-J352", "J352/01", None, "2023-Jun", "MS", f"{_OCR}/705075-mark-scheme-exploring-modern-and-literary-heritage-texts.pdf"),
+    # ── A-level Maths H240 / English Lit H472 / Lang H470 (round 2) ──
+    ("OCR-MATH-H240", "H240/01", None, "2024-Jun", "QP", f"{_OCR}/726654-question-paper-pure-mathematics.pdf"),
+    ("OCR-MATH-H240", "H240/01", None, "2024-Jun", "MS", f"{_OCR}/726795-mark-scheme-pure-mathematics.pdf"),
+    ("OCR-MATH-H240", "H240/02", None, "2024-Jun", "QP", f"{_OCR}/726656-question-paper-pure-mathematics-and-statistics.pdf"),
+    ("OCR-MATH-H240", "H240/02", None, "2024-Jun", "MS", f"{_OCR}/726796-mark-scheme-pure-mathematics-and-statistics.pdf"),
+    ("OCR-MATH-H240", "H240/01", None, "2023-Jun", "QP", f"{_OCR}/703866-question-paper-pure-mathematics.pdf"),
+    ("OCR-MATH-H240", "H240/01", None, "2023-Jun", "MS", f"{_OCR}/704008-mark-scheme-pure-mathematics.pdf"),
+    ("OCR-MATH-H240", "H240/01", None, "2022-Jun", "QP", f"{_OCR}/676845-question-paper-pure-mathematics.pdf"),
+    ("OCR-MATH-H240", "H240/01", None, "2022-Jun", "MS", f"{_OCR}/677005-mark-scheme-pure-mathematics.pdf"),
+    ("OCR-ENGLIT-H472", "H472/01", None, "2024-Jun", "QP", f"{_OCR}/726602-question-paper-drama-and-poetry-pre-1900.pdf"),
+    ("OCR-ENGLIT-H472", "H472/01", None, "2024-Jun", "MS", f"{_OCR}/726762-mark-scheme-drama-and-poetry-pre-1900.pdf"),
+    ("OCR-ENGLIT-H472", "H472/01", None, "2023-Jun", "QP", f"{_OCR}/703813-question-paper-drama-and-poetry-pre-1900.pdf"),
+    ("OCR-ENGLIT-H472", "H472/01", None, "2023-Jun", "MS", f"{_OCR}/703974-mark-scheme-drama-and-poetry-pre-1900.pdf"),
+    ("OCR-ENGLIT-H472", "H472/01", None, "2022-Jun", "QP", f"{_OCR}/676783-question-paper-drama-and-poetry-pre-1900.pdf"),
+    ("OCR-ENGLIT-H472", "H472/01", None, "2022-Jun", "MS", f"{_OCR}/676965-mark-scheme-drama-and-poetry-pre-1900.pdf"),
+    ("OCR-ENGL-H470", "H470/01", None, "2024-Jun", "QP", f"{_OCR}/726595-question-paper-exploring-language.pdf"),
+    ("OCR-ENGL-H470", "H470/01", None, "2024-Jun", "MS", f"{_OCR}/726764-mark-scheme-exploring-language.pdf"),
+    ("OCR-ENGL-H470", "H470/01", None, "2023-Jun", "QP", f"{_OCR}/703806-question-paper-exploring-language.pdf"),
+    ("OCR-ENGL-H470", "H470/01", None, "2023-Jun", "MS", f"{_OCR}/703976-mark-scheme-exploring-language.pdf"),
+    ("OCR-ENGL-H470", "H470/01", None, "2022-Jun", "QP", f"{_OCR}/676772-question-paper-exploring-language.pdf"),
+    ("OCR-ENGL-H470", "H470/01", None, "2022-Jun", "MS", f"{_OCR}/676967-mark-scheme-exploring-language.pdf"),
+    # ── Humanities (round 2) ──
+    ("OCR-COMP-J277", "J277/01", None, "2024-Jun", "QP", f"{_OCR}/727534-question-paper-computer-systems.pdf"),
+    ("OCR-COMP-J277", "J277/01", None, "2024-Jun", "MS", f"{_OCR}/727652-mark-scheme-computer-systems.pdf"),
+    ("OCR-COMP-J277", "J277/02", None, "2024-Jun", "QP", f"{_OCR}/727535-question-paper-computational-thinking-algorithms-and-programming.pdf"),
+    ("OCR-COMP-J277", "J277/02", None, "2024-Jun", "MS", f"{_OCR}/727653-mark-scheme-computational-thinking-algorithms-and-programming.pdf"),
+    ("OCR-GEOG-J383", "J383/01", None, "2024-Jun", "QP", f"{_OCR}/727564-question-paper-living-in-the-uk-today.pdf"),
+    ("OCR-GEOG-J383", "J383/01", None, "2024-Jun", "MS", f"{_OCR}/727661-mark-scheme-living-in-the-uk-today.pdf"),
+    ("OCR-GEOG-J383", "J383/02", None, "2024-Jun", "QP", f"{_OCR}/727566-question-paper-the-world-around-us.pdf"),
+    ("OCR-GEOG-J383", "J383/02", None, "2024-Jun", "MS", f"{_OCR}/727662-mark-scheme-the-world-around-us.pdf"),
+    ("OCR-BUS-J204", "J204/01", None, "2024-Jun", "QP", f"{_OCR}/727519-question-paper-business-1-business-activity-marketing-and-people.pdf"),
+    ("OCR-BUS-J204", "J204/01", None, "2024-Jun", "MS", f"{_OCR}/727634-mark-scheme-business-1-business-activity-marketing-and-people.pdf"),
+    ("OCR-BUS-J204", "J204/02", None, "2024-Jun", "QP", f"{_OCR}/727520-question-paper-business-2-operations-finance-and-influences-on-business.pdf"),
+    ("OCR-BUS-J204", "J204/02", None, "2024-Jun", "MS", f"{_OCR}/727635-mark-scheme-business-2-operations-finance-and-influences-on-business.pdf"),
+    ("OCR-BUS-J204", "J204/01", None, "2023-Jun", "QP", f"{_OCR}/704745-question-paper-business-1-business-activity-marketing-and-people.pdf"),
+    ("OCR-BUS-J204", "J204/01", None, "2023-Jun", "MS", f"{_OCR}/704864-mark-scheme-business-1-business-activity-marketing-and-people.pdf"),
+    ("OCR-HIST-J411", "J411/11", None, "2024-Jun", "QP", f"{_OCR}/727590-question-paper-the-people-s-health-c.1250-to-present-with-the-norman-conquest-1065-1087.pdf"),
+    ("OCR-HIST-J411", "J411/11", None, "2024-Jun", "MS", f"{_OCR}/727678-mark-scheme-the-people-s-health-c.1250-to-present-with-the-norman-conquest-1065-1087.pdf"),
+]
+
+
+def build_board(board_code: str, specs_meta: Dict, papers: List) -> Dict[str, Any]:
+    prefix = EXAM_CODE_PREFIX[board_code]
+    print(f"[{board_code}] re-verifying {len(papers)} confirmed URLs...", file=sys.stderr)
+    live: Dict[int, bool] = {}
+    with cf.ThreadPoolExecutor(max_workers=24) as ex:
+        futs = {ex.submit(head_ok, p[5]): i for i, p in enumerate(papers)}
+        for fut in cf.as_completed(futs):
+            live[futs[fut]] = fut.result()
+    by_spec: Dict[str, List[Dict[str, Any]]] = {}
+    for i, (spec_code, paper_code, tier, session, role, url) in enumerate(papers):
+        if not live.get(i):
+            print(f"  DROP (not live): {url}", file=sys.stderr)
+            continue
+        award = specs_meta[spec_code][1]
+        by_spec.setdefault(spec_code, []).append({
+            "exam_code": _mk_exam_code(prefix, award, paper_code, session, role),
+            "paper_code": paper_code, "tier": tier,
+            "session": session, "doc_type": role,
+            "file": {"source": f"url:{url}", "original_name": os.path.basename(url),
+                     "provenance": {"source_url": url, "fetched": FETCHED,
+                                    "license": f"{board_code} public past paper"}},
+        })
+    spec_list = []
+    for spec_code, (subject, award, level, first_teach) in specs_meta.items():
+        if spec_code not in by_spec:
+            continue
+        spec_list.append({
+            "spec_code": spec_code, "exam_board_code": board_code, "subject_code": subject,
+            "award_code": award, "award_level": level, "first_teach": first_teach,
+            "papers": sorted(by_spec[spec_code], key=lambda p: p["exam_code"]),
+        })
+        print(f"[{board_code}] {spec_code}: {len(by_spec[spec_code])} live papers", file=sys.stderr)
+    return {"exam_board_code": board_code, "specifications": spec_list}
+
+
+def main() -> None:
+    out_path = os.path.join(os.path.dirname(__file__), "exam-corpus.yaml")
+    boards = [
+        build_aqa(),
+        build_board("EDEXCEL", EDEXCEL_SPECS, EDEXCEL_PAPERS),
+        build_board("OCR", OCR_SPECS, OCR_PAPERS),
+    ]
+    n_specs = sum(len(b["specifications"]) for b in boards)
+    n_papers = sum(len(s["papers"]) for b in boards for s in b["specifications"])
+    manifest = {
+        "version": 1,
+        "defaults": {"bucket": "cc.examboards"},
+        "provenance": {
+            "collected_by": "kcar",
+            "collected_at": FETCHED,
+            "license_posture": ("Public exam-board past papers downloaded from each board's own "
+                                "official site (AQA filestore, Pearson DAM, OCR Images). Stored in "
+                                "the private dev cc.examboards bucket for internal exam-marker dev/test. "
+                                "Each item records its source_url. Review redistribution rights before "
+                                "any public exposure."),
+            "sources": {
+                "AQA": "https://filestore.aqa.org.uk/sample-papers-and-mark-schemes/",
+                "EDEXCEL": "https://qualifications.pearson.com/en/support/support-topics/exams/past-papers.html",
+                "OCR": "https://www.ocr.org.uk/qualifications/past-paper-finder/",
+            },
+        },
+        # Optional: uncomment + set on dev .94 to exercise user-side flows / first-sweep.
+        # "test_subset": {"user_email": "[email protected]", "papers": 2},
+        # "system_identity": {"user_email": "[email protected]"},
+        "boards": boards,
+    }
+    with open(out_path, "w") as fh:
+        yaml.safe_dump(manifest, fh, sort_keys=False, default_flow_style=False, width=120)
+    print(f"\nWROTE {out_path}: {n_specs} specs, {n_papers} papers across {len(boards)} boards",
+          file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

run/initialization/reset_environment.py

@@ -82,6 +82,41 @@ SUPABASE_TABLES_TO_CLEAR = [
     "admin_profiles",
 ]
 
+# Exam subsystem tables, FK child-first. NOT in the list above — the previous full reset()
+# never cleared exam data or storage at all; the granular scopes below fold it in.
+EXAM_CORPUS_TABLES = [
+    "mark_entries",
+    "student_submissions",
+    "marking_batches",
+    "exam_response_areas",
+    "exam_boundaries",
+    "exam_template_layout",
+    "exam_questions",
+    "exam_templates",
+    "eb_exams",
+    "eb_specifications",
+]
+
+# Timetable / calendar materialization subset (for scope='timetable').
+TIMETABLE_TABLES = [
+    "lesson_deliveries",
+    "lesson_collaborators",
+    "taught_lessons",
+    "academic_periods",
+    "academic_days",
+    "academic_weeks",
+    "academic_term_breaks",
+    "academic_terms",
+    "academic_years",
+    "teacher_timetable_slots",
+    "teacher_timetables",
+    "school_timetables",
+    "planned_lessons",
+]
+
+# Buckets whose objects the exam-corpus reset clears (Storage API — protect_delete blocks raw SQL).
+EXAM_STORAGE_BUCKET = "cc.examboards"
+
 
 def _sb_headers():
     url = os.environ["SUPABASE_URL"]
@@ -146,13 +181,84 @@ def _supabase_delete_auth_user(url: str, headers: dict, uid: str):
         logger.warning(f"  Delete auth user {uid}: {r.status_code} {r.text[:80]}")
 
 
+# ─── Granular helpers ───────────────────────────────────────────────────────────
+
+def _clear_tables(url: str, headers: dict, tables: List[str]) -> "tuple[List[str], List[str]]":
+    cleared, failed = [], []
+    for table in tables:
+        if _sb_clear_table(url, headers, table) in (200, 204):
+            cleared.append(table)
+            logger.info(f"  ✓ {table}")
+        else:
+            failed.append(table)
+    return cleared, failed
+
+
+def _clear_exam_storage() -> Dict[str, Any]:
+    """Remove cc.examboards objects via the Storage API (protect_delete blocks raw SQL deletes).
+    Gathers storage_loc from eb_exams/eb_specifications BEFORE the rows are cleared."""
+    try:
+        from modules.database.supabase.utils.client import SupabaseServiceRoleClient
+        from modules.database.supabase.utils.storage import StorageAdmin
+    except Exception as exc:
+        logger.warning(f"  exam storage clear skipped (import): {exc}")
+        return {"removed": 0, "error": str(exc)}
+    sb = SupabaseServiceRoleClient().supabase
+    storage = StorageAdmin()
+    locs: List[str] = []
+    for table in ("eb_exams", "eb_specifications"):
+        try:
+            rows = sb.table(table).select("storage_loc").execute().data or []
+            locs += [r["storage_loc"] for r in rows if r.get("storage_loc")]
+        except Exception as exc:
+            logger.warning(f"  storage_loc gather {table}: {exc}")
+    by_bucket: Dict[str, List[str]] = {}
+    for loc in locs:
+        if "/" in loc:
+            b, _, p = loc.partition("/")
+            by_bucket.setdefault(b, []).append(p)
+    removed = 0
+    for b, paths in by_bucket.items():
+        for i in range(0, len(paths), 100):
+            chunk = paths[i:i + 100]
+            try:
+                storage.client.supabase.storage.from_(b).remove(chunk)
+                removed += len(chunk)
+            except Exception as exc:
+                logger.warning(f"  storage remove {b}: {exc}")
+    logger.info(f"  exam storage removed {removed} objects from {list(by_bucket)}")
+    return {"removed": removed, "buckets": list(by_bucket)}
+
+
 # ─── Main reset ───────────────────────────────────────────────────────────────
 
-def reset() -> Dict[str, Any]:
+def reset(scope: str = "all") -> Dict[str, Any]:
+    """Destructive reset. scope ∈ {all, exam-corpus, timetable}.
+
+    - all          : full wipe (Neo4j + Supabase data + auth users) AND the exam subsystem + storage.
+    - exam-corpus  : ONLY eb_*/exam_* tables + cc.examboards storage objects (load/unload the corpus).
+    - timetable    : ONLY timetable/calendar materialization tables.
+    """
+    scope = (scope or "all").lower()
+    if scope not in ("all", "exam-corpus", "timetable"):
+        raise ValueError(f"invalid scope {scope!r} (want all|exam-corpus|timetable)")
+    url, headers = _sb_headers()
+
+    if scope == "exam-corpus":
+        logger.info("RESET (scope=exam-corpus) — exam tables + cc.examboards storage")
+        storage = _clear_exam_storage()
+        cleared, failed = _clear_tables(url, headers, EXAM_CORPUS_TABLES)
+        return {"scope": scope, "exam_storage": storage, "tables_cleared": cleared, "tables_failed": failed}
+
+    if scope == "timetable":
+        logger.info("RESET (scope=timetable) — timetable/calendar tables")
+        cleared, failed = _clear_tables(url, headers, TIMETABLE_TABLES)
+        return {"scope": scope, "tables_cleared": cleared, "tables_failed": failed}
+
     logger.info("=" * 60)
     logger.info("RESET ENVIRONMENT — full destructive wipe starting")
     logger.info("=" * 60)
-    results: Dict[str, Any] = {}
+    results: Dict[str, Any] = {"scope": scope}
 
     # ── 1. Neo4j: drop everything except system + neo4j ──────────────────────
     logger.info("\n[Neo4j] Dropping all non-system databases...")
@@ -213,11 +319,22 @@ def reset() -> Dict[str, Any]:
     )
     logger.info("  kcar → admin_profiles restored ✓")
 
+    # ── 5. Exam subsystem: storage objects (Storage API) + exam tables ───────────
+    # (The legacy full reset cleared neither exam tables nor storage — folded in here.)
+    logger.info("\n[Supabase] Clearing exam subsystem (storage + eb_*/exam_* tables)...")
+    exam_storage = _clear_exam_storage()
+    exam_cleared, exam_failed = _clear_tables(url, headers, EXAM_CORPUS_TABLES)
+
     results["supabase"] = {
         "tables_cleared": cleared,
         "tables_failed": failed,
         "deleted_users": deleted_emails,
     }
+    results["exam"] = {
+        "storage": exam_storage,
+        "tables_cleared": exam_cleared,
+        "tables_failed": exam_failed,
+    }
 
     logger.info("\n" + "=" * 60)
     logger.info("RESET COMPLETE")

run/initialization/seed_curriculum.py

@@ -1,15 +1,20 @@
 """
-seed_curriculum.py — Create curriculum data: exam board specifications and exams.
+seed_curriculum.py — DEPRECATED hardcoded curriculum/exam seeder.
 
-Seeds eb_specifications and eb_exams tables with realistic UK exam board data
-(AQA, Edexcel, OCR) for Physics, Maths, and Computer Science across both schools.
+⚠️ SUPERSEDED (2026-06-07) by the manifest-driven corpus loader:
+    run/initialization/seed_exam_corpus.py  (+ manifests/exam-corpus.yaml)
 
-Also seeds curriculum_topics in Neo4j for the school databases.
+The exam-board parts of this file (eb_specifications / eb_exams) are now seeded from a
+verified, provenance-bearing manifest with real uploaded PDFs — not the hardcoded rows
+below. This module also had a storage_loc inconsistency the overhaul standardises away:
+exam-board files belong in the `cc.examboards` bucket at the canonical path
+`cc.examboards/{board}/{subject}/{award}/{paper}/{session}/{role}.pdf`, NOT under
+`cc.public.snapshots/curriculum/...` (the placeholder rows below still show the old path).
 
-Tables: eb_specifications, eb_exams
-Neo4j: curriculum topic nodes in school databases
+KEEP ONLY for the Neo4j `curriculum_topics` seed (step [3]) which has no replacement yet.
+Do NOT use the eb_specifications/eb_exams blocks for new work — use seed_exam_corpus.py.
 
-Run inside ccapi container:
+Run (Neo4j curriculum topics only is the supported remaining use):
     python3 -c "from run.initialization.seed_curriculum import seed; seed()"
 """
 import os

run/initialization/seed_exam_corpus.py

@@ -0,0 +1,778 @@
+"""
+seed_exam_corpus.py — manifest-driven loader for the public exam-paper corpus.
+
+SCOPE (separate from infra): assumes storage buckets already exist (provisioned by
+run/initialization/buckets.py during infra init). This loader UPLOADS papers and
+SEEDS the catalogue; it does NOT create buckets.
+
+Pipeline per manifest item:
+  validate -> resolve source bytes (local path | url:, cached) -> upload file to
+  cc.examboards (canonical path, skip-if-exists unless --force) -> upsert
+  eb_specifications / eb_exams (catalogue) -> (optional, --user-subset) copy a subset
+  into a test user's exam space so user-side flows are testable -> (optional,
+  --first-sweep) run the docling/auto-map first pass to gather structure.
+
+Manifest template: ~/cc/specs/exam-corpus-manifest.example.yaml
+
+Catalogue columns (real — verified against volumes/db/cc/61-core-schema.sql):
+  eb_specifications(spec_code UNIQUE, exam_board_code, award_code, subject_code,
+                    first_teach, spec_ver, storage_loc, doc_type CHECK(pdf|json|...),
+                    doc_details jsonb, docling_docs jsonb)
+  eb_exams(exam_code UNIQUE, spec_code FK, paper_code, tier, session, type_code,
+           storage_loc, doc_type CHECK(pdf|json|...), doc_details jsonb, docling_docs jsonb)
+
+IMPORTANT schema note: the QP/MS/INSERT/ER *document role* is stored in `type_code`
+(the `/catalogue` endpoint filters `type_code == 'QP'`). The `doc_type` column is the
+*file format* and is CHECK-constrained to {pdf,json,md,html,txt,doctags} — so it is
+always 'pdf' here. (The manifest field is named `doc_type` for the role; the loader
+maps manifest.doc_type -> DB.type_code and sets DB.doc_type = 'pdf'.)
+
+Locked conventions (see ~/cc/ideas/2026-06-07-exam-paper-ingestion.md):
+  session    = "YYYY-Mon"  e.g. "2022-Jun", "2021-Nov"
+  exam_code  = "{BOARD}-{award}-{paper_safe}-{SESSIONCOMPACT}-{ROLE}"  e.g. AQA-8463-1H-2022JUN-QP
+  spec path  = cc.examboards/{board}/{subject}/{award}/spec/{spec_ver}.pdf
+  paper path = cc.examboards/{board}/{subject}/{award}/{paper_safe}/{session}/{role}.pdf
+
+Run inside the api container (env: SUPABASE_URL + SERVICE_ROLE_KEY for dev .94), e.g.:
+  python3 -m run.initialization.seed_exam_corpus --manifest /path/exam-corpus.yaml --dry-run
+  python3 -m run.initialization.seed_exam_corpus --manifest ... --board AQA
+  python3 -m run.initialization.seed_exam_corpus --manifest ... --first-sweep
+"""
+from __future__ import annotations
+import argparse
+import hashlib
+import os
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Tuple
+
+import requests
+import yaml  # PyYAML
+
+from modules.logger_tool import initialise_logger
+from modules.database.supabase.utils.client import SupabaseServiceRoleClient
+from modules.database.supabase.utils.storage import StorageAdmin, StorageError
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+
+EXAM_BUCKET = "cc.examboards"
+# Manifest `doc_type` carries the document ROLE (stored in eb_exams.type_code).
+DOC_ROLES = {"QP", "MS", "INSERT", "ER", "SPECIMEN", "GRADE_BOUNDARIES", "DATA_SHEET"}
+TIERS = {"H", "F", None}
+# Default working dir for cached url: downloads (override with --cache-dir / EXAM_CORPUS_CACHE).
+DEFAULT_CACHE_DIR = os.getenv("EXAM_CORPUS_CACHE", "/tmp/exam-corpus-cache")
+# Persistent, mountable local store laid out exactly like the bucket (download once, seed many,
+# offline-repeatable). Override with --store-dir / EXAM_CORPUS_STORE. Distinct from --cache-dir,
+# which is a throwaway url hash-cache.
+DEFAULT_STORE_DIR = os.getenv(
+    "EXAM_CORPUS_STORE",
+    os.path.join(os.path.dirname(os.path.abspath(__file__)), "manifests", "_corpus_store"),
+)
+
+
+# ─────────────────────────────── canonical storage paths ───────────────────────────────
+def _lc(s: str) -> str:
+    return (s or "").strip().lower().replace(" ", "-")
+
+def _paper_safe(paper_code: str) -> str:
+    # Drop the award prefix, keep all remaining segments so combined-science sub-papers
+    # don't collide on the storage path:
+    #   "8463/1H"    -> "1h"
+    #   "8464/B/1H"  -> "b-1h"   (Trilogy: subject letter + paper + tier)
+    #   "7408/1"     -> "1"
+    parts = _lc(paper_code).split("/")
+    return "-".join(parts[1:]) if len(parts) > 1 else parts[0]
+
+def spec_storage_loc(board: str, subject: str, award: str, spec_ver: str) -> str:
+    # e.g. cc.examboards/aqa/physics/8463/spec/1.1.pdf
+    return f"{EXAM_BUCKET}/{_lc(board)}/{_lc(subject)}/{_lc(award)}/spec/{_lc(spec_ver or 'spec')}.pdf"
+
+def paper_storage_loc(board: str, subject: str, award: str, paper_code: str, session: str, doc_role: str) -> str:
+    # e.g. cc.examboards/aqa/physics/8463/1h/2022-jun/qp.pdf
+    return f"{EXAM_BUCKET}/{_lc(board)}/{_lc(subject)}/{_lc(award)}/{_paper_safe(paper_code)}/{_lc(session)}/{_lc(doc_role)}.pdf"
+
+
+# ─────────────────────────────── report ───────────────────────────────
+@dataclass
+class LoadReport:
+    specs_upserted: int = 0
+    papers_upserted: int = 0
+    files_uploaded: int = 0
+    files_skipped: int = 0
+    files_failed: int = 0
+    user_copies: int = 0
+    swept: int = 0
+    sweep_failed: int = 0
+    downloaded: int = 0
+    download_cached: int = 0
+    unseed_objects: int = 0
+    unseed_exams: int = 0
+    unseed_specs: int = 0
+    unseed_templates: int = 0
+    errors: List[str] = field(default_factory=list)
+
+    def as_dict(self) -> Dict[str, Any]:
+        return {
+            "specs_upserted": self.specs_upserted,
+            "papers_upserted": self.papers_upserted,
+            "downloaded": self.downloaded,
+            "download_cached": self.download_cached,
+            "unseed_objects": self.unseed_objects,
+            "unseed_exams": self.unseed_exams,
+            "unseed_specs": self.unseed_specs,
+            "unseed_templates": self.unseed_templates,
+            "files_uploaded": self.files_uploaded,
+            "files_skipped": self.files_skipped,
+            "files_failed": self.files_failed,
+            "user_copies": self.user_copies,
+            "swept": self.swept,
+            "sweep_failed": self.sweep_failed,
+            "errors": self.errors,
+        }
+
+
+# ─────────────────────────────── validation ───────────────────────────────
+def validate_manifest(m: Dict[str, Any]) -> List[str]:
+    errs: List[str] = []
+    seen_specs, seen_exams = set(), set()
+    for board in m.get("boards", []):
+        bcode = board.get("exam_board_code")
+        if not bcode:
+            errs.append("board missing exam_board_code")
+        for spec in board.get("specifications", []):
+            sc = spec.get("spec_code")
+            if not sc or sc in seen_specs:
+                errs.append(f"spec_code missing/duplicate: {sc!r}")
+            seen_specs.add(sc)
+            for field_name in ("award_code", "subject_code"):
+                if not spec.get(field_name):
+                    errs.append(f"{sc}: missing {field_name}")
+            for p in spec.get("papers", []):
+                ec = p.get("exam_code")
+                if not ec or ec in seen_exams:
+                    errs.append(f"exam_code missing/duplicate: {ec!r}")
+                seen_exams.add(ec)
+                if p.get("doc_type") not in DOC_ROLES:
+                    errs.append(f"{ec}: bad doc_type/role {p.get('doc_type')!r} (want one of {sorted(DOC_ROLES)})")
+                if p.get("tier") not in TIERS:
+                    errs.append(f"{ec}: bad tier {p.get('tier')!r} (want H|F|null)")
+                if not p.get("paper_code"):
+                    errs.append(f"{ec}: missing paper_code")
+                if not p.get("session"):
+                    errs.append(f"{ec}: missing session")
+                src = (p.get("file") or {}).get("source")
+                if not src:
+                    errs.append(f"{ec}: missing file.source")
+                elif not src.startswith("url:") and not os.path.exists(src):
+                    errs.append(f"{ec}: local source not found: {src}")
+    return errs
+
+
+# ─────────────────────────────── source resolution (local | url:, cached) ───────────────────────────────
+def _resolve_source_bytes(source: str, *, cache_dir: str) -> bytes:
+    """Resolve a manifest file source to bytes.
+
+    'url:https://...'  -> fetch (cached to cache_dir by url hash) ; verifies non-empty.
+    '<local path>'     -> read from disk.
+    """
+    if source.startswith("url:"):
+        url = source[len("url:"):]
+        os.makedirs(cache_dir, exist_ok=True)
+        cache_key = hashlib.sha1(url.encode("utf-8")).hexdigest()
+        cache_path = os.path.join(cache_dir, f"{cache_key}.pdf")
+        if os.path.exists(cache_path) and os.path.getsize(cache_path) > 0:
+            with open(cache_path, "rb") as fh:
+                return fh.read()
+        logger.info(f"[fetch] {url}")
+        resp = requests.get(url, timeout=60, allow_redirects=True)
+        resp.raise_for_status()
+        data = resp.content
+        ctype = resp.headers.get("content-type", "")
+        if not data:
+            raise ValueError(f"empty download: {url}")
+        if "pdf" not in ctype.lower() and not data[:5].startswith(b"%PDF"):
+            raise ValueError(f"not a PDF (content-type={ctype!r}): {url}")
+        tmp = cache_path + ".part"
+        with open(tmp, "wb") as fh:
+            fh.write(data)
+        os.replace(tmp, cache_path)
+        return data
+    with open(source, "rb") as fh:
+        return fh.read()
+
+
+# ─────────────────────── persistent local store (download-once, seed-many) ───────────────────────
+def _store_path(store_dir: str, storage_loc: str) -> str:
+    """Local path mirroring the bucket layout (so the store is directly mountable as the corpus):
+    storage_loc 'cc.examboards/aqa/physics/8463/1h/2022-jun/qp.pdf'
+      -> {store_dir}/aqa/physics/8463/1h/2022-jun/qp.pdf
+    """
+    _, _, path = storage_loc.partition("/")
+    return os.path.join(store_dir, path)
+
+def _item_bytes(source: str, storage_loc: str, *, store_dir: Optional[str], cache_dir: str,
+                populate: bool = True, rep: Optional[LoadReport] = None) -> bytes:
+    """Resolve bytes for an item, preferring the persistent local store when present.
+
+    If store_dir holds the file → read it (offline). Otherwise resolve the source (local|url:) and,
+    when populate=True, write it into the store at its canonical path for future offline runs.
+    """
+    if store_dir:
+        sp = _store_path(store_dir, storage_loc)
+        if os.path.exists(sp) and os.path.getsize(sp) > 0:
+            if rep is not None:
+                rep.download_cached += 1
+            with open(sp, "rb") as fh:
+                return fh.read()
+    data = _resolve_source_bytes(source, cache_dir=cache_dir)
+    if store_dir and populate:
+        sp = _store_path(store_dir, storage_loc)
+        os.makedirs(os.path.dirname(sp), exist_ok=True)
+        tmp = sp + ".part"
+        with open(tmp, "wb") as fh:
+            fh.write(data)
+        os.replace(tmp, sp)
+        if rep is not None:
+            rep.downloaded += 1
+    return data
+
+def download_corpus(m: Dict[str, Any], *, store_dir: str, board_filter: Optional[str],
+                    spec_filter: Optional[str], cache_dir: str, rep: LoadReport) -> None:
+    """--download-only: populate the persistent local store from the manifest. No DB/bucket writes.
+    A later run with the same --store-dir (e.g. mounted into the container) seeds offline from it."""
+    for board in m.get("boards", []):
+        if board_filter and board.get("exam_board_code") != board_filter:
+            continue
+        for spec in board.get("specifications", []):
+            if spec_filter and spec.get("spec_code") != spec_filter:
+                continue
+            sf = spec.get("spec_file")
+            if sf and sf.get("source"):
+                sloc = spec_storage_loc(board["exam_board_code"], spec.get("subject_code", ""),
+                                        spec.get("award_code", ""), spec.get("spec_ver", ""))
+                try:
+                    _item_bytes(sf["source"], sloc, store_dir=store_dir, cache_dir=cache_dir, rep=rep)
+                except Exception as exc:
+                    rep.errors.append(f"download spec {spec.get('spec_code')}: {exc}")
+            for p in spec.get("papers", []):
+                ploc = paper_storage_loc(board["exam_board_code"], spec.get("subject_code", ""),
+                                         spec.get("award_code", ""), p["paper_code"], p["session"], p["doc_type"])
+                try:
+                    _item_bytes(p["file"]["source"], ploc, store_dir=store_dir, cache_dir=cache_dir, rep=rep)
+                except Exception as exc:
+                    rep.errors.append(f"download {p.get('exam_code')}: {exc}")
+    logger.info(f"download-only done: downloaded={rep.downloaded} already_in_store={rep.download_cached} "
+                f"errors={len(rep.errors)} store={store_dir}")
+
+
+# ─────────────────────────────── storage upload (skip-if-exists + sha256) ───────────────────────────────
+def _split_loc(storage_loc: str) -> Tuple[str, str]:
+    bucket, _, path = storage_loc.partition("/")
+    return bucket, path
+
+def _object_exists(storage: StorageAdmin, bucket: str, path: str) -> bool:
+    """Existence check by listing the object's parent folder (Supabase storage has no stat)."""
+    parent, _, name = path.rpartition("/")
+    try:
+        listing = storage.client.supabase.storage.from_(bucket).list(parent)
+    except Exception as exc:
+        logger.warning(f"[exists?] list failed for {bucket}/{parent}: {exc}")
+        return False
+    return any((item.get("name") == name) for item in (listing or []))
+
+def upload_file(storage: StorageAdmin, storage_loc: str, data: bytes, *, force: bool, rep: LoadReport) -> str:
+    """Upload PDF bytes to storage at storage_loc. Returns the sha256 of the bytes.
+
+    Idempotent: if the object already exists and --force was not given, skips the upload
+    (the catalogue upsert still runs and records the checksum). With --force, overwrites.
+    """
+    sha = hashlib.sha256(data).hexdigest()
+    bucket, path = _split_loc(storage_loc)
+    if not force and _object_exists(storage, bucket, path):
+        logger.info(f"[upload] skip-exists {storage_loc} (sha256={sha[:12]})")
+        rep.files_skipped += 1
+        return sha
+    try:
+        storage.upload_file(bucket, path, data, "application/pdf", upsert=True)
+        logger.info(f"[upload] {storage_loc} ({len(data)} bytes, sha256={sha[:12]}) force={force}")
+        rep.files_uploaded += 1
+    except StorageError as exc:
+        logger.error(f"[upload] FAILED {storage_loc}: {exc}")
+        rep.files_failed += 1
+        rep.errors.append(f"upload {storage_loc}: {exc}")
+    return sha
+
+
+# ─────────────────────────────── catalogue upserts ───────────────────────────────
+def upsert_specification(client: SupabaseServiceRoleClient, spec: Dict[str, Any],
+                         storage_loc: Optional[str], sha: Optional[str], rep: LoadReport) -> None:
+    sf = spec.get("spec_file") or {}
+    doc_details = {
+        "award_level": spec.get("award_level"),
+        "provenance": sf.get("provenance"),
+        "original_name": sf.get("original_name"),
+        "sha256": sha,
+    }
+    row = {
+        "spec_code": spec["spec_code"],
+        "exam_board_code": spec["exam_board_code"],
+        "award_code": spec.get("award_code"),
+        "subject_code": spec.get("subject_code"),
+        "first_teach": spec.get("first_teach"),
+        "spec_ver": spec.get("spec_ver"),
+        "storage_loc": storage_loc,
+        "doc_type": "pdf",  # file format (CHECK-constrained); the role lives on eb_exams.type_code
+        "doc_details": {k: v for k, v in doc_details.items() if v is not None},
+    }
+    try:
+        client.supabase.table("eb_specifications").upsert(row, on_conflict="spec_code").execute()
+        logger.info(f"[spec] upsert {row['spec_code']}")
+        rep.specs_upserted += 1
+    except Exception as exc:
+        logger.error(f"[spec] FAILED {row['spec_code']}: {exc}")
+        rep.errors.append(f"spec {row['spec_code']}: {exc}")
+
+def upsert_paper(client: SupabaseServiceRoleClient, spec_code: str, p: Dict[str, Any],
+                 storage_loc: str, sha: Optional[str], rep: LoadReport) -> None:
+    f = p.get("file") or {}
+    doc_role = p["doc_type"]  # manifest role: QP|MS|INSERT|ER...
+    doc_details = {
+        "doc_role": doc_role,                # mirror of type_code for clarity
+        "original_name": f.get("original_name"),
+        "provenance": f.get("provenance"),
+        "sha256": sha,
+    }
+    row = {
+        "exam_code": p["exam_code"],
+        "spec_code": spec_code,
+        "paper_code": p.get("paper_code"),
+        "tier": p.get("tier"),
+        "session": p.get("session"),
+        "type_code": doc_role,               # ROLE goes here (QP/MS/INSERT/ER)
+        "doc_type": "pdf",                   # file format (CHECK-constrained)
+        "storage_loc": storage_loc,
+        "doc_details": {k: v for k, v in doc_details.items() if v is not None},
+    }
+    try:
+        client.supabase.table("eb_exams").upsert(row, on_conflict="exam_code").execute()
+        logger.info(f"[paper] upsert {row['exam_code']} type_code={doc_role}")
+        rep.papers_upserted += 1
+    except Exception as exc:
+        logger.error(f"[paper] FAILED {row['exam_code']}: {exc}")
+        rep.errors.append(f"paper {row['exam_code']}: {exc}")
+
+
+# ─────────────────────────────── user-side test subset ───────────────────────────────
+def _resolve_test_user(client: SupabaseServiceRoleClient, cfg: Dict[str, Any]) -> Optional[Tuple[str, str]]:
+    """Resolve (user_id, institute_id) for the user-side subset from config, with discovery fallback."""
+    user_id = cfg.get("user_id")
+    if not user_id and cfg.get("user_email"):
+        res = client.supabase.table("profiles").select("id").eq("email", cfg["user_email"]).limit(1).execute()
+        rows = getattr(res, "data", None) or []
+        user_id = rows[0]["id"] if rows else None
+    if not user_id:
+        logger.warning("[user-subset] no test user resolvable (set test_subset.user_id or user_email); skipping")
+        return None
+    institute_id = cfg.get("institute_id")
+    if not institute_id:
+        res = client.supabase.table("institute_memberships").select("institute_id").eq("profile_id", user_id).limit(1).execute()
+        rows = getattr(res, "data", None) or []
+        institute_id = rows[0]["institute_id"] if rows else None
+    if not institute_id:
+        logger.warning(f"[user-subset] no institute for user {user_id}; skipping")
+        return None
+    return user_id, institute_id
+
+def copy_user_test_subset(client: SupabaseServiceRoleClient, storage: StorageAdmin,
+                          m: Dict[str, Any], rep: LoadReport) -> None:
+    """Copy a small subset of admin papers into a test user's exam space so user-side flows
+    (upload-as-exam / promote-from-cabinet / mark) are testable.
+
+    Driven by an optional manifest `test_subset:` block:
+        test_subset:
+          user_id: <uuid>            # or user_email: <email>
+          institute_id: <uuid>       # optional; discovered from membership if omitted
+          papers: 2                  # how many QP papers to copy (default 2)
+    Degrades gracefully (logs + skips) if no test user is resolvable on this env.
+    """
+    cfg = m.get("test_subset") or {}
+    resolved = _resolve_test_user(client, cfg)
+    if not resolved:
+        return
+    user_id, institute_id = resolved
+    limit = int(cfg.get("papers", 2))
+
+    # Gather candidate QP papers (admin corpus already uploaded to cc.examboards).
+    candidates: List[Tuple[str, Dict[str, Any]]] = []
+    for board in m.get("boards", []):
+        for spec in board.get("specifications", []):
+            for p in spec.get("papers", []):
+                if p.get("doc_type") == "QP":
+                    candidates.append((board["exam_board_code"], spec, p))
+    candidates = candidates[:limit]
+    if not candidates:
+        logger.info("[user-subset] no QP papers to copy")
+        return
+
+    # Ensure a cabinet for the user.
+    cab_name = "Exam Marker Template Sources"
+    res = client.supabase.table("file_cabinets").select("id").eq("user_id", user_id).eq("name", cab_name).limit(1).execute()
+    rows = getattr(res, "data", None) or []
+    if rows:
+        cabinet_id = rows[0]["id"]
+    else:
+        ins = client.supabase.table("file_cabinets").insert({"user_id": user_id, "name": cab_name}).execute()
+        cabinet_id = (getattr(ins, "data", None) or [{}])[0].get("id")
+    if not cabinet_id:
+        logger.warning("[user-subset] could not ensure cabinet; skipping")
+        return
+
+    import uuid as _uuid
+    for board_code, spec, p in candidates:
+        src_loc = paper_storage_loc(board_code, spec.get("subject_code", ""), spec.get("award_code", ""),
+                                    p["paper_code"], p["session"], p["doc_type"])
+        sbucket, spath = _split_loc(src_loc)
+        try:
+            data = storage.download_file(sbucket, spath)
+        except Exception as exc:
+            logger.warning(f"[user-subset] source missing {src_loc}: {exc}; skipping {p['exam_code']}")
+            continue
+        file_id = str(_uuid.uuid4())
+        safe_name = f"{p['exam_code']}.pdf"
+        dst_bucket = "cc.users"
+        dst_path = f"exam-marker/{institute_id}/{cabinet_id}/{file_id}/{safe_name}"
+        try:
+            storage.upload_file(dst_bucket, dst_path, data, "application/pdf", upsert=True)
+        except Exception as exc:
+            logger.warning(f"[user-subset] copy upload failed {dst_path}: {exc}")
+            continue
+        client.supabase.table("files").upsert({
+            "id": file_id, "cabinet_id": cabinet_id, "name": safe_name, "path": dst_path,
+            "bucket": dst_bucket, "mime_type": "application/pdf", "uploaded_by": user_id,
+            "size_bytes": len(data), "source": "exam-corpus-seed", "is_directory": False,
+            "relative_path": safe_name, "processing_status": "uploaded",
+        }).execute()
+        logger.info(f"[user-subset] copied {p['exam_code']} -> {dst_bucket}/{dst_path}")
+        rep.user_copies += 1
+
+
+# ─────────────────────────────── first sweep (docling auto-map) ───────────────────────────────
+def _resolve_system_identity(client: SupabaseServiceRoleClient, m: Dict[str, Any]) -> Optional[Tuple[str, str]]:
+    cfg = m.get("system_identity") or m.get("test_subset") or {}
+    user_id = cfg.get("teacher_id") or cfg.get("user_id")
+    if not user_id and cfg.get("user_email"):
+        res = client.supabase.table("profiles").select("id").eq("email", cfg["user_email"]).limit(1).execute()
+        rows = getattr(res, "data", None) or []
+        user_id = rows[0]["id"] if rows else None
+    institute_id = cfg.get("institute_id")
+    if user_id and not institute_id:
+        res = client.supabase.table("institute_memberships").select("institute_id").eq("profile_id", user_id).limit(1).execute()
+        rows = getattr(res, "data", None) or []
+        institute_id = rows[0]["institute_id"] if rows else None
+    if not user_id or not institute_id:
+        logger.warning("[first-sweep] no system identity (set system_identity.teacher_id+institute_id); skipping sweep")
+        return None
+    return user_id, institute_id
+
+def first_sweep(client: SupabaseServiceRoleClient, storage: StorageAdmin,
+                m: Dict[str, Any], board_filter: Optional[str], spec_filter: Optional[str],
+                cache_dir: str, rep: LoadReport) -> None:
+    """Run the docling/auto_map first pass over seeded QP papers and persist the resulting
+    template structure (questions/response areas/boundaries/layout) via the same mapping the
+    /auto-map endpoint uses. System-owned exam_templates are created per QP paper.
+
+    Requires a resolvable `system_identity` (teacher_id/user_email + institute_id) on this env.
+    """
+    identity = _resolve_system_identity(client, m)
+    if not identity:
+        return
+    teacher_id, institute_id = identity
+
+    # Import the auto-map mapping helpers lazily (pulls fastapi/router only when sweeping).
+    try:
+        from api.services.docling import auto_map, AutoMapError
+        from routers.exam.templates import _map_first_pass_to_rows
+    except Exception as exc:
+        logger.error(f"[first-sweep] could not import auto-map pipeline: {exc}")
+        rep.errors.append(f"first-sweep import: {exc}")
+        return
+
+    sb = client.supabase
+    for board in m.get("boards", []):
+        if board_filter and board.get("exam_board_code") != board_filter:
+            continue
+        for spec in board.get("specifications", []):
+            if spec_filter and spec.get("spec_code") != spec_filter:
+                continue
+            for p in spec.get("papers", []):
+                if p.get("doc_type") != "QP":
+                    continue
+                # Resolve the seeded eb_exams row (id) for the template join.
+                ex = sb.table("eb_exams").select("id, exam_code").eq("exam_code", p["exam_code"]).limit(1).execute()
+                ex_rows = getattr(ex, "data", None) or []
+                exam_id = ex_rows[0]["id"] if ex_rows else None
+
+                loc = paper_storage_loc(board["exam_board_code"], spec.get("subject_code", ""),
+                                        spec.get("award_code", ""), p["paper_code"], p["session"], p["doc_type"])
+                bkt, path = _split_loc(loc)
+                try:
+                    pdf_bytes = storage.download_file(bkt, path)
+                except Exception as exc:
+                    logger.warning(f"[first-sweep] source missing {loc}: {exc}; skipping {p['exam_code']}")
+                    continue
+
+                # Ensure a system-owned template for this paper (idempotent on exam_code+teacher).
+                tpl = sb.table("exam_templates").select("id").eq("exam_code", p["exam_code"]).eq("teacher_id", teacher_id).limit(1).execute()
+                tpl_rows = getattr(tpl, "data", None) or []
+                if tpl_rows:
+                    template_id = tpl_rows[0]["id"]
+                else:
+                    new_tpl = sb.table("exam_templates").insert({
+                        "exam_id": exam_id, "exam_code": p["exam_code"], "institute_id": institute_id,
+                        "teacher_id": teacher_id, "title": f"{p['exam_code']} (auto-map seed)",
+                        "subject": spec.get("subject_code"), "status": "draft",
+                    }).execute()
+                    template_id = (getattr(new_tpl, "data", None) or [{}])[0].get("id")
+                if not template_id:
+                    logger.warning(f"[first-sweep] could not ensure template for {p['exam_code']}; skipping")
+                    continue
+
+                try:
+                    first_pass = auto_map(pdf_bytes, source_pdf=loc)
+                    rows = _map_first_pass_to_rows(template_id, first_pass, pdf_bytes)
+                except (AutoMapError, ValueError) as exc:
+                    logger.warning(f"[first-sweep] auto-map failed for {p['exam_code']}: {exc}")
+                    rep.sweep_failed += 1
+                    continue
+                except Exception as exc:
+                    logger.exception(f"[first-sweep] unexpected error for {p['exam_code']}: {exc}")
+                    rep.sweep_failed += 1
+                    continue
+
+                # Refresh derived rows. Seed templates are system-owned with no human edits to
+                # preserve, so we clear ALL child rows for the template (not just ai/unconfirmed)
+                # and re-insert id-deduped payloads — idempotent across re-runs and robust to the
+                # deterministic uuid5 ids the mapper can repeat within a batch.
+                for table in ("exam_response_areas", "exam_boundaries", "exam_template_layout", "exam_questions"):
+                    sb.table(table).delete().eq("template_id", template_id).execute()
+                for table, key in (("exam_questions", "questions"), ("exam_response_areas", "response_areas"),
+                                   ("exam_boundaries", "boundaries"), ("exam_template_layout", "layout")):
+                    seen_ids: set = set()
+                    payload = []
+                    for r in (rows.get(key) or []):
+                        rid = r.get("id")
+                        if rid is not None and rid in seen_ids:
+                            continue
+                        if rid is not None:
+                            seen_ids.add(rid)
+                        payload.append(r)
+                    if payload:
+                        sb.table(table).insert(payload).execute()
+                updates = {"page_count": first_pass.get("meta", {}).get("n_pages")}
+                sb.table("exam_templates").update({k: v for k, v in updates.items() if v is not None}).eq("id", template_id).execute()
+                logger.info(f"[first-sweep] swept {p['exam_code']} -> template {template_id} "
+                            f"(q={len(rows.get('questions', []))} ra={len(rows.get('response_areas', []))})")
+                rep.swept += 1
+
+
+# ─────────────────────────────── unseed (inverse of the loader) ───────────────────────────────
+def _chunks(seq: List[Any], n: int = 100):
+    for i in range(0, len(seq), n):
+        yield seq[i:i + n]
+
+def unseed(client: SupabaseServiceRoleClient, storage: StorageAdmin, *,
+           board_filter: Optional[str], spec_filter: Optional[str],
+           drop_specs: bool = True, drop_seed_templates: bool = True, rep: LoadReport) -> None:
+    """Inverse of the loader: remove the seeded public corpus, scoped by --board/--spec (or all).
+
+    Deletes (in FK-safe order): cc.examboards storage objects (via the Storage API, since the
+    protect_delete trigger blocks direct SQL deletes), first-sweep exam_templates created by the
+    seed (title '... (auto-map seed)', cascades children), eb_exams rows, then eb_specifications.
+    """
+    sb = client.supabase
+    q = sb.table("eb_specifications").select("spec_code, storage_loc, exam_board_code")
+    if board_filter:
+        q = q.eq("exam_board_code", board_filter)
+    if spec_filter:
+        q = q.eq("spec_code", spec_filter)
+    specs = getattr(q.execute(), "data", None) or []
+    spec_codes = [s["spec_code"] for s in specs]
+    if not spec_codes:
+        logger.info("[unseed] no matching specifications; nothing to do")
+        return
+
+    exams: List[Dict[str, Any]] = []
+    for chunk in _chunks(spec_codes):
+        res = sb.table("eb_exams").select("id, exam_code, storage_loc").in_("spec_code", chunk).execute()
+        exams.extend(getattr(res, "data", None) or [])
+
+    # 1) Storage objects (Storage API; batch-remove per bucket). Specs may carry a spec PDF too.
+    by_bucket: Dict[str, List[str]] = {}
+    for row in exams + specs:
+        loc = row.get("storage_loc")
+        if not loc or "/" not in loc:
+            continue
+        bkt, _, path = loc.partition("/")
+        by_bucket.setdefault(bkt, []).append(path)
+    for bkt, paths in by_bucket.items():
+        for chunk in _chunks(paths, 100):
+            try:
+                storage.client.supabase.storage.from_(bkt).remove(chunk)
+                rep.unseed_objects += len(chunk)
+            except Exception as exc:
+                logger.warning(f"[unseed] storage remove failed ({bkt}, {len(chunk)} objs): {exc}")
+
+    # 2) First-sweep templates created by the seed (cascades questions/regions/boundaries/layout).
+    if drop_seed_templates and exams:
+        exam_codes = [e["exam_code"] for e in exams if e.get("exam_code")]
+        for chunk in _chunks(exam_codes, 100):
+            try:
+                res = sb.table("exam_templates").delete(count="exact") \
+                    .in_("exam_code", chunk).like("title", "%(auto-map seed)%").execute()
+                rep.unseed_templates += getattr(res, "count", None) or len(getattr(res, "data", []) or [])
+            except Exception as exc:
+                logger.warning(f"[unseed] template delete failed: {exc}")
+
+    # 3) Catalogue rows: eb_exams (by id), then eb_specifications (by spec_code).
+    exam_ids = [e["id"] for e in exams]
+    for chunk in _chunks(exam_ids, 100):
+        try:
+            sb.table("eb_exams").delete().in_("id", chunk).execute()
+            rep.unseed_exams += len(chunk)
+        except Exception as exc:
+            logger.warning(f"[unseed] eb_exams delete failed: {exc}")
+    if drop_specs:
+        for chunk in _chunks(spec_codes, 100):
+            try:
+                sb.table("eb_specifications").delete().in_("spec_code", chunk).execute()
+                rep.unseed_specs += len(chunk)
+            except Exception as exc:
+                logger.warning(f"[unseed] eb_specifications delete failed: {exc}")
+
+    logger.info(f"unseed done: storage_objects={rep.unseed_objects} templates={rep.unseed_templates} "
+                f"exams={rep.unseed_exams} specs={rep.unseed_specs}")
+
+
+# ─────────────────────────────── orchestration ───────────────────────────────
+def load(manifest_path: str, *, dry_run: bool, force: bool, board_filter: Optional[str],
+         spec_filter: Optional[str], user_subset: bool, do_first_sweep: bool,
+         cache_dir: str = DEFAULT_CACHE_DIR, store_dir: Optional[str] = None) -> LoadReport:
+    with open(manifest_path) as f:
+        m = yaml.safe_load(f)
+    rep = LoadReport()
+
+    errs = validate_manifest(m)
+    if errs:
+        rep.errors = list(errs)
+        logger.error(f"manifest validation failed: {len(errs)} error(s)")
+        for e in errs[:40]:
+            logger.error(f"  - {e}")
+        if not dry_run:
+            return rep
+
+    client = None if dry_run else SupabaseServiceRoleClient()
+    storage = None if dry_run else StorageAdmin()
+
+    for board in m.get("boards", []):
+        if board_filter and board.get("exam_board_code") != board_filter:
+            continue
+        for spec in board.get("specifications", []):
+            if spec_filter and spec.get("spec_code") != spec_filter:
+                continue
+            # Specification document (optional).
+            sloc = None
+            spec_sha = None
+            sf = spec.get("spec_file")
+            if sf and sf.get("source"):
+                sloc = spec_storage_loc(board["exam_board_code"], spec.get("subject_code", ""),
+                                        spec.get("award_code", ""), spec.get("spec_ver", ""))
+                if not dry_run:
+                    try:
+                        spec_sha = upload_file(storage, sloc,
+                                               _item_bytes(sf["source"], sloc, store_dir=store_dir,
+                                                           cache_dir=cache_dir, rep=rep),
+                                               force=force, rep=rep)
+                    except Exception as exc:
+                        logger.error(f"[spec-file] {spec.get('spec_code')}: {exc}")
+                        rep.files_failed += 1
+                        rep.errors.append(f"spec-file {spec.get('spec_code')}: {exc}")
+            if not dry_run:
+                upsert_specification(client, spec, sloc, spec_sha, rep)
+
+            # Papers.
+            for p in spec.get("papers", []):
+                ploc = paper_storage_loc(board["exam_board_code"], spec.get("subject_code", ""),
+                                         spec.get("award_code", ""), p["paper_code"], p["session"], p["doc_type"])
+                if dry_run:
+                    continue
+                psha = None
+                try:
+                    psha = upload_file(storage, ploc,
+                                       _item_bytes(p["file"]["source"], ploc, store_dir=store_dir,
+                                                   cache_dir=cache_dir, rep=rep),
+                                       force=force, rep=rep)
+                except Exception as exc:
+                    logger.error(f"[paper-file] {p.get('exam_code')}: {exc}")
+                    rep.files_failed += 1
+                    rep.errors.append(f"paper-file {p.get('exam_code')}: {exc}")
+                upsert_paper(client, spec["spec_code"], p, ploc, psha, rep)
+
+    if user_subset and not dry_run:
+        copy_user_test_subset(client, storage, m, rep)
+    if do_first_sweep and not dry_run:
+        first_sweep(client, storage, m, board_filter, spec_filter, cache_dir, rep)
+
+    logger.info(f"corpus load done: specs={rep.specs_upserted} papers={rep.papers_upserted} "
+                f"uploaded={rep.files_uploaded} skipped={rep.files_skipped} failed={rep.files_failed} "
+                f"user_copies={rep.user_copies} swept={rep.swept} errors={len(rep.errors)}")
+    return rep
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="Seed (or unseed) the public exam-paper corpus from a manifest.")
+    ap.add_argument("--manifest", help="corpus manifest (required except for --unseed)")
+    ap.add_argument("--dry-run", action="store_true", help="validate + report, no writes")
+    ap.add_argument("--force", action="store_true", help="re-upload/overwrite existing storage objects")
+    ap.add_argument("--board", default=None, help="only this exam_board_code")
+    ap.add_argument("--spec", default=None, help="only this spec_code")
+    ap.add_argument("--user-subset", action="store_true", help="also seed a user-side test subset")
+    ap.add_argument("--first-sweep", action="store_true", help="run docling/auto-map first pass on seeded papers")
+    ap.add_argument("--cache-dir", default=DEFAULT_CACHE_DIR, help="throwaway url-hash cache dir")
+    ap.add_argument("--store-dir", default=DEFAULT_STORE_DIR,
+                    help="persistent, bucket-shaped local store (download-once, seed-many)")
+    ap.add_argument("--no-store", action="store_true",
+                    help="ignore the local store; always fetch from source (don't read/populate the store)")
+    ap.add_argument("--download-only", action="store_true",
+                    help="populate the local store from the manifest; no DB/bucket writes")
+    ap.add_argument("--unseed", action="store_true",
+                    help="INVERSE: remove seeded eb_*/storage/first-sweep templates (scoped by --board/--spec)")
+    a = ap.parse_args()
+    store_dir = None if a.no_store else a.store_dir
+    import json
+
+    if a.unseed:
+        rep = LoadReport()
+        unseed(SupabaseServiceRoleClient(), StorageAdmin(),
+               board_filter=a.board, spec_filter=a.spec, rep=rep)
+        print(json.dumps(rep.as_dict(), indent=2))
+        return
+
+    if not a.manifest:
+        ap.error("--manifest is required unless --unseed is given")
+
+    if a.download_only:
+        with open(a.manifest) as f:
+            m = yaml.safe_load(f)
+        rep = LoadReport()
+        download_corpus(m, store_dir=(a.store_dir), board_filter=a.board, spec_filter=a.spec,
+                        cache_dir=a.cache_dir, rep=rep)
+        print(json.dumps(rep.as_dict(), indent=2))
+        return
+
+    rep = load(a.manifest, dry_run=a.dry_run, force=a.force, board_filter=a.board, spec_filter=a.spec,
+               user_subset=a.user_subset, do_first_sweep=a.first_sweep, cache_dir=a.cache_dir,
+               store_dir=store_dir)
+    print(json.dumps(rep.as_dict(), indent=2))
+
+
+if __name__ == "__main__":
+    main()

tests/test_docling_auto_map.py

@@ -26,7 +26,9 @@ def test_auto_map_matches_spike_physics_template_shape():
     assert len(result["margins"]) == len(expected["margins"])
     assert set(result["pages"].keys()) == set(expected["pages"].keys())
     assert result["pages"]["2"]["role"] == expected["pages"]["2"]["role"]
-    assert result["pages"]["2"]["part_bands"][0].keys() == expected["pages"]["2"]["part_bands"][0].keys()
+    part_band = result["pages"]["2"]["part_bands"][0]
+    assert set(expected["pages"]["2"]["part_bands"][0].keys()).issubset(part_band.keys())
+    assert part_band["box"]
 
 
 @pytest.mark.skipif(not BORN_DIGITAL_PDF.exists(), reason="born-digital spike PDF not present")

tests/test_docling_template.py

@@ -0,0 +1,55 @@
+from api.services.docling.template import build, synthesize_part_box
+
+
+def test_synthesize_part_box_uses_content_margins_and_band_y():
+    part_band = {"label": "01.1", "y_start": 712.34, "y_end": 601.27}
+    content_x_band = {"x_left": 54.04, "x_right": 521.96}
+
+    assert synthesize_part_box(part_band, content_x_band) == {
+        "l": 54.0,
+        "t": 712.3,
+        "r": 522.0,
+        "b": 601.3,
+        "coord_origin": "BOTTOMLEFT",
+    }
+
+
+def test_synthesize_part_box_returns_none_without_margin_contract():
+    assert synthesize_part_box({"y_start": 700, "y_end": 650}, {}) is None
+    assert synthesize_part_box({"y_start": 700}, {"x_left": 50, "x_right": 520}) is None
+
+
+def test_build_carries_label_box_as_anchor_and_single_synthesized_part_box():
+    structured = {
+        "board": "aqa",
+        "paper_code": "8463/1",
+        "questions": [{
+            "question": "01",
+            "parts": [{
+                "label": "01.1",
+                "page": 2,
+                "bbox": {"l": 40, "t": 720, "r": 70, "b": 705},
+            }],
+        }],
+    }
+    bands = {
+        "pages": {
+            "2": {
+                "main": [{"question": "01", "y_start": 730, "y_end": 0, "is_start": True}],
+                "part": [{"label": "01.1", "question": "01", "y_start": 720, "y_end": 610}],
+            }
+        }
+    }
+    furniture = {
+        "n_pages": 2,
+        "content_margins": {
+            "content_x_band": {"x_left": 55, "x_right": 515},
+            "per_page": {"2": {"top": 760, "bottom": 40, "left": 55, "right": 515}},
+        },
+        "items": [],
+    }
+
+    part = build(structured, bands, furniture)["pages"]["2"]["part_bands"][0]
+
+    assert part["label_box"] == {"l": 40, "t": 720, "r": 70, "b": 705}
+    assert part["box"] == {"l": 55, "t": 720, "r": 515, "b": 610, "coord_origin": "BOTTOMLEFT"}

tests/test_exam_templates.py

@@ -481,3 +481,144 @@ def test_neo4j_sync_non_owner_403():
 def test_neo4j_sync_404():
     client, _ = make_client(store={"exam_templates": []})
     assert client.post("/api/exam/templates/does-not-exist/neo4j-sync").status_code == 404
+
+
+# ─── S5 auto-map endpoint ────────────────────────────────────────────────────
+
+def _first_pass_template():
+    return {
+        "meta": {"schema": "exam-template/first-pass/v1", "paper_code": "8463/1", "n_pages": 1},
+        "margins": [
+            {"edge": "left", "axis": "x", "value": 50, "scope": "document", "source": "auto", "confirmed": False},
+            {"edge": "right", "axis": "x", "value": 550, "scope": "document", "source": "auto", "confirmed": False},
+            {"edge": "top", "axis": "y", "value": 780, "scope": "page", "page": 1, "source": "auto", "confirmed": False},
+            {"edge": "bottom", "axis": "y", "value": 60, "scope": "page", "page": 1, "source": "auto", "confirmed": False},
+        ],
+        "pages": {
+            "1": {
+                "role": "question", "role_source": "auto", "margins_enabled": True,
+                "main_bands": [{"question": "01", "y_start": 780, "y_end": 60, "source": "auto", "confirmed": False}],
+                "part_bands": [{"label": "01.1", "question": "01", "y_start": 700, "y_end": 500, "label_box": {"l": 50, "t": 700, "r": 90, "b": 680, "coord_origin": "BOTTOMLEFT"}, "source": "auto", "confirmed": False}],
+                "furniture": [], "figures": [], "tables": [],
+            }
+        },
+    }
+
+
+def _patch_auto_map(monkeypatch, store, *, fast=True):
+    monkeypatch.setattr(templates_mod, "StorageAdmin", _FakeStorageAdmin)
+    monkeypatch.setattr(templates_mod, "SupabaseServiceRoleClient", lambda: _FakeServiceRoleClient(store))
+    monkeypatch.setattr(templates_mod, "_pdf_has_text_layer", lambda _pdf: fast)
+    monkeypatch.setattr(templates_mod, "auto_map", lambda *_a, **_k: _first_pass_template())
+    monkeypatch.setattr(templates_mod, "detect_response_regions_from_pdf", lambda *_a, **_k: [])
+    monkeypatch.setattr(templates_mod, "_pdf_page_geometry", lambda _pdf: [{"media_x0": 0.0, "crop_x0": 0.0, "crop_y0": 0.0, "page_pt_w": 600.0, "page_pt_h": 800.0, "rendered_w": 600.0, "rendered_h": 800.0, "page_top": 0.0}])
+    templates_mod._AUTO_MAP_JOB_STATUS.clear()
+
+
+def _template_with_source(owner=TEACHER):
+    return {
+        "exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": owner, "source_file_id": "f1"}],
+        "files": [{"id": "f1", "bucket": "cc.users", "path": "exam-marker/i/c/f1/paper.pdf", "name": "paper.pdf"}],
+    }
+
+
+def test_box_to_canvas_uses_cropbox_as_page_origin():
+    pages = [{
+        "media_x0": 0.0, "crop_x0": 100.0, "crop_y0": 200.0,
+        "page_pt_w": 400.0, "page_pt_h": 600.0,
+        "rendered_w": 400.0, "rendered_h": 600.0,
+        "page_top": 25.0,
+    }]
+    box = {"l": 100.0, "t": 800.0, "r": 180.0, "b": 760.0, "coord_origin": "BOTTOMLEFT"}
+    assert templates_mod._box_to_canvas(box, 1, pages) == {"x": 0.0, "y": 25.0, "w": 80.0, "h": 40.0}
+
+
+def test_response_region_types_are_mapped_to_response_form_enum(monkeypatch):
+    monkeypatch.setattr(templates_mod, "_pdf_page_geometry", lambda _pdf: [{"media_x0": 0.0, "crop_x0": 0.0, "crop_y0": 0.0, "page_pt_w": 600.0, "page_pt_h": 800.0, "rendered_w": 600.0, "rendered_h": 800.0, "page_top": 0.0}])
+    first_pass = _first_pass_template()
+    regions = [
+        {"page_index": 0, "bbox": {"l": 50, "t": 700, "r": 100, "b": 680, "coord_origin": "BOTTOMLEFT"}, "region_type": "answer_lines", "confidence": 0.9},
+        {"page_index": 0, "bbox": {"l": 50, "t": 650, "r": 100, "b": 620, "coord_origin": "BOTTOMLEFT"}, "region_type": "answer_box", "confidence": 0.9},
+        {"page_index": 0, "bbox": {"l": 50, "t": 600, "r": 100, "b": 560, "coord_origin": "BOTTOMLEFT"}, "region_type": "working_space", "confidence": 0.9},
+    ]
+    rows = templates_mod._map_first_pass_to_rows("t1", first_pass, b"%PDF", regions)
+    forms = [r.get("response_form") for r in rows["response_areas"] if r.get("derivation") == "opencv-response-region"]
+    assert forms == ["lines", "answer-box", "working"]
+
+
+def test_auto_map_fast_path_merges_ai_rows_and_returns_detail(monkeypatch):
+    store = _template_with_source()
+    client, store = make_client(store=store)
+    _patch_auto_map(monkeypatch, store, fast=True)
+    resp = client.post("/api/exam/templates/t1/auto-map")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["exam_code"] == "8463/1"
+    assert body["layout"] and body["layout"][0]["source"] == "ai"
+    assert any(q["label"] == "01.1" and q["source"] == "ai" and q["confirmed"] is False for q in store["exam_questions"])
+    assert store["exam_boundaries"] and store["exam_boundaries"][0]["derivation"] == "docling-main-band"
+
+
+def test_auto_map_preserves_manual_and_confirmed_rows_on_rerun(monkeypatch):
+    store = _template_with_source()
+    store.update({
+        "exam_questions": [
+            {"id": "manual", "template_id": "t1", "label": "manual", "order": 0, "source": "manual", "confirmed": True},
+            {"id": "accepted-ai", "template_id": "t1", "label": "accepted", "order": 1, "source": "ai", "confirmed": True},
+            {"id": "old-ai", "template_id": "t1", "label": "old", "order": 2, "source": "ai", "confirmed": False},
+        ],
+        "exam_response_areas": [], "exam_boundaries": [], "exam_template_layout": [],
+    })
+    client, store = make_client(store=store)
+    _patch_auto_map(monkeypatch, store, fast=True)
+    assert client.post("/api/exam/templates/t1/auto-map").status_code == 200
+    ids = {q["id"] for q in store["exam_questions"]}
+    assert {"manual", "accepted-ai"}.issubset(ids)
+    assert "old-ai" not in ids
+
+
+def test_auto_map_non_owner_is_403_before_download(monkeypatch):
+    store = _template_with_source(owner=OTHER_TEACHER)
+    client, store = make_client(user_id=TEACHER, institute_ids=(INST_A,), store=store)
+    def _no_download(*_a, **_k):
+        raise AssertionError("download should not run before owner gate")
+    monkeypatch.setattr(templates_mod, "StorageAdmin", _no_download)
+    resp = client.post("/api/exam/templates/t1/auto-map")
+    assert resp.status_code == 403
+
+
+def test_auto_map_owner_lost_institute_membership_is_404_before_download(monkeypatch):
+    store = _template_with_source(owner=TEACHER)
+    client, store = make_client(user_id=TEACHER, institute_ids=(INST_B,), store=store)
+    def _no_download(*_a, **_k):
+        raise AssertionError("download should not run before visibility gate")
+    monkeypatch.setattr(templates_mod, "StorageAdmin", _no_download)
+    resp = client.post("/api/exam/templates/t1/auto-map")
+    assert resp.status_code == 404
+
+
+def test_auto_map_blocks_when_marks_recorded(monkeypatch):
+    store = _template_with_source()
+    store.update({
+        "marking_batches": [{"id": "b1", "template_id": "t1"}],
+        "mark_entries": [{"id": "m1", "batch_id": "b1"}],
+    })
+    client, store = make_client(store=store)
+    _patch_auto_map(monkeypatch, store, fast=True)
+    resp = client.post("/api/exam/templates/t1/auto-map")
+    assert resp.status_code == 409
+
+
+def test_auto_map_ocr_returns_job_id_and_status_completes(monkeypatch):
+    store = _template_with_source()
+    client, store = make_client(store=store)
+    _patch_auto_map(monkeypatch, store, fast=False)
+    resp = client.post("/api/exam/templates/t1/auto-map")
+    assert resp.status_code == 202
+    job_id = resp.json()["job_id"]
+    status = client.get(f"/api/exam/templates/t1/auto-map/{job_id}/status")
+    assert status.status_code == 200
+    body = status.json()
+    assert body["status"] == "completed"
+    assert body["counts"]["questions"] >= 2
+    assert body["template"]["layout"]