Merge P2: auto-map via extraction service

When EXAM_EXTRACT_URL is set, POST /templates/{id}/auto-map now runs the spike's
full recognition pipeline (via the P1 service) instead of the thin first-pass, as
an async job. New modules/services/exam_extract.py (HTTP client: POST paper, poll,
return the analyse contract). templates.py: _map_service_contract_to_rows adapts the
page-fraction analyse contract onto the app's 780-wide stacked canvas (fraction ×
rendered dims + page_top), re-namespaces the service's uuid5s per template (stable
re-map, no cross-template collision), and is FK-safe (drops orphan regions, de-parents
dangling parents). Rich meta (OMR boxes, select_n, unit/quantity, n_options) persists
to the new exam_response_areas.meta column (migration 75). Typed response_form + mcq
answer_type + kind:context reference links map natively (schema already allows them).

The thin path is preserved as fallback when EXAM_EXTRACT_URL is unset. Unit test
covers the mapping (coord conversion, id remap, FK-safety, meta passthrough).

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

api/services/docling/bands.py

@@ -67,11 +67,13 @@ def derive_bands(result, doc=None, rapid_glob=None):
     topnum = _topnumber_boxes(docs)
     # gather parts with geometry, grouped by page
     by_page = defaultdict(list)               # page -> [(q, label, t, b)]
+    part_marks = {}                           # (question, part label) -> parsed marks (born-digital grammar)
     for q in result.get("questions", []):
         for p in q["parts"]:
             bb, pg = p.get("bbox"), p.get("page")
             if bb and pg:
                 by_page[pg].append((q["question"], p["label"], bb["t"], bb["b"]))
+                part_marks[(q["question"], p["label"])] = p.get("marks")
 
     # global first page each question appears on (to mark the true start vs continuation pages)
     q_first_page = {}
@@ -104,7 +106,8 @@ def derive_bands(result, doc=None, rapid_glob=None):
         for (q, lab), st, en, _ in _ends(part_items):
             qen = main_band.get(q, (st, 0))[1]                      # don't run past the question end
             part.append({"label": lab, "question": q,
-                         "y_start": round(st, 1), "y_end": round(max(en, qen), 1)})
+                         "y_start": round(st, 1), "y_end": round(max(en, qen), 1),
+                         "marks": part_marks.get((q, lab))})
         pages[pg] = {"main": main, "part": part}
 
     return {"board": result.get("board"), "paper_code": result.get("paper_code"),

api/services/docling/template.py

@@ -159,6 +159,7 @@ def build(structured, bands, furniture, pdf=None, page_roles=None):
                 "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),
+                "marks": p.get("marks"),                 # parsed per-part marks (born-digital)
                 "source": "auto", "confirmed": False,
             })
         pr = page_roles.get(pgs) or page_roles.get(pg) or {}

modules/services/exam_extract.py

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

routers/exam/__init__.py

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

routers/exam/bank.py

@@ -0,0 +1,152 @@
+"""Question bank + custom-paper assembly (/api/exam/bank, /api/exam/custom-papers) — mode 3.
+
+Build-your-own-from-the-spec. The BANK is a browsable index over the leaf questions across the templates
+the caller can see (RLS-scoped, institute-wide); a CUSTOM PAPER is a normal exam_template whose questions
+are COPIES of the selected bank questions, so it flows through the existing setup / marking / projection
+pipeline unchanged. Copy-on-assemble (v1) avoids decoupling question identity from a template; a shared-
+question model is Phase-2. Design: ~/cc/ideas/2026-07-02-mode3-question-bank-design.md.
+
+All access is as-the-user (E1): the bank query returns only questions in templates RLS lets the caller see,
+and a custom paper is created owned by the caller + institute-scoped (writes pass the same RLS with-check).
+"""
+from __future__ import annotations
+
+import os
+import uuid
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from modules.database.services.exam_projection import project_template_safe
+from modules.logger_tool import initialise_logger
+from routers.exam.dependencies import ExamContext, get_exam_context
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+
+router = APIRouter()
+
+
+def _rows(result: Any) -> List[Dict[str, Any]]:
+    data = getattr(result, "data", None)
+    if not data:
+        return []
+    return data if isinstance(data, list) else [data]
+
+
+def _first(result: Any) -> Optional[Dict[str, Any]]:
+    rows = _rows(result)
+    return rows[0] if rows else None
+
+
+@router.get("/bank")
+async def list_bank(
+    spec_ref: Optional[str] = None,
+    subject: Optional[str] = None,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    """Leaf questions across the caller's institute templates, with their source-paper context.
+
+    Filter by `spec_ref` (spec point) and/or `subject`. Facets over the full visible set drive the UI
+    filters (so filtering by one axis doesn't hide the others' options). This is the spec-planning surface:
+    a teacher picks a spec point → gets every question tagged it across their papers → assembles a paper.
+    """
+    sel = (
+        "id, template_id, label, max_marks, answer_type, spec_ref, bounds, page, "
+        "exam_templates(id, title, subject, exam_code)"
+    )
+    query = ctx.supabase.table("exam_questions").select(sel).eq("is_container", False)
+    if spec_ref:
+        query = query.eq("spec_ref", spec_ref)
+    rows = _rows(query.execute())
+
+    facets_spec: Dict[str, int] = {}
+    facets_subject: Dict[str, int] = {}
+    items: List[Dict[str, Any]] = []
+    for r in rows:
+        tmpl = r.get("exam_templates") or {}
+        subj = tmpl.get("subject")
+        if r.get("spec_ref"):
+            facets_spec[r["spec_ref"]] = facets_spec.get(r["spec_ref"], 0) + 1
+        if subj:
+            facets_subject[subj] = facets_subject.get(subj, 0) + 1
+        if subject and subj != subject:
+            continue
+        items.append({
+            "id": r["id"], "template_id": r["template_id"], "label": r.get("label"),
+            "max_marks": r.get("max_marks"), "answer_type": r.get("answer_type"),
+            "spec_ref": r.get("spec_ref"), "bounds": r.get("bounds"), "page": r.get("page"),
+            "paper": {"id": tmpl.get("id"), "title": tmpl.get("title"),
+                      "subject": subj, "exam_code": tmpl.get("exam_code")},
+        })
+    return {"questions": items, "n": len(items),
+            "facets": {"spec_ref": facets_spec, "subject": facets_subject}}
+
+
+class CustomPaperRequest(BaseModel):
+    title: str
+    subject: Optional[str] = None
+    institute_id: Optional[str] = None
+    question_ids: List[str]
+
+
+@router.post("/custom-papers")
+async def create_custom_paper(
+    body: CustomPaperRequest,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    """Assemble the selected bank questions into a new template (copy-on-assemble), then project it."""
+    if not body.question_ids:
+        raise HTTPException(status_code=400, detail="question_ids is required")
+    institute_id = ctx.resolve_institute(body.institute_id)
+
+    # RLS makes this return only leaf questions the caller may see; unknown/forbidden ids are silently dropped.
+    srcs = _rows(
+        ctx.supabase.table("exam_questions").select("*")
+        .in_("id", body.question_ids).eq("is_container", False).execute()
+    )
+    by_id = {s["id"]: s for s in srcs}
+    ordered = [by_id[qid] for qid in body.question_ids if qid in by_id]  # preserve the caller's order
+    if not ordered:
+        raise HTTPException(status_code=404, detail="No accessible questions for the given ids")
+
+    template_id = str(uuid.uuid4())
+    ctx.supabase.table("exam_templates").insert({
+        "id": template_id, "title": body.title, "subject": body.subject,
+        "institute_id": institute_id, "teacher_id": ctx.user_id, "status": "draft",
+    }).execute()
+
+    id_map: Dict[str, str] = {}
+    q_rows: List[Dict[str, Any]] = []
+    for order, s in enumerate(ordered):
+        new_id = str(uuid.uuid4())
+        id_map[s["id"]] = new_id
+        q_rows.append({
+            "id": new_id, "template_id": template_id, "parent_id": None,
+            "label": s.get("label"), "order": order, "max_marks": s.get("max_marks") or 0,
+            "answer_type": s.get("answer_type"), "mcq_options": s.get("mcq_options"),
+            "mark_scheme": s.get("mark_scheme") or {}, "is_container": False,
+            "spec_ref": s.get("spec_ref"), "bounds": s.get("bounds"), "page": s.get("page"),
+            "source": "manual", "confirmed": True,
+        })
+    ctx.supabase.table("exam_questions").insert(q_rows).execute()
+
+    # Copy each selected question's response areas onto its new copy (geometry travels with the question).
+    ra_rows: List[Dict[str, Any]] = []
+    if id_map:
+        for ra in _rows(
+            ctx.supabase.table("exam_response_areas").select("*")
+            .in_("question_id", list(id_map.keys())).execute()
+        ):
+            ra_rows.append({
+                "id": str(uuid.uuid4()), "question_id": id_map[ra["question_id"]], "template_id": template_id,
+                "page": ra.get("page"), "bounds": ra.get("bounds"), "kind": ra.get("kind"),
+                "response_form": ra.get("response_form"), "context_type": ra.get("context_type"),
+                "source": "manual", "confirmed": True,
+            })
+        if ra_rows:
+            ctx.supabase.table("exam_response_areas").insert(ra_rows).execute()
+
+    project_template_safe(template_id)  # a custom paper is a normal paper in the graph
+    return {"id": template_id, "title": body.title, "subject": body.subject,
+            "n_questions": len(q_rows), "n_response_areas": len(ra_rows)}

routers/exam/batches.py

@@ -245,6 +245,36 @@ async def batch_csv(
 
 # ─── marks ───────────────────────────────────────────────────────────────────
 
+def _advance_completion(ctx: ExamContext, batch_id: str, submission_id: str) -> None:
+    """After a mark upsert, advance statuses: a submission with a mark for every markable (leaf)
+    question → complete; a batch whose every non-absent submission is complete → complete. Nothing
+    here regresses a status (only promotes to complete), so it is safe to run on every upsert."""
+    batch = _first(
+        ctx.supabase.table("marking_batches").select("id, template_id, status").eq("id", batch_id).limit(1).execute()
+    )
+    if not batch:
+        return
+    markable = {
+        q["id"] for q in _rows(
+            ctx.supabase.table("exam_questions").select("id, is_container").eq("template_id", batch["template_id"]).execute()
+        ) if not q.get("is_container")
+    }
+    if not markable:
+        return
+    marked = {
+        m["question_id"] for m in _rows(
+            ctx.supabase.table("mark_entries").select("question_id").eq("submission_id", submission_id).execute()
+        )
+    }
+    if not markable.issubset(marked):
+        return
+    ctx.supabase.table("student_submissions").update({"status": "complete"}).eq("id", submission_id).execute()
+    subs = _rows(ctx.supabase.table("student_submissions").select("status").eq("batch_id", batch_id).execute())
+    active = [s for s in subs if s.get("status") != "absent"]
+    if active and all(s.get("status") == "complete" for s in active) and batch.get("status") != "complete":
+        ctx.supabase.table("marking_batches").update({"status": "complete"}).eq("id", batch_id).execute()
+
+
 @router.put("/marks/{mark_id}")
 async def upsert_mark(
     mark_id: str,
@@ -259,6 +289,15 @@ async def upsert_mark(
     if not submission:
         raise HTTPException(status_code=404, detail="Submission not found")
 
+    # Reject an award that exceeds the question's max (only when a max is actually set; 0/None means
+    # "not scored yet" for AI/unmapped questions, so we can't validate those).
+    question = _first(
+        ctx.supabase.table("exam_questions").select("id, max_marks").eq("id", body.question_id).limit(1).execute()
+    )
+    max_marks = (question or {}).get("max_marks")
+    if isinstance(max_marks, (int, float)) and max_marks > 0 and body.awarded_marks is not None and body.awarded_marks > max_marks:
+        raise HTTPException(status_code=422, detail=f"awarded_marks {body.awarded_marks} exceeds max_marks {max_marks} for this question")
+
     row = {
         "id": mark_id,
         "submission_id": body.submission_id,
@@ -285,6 +324,9 @@ async def upsert_mark(
     if submission.get("status") in ("absent", "unmatched"):
         ctx.supabase.table("student_submissions").update({"status": "marking"}).eq("id", body.submission_id).execute()
 
+    # Promote the submission/batch to complete once every markable question has a mark.
+    _advance_completion(ctx, submission["batch_id"], body.submission_id)
+
     return upserted
 
 

routers/exam/corpus.py

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

routers/exam/templates.py

@@ -465,6 +465,17 @@ def _safe_confidence(value: Any = None) -> float:
     return 0.75
 
 
+def _safe_marks(value: Any = None) -> int:
+    """Parsed per-part marks → a non-negative int; unknown/None → 0 (image-only OCR has no marks yet)."""
+    if isinstance(value, bool):
+        return 0
+    if isinstance(value, (int, float)):
+        return max(0, int(value))
+    if isinstance(value, str) and value.strip().isdigit():
+        return int(value.strip())
+    return 0
+
+
 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 []:
@@ -541,7 +552,7 @@ def _map_first_pass_to_rows(template_id: str, first_pass: Dict[str, Any], pdf_by
                 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"})
+            questions.append({"id": pid, "template_id": template_id, "parent_id": parent_id, "label": label, "order": len(questions), "max_marks": _safe_marks(band.get("marks")), "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)):
@@ -602,7 +613,13 @@ def _refresh_ai_rows(ctx: ExamContext, template_id: str, rows: Dict[str, List[Di
     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 = _dedupe_rows_by_id(rows.get(key) or [])
+        # A row that survived the delete above is confirmed-AI or manual — the teacher's curated version.
+        # AI ids are a deterministic uuid5 of (template_id, semantic key), so a re-run re-emits the SAME id
+        # for a confirmed ghost; re-inserting it would PK-collide and fail the whole batch. Skip those ids so
+        # confirmed work is preserved and the insert is safe (fixes the re-map-after-confirm collision).
+        kept = sb.table(table).select("id").eq("template_id", template_id).execute()
+        kept_ids = {str(r["id"]) for r in (getattr(kept, "data", None) or []) if isinstance(r, dict) and r.get("id")}
+        payload = [r for r in _dedupe_rows_by_id(rows.get(key) or []) if str(r.get("id")) not in kept_ids]
         if payload:
             sb.table(table).insert(payload).execute()
 

run/initialization/init_exam_graph.py

@@ -1,17 +1,19 @@
 """
 init_exam_graph.py — Initialise the cc.public.exams Neo4j knowledge graph.
 
-Creates the shared, public exam database, its uniqueness constraints, and seeds the AQA exam
-board + AQA GCSE Physics (8463) specification with its 8 top-level topic SpecPoints. Idempotent
-(CREATE DATABASE IF NOT EXISTS / CREATE CONSTRAINT IF NOT EXISTS / MERGE).
+Creates the shared, public exam database, its uniqueness constraints, and seeds the AQA exam board
++ the 6 current test specifications (GCSE & A-level Physics/Chemistry/Biology) with their top-level
+topic SpecPoints (44 in total). Idempotent (CREATE DATABASE IF NOT EXISTS / CREATE CONSTRAINT IF NOT
+EXISTS / MERGE).
 
 Run inside the ccapi container:
     python3 -c "from run.initialization.init_exam_graph import init; import json; print(json.dumps(init()))"
 
-NOTE: the 8 SpecPoints seeded here are the real AQA GCSE Physics *top-level* topics. The full
-sub-point breakdown (e.g. 4.1.1.1 ...) is a later data-population task (sourceable from the AQA
-spec PDF via Docling). spec_code AQA-PHYS-8463 is the standalone GCSE Physics code that matches
-"AQA Physics Paper 1H"; the eb_exams/eb_specifications seed (card S4-3) must use the same code.
+NOTE: only *top-level* topics are seeded (the granularity a teacher plans against). The full sub-point
+breakdown (e.g. 4.1.1.1 ...) is a later data-population task (sourceable from the AQA spec PDF via
+Docling). Seeding all 6 specs means a template's spec_ref finds a matching SpecPoint so
+(:Part)-[:ASSESSES]->(:SpecPoint) fires beyond GCSE Physics; spec_code (e.g. AQA-PHYS-8463) must match
+the eb_exams/eb_specifications seed (card S4-3) and the app's deriveSpecCode.
 """
 import uuid
 from typing import Dict, Any
@@ -41,6 +43,37 @@ SPEC_POINTS = [
     ("4.8", "Space physics"),
 ]
 
+# Full AQA catalogue for the current test specs (top-level topics; ref = topic number). Seeding all of
+# them means a template's spec_ref finds a matching SpecPoint so (:Part)-[:ASSESSES]->(:SpecPoint) fires
+# beyond AQA GCSE Physics. Sub-point granularity (e.g. 4.1.1.1) remains a later data-population task.
+SPECIFICATIONS = [
+    {**SPEC, "topics": SPEC_POINTS},
+    {"spec_code": "AQA-CHEM-8462", "exam_board_code": "AQA", "subject_code": "CHEM", "award_code": "GCSE",
+     "title": "AQA GCSE Chemistry (8462)", "topics": [
+        ("4.1", "Atomic structure and the periodic table"), ("4.2", "Bonding, structure, and the properties of matter"),
+        ("4.3", "Quantitative chemistry"), ("4.4", "Chemical changes"), ("4.5", "Energy changes"),
+        ("4.6", "The rate and extent of chemical change"), ("4.7", "Organic chemistry"), ("4.8", "Chemical analysis"),
+        ("4.9", "Chemistry of the atmosphere"), ("4.10", "Using resources")]},
+    {"spec_code": "AQA-BIOL-8461", "exam_board_code": "AQA", "subject_code": "BIOL", "award_code": "GCSE",
+     "title": "AQA GCSE Biology (8461)", "topics": [
+        ("4.1", "Cell biology"), ("4.2", "Organisation"), ("4.3", "Infection and response"), ("4.4", "Bioenergetics"),
+        ("4.5", "Homeostasis and response"), ("4.6", "Inheritance, variation and evolution"), ("4.7", "Ecology")]},
+    {"spec_code": "AQA-PHYS-7408", "exam_board_code": "AQA", "subject_code": "PHYS", "award_code": "A-level",
+     "title": "AQA A-level Physics (7408)", "topics": [
+        ("3.1", "Measurements and their errors"), ("3.2", "Particles and radiation"), ("3.3", "Waves"),
+        ("3.4", "Mechanics and materials"), ("3.5", "Electricity"), ("3.6", "Further mechanics and thermal physics"),
+        ("3.7", "Fields and their consequences"), ("3.8", "Nuclear physics")]},
+    {"spec_code": "AQA-CHEM-7405", "exam_board_code": "AQA", "subject_code": "CHEM", "award_code": "A-level",
+     "title": "AQA A-level Chemistry (7405)", "topics": [
+        ("3.1", "Physical chemistry"), ("3.2", "Inorganic chemistry"), ("3.3", "Organic chemistry")]},
+    {"spec_code": "AQA-BIOL-7402", "exam_board_code": "AQA", "subject_code": "BIOL", "award_code": "A-level",
+     "title": "AQA A-level Biology (7402)", "topics": [
+        ("3.1", "Biological molecules"), ("3.2", "Cells"), ("3.3", "Organisms exchange substances with their environment"),
+        ("3.4", "Genetic information, variation and relationships between organisms"),
+        ("3.5", "Energy transfers in and between organisms"), ("3.6", "Organisms respond to changes"),
+        ("3.7", "Genetics, populations, evolution and ecosystems"), ("3.8", "The control of gene expression")]},
+]
+
 CONSTRAINTS = [
     "CREATE CONSTRAINT exam_board_uid    IF NOT EXISTS FOR (n:ExamBoard)     REQUIRE n.uuid_string IS UNIQUE",
     "CREATE CONSTRAINT spec_uid          IF NOT EXISTS FOR (n:Specification) REQUIRE n.uuid_string IS UNIQUE",
@@ -81,35 +114,36 @@ def init() -> Dict[str, Any]:
             s.run(c).consume()
             result["constraints"] += 1
 
-        # 3. board + spec
+        # 3. board (once)
         board_uid = _uid("ExamBoard", BOARD["code"])
-        spec_uid = _uid("Specification", SPEC["spec_code"])
         s.run(
             "MERGE (b:ExamBoard {uuid_string:$uid}) "
             "SET b.code=$code, b.name=$name, b.node_storage_path=$nsp",
             uid=board_uid, code=BOARD["code"], name=BOARD["name"],
             nsp=f"{EXAM_DB}/ExamBoard/{BOARD['code']}",
         ).consume()
+
+        # 4. each specification + its top-level spec points (idempotent MERGE)
+        for spec in SPECIFICATIONS:
+            spec_uid = _uid("Specification", spec["spec_code"])
             s.run(
                 "MERGE (sp:Specification {uuid_string:$uid}) "
                 "SET sp.spec_code=$sc, sp.exam_board_code=$ebc, sp.subject_code=$subj, "
                 "    sp.award_code=$award, sp.title=$title, sp.node_storage_path=$nsp "
                 "WITH sp MATCH (b:ExamBoard {code:$ebc}) MERGE (b)-[:PUBLISHES]->(sp)",
-            uid=spec_uid, sc=SPEC["spec_code"], ebc=SPEC["exam_board_code"],
-            subj=SPEC["subject_code"], award=SPEC["award_code"], title=SPEC["title"],
-            nsp=f"{EXAM_DB}/Specification/{SPEC['spec_code']}",
+                uid=spec_uid, sc=spec["spec_code"], ebc=spec["exam_board_code"],
+                subj=spec["subject_code"], award=spec["award_code"], title=spec["title"],
+                nsp=f"{EXAM_DB}/Specification/{spec['spec_code']}",
             ).consume()
-
-        # 4. spec points
-        for ref, desc in SPEC_POINTS:
-            sp_uid = _uid("SpecPoint", SPEC["spec_code"], ref)
+            for ref, desc in spec["topics"]:
+                sp_uid = _uid("SpecPoint", spec["spec_code"], ref)
                 s.run(
                     "MERGE (p:SpecPoint {uuid_string:$uid}) "
                     "SET p.ref=$ref, p.description=$desc, p.spec_code=$sc, "
                     "    p.exam_board_code=$ebc, p.node_storage_path=$nsp "
                     "WITH p MATCH (s:Specification {spec_code:$sc}) MERGE (s)-[:HAS_SPEC_POINT]->(p)",
-                uid=sp_uid, ref=ref, desc=desc, sc=SPEC["spec_code"],
-                ebc=SPEC["exam_board_code"], nsp=f"{EXAM_DB}/SpecPoint/{SPEC['spec_code']}/{ref}",
+                    uid=sp_uid, ref=ref, desc=desc, sc=spec["spec_code"],
+                    ebc=spec["exam_board_code"], nsp=f"{EXAM_DB}/SpecPoint/{spec['spec_code']}/{ref}",
                 ).consume()
                 result["spec_points"] += 1
 

tests/test_exam_bank.py

@@ -0,0 +1,134 @@
+"""Tests for the question-bank + custom-paper assembly router (/api/exam/bank, /custom-papers) — mode 3.
+
+Same FakeSupabase + dependency-override pattern as test_exam_templates.py; the as-user RLS itself is
+verified against the live dev DB (.94), not here.
+"""
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+import routers.exam.bank as bank_mod
+from routers.exam.bank import router
+from routers.exam.dependencies import ExamContext, get_exam_context
+
+
+TEACHER = "00000000-0000-0000-0000-000000000001"
+INST_A = "10000000-0000-0000-0000-000000000001"
+
+
+@pytest.fixture(autouse=True)
+def _stub_projection(monkeypatch):
+    monkeypatch.setattr(bank_mod, "project_template_safe", lambda tid: None)
+
+
+class FakeResult:
+    def __init__(self, data):
+        self.data = data
+
+
+class FakeQuery:
+    def __init__(self, store, table):
+        self.store = store
+        self.table = table
+        self.rows = list(store.get(table, []))
+        self._op = None
+        self._payload = None
+
+    def select(self, *_a, **_k):
+        self._op = "select"
+        return self
+
+    def insert(self, payload):
+        self._op = "insert"
+        self._payload = payload
+        return self
+
+    def eq(self, key, value):
+        self.rows = [r for r in self.rows if r.get(key) == value]
+        return self
+
+    def in_(self, key, values):
+        values = set(values)
+        self.rows = [r for r in self.rows if r.get(key) in values]
+        return self
+
+    def execute(self):
+        backing = self.store.setdefault(self.table, [])
+        if self._op == "insert":
+            payloads = self._payload if isinstance(self._payload, list) else [self._payload]
+            inserted = []
+            for p in payloads:
+                row = dict(p)
+                row.setdefault("id", f"gen-{self.table}-{len(backing)}")
+                backing.append(row)
+                inserted.append(row)
+            return FakeResult(inserted)
+        return FakeResult(self.rows)
+
+
+class FakeSupabase:
+    def __init__(self, store):
+        self.store = store
+
+    def table(self, name):
+        return FakeQuery(self.store, name)
+
+
+def make_client(store, user_id=TEACHER, institute_ids=(INST_A,)):
+    app = FastAPI()
+    app.include_router(router, prefix="/api/exam")
+    app.dependency_overrides[get_exam_context] = lambda: ExamContext(user_id, "tok", FakeSupabase(store), list(institute_ids))
+    return TestClient(app), store
+
+
+def test_bank_lists_leaf_questions_with_facets_excluding_containers():
+    store = {"exam_questions": [
+        {"id": "c1", "template_id": "t1", "label": "Q1", "is_container": True, "spec_ref": None,
+         "exam_templates": {"id": "t1", "title": "Physics 1H", "subject": "physics", "exam_code": "AQA-PHYS-8463"}},
+        {"id": "q1", "template_id": "t1", "label": "01.1", "is_container": False, "max_marks": 3, "spec_ref": "4.2",
+         "exam_templates": {"id": "t1", "title": "Physics 1H", "subject": "physics", "exam_code": "AQA-PHYS-8463"}},
+        {"id": "q2", "template_id": "t1", "label": "01.2", "is_container": False, "max_marks": 2, "spec_ref": "4.2",
+         "exam_templates": {"id": "t1", "title": "Physics 1H", "subject": "physics", "exam_code": "AQA-PHYS-8463"}},
+    ]}
+    c, _ = make_client(store)
+    body = c.get("/api/exam/bank").json()
+    labels = {q["label"] for q in body["questions"]}
+    assert labels == {"01.1", "01.2"}  # container excluded
+    assert body["facets"]["spec_ref"] == {"4.2": 2}
+    assert body["questions"][0]["paper"]["exam_code"] == "AQA-PHYS-8463"
+
+
+def test_bank_filters_by_spec_ref():
+    store = {"exam_questions": [
+        {"id": "q1", "template_id": "t1", "label": "a", "is_container": False, "spec_ref": "4.2", "exam_templates": {"subject": "physics"}},
+        {"id": "q2", "template_id": "t1", "label": "b", "is_container": False, "spec_ref": "4.5", "exam_templates": {"subject": "physics"}},
+    ]}
+    c, _ = make_client(store)
+    body = c.get("/api/exam/bank?spec_ref=4.2").json()
+    assert [q["id"] for q in body["questions"]] == ["q1"]
+
+
+def test_custom_paper_copies_selected_questions_in_order():
+    store = {"exam_questions": [
+        {"id": "q1", "template_id": "t1", "label": "01.1", "is_container": False, "max_marks": 3, "spec_ref": "4.2", "answer_type": "short"},
+        {"id": "q2", "template_id": "t2", "label": "05.1", "is_container": False, "max_marks": 6, "spec_ref": "4.5", "answer_type": "written"},
+    ]}
+    c, store = make_client(store)
+    r = c.post("/api/exam/custom-papers", json={"title": "Electricity mini-test", "subject": "physics", "question_ids": ["q2", "q1"]})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["n_questions"] == 2
+    # a new template row was created, owned + institute-scoped
+    new_tpls = [t for t in store["exam_templates"] if t["id"] == body["id"]]
+    assert new_tpls and new_tpls[0]["teacher_id"] == TEACHER and new_tpls[0]["institute_id"] == INST_A
+    # two COPIES exist under the new template, order preserving the requested [q2, q1], with fresh ids
+    copies = sorted([q for q in store["exam_questions"] if q["template_id"] == body["id"]], key=lambda q: q["order"])
+    assert [q["label"] for q in copies] == ["05.1", "01.1"]
+    assert all(q["id"] not in ("q1", "q2") for q in copies)  # not the originals
+    assert copies[0]["max_marks"] == 6 and copies[0]["spec_ref"] == "4.5"
+
+
+def test_custom_paper_requires_accessible_questions():
+    c, _ = make_client({"exam_questions": []})
+    assert c.post("/api/exam/custom-papers", json={"title": "x", "question_ids": ["nope"]}).status_code == 404
+    assert c.post("/api/exam/custom-papers", json={"title": "x", "question_ids": []}).status_code == 400

tests/test_exam_batches.py

@@ -249,6 +249,30 @@ def test_upsert_mark_submission_404():
     assert c.put("/api/exam/marks/mk-x", json={"submission_id": "nope", "question_id": "q1", "awarded_marks": 1}).status_code == 404
 
 
+def test_upsert_mark_rejects_over_max():
+    c = make_client(_batch_with_cohort())  # q1 max_marks = 3
+    r = c.put("/api/exam/marks/mk-over", json={"submission_id": "sub2", "question_id": "q1", "awarded_marks": 4})
+    assert r.status_code == 422
+
+
+def test_upsert_mark_completes_submission_and_batch():
+    store = base_store(
+        marking_batches=[{"id": "b1", "template_id": TPL, "institute_id": INST_A, "teacher_id": TEACHER, "status": "marking"}],
+        exam_questions=[
+            {"id": "q0", "template_id": TPL, "label": "Q1", "max_marks": 0, "order": 0, "is_container": True},
+            {"id": "q1", "template_id": TPL, "label": "01", "max_marks": 3, "order": 1, "is_container": False},
+            {"id": "q2", "template_id": TPL, "label": "02", "max_marks": 5, "order": 2, "is_container": False},
+        ],
+        student_submissions=[{"id": "sub1", "batch_id": "b1", "student_id": "s1", "status": "marking"}],
+        mark_entries=[{"id": "m1", "batch_id": "b1", "submission_id": "sub1", "question_id": "q1", "awarded_marks": 2}],
+    )
+    c = make_client(store)
+    # marking the last leaf question completes the submission (container q0 doesn't block) and the batch
+    assert c.put("/api/exam/marks/m2", json={"submission_id": "sub1", "question_id": "q2", "awarded_marks": 4}).status_code == 200
+    assert next(s for s in store["student_submissions"] if s["id"] == "sub1")["status"] == "complete"
+    assert next(b for b in store["marking_batches"] if b["id"] == "b1")["status"] == "complete"
+
+
 # ─── scans (E3 guards) ───────────────────────────────────────────────────────
 
 def _batch_store():

tests/test_exam_corpus.py

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

tests/test_exam_extract_mapping.py

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

tests/test_exam_templates.py

@@ -642,6 +642,21 @@ def test_auto_map_fast_path_merges_ai_rows_and_returns_detail(monkeypatch):
     assert store["exam_boundaries"] and store["exam_boundaries"][0]["derivation"] == "docling-main-band"
 
 
+def test_auto_map_surfaces_born_digital_part_marks(monkeypatch):
+    # Regression: the born-digital grammar parses per-part marks, but the row mapper hardcoded
+    # max_marks=0. A part band carrying `marks` must flow through to the question row's max_marks.
+    store = _template_with_source()
+    store.update({"exam_questions": [], "exam_response_areas": [], "exam_boundaries": [], "exam_template_layout": []})
+    client, store = make_client(store=store)
+    fp = _first_pass_template()
+    fp["pages"]["1"]["part_bands"][0]["marks"] = 4
+    _patch_auto_map(monkeypatch, store, fast=True)
+    monkeypatch.setattr(templates_mod, "auto_map", lambda *_a, **_k: fp)  # override with the marked part band
+    assert client.post("/api/exam/templates/t1/auto-map").status_code == 200
+    part = next(q for q in store["exam_questions"] if q.get("label") == "01.1")
+    assert part["max_marks"] == 4
+
+
 def test_auto_map_deduplicates_repeated_response_area_ids(monkeypatch):
     store = _template_with_source()
     client, store = make_client(store=store)
@@ -674,6 +689,27 @@ def test_auto_map_preserves_manual_and_confirmed_rows_on_rerun(monkeypatch):
     assert "old-ai" not in ids
 
 
+def test_auto_map_rerun_after_confirm_does_not_pk_collide(monkeypatch):
+    # Regression: a confirmed ghost keeps its DETERMINISTIC uuid5 id, which auto-map re-emits on a
+    # re-run — the old code re-inserted that id and PK-collided. Use the real generated id (not an
+    # arbitrary one, unlike the test above) so the collision path is actually exercised.
+    store = _template_with_source()
+    store.update({"exam_questions": [], "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
+    ghost_id = next(q["id"] for q in store["exam_questions"] if q["source"] == "ai")
+    # teacher confirms that ghost (row kept, deterministic id unchanged)
+    for q in store["exam_questions"]:
+        if q["id"] == ghost_id:
+            q["confirmed"] = True
+    # re-run auto-map: must not re-insert the same id, and must preserve the teacher's confirmation
+    assert client.post("/api/exam/templates/t1/auto-map").status_code == 200
+    same = [r for r in store["exam_questions"] if r["id"] == ghost_id]
+    assert len(same) == 1, "confirmed ghost must not be duplicated (PK collision) on re-run"
+    assert same[0]["confirmed"] is True, "teacher's confirmation must survive a re-run"
+
+
 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)