feat(exam): add metadata patch for templates

The download path now resolves the files row via SupabaseServiceRoleClient (to
sidestep the cabinet_memberships RLS recursion); the test must mock it like the
upload test does. Test-only.

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

modules/database/schemas/nodes/exams/exam_nodes.py

@@ -0,0 +1,92 @@
+"""
+Neontology node schemas for the cc.public.exams knowledge graph.
+
+cc.public.exams is a dedicated, shared, public Neo4j database — co-primary/authoritative for the
+exam knowledge graph (specs, spec-points, paper→question→part→region structure, ASSESSES links).
+Supabase remains source of truth for operational data (geometry, marks, submissions); the two
+layers join on shared UUIDs:
+
+    exam_questions.id        <-> Question|Part.uuid_string   (container -> Question, leaf -> Part)
+    exam_response_areas.id   <-> Region.uuid_string
+    eb_exams.exam_code       <-> ExamPaper.exam_code
+    eb_specifications.spec_code <-> Specification.spec_code
+
+Ownership: created by an infra-init step; read by all authenticated API calls; written by the API
+service role only (no direct client writes).
+"""
+from typing import ClassVar, Optional
+from ..base_nodes import CCBaseNode
+
+
+class ExamBaseNode(CCBaseNode):
+    __primarylabel__: ClassVar[str] = ''
+
+
+class ExamBoardNode(ExamBaseNode):
+    __primarylabel__: ClassVar[str] = 'ExamBoard'
+    code: str                      # 'AQA'
+    name: str
+
+
+class SpecificationNode(ExamBaseNode):
+    __primarylabel__: ClassVar[str] = 'Specification'
+    spec_code: str                 # 'AQA-PHYS-8463'  (== Supabase eb_specifications.spec_code)
+    exam_board_code: str
+    subject_code: Optional[str] = None
+    award_code: Optional[str] = None
+    title: Optional[str] = None
+
+
+class SpecPointNode(ExamBaseNode):
+    __primarylabel__: ClassVar[str] = 'SpecPoint'
+    ref: str                       # '4.1', '4.2.1'
+    description: str
+    spec_code: str
+    exam_board_code: str
+
+
+class ExamPaperNode(ExamBaseNode):
+    __primarylabel__: ClassVar[str] = 'ExamPaper'
+    exam_code: str                 # == Supabase eb_exams.exam_code
+    spec_code: str
+    paper_code: Optional[str] = None
+    tier: Optional[str] = None
+    session: Optional[str] = None
+    title: Optional[str] = None
+    page_count: Optional[int] = None
+
+
+class QuestionNode(ExamBaseNode):  # roll-up container; uuid_string == exam_questions.id
+    __primarylabel__: ClassVar[str] = 'Question'
+    exam_code: str
+    label: str                     # '01'
+    order: int
+    max_marks: float
+
+
+class PartNode(ExamBaseNode):      # leaf; uuid_string == exam_questions.id
+    __primarylabel__: ClassVar[str] = 'Part'
+    exam_code: str
+    label: str                     # '01.1'
+    order: int
+    max_marks: float
+    answer_type: str
+    mark_scheme_type: str
+
+
+class RegionNode(ExamBaseNode):    # uuid_string == exam_response_areas.id
+    __primarylabel__: ClassVar[str] = 'Region'
+    page: int
+    kind: str                      # 'response' | 'context'
+    response_form: str
+
+
+# Relationship reference (written by the projection / linker, not modelled as classes here):
+#   (:ExamBoard)-[:PUBLISHES]->(:Specification)
+#   (:Specification)-[:HAS_SPEC_POINT]->(:SpecPoint)
+#   (:Specification)-[:HAS_PAPER]->(:ExamPaper)
+#   (:ExamPaper)-[:HAS_QUESTION]->(:Question)
+#   (:Question)-[:HAS_PART]->(:Part)        # nested questions allowed
+#   (:Part)-[:HAS_REGION]->(:Region)
+#   (:Part)-[:ASSESSES]->(:SpecPoint)       # from exam_questions.spec_ref
+#   (:SpecPoint)-[:TEACHES]->(:LearningStatement)   # DEFERRED cross-db bridge

modules/database/services/exam_projection.py

@@ -0,0 +1,168 @@
+"""Project a saved exam template into the cc.public.exams Neo4j graph (card S4-7).
+
+Supabase is source of truth for the operational template (geometry, marks); cc.public.exams is
+the co-primary knowledge graph (spec §2/S2). On template save the structural skeleton —
+ExamPaper → Question/Part → Region, plus Part-[:ASSESSES]->SpecPoint — is projected here.
+
+Ownership model (R3.5.1): the graph is written by the API SERVICE ROLE only (no client writes),
+so this task reads the template via service role and writes Neo4j with the system driver. It is
+the sanctioned service-role path (documented in the ADR), distinct from the as-user request path.
+
+Join keys (never regenerated):
+    exam_questions.id       -> Question.uuid_string (container) | Part.uuid_string (leaf)
+    exam_response_areas.id  -> Region.uuid_string
+    eb_exams.exam_code      -> ExamPaper.exam_code
+    eb_specifications.spec_code -> Specification.spec_code
+
+Projection is a full re-sync per exam_code (delete this paper's Question/Part/Region, recreate),
+matching the PUT full-replace semantics — idempotent and safe to re-run.
+"""
+import os
+import uuid
+from typing import Any, Dict, List, Optional
+
+from modules.database.supabase.utils.client import SupabaseServiceRoleClient
+from modules.database.tools.neo4j_driver_tools import get_session
+from modules.logger_tool import initialise_logger
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+
+# MUST match run/initialization/init_exam_graph.py (shared DB name + deterministic uuid namespace).
+EXAM_DB = "cc.public.exams"
+NS = uuid.UUID("00000000-0000-0000-0000-00000000e8a1")
+
+
+def _uid(*parts: str) -> str:
+    return str(uuid.uuid5(NS, ":".join(parts)))
+
+
+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 project_template(template_id: str) -> Dict[str, Any]:
+    """Read the template (service role) and (re)project its structure into cc.public.exams.
+
+    Returns a counts dict. Raises on hard failure (caller decides whether to swallow — a
+    BackgroundTask logs and drops; the manual /neo4j-sync endpoint surfaces the error).
+    """
+    sb = SupabaseServiceRoleClient().supabase
+    template = (sb.table("exam_templates").select("*").eq("id", template_id).limit(1).execute().data or [None])[0]
+    if not template:
+        raise ValueError(f"template {template_id} not found")
+
+    questions = _rows(sb.table("exam_questions").select("*").eq("template_id", template_id).order("order").execute())
+    regions = _rows(sb.table("exam_response_areas").select("*").eq("template_id", template_id).execute())
+
+    # Resolve the paper's exam_code + spec metadata. Catalogue paper → from eb_exams; ad-hoc upload
+    # (no exam_code) → a stable synthetic code so the paper still has a unique graph key.
+    exam_code = template.get("exam_code")
+    spec_code = None
+    paper_meta: Dict[str, Any] = {}
+    if template.get("exam_id"):
+        eb = (sb.table("eb_exams").select("exam_code, spec_code, paper_code, tier, session")
+              .eq("id", template["exam_id"]).limit(1).execute().data or [None])[0]
+        if eb:
+            exam_code = exam_code or eb.get("exam_code")
+            spec_code = eb.get("spec_code")
+            paper_meta = eb
+    if not exam_code:
+        exam_code = f"tpl:{template_id}"
+
+    paper_uid = _uid("ExamPaper", exam_code)
+    counts = {"exam_code": exam_code, "questions": 0, "parts": 0, "regions": 0, "assesses": 0, "spec_linked": False}
+
+    with get_session(database=EXAM_DB) as s:
+        # 1. Clean this paper's existing children (full re-sync), keep the ExamPaper node itself.
+        s.run("MATCH (r:Region {exam_code:$ec}) DETACH DELETE r", ec=exam_code).consume()
+        s.run("MATCH (n {exam_code:$ec}) WHERE n:Question OR n:Part DETACH DELETE n", ec=exam_code).consume()
+
+        # 2. ExamPaper node.
+        s.run(
+            "MERGE (p:ExamPaper {uuid_string:$uid}) "
+            "SET p.exam_code=$ec, p.spec_code=$sc, p.title=$title, p.page_count=$pc, "
+            "    p.paper_code=$paper_code, p.tier=$tier, p.session=$session, p.node_storage_path=$nsp",
+            uid=paper_uid, ec=exam_code, sc=spec_code, title=template.get("title"),
+            pc=template.get("page_count"), paper_code=paper_meta.get("paper_code"),
+            tier=paper_meta.get("tier"), session=paper_meta.get("session"),
+            nsp=f"{EXAM_DB}/ExamPaper/{exam_code}",
+        ).consume()
+
+        # 3. Link to its Specification (seeded separately) when known.
+        if spec_code:
+            r = s.run(
+                "MATCH (sp:Specification {spec_code:$sc}), (p:ExamPaper {exam_code:$ec}) "
+                "MERGE (sp)-[:HAS_PAPER]->(p) RETURN count(*) AS n",
+                sc=spec_code, ec=exam_code,
+            ).single()
+            counts["spec_linked"] = bool(r and r["n"])
+
+        # 4. Question/Part nodes — pass 1: create all nodes (so parents exist before linking).
+        for q in questions:
+            label = "Question" if q.get("is_container") else "Part"
+            s.run(
+                f"MERGE (n:{label} {{uuid_string:$uid}}) "
+                "SET n.exam_code=$ec, n.label=$label, n.order=$order, n.max_marks=$mm, "
+                "    n.answer_type=$at, n.mark_scheme_type=$mst, n.spec_ref=$sref, "
+                "    n.node_storage_path=$nsp",
+                uid=q["id"], ec=exam_code, label=q.get("label"), order=q.get("order") or 0,
+                mm=q.get("max_marks") or 0, at=q.get("answer_type"),
+                mst=(q.get("mark_scheme") or {}).get("type") if isinstance(q.get("mark_scheme"), dict) else None,
+                sref=q.get("spec_ref"), nsp=f"{EXAM_DB}/{label}/{q['id']}",
+            ).consume()
+            counts["parts" if label == "Part" else "questions"] += 1
+
+        # 5. Structural + ASSESSES edges — pass 2.
+        for q in questions:
+            if q.get("parent_id"):
+                s.run(
+                    "MATCH (parent {uuid_string:$pid}), (n {uuid_string:$uid}) MERGE (parent)-[:HAS_PART]->(n)",
+                    pid=q["parent_id"], uid=q["id"],
+                ).consume()
+            else:
+                s.run(
+                    "MATCH (p:ExamPaper {exam_code:$ec}), (n {uuid_string:$uid}) MERGE (p)-[:HAS_QUESTION]->(n)",
+                    ec=exam_code, uid=q["id"],
+                ).consume()
+            if q.get("spec_ref"):
+                # SpecPoints are seeded per spec_code; match within this paper's spec when known.
+                r = s.run(
+                    "MATCH (n {uuid_string:$uid}), (sp:SpecPoint {ref:$ref}) "
+                    + ("WHERE sp.spec_code=$sc " if spec_code else "")
+                    + "MERGE (n)-[:ASSESSES]->(sp) RETURN count(*) AS n",
+                    uid=q["id"], ref=q["spec_ref"], sc=spec_code,
+                ).single()
+                counts["assesses"] += (r["n"] if r else 0)
+
+        # 6. Region nodes + HAS_REGION edges.
+        # Only response/context regions are part of the knowledge graph (RegionNode.kind). The other
+        # S4-9 kinds (question_number, mark_area, reference, furniture) are physical-layer metadata
+        # about the paper, not curriculum structure — they stay in Supabase, out of cc.public.exams.
+        for rg in regions:
+            if rg.get("kind") not in ("response", "context"):
+                continue
+            s.run(
+                "MERGE (r:Region {uuid_string:$uid}) "
+                "SET r.exam_code=$ec, r.page=$page, r.kind=$kind, r.response_form=$rf, r.node_storage_path=$nsp",
+                uid=rg["id"], ec=exam_code, page=rg.get("page"), kind=rg.get("kind"),
+                rf=rg.get("response_form"), nsp=f"{EXAM_DB}/Region/{rg['id']}",
+            ).consume()
+            s.run(
+                "MATCH (q {uuid_string:$qid}), (r:Region {uuid_string:$uid}) MERGE (q)-[:HAS_REGION]->(r)",
+                qid=rg["question_id"], uid=rg["id"],
+            ).consume()
+            counts["regions"] += 1
+
+    logger.info(f"Projected template {template_id} → cc.public.exams: {counts}")
+    return counts
+
+
+def project_template_safe(template_id: str) -> None:
+    """BackgroundTask wrapper: never raises (a failed projection must not break the HTTP save)."""
+    try:
+        project_template(template_id)
+    except Exception as exc:
+        logger.error(f"Background Neo4j projection failed for template {template_id}: {exc}")

modules/database/supabase/utils/client.py

@@ -24,6 +24,10 @@ def _create_base_client(url: str, key: str, access_token: Optional[str] = None,
     # Otherwise fall back to the API key
     auth_header = f"Bearer {access_token}" if access_token else f"Bearer {key}"
 
+    # Only override Authorization here. apikey is supplied to create_client via the `key` arg and
+    # set by supabase-py itself; setting it again here sends a DUPLICATE apikey header that the
+    # Supabase gateway (Kong) rejects with 401 "Duplicate API key found". For a per-user client
+    # apikey stays the anon key (from `key`) while this Authorization carries the user JWT.
     headers = {
         "Authorization": auth_header,
     }

modules/database/supabase/utils/storage.py

@@ -23,76 +23,76 @@ class StorageManager:
     def check_bucket_exists(self, bucket_id: str) -> bool:
         """Check if a storage bucket exists"""
         try:
-            self.logger.info(f"Checking if bucket {{bucket_id}} exists")
+            self.logger.info(f"Checking if bucket {bucket_id} exists")
             buckets = self.client.supabase.storage.list_buckets()
             return any(bucket.name == bucket_id for bucket in buckets)
         except Exception as e:
-            self.logger.error(f"Error checking bucket {{bucket_id}}: {{str(e)}}")
+            self.logger.error(f"Error checking bucket {bucket_id}: {str(e)}")
             return False
 
     def list_bucket_contents(self, bucket_id: str, path: str = "") -> Dict:
         """List contents of a bucket at specified path"""
         try:
-            self.logger.info(f"Listing contents of bucket {{bucket_id}} at path {{path}}")
+            self.logger.info(f"Listing contents of bucket {bucket_id} at path {path}")
             contents = self.client.supabase.storage.from_(bucket_id).list(path)
-            return {{
+            return {
                 "folders": [item for item in contents if item.get("id", "").endswith("/")],
                 "files": [item for item in contents if not item.get("id", "").endswith("/")]
-            }}
+            }
         except Exception as e:
-            self.logger.error(f"Error listing bucket contents: {{str(e)}}")
+            self.logger.error(f"Error listing bucket contents: {str(e)}")
             raise StorageError(str(e))
 
     def upload_file(self, bucket_id: str, file_path: str, file_data: bytes, content_type: str, upsert: bool = True) -> Any:
         """Upload a file to a storage bucket"""
         try:
-            self.logger.info(f"Uploading file to {{bucket_id}} at path {{file_path}}")
+            self.logger.info(f"Uploading file to {bucket_id} at path {file_path}")
             return self.client.supabase.storage.from_(bucket_id).upload(
                 path=file_path,
                 file=file_data,
-                file_options={{
+                file_options={
                     "content-type": content_type,
                     "x-upsert": "true" if upsert else "false"
-                }}
+                }
             )
         except Exception as e:
-            self.logger.error(f"Error uploading file: {{str(e)}}")
+            self.logger.error(f"Error uploading file: {str(e)}")
             raise StorageError(str(e))
 
     def download_file(self, bucket_id: str, file_path: str) -> bytes:
         """Download a file from a storage bucket"""
         try:
-            self.logger.info(f"Downloading file from {{bucket_id}} at path {{file_path}}")
+            self.logger.info(f"Downloading file from {bucket_id} at path {file_path}")
             return self.client.supabase.storage.from_(bucket_id).download(file_path)
         except Exception as e:
-            self.logger.error(f"Error downloading file: {{str(e)}}")
+            self.logger.error(f"Error downloading file: {str(e)}")
             raise StorageError(str(e))
 
     def delete_file(self, bucket_id: str, file_path: str) -> None:
         """Delete a file from a storage bucket"""
         try:
-            self.logger.info(f"Deleting file from {{bucket_id}} at path {{file_path}}")
+            self.logger.info(f"Deleting file from {bucket_id} at path {file_path}")
             self.client.supabase.storage.from_(bucket_id).remove([file_path])
         except Exception as e:
-            self.logger.error(f"Error deleting file: {{str(e)}}")
+            self.logger.error(f"Error deleting file: {str(e)}")
             raise StorageError(str(e))
 
     def get_public_url(self, bucket_id: str, file_path: str) -> str:
         """Get public URL for a file"""
         try:
-            self.logger.info(f"Getting public URL for file in {{bucket_id}} at path {{file_path}}")
+            self.logger.info(f"Getting public URL for file in {bucket_id} at path {file_path}")
             return self.client.supabase.storage.from_(bucket_id).get_public_url(file_path)
         except Exception as e:
-            self.logger.error(f"Error getting public URL: {{str(e)}}")
+            self.logger.error(f"Error getting public URL: {str(e)}")
             raise StorageError(str(e))
 
     def create_signed_url(self, bucket_id: str, file_path: str, expires_in: int = 3600) -> Any:
         """Create a signed URL for temporary file access"""
         try:
-            self.logger.info(f"Creating signed URL for file in {{bucket_id}} at path {{file_path}}")
+            self.logger.info(f"Creating signed URL for file in {bucket_id} at path {file_path}")
             return self.client.supabase.storage.from_(bucket_id).create_signed_url(file_path, expires_in)
         except Exception as e:
-            self.logger.error(f"Error creating signed URL: {{str(e)}}")
+            self.logger.error(f"Error creating signed URL: {str(e)}")
             raise StorageError(str(e))
 
 class StorageAdmin(StorageManager):
@@ -115,9 +115,9 @@ class StorageAdmin(StorageManager):
     ) -> Dict[str, Any]:
         """Create a new storage bucket with supported parameters."""
         try:
-            self.logger.info(f"Creating bucket {{id}} with name {{name}}")
+            self.logger.info(f"Creating bucket {id} with name {name}")
             
-            options: Optional[CreateBucketOptions] = {{}}
+            options: Optional[CreateBucketOptions] = {}
             if public:
                 options["public"] = public
             if file_size_limit is not None:
@@ -133,7 +133,7 @@ class StorageAdmin(StorageManager):
             return bucket
             
         except Exception as e:
-            self.logger.error(f"Error creating bucket {{id}}: {{str(e)}}")
+            self.logger.error(f"Error creating bucket {id}: {str(e)}")
             raise StorageError(str(e))
 
     def initialize_core_buckets(self, admin_user_id: Optional[str] = None) -> List[Dict[str, Any]]:
@@ -144,7 +144,7 @@ class StorageAdmin(StorageManager):
                 raise ValueError("Admin user ID is required for bucket initialization")
 
             core_buckets = [
-                {{
+                {
                     "id": "cc.users",
                     "name": "CC Users",
                     "public": False,
@@ -156,8 +156,8 @@ class StorageAdmin(StorageManager):
                         'application/msword', 'application/vnd.openxmlformats-officedocument.*',
                         'text/plain', 'text/csv', 'application/json'
                     ]
-                }},
+                },
-                {{
+                {
                     "id": "cc.institutes",
                     "name": "CC Institutes",
                     "public": False,
@@ -169,7 +169,7 @@ class StorageAdmin(StorageManager):
                         'application/msword', 'application/vnd.openxmlformats-officedocument.*',
                         'text/plain', 'text/csv', 'application/json'
                     ]
-                }}
+                }
             ]
             
             results = []
@@ -177,30 +177,30 @@ class StorageAdmin(StorageManager):
                 try:
                     bucket_name = bucket.pop("name")
                     result = self.create_bucket(name=bucket_name, **bucket)
-                    results.append({{
+                    results.append({
                         "bucket": bucket["id"],
                         "status": "success",
                         "result": result
-                    }})
+                    })
                 except Exception as e:
-                    self.logger.error(f"Error creating bucket {{bucket['id']}}: {{str(e)}}")
+                    self.logger.error(f"Error creating bucket {bucket['id']}: {str(e)}")
-                    results.append({{
+                    results.append({
                         "bucket": bucket["id"],
                         "status": "error",
                         "error": str(e)
-                    }})
+                    })
             
             return results
             
         except Exception as e:
-            self.logger.error(f"Error initializing core buckets: {{str(e)}}")
+            self.logger.error(f"Error initializing core buckets: {str(e)}")
             raise StorageError(str(e))
 
     def create_user_bucket(self, user_id: str, username: str) -> Dict[str, Any]:
         """Create a storage bucket for a specific user."""
         try:
-            bucket_id = f"cc.users.admin.{{username}}"
+            bucket_id = f"cc.users.admin.{username}"
-            bucket_name = f"User Files - {{username}}"
+            bucket_name = f"User Files - {username}"
             
             return self.create_bucket(
                 id=bucket_id,
@@ -217,7 +217,7 @@ class StorageAdmin(StorageManager):
             )
             
         except Exception as e:
-            self.logger.error(f"Error creating user bucket for {{username}}: {{str(e)}}")
+            self.logger.error(f"Error creating user bucket for {username}: {str(e)}")
             raise StorageError(str(e))
 
     def create_school_buckets(self, school_id: str, school_name: str, admin_user_id: Optional[str] = None) -> Dict[str, Any]:
@@ -228,9 +228,9 @@ class StorageAdmin(StorageManager):
                 raise ValueError("Admin user ID is required for school bucket creation")
 
             school_buckets = [
-                {{
+                {
-                    "id": f"cc.institutes.{{school_id}}.public",
+                    "id": f"cc.institutes.{school_id}.public",
-                    "name": f"{{school_name}} - Public Files",
+                    "name": f"{school_name} - Public Files",
                     "public": True,
                     "owner": owner_id,
                     "owner_id": school_id,
@@ -240,10 +240,10 @@ class StorageAdmin(StorageManager):
                         'application/msword', 'application/vnd.openxmlformats-officedocument.*',
                         'text/plain', 'text/csv', 'application/json'
                     ]
-                }},
+                },
-                {{
+                {
-                    "id": f"cc.institutes.{{school_id}}.private",
+                    "id": f"cc.institutes.{school_id}.private",
-                    "name": f"{{school_name}} - Private Files",
+                    "name": f"{school_name} - Private Files",
                     "public": False,
                     "owner": owner_id,
                     "owner_id": school_id,
@@ -253,29 +253,29 @@ class StorageAdmin(StorageManager):
                         'application/msword', 'application/vnd.openxmlformats-officedocument.*',
                         'text/plain', 'text/csv', 'application/json'
                     ]
-                }}
+                }
             ]
             
-            results = {{}}
+            results = {}
             for bucket in school_buckets:
                 try:
                     bucket_name = bucket.pop("name")
                     result = self.create_bucket(name=bucket_name, **bucket)
-                    results[bucket["id"]] = {{
+                    results[bucket["id"]] = {
                         "status": "success",
                         "result": result
-                    }}
+                    }
                 except Exception as e:
-                    self.logger.error(f"Error creating school bucket {{bucket['id']}}: {{str(e)}}")
+                    self.logger.error(f"Error creating school bucket {bucket['id']}: {str(e)}")
-                    results[bucket["id"]] = {{
+                    results[bucket["id"]] = {
                         "status": "error",
                         "error": str(e)
-                    }}
+                    }
             
             return results
             
         except Exception as e:
-            self.logger.error(f"Error creating school buckets: {{str(e)}}")
+            self.logger.error(f"Error creating school buckets: {str(e)}")
             raise StorageError(str(e))
 
 class StorageUser(StorageManager):

routers/exam/__init__.py

@@ -0,0 +1,16 @@
+"""Exam-marker API package (/api/exam/).
+
+A clean top-level router group (R5.1/E5), deliberately NOT nested under /database/. Every
+endpoint authenticates the JWT and calls Supabase as-the-user so the RLS in
+volumes/db/cc/72-exam-marker.sql is enforced (spec E1/E2 fixes).
+"""
+from fastapi import APIRouter
+
+from routers.exam.templates import router as templates_router
+from routers.exam.batches import router as batches_router
+
+router = APIRouter()
+router.include_router(templates_router)
+router.include_router(batches_router)
+
+__all__ = ["router"]

routers/exam/batches.py

@@ -0,0 +1,362 @@
+"""Marking batches, scans, marks, results & CSV (/api/exam/batches..., /api/exam/marks/...) — S4-6.
+
+As with templates, all user-facing access is as-the-user (RLS-enforced; E1). A batch is owned by
+the teacher who creates it (R2.4); colleagues in the same institute can read it
+(marking_batches_read), a teacher in another institute cannot (→ 404, IDOR-safe).
+
+Roster→cohort (R4.3/A7): creating a batch from a class materialises one student_submissions row
+per active enrollee (status='absent'), so every enrolled student is present in results/CSV from
+the start and a no-show is never silently dropped. The roster ids are read AS THE USER from
+class_students (cs_read requires the caller to teach/admin the class); only the display names are
+resolved via service role (profiles is deny-all as-user, E4 — see resolve_student_names).
+
+Scans (R2.3/E3): the upload endpoint enforces a max size and validates that the bytes are a PDF
+before storing. QR-decode + automatic student-matching is a follow-on (no QR'd fixtures exist
+until the PrintGenerator card); v1 supports explicit (manual) and ordered matching.
+"""
+from __future__ import annotations
+
+import csv
+import io
+import os
+import uuid
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
+from fastapi.responses import Response
+
+from modules.database.supabase.utils.client import SupabaseServiceRoleClient
+from modules.database.supabase.utils.storage import StorageAdmin
+from modules.logger_tool import initialise_logger
+from routers.exam.dependencies import ExamContext, get_exam_context, resolve_student_names
+from routers.exam.schemas import CreateBatchRequest, MarkUpsertRequest
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+
+router = APIRouter()
+
+# E3: bound the upload so a 36-page scan batch can't exhaust memory / be a DoS vector.
+MAX_SCAN_BYTES = int(os.getenv("EXAM_SCAN_MAX_BYTES", str(50 * 1024 * 1024)))  # 50 MB default
+SCANS_BUCKET = os.getenv("EXAM_SCANS_BUCKET", "cc.users")
+SCANS_PREFIX = "exam-submissions"
+
+
+# ─── helpers ─────────────────────────────────────────────────────────────────
+
+def _rows(result: Any) -> List[Dict[str, Any]]:
+    data = getattr(result, "data", None)
+    if not data:
+        return []
+    return data if isinstance(data, list) else [data]
+
+
+def _first(result: Any) -> Optional[Dict[str, Any]]:
+    rows = _rows(result)
+    return rows[0] if rows else None
+
+
+def _fetch_batch_or_404(ctx: ExamContext, batch_id: str) -> Dict[str, Any]:
+    row = _first(ctx.supabase.table("marking_batches").select("*").eq("id", batch_id).limit(1).execute())
+    if not row:
+        raise HTTPException(status_code=404, detail="Batch not found")
+    return row
+
+
+def _require_owner(ctx: ExamContext, batch: Dict[str, Any]) -> None:
+    if batch.get("teacher_id") != ctx.user_id:
+        raise HTTPException(status_code=403, detail="Only the batch owner can modify it")
+
+
+# ─── batches ─────────────────────────────────────────────────────────────────
+
+@router.post("/batches")
+async def create_batch(
+    body: CreateBatchRequest,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    # The batch inherits the template's institute; reading the template as-user also proves the
+    # caller may see it (RLS) — an unseeable template → 404.
+    template = _first(
+        ctx.supabase.table("exam_templates").select("id, institute_id").eq("id", body.template_id).limit(1).execute()
+    )
+    if not template:
+        raise HTTPException(status_code=404, detail="Template not found")
+
+    batch_row = {
+        "template_id": body.template_id,
+        "class_id": body.class_id,
+        "institute_id": template["institute_id"],
+        "teacher_id": ctx.user_id,
+        "title": body.title,
+        "status": "open",
+    }
+    batch_row = {k: v for k, v in batch_row.items() if v is not None}
+    batch = _first(ctx.supabase.table("marking_batches").insert(batch_row).execute())
+    if not batch:
+        raise HTTPException(status_code=500, detail="Failed to create batch")
+    batch_id = batch["id"]
+
+    seeded = 0
+    if body.class_id:
+        # Roster read is AS THE USER → cs_read requires the caller to teach/admin the class.
+        roster = _rows(
+            ctx.supabase.table("class_students")
+            .select("student_id")
+            .eq("class_id", body.class_id)
+            .eq("status", "active")
+            .execute()
+        )
+        student_ids = [r["student_id"] for r in roster if r.get("student_id")]
+        names = resolve_student_names(student_ids)
+        if student_ids:
+            sub_rows = [
+                {
+                    "batch_id": batch_id,
+                    "student_id": sid,
+                    "student_name": names.get(sid),
+                    "status": "absent",  # A7: present in results until a scan is matched
+                }
+                for sid in student_ids
+            ]
+            ctx.supabase.table("student_submissions").insert(sub_rows).execute()
+            seeded = len(sub_rows)
+
+    logger.info(f"Marking batch {batch_id} created by {ctx.user_id}; {seeded} roster submissions seeded")
+    return {**batch, "submission_count": seeded}
+
+
+@router.get("/batches")
+async def list_batches(
+    include_archived: bool = False,
+    template_id: Optional[str] = None,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    q = ctx.supabase.table("marking_batches").select("*")
+    if template_id:
+        q = q.eq("template_id", template_id)
+    if not include_archived:
+        q = q.neq("status", "archived")
+    return {"batches": _rows(q.order("created_at", desc=True).execute())}
+
+
+@router.get("/batches/{batch_id}/queue")
+async def batch_queue(
+    batch_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    batch = _fetch_batch_or_404(ctx, batch_id)
+    submissions = _rows(
+        ctx.supabase.table("student_submissions").select("*").eq("batch_id", batch_id).execute()
+    )
+    marks = _rows(ctx.supabase.table("mark_entries").select("submission_id").eq("batch_id", batch_id).execute())
+    marked_counts: Dict[str, int] = {}
+    for m in marks:
+        sid = m.get("submission_id")
+        marked_counts[sid] = marked_counts.get(sid, 0) + 1
+
+    enriched = [{**s, "mark_entry_count": marked_counts.get(s["id"], 0)} for s in submissions]
+    progress = {
+        "total": len(submissions),
+        "absent": sum(1 for s in submissions if s.get("status") == "absent"),
+        "complete": sum(1 for s in submissions if s.get("status") == "complete"),
+        "in_progress": sum(1 for s in submissions if s.get("status") in ("matched", "marking")),
+    }
+    return {"batch": batch, "submissions": enriched, "progress": progress}
+
+
+# ─── results & CSV (A7) ──────────────────────────────────────────────────────
+
+def _assemble_results(ctx: ExamContext, batch: Dict[str, Any]) -> Dict[str, Any]:
+    batch_id = batch["id"]
+    questions = _rows(
+        ctx.supabase.table("exam_questions")
+        .select("id, label, max_marks, order")
+        .eq("template_id", batch["template_id"])
+        .order("order")
+        .execute()
+    )
+    submissions = _rows(
+        ctx.supabase.table("student_submissions").select("*").eq("batch_id", batch_id).execute()
+    )
+    marks = _rows(ctx.supabase.table("mark_entries").select("*").eq("batch_id", batch_id).execute())
+
+    by_sub: Dict[str, Dict[str, float]] = {}
+    for m in marks:
+        by_sub.setdefault(m["submission_id"], {})[m["question_id"]] = m.get("awarded_marks")
+
+    results = []
+    for s in submissions:  # every submission incl. absent → A7
+        sub_marks = by_sub.get(s["id"], {})
+        # Blank total ONLY for a genuine no-show (absent AND nothing marked). A student with any
+        # mark gets a real total regardless of status; a present-but-unmarked student totals 0.
+        if sub_marks:
+            total = sum(v or 0 for v in sub_marks.values())
+        elif s.get("status") == "absent":
+            total = None
+        else:
+            total = 0
+        results.append({
+            "submission_id": s["id"],
+            "student_id": s.get("student_id"),
+            "student_name": s.get("student_name"),
+            "status": s.get("status"),
+            "marks": {qid: sub_marks.get(qid) for qid in (q["id"] for q in questions)},
+            "total": total,
+        })
+    return {"batch": batch, "questions": questions, "results": results}
+
+
+@router.get("/batches/{batch_id}/results")
+async def batch_results(
+    batch_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    batch = _fetch_batch_or_404(ctx, batch_id)
+    return _assemble_results(ctx, batch)
+
+
+@router.get("/batches/{batch_id}/csv")
+async def batch_csv(
+    batch_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Response:
+    batch = _fetch_batch_or_404(ctx, batch_id)
+    data = _assemble_results(ctx, batch)
+    questions = data["questions"]
+
+    buf = io.StringIO()
+    writer = csv.writer(buf)
+    writer.writerow(["student_name", "student_id", "status"] + [q["label"] for q in questions] + ["total"])
+    for r in data["results"]:
+        # Absent students: blank marks + blank total, but the row is ALWAYS present (A7).
+        cells = [
+            "" if r["marks"].get(q["id"]) is None else r["marks"].get(q["id"])
+            for q in questions
+        ]
+        total = "" if r["total"] is None else r["total"]
+        writer.writerow([r.get("student_name") or "", r.get("student_id") or "", r.get("status")] + cells + [total])
+
+    return Response(
+        content=buf.getvalue(),
+        media_type="text/csv",
+        headers={"Content-Disposition": f'attachment; filename="batch-{batch_id}.csv"'},
+    )
+
+
+# ─── marks ───────────────────────────────────────────────────────────────────
+
+@router.put("/marks/{mark_id}")
+async def upsert_mark(
+    mark_id: str,
+    body: MarkUpsertRequest,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    # Derive batch_id from the submission (as-user read → also enforces the caller owns the batch
+    # the submission belongs to). The client never supplies the RLS scoping key directly.
+    submission = _first(
+        ctx.supabase.table("student_submissions").select("id, batch_id, status").eq("id", body.submission_id).limit(1).execute()
+    )
+    if not submission:
+        raise HTTPException(status_code=404, detail="Submission not found")
+
+    row = {
+        "id": mark_id,
+        "submission_id": body.submission_id,
+        "question_id": body.question_id,
+        "batch_id": submission["batch_id"],
+        "awarded_marks": body.awarded_marks,
+        "marked_by": "teacher",
+    }
+    if body.mark_scheme_detail is not None:
+        row["mark_scheme_detail"] = body.mark_scheme_detail
+    if body.annotation_shape_ids is not None:
+        row["annotation_shape_ids"] = body.annotation_shape_ids
+    if body.comment is not None:
+        row["comment"] = body.comment
+    if body.confirmed is not None:
+        row["confirmed"] = body.confirmed
+
+    upserted = _first(ctx.supabase.table("mark_entries").upsert(row).execute())
+    if not upserted:
+        raise HTTPException(status_code=500, detail="Failed to upsert mark")
+
+    # A marked student is, by definition, not absent — advance the submission out of the
+    # no-submission states so results/queue reflect that marking has started.
+    if submission.get("status") in ("absent", "unmatched"):
+        ctx.supabase.table("student_submissions").update({"status": "marking"}).eq("id", body.submission_id).execute()
+
+    return upserted
+
+
+# ─── scans (R2.3 / E3) ───────────────────────────────────────────────────────
+
+@router.post("/batches/{batch_id}/scans")
+async def upload_scan(
+    batch_id: str,
+    file: UploadFile = File(...),
+    student_id: Optional[str] = Form(default=None),
+    matching_method: str = Form(default="manual"),
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    batch = _fetch_batch_or_404(ctx, batch_id)
+    _require_owner(ctx, batch)
+
+    # E3: validate MIME (client-declared) before reading the body.
+    if (file.content_type or "").lower() not in ("application/pdf", "application/x-pdf"):
+        raise HTTPException(status_code=415, detail="Only application/pdf scans are accepted")
+
+    # E3: read with a hard size ceiling instead of buffering an unbounded upload.
+    chunks: List[bytes] = []
+    total = 0
+    while True:
+        chunk = await file.read(1024 * 1024)
+        if not chunk:
+            break
+        total += len(chunk)
+        if total > MAX_SCAN_BYTES:
+            raise HTTPException(status_code=413, detail=f"Scan exceeds max size ({MAX_SCAN_BYTES} bytes)")
+        chunks.append(chunk)
+    data = b"".join(chunks)
+    # E3: content-sniff — declared type can be spoofed; require the PDF magic header.
+    if not data.startswith(b"%PDF-"):
+        raise HTTPException(status_code=415, detail="Uploaded file is not a valid PDF")
+
+    # Store via service role (documented): no submissions-bucket storage RLS exists yet; the
+    # endpoint already authorised the caller as the batch owner above.
+    storage_path = f"{SCANS_PREFIX}/{batch_id}/{uuid.uuid4()}.pdf"
+    try:
+        StorageAdmin().upload_file(SCANS_BUCKET, storage_path, data, "application/pdf", upsert=True)
+    except Exception as exc:
+        logger.error(f"scan storage upload failed (batch={batch_id}): {exc}")
+        raise HTTPException(status_code=502, detail="Failed to store scan")
+
+    sb = ctx.supabase
+    submission: Optional[Dict[str, Any]] = None
+    if matching_method == "manual" and student_id:
+        submission = _first(
+            sb.table("student_submissions").select("*").eq("batch_id", batch_id).eq("student_id", student_id).limit(1).execute()
+        )
+    elif matching_method == "ordered":
+        # Assign to the next not-yet-submitted roster slot.
+        pending = _rows(
+            sb.table("student_submissions").select("*").eq("batch_id", batch_id).in_("status", ["absent", "unmatched"]).execute()
+        )
+        submission = pending[0] if pending else None
+
+    payload = {
+        "scan_url": storage_path,
+        "qr_code": None,
+        "matching_method": matching_method if (student_id or matching_method == "ordered") else "manual",
+        "page_count": None,
+        "status": "matched" if submission else "unmatched",
+    }
+
+    if submission:
+        updated = _first(sb.table("student_submissions").update(payload).eq("id", submission["id"]).execute())
+        return updated or submission
+    # No roster slot matched → create an unmatched submission to be reconciled later.
+    new_row = {"batch_id": batch_id, **payload}
+    created = _first(sb.table("student_submissions").insert(new_row).execute())
+    if not created:
+        raise HTTPException(status_code=500, detail="Failed to record scan submission")
+    return created

routers/exam/dependencies.py

@@ -0,0 +1,146 @@
+"""Auth + data-access plumbing for the /api/exam/ router.
+
+Per the audit (spec S1/E1): the exam API calls Supabase **as the user** so the RLS in
+72-exam-marker.sql is actually enforced — it does NOT use the service role for user-facing
+reads/writes the way files.py / classes_router.py do. The bearer already attaches the raw
+JWT as payload["_access_token"] (supabase_bearer.py) precisely for this.
+
+Institute resolution is the one wrinkle: institute_memberships and profiles are RLS
+deny-all to a normal authenticated user (E4), so we cannot read them as-user. Instead we
+call public.user_institute_ids() — a SECURITY DEFINER function (71-class-management.sql) that
+PostgREST exposes as an RPC — which returns the caller's institute ids regardless of those
+table policies. This is the same function the RLS policies themselves key off, so the API's
+view of "which institutes is this user in" is guaranteed consistent with what RLS will allow.
+"""
+from __future__ import annotations
+
+import os
+from typing import Any, Dict, List, Optional
+
+from fastapi import Depends, HTTPException
+
+from modules.auth.supabase_bearer import SupabaseBearer
+from modules.database.supabase.utils.client import (
+    SupabaseAnonClient,
+    SupabaseServiceRoleClient,
+)
+from modules.logger_tool import initialise_logger
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+
+auth = SupabaseBearer()
+
+
+class ExamContext:
+    """The per-request handle every exam endpoint works through.
+
+    Bundles the caller's id, an as-user Supabase client (RLS-enforced), and the set of
+    institute ids the caller belongs to (for R5.5 institute validation on writes).
+    """
+
+    def __init__(self, user_id: str, access_token: str, supabase: Any, institute_ids: List[str]):
+        self.user_id = user_id
+        self.access_token = access_token
+        self.supabase = supabase
+        self.institute_ids = institute_ids
+
+    def resolve_institute(self, requested: Optional[str]) -> str:
+        """Validate a client-supplied institute_id, or pick the sole membership.
+
+        R5.5: a client-supplied institute_id is never trusted as the authz signal — it must
+        be one the caller actually belongs to. RLS would reject a bad value at write time
+        anyway; resolving here turns that into a clean 400/403 instead of an opaque DB error.
+        """
+        if requested:
+            if requested not in self.institute_ids:
+                raise HTTPException(status_code=403, detail="Not a member of the requested institute")
+            return requested
+        if len(self.institute_ids) == 1:
+            return self.institute_ids[0]
+        if not self.institute_ids:
+            raise HTTPException(status_code=403, detail="Caller has no institute membership")
+        raise HTTPException(
+            status_code=400,
+            detail="institute_id is required when the caller belongs to multiple institutes",
+        )
+
+
+def _extract_institute_ids(rpc_data: Any) -> List[str]:
+    """Normalise the user_institute_ids() RPC result to a list of uuid strings.
+
+    A `returns setof uuid` function comes back from PostgREST as a JSON array of scalars,
+    but tolerate the `[{"user_institute_ids": "..."}]` shape too in case of driver quirks.
+    """
+    out: List[str] = []
+    for row in rpc_data or []:
+        if isinstance(row, dict):
+            val = row.get("user_institute_ids") or next(iter(row.values()), None)
+        else:
+            val = row
+        if val:
+            out.append(str(val))
+    return out
+
+
+async def get_exam_context(payload: Dict[str, Any] = Depends(auth)) -> ExamContext:
+    user_id = payload.get("sub") or payload.get("user_id")
+    access_token = payload.get("_access_token")
+    if not user_id or not access_token:
+        raise HTTPException(status_code=401, detail="Invalid token payload")
+
+    supabase = SupabaseAnonClient.for_user(access_token).supabase
+
+    try:
+        res = supabase.rpc("user_institute_ids").execute()
+        institute_ids = _extract_institute_ids(getattr(res, "data", None))
+    except Exception as exc:
+        logger.error(f"Failed to resolve institute memberships: {exc}")
+        raise HTTPException(status_code=502, detail="Could not resolve institute membership")
+
+    return ExamContext(user_id, access_token, supabase, institute_ids)
+
+
+def resolve_student_names(student_ids: List[str]) -> Dict[str, str]:
+    """Map profile id → display name for roster students (batch-creation denormalisation).
+
+    Documented service-role exception (S1, mirrors lookup_exam_code): `profiles` has no as-user
+    SELECT policy (E4), so the roster's display names can't be read as-the-user. The caller's
+    right to the roster itself is already enforced as-user (class_students.cs_read requires the
+    caller to teach/admin the class); this only resolves names for ids already authorised, and
+    the result is denormalised onto student_submissions so later reads need no profiles access.
+    """
+    if not student_ids:
+        return {}
+    try:
+        sb = SupabaseServiceRoleClient().supabase
+        res = (
+            sb.table("profiles")
+            .select("id, full_name, display_name, email")
+            .in_("id", list(student_ids))
+            .execute()
+        )
+        out: Dict[str, str] = {}
+        for p in getattr(res, "data", None) or []:
+            out[p["id"]] = p.get("full_name") or p.get("display_name") or p.get("email") or ""
+        return out
+    except Exception as exc:
+        logger.warning(f"student name resolution failed: {exc}")
+        return {}
+
+
+def lookup_exam_code(exam_id: str) -> Optional[str]:
+    """Resolve eb_exams.exam_code for a catalogue paper (denormalised onto the template).
+
+    Documented service-role exception (S1): eb_exams is shared exam-board reference data with
+    no as-user SELECT policy (E4), so a normal user cannot read it. This is a read of public
+    catalogue metadata only — not user-scoped data — and is used solely to keep the Neo4j join
+    key (exam_code) correct on the template row.
+    """
+    try:
+        sb = SupabaseServiceRoleClient().supabase
+        res = sb.table("eb_exams").select("exam_code").eq("id", exam_id).limit(1).execute()
+        rows = getattr(res, "data", None) or []
+        return rows[0].get("exam_code") if rows else None
+    except Exception as exc:
+        logger.warning(f"exam_code lookup failed for exam_id={exam_id}: {exc}")
+        return None

routers/exam/schemas.py

@@ -0,0 +1,136 @@
+"""Pydantic request/response models for the /api/exam/ router (S4-5).
+
+Templates are saved from the canvas with a full-replace PUT (R5.2): the client owns
+stable UUIDs for questions / response areas / boundaries so the Supabase ids line up
+with the Neo4j join keys (exam_questions.id ↔ Question|Part.uuid_string,
+exam_response_areas.id ↔ Region.uuid_string — see spec §2). Granular mark-scheme edits
+go through PATCH /api/exam/questions/{qid}.
+
+Models mirror the columns in volumes/db/cc/72-exam-marker.sql. They are intentionally
+permissive (most fields optional) so the canvas can round-trip partial state during
+authoring without the API rejecting work-in-progress.
+"""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+
+# ─── Templates ─────────────────────────────────────────────────────────────────
+
+class CreateTemplateRequest(BaseModel):
+    title: str
+    subject: Optional[str] = None
+    # Catalogue paper (eb_exams) the template maps, when chosen from the catalogue (R2.2).
+    exam_id: Optional[str] = None
+    # Denormalised onto the template for the Neo4j join (eb_exams.exam_code ↔ ExamPaper.exam_code).
+    # If exam_id is given but exam_code is omitted, the API resolves it from the catalogue.
+    exam_code: Optional[str] = None
+    # Uploaded PDF (files.id) for an ad-hoc paper (R2.2).
+    source_file_id: Optional[str] = None
+    page_count: Optional[int] = None
+    # Active institute (R1.4/R5.5). Validated against the caller's memberships; never trusted
+    # as the authorization signal. Optional when the caller belongs to exactly one institute.
+    institute_id: Optional[str] = None
+
+
+class UpdateTemplateMetaRequest(BaseModel):
+    """Template-level fields that a full-replace PUT may also update alongside the canvas."""
+    title: Optional[str] = None
+    subject: Optional[str] = None
+    page_count: Optional[int] = None
+    status: Optional[Literal["draft", "ready", "archived"]] = None
+
+
+# ─── Canvas entities (children of a template) ────────────────────────────────────
+
+class QuestionPayload(BaseModel):
+    # Client-supplied stable UUID (== Neo4j Question|Part.uuid_string). Optional on first save.
+    id: Optional[str] = None
+    parent_id: Optional[str] = None
+    label: str
+    order: int = 0
+    max_marks: float = 0
+    answer_type: Optional[Literal["written", "mcq", "short", "diagram"]] = None
+    mcq_options: Optional[Any] = None
+    mark_scheme: Dict[str, Any] = Field(default_factory=dict)
+    is_container: bool = False
+    spec_ref: Optional[str] = None
+    # Drawn Part box geometry (73-exam-marker-regions.sql). Null for derived main questions.
+    bounds: Optional[Dict[str, Any]] = None  # {x,y,w,h}
+    page: Optional[int] = None
+
+
+class ResponseAreaPayload(BaseModel):
+    id: Optional[str] = None  # == Neo4j Region.uuid_string (only response/context project)
+    question_id: str
+    page: int
+    bounds: Dict[str, Any]  # {x,y,w,h}
+    # S4-9 taxonomy (73-exam-marker-regions.sql): response/context graded-or-stimulus;
+    # question_number/mark_area = physical metadata; reference = student resource; furniture = ignore.
+    kind: Literal["response", "context", "question_number", "mark_area", "reference", "furniture"]
+    response_form: Optional[
+        Literal["lines", "answer-box", "working", "diagram", "tick-boxes", "table", "blanks"]
+    ] = None
+    # Optional Context differentiation (v1 generic; future graph/chart/data_table/diagram/code_block/passage).
+    context_type: Optional[str] = None
+    source: Literal["manual", "ai"] = "manual"
+    confirmed: bool = True
+    confidence: Optional[float] = None
+
+
+class BoundaryPayload(BaseModel):
+    id: Optional[str] = None
+    question_id: Optional[str] = None
+    label: Optional[str] = None
+    page_index: int
+    y: float
+    bounds: Optional[Dict[str, Any]] = None
+    source: Literal["manual", "ai"] = "manual"
+    confirmed: bool = True
+
+
+class TemplateReplaceRequest(BaseModel):
+    """Full-replace canvas save (R5.2 primary path). All children are replaced wholesale."""
+    meta: Optional[UpdateTemplateMetaRequest] = None
+    questions: List[QuestionPayload] = Field(default_factory=list)
+    response_areas: List[ResponseAreaPayload] = Field(default_factory=list)
+    boundaries: List[BoundaryPayload] = Field(default_factory=list)
+
+
+class PatchQuestionRequest(BaseModel):
+    """Incremental mark-scheme / spec-ref edit (R5.2 granular path)."""
+    label: Optional[str] = None
+    order: Optional[int] = None
+    max_marks: Optional[float] = None
+    answer_type: Optional[Literal["written", "mcq", "short", "diagram"]] = None
+    mcq_options: Optional[Any] = None
+    mark_scheme: Optional[Dict[str, Any]] = None
+    is_container: Optional[bool] = None
+    spec_ref: Optional[str] = None
+
+
+# ─── Marking batches & marks ─────────────────────────────────────────────────
+
+class CreateBatchRequest(BaseModel):
+    template_id: str
+    # When a class is given, the roster (class_students, status='active') is materialised as
+    # student_submissions(status='absent') so every enrolled student appears in results (A7).
+    class_id: Optional[str] = None
+    title: Optional[str] = None
+
+
+class MarkUpsertRequest(BaseModel):
+    """Upsert one mark entry (PUT /marks/{id}; id is the mark_entry uuid).
+
+    batch_id is derived server-side from the submission, so the client never sets the RLS
+    scoping key. submission_id + question_id identify what is being marked.
+    """
+    submission_id: str
+    question_id: str
+    awarded_marks: float = 0
+    mark_scheme_detail: Optional[Dict[str, Any]] = None
+    annotation_shape_ids: Optional[Any] = None
+    comment: Optional[str] = None
+    confirmed: Optional[bool] = None

routers/exam/templates.py

@@ -0,0 +1,566 @@
+"""Template CRUD for the exam-marker (/api/exam/templates...) — card S4-5.
+
+All access is as-the-user (RLS-enforced; spec E1 fix) via ExamContext. Ownership is also
+checked explicitly before mutating (E2: never trust a client-supplied id as authorization) —
+defence in depth on top of RLS. A row the caller cannot see under RLS reads back as absent,
+so cross-institute access surfaces as 404, never a data leak (IDOR-safe).
+
+Hybrid persistence (R5.2): PUT /templates/{id} is a full-replace of the canvas children
+(questions + response areas + boundaries); PATCH /questions/{qid} is the granular mark-scheme
+edit path. Client-supplied UUIDs are preserved so Supabase ids stay aligned with the Neo4j
+join keys (spec §2).
+"""
+from __future__ import annotations
+
+import os
+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 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
+from modules.logger_tool import initialise_logger
+from routers.exam.dependencies import ExamContext, get_exam_context, lookup_exam_code
+from routers.exam.schemas import (
+    CreateTemplateRequest,
+    PatchQuestionRequest,
+    TemplateReplaceRequest,
+    UpdateTemplateMetaRequest,
+)
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+
+router = APIRouter()
+
+SOURCE_CABINET_NAME = "Exam Marker Template Sources"
+SOURCE_BUCKET_FALLBACK = "cc.users"
+
+
+# ─── helpers ─────────────────────────────────────────────────────────────────
+
+def _rows(result: Any) -> List[Dict[str, Any]]:
+    data = getattr(result, "data", None)
+    if not data:
+        return []
+    return data if isinstance(data, list) else [data]
+
+
+def _first(result: Any) -> Optional[Dict[str, Any]]:
+    rows = _rows(result)
+    return rows[0] if rows else None
+
+
+def _fetch_template_or_404(ctx: ExamContext, template_id: str) -> Dict[str, Any]:
+    """Load a template the caller can see (RLS-scoped). Missing/forbidden → 404."""
+    res = ctx.supabase.table("exam_templates").select("*").eq("id", template_id).limit(1).execute()
+    row = _first(res)
+    if not row:
+        raise HTTPException(status_code=404, detail="Template not found")
+    return row
+
+
+def _require_owner(ctx: ExamContext, template: Dict[str, Any]) -> None:
+    """Writes are limited to the owning teacher (R2.4)."""
+    if template.get("teacher_id") != ctx.user_id:
+        raise HTTPException(status_code=403, detail="Only the template owner can modify it")
+
+
+def _require_source_visibility_or_404(ctx: ExamContext, template: Dict[str, Any]) -> None:
+    """Template source reads must not leak existence across institutes or non-owners."""
+    if template.get("teacher_id") != ctx.user_id:
+        raise HTTPException(status_code=404, detail="Template not found")
+    if template.get("institute_id") not in ctx.institute_ids:
+        raise HTTPException(status_code=404, detail="Template not found")
+
+
+def _template_has_recorded_marks(ctx: ExamContext, template_id: str) -> bool:
+    """True if any mark_entry exists for a batch of this template (→ destructive PUT is unsafe)."""
+    batches = _rows(
+        ctx.supabase.table("marking_batches").select("id").eq("template_id", template_id).execute()
+    )
+    batch_ids = [b["id"] for b in batches]
+    if not batch_ids:
+        return False
+    marks = _rows(
+        ctx.supabase.table("mark_entries").select("id").in_("batch_id", batch_ids).limit(1).execute()
+    )
+    return bool(marks)
+
+
+
+def _model_fields_set(model: Any) -> set[str]:
+    """Return fields explicitly provided by the client (Pydantic v1/v2 compatible)."""
+    if hasattr(model, "model_fields_set"):
+        return set(model.model_fields_set)
+    return set(getattr(model, "__fields_set__", set()))
+
+
+def _model_dump(model: Any) -> Dict[str, Any]:
+    """Dump a Pydantic model without pinning this router to v1/v2 APIs."""
+    if hasattr(model, "model_dump"):
+        return model.model_dump()
+    return model.dict()
+
+
+def _template_meta_updates(body: UpdateTemplateMetaRequest, *, include_explicit_nulls: bool = False) -> Dict[str, Any]:
+    data = _model_dump(body)
+    if include_explicit_nulls:
+        fields = _model_fields_set(body)
+        return {k: data[k] for k in fields if k in data}
+    return {k: v for k, v in data.items() if v is not None}
+
+def _parse_storage_loc(storage_loc: str) -> Tuple[str, str]:
+    bucket, sep, path = (storage_loc or "").partition("/")
+    if not bucket or not sep or not path:
+        raise ValueError(f"Invalid storage_loc: {storage_loc!r}")
+    return bucket, path
+
+
+def _lookup_exam_storage_loc(exam_id: str) -> Optional[str]:
+    try:
+        sb = SupabaseServiceRoleClient().supabase
+        res = sb.table("eb_exams").select("storage_loc").eq("id", exam_id).limit(1).execute()
+        row = _first(res)
+        return row.get("storage_loc") if row else None
+    except Exception as exc:
+        logger.warning(f"storage_loc lookup failed for exam_id={exam_id}: {exc}")
+        return None
+
+
+async def _parse_create_template_request(request: Request) -> tuple[CreateTemplateRequest, Optional[UploadFile]]:
+    content_type = request.headers.get("content-type", "")
+    if "multipart/form-data" in content_type:
+        form = await request.form()
+        payload: Dict[str, Any] = {}
+        for key in ("title", "subject", "exam_id", "exam_code", "source_file_id", "page_count", "institute_id"):
+            value = form.get(key)
+            if value is not None and value != "":
+                payload[key] = value
+        upload = form.get("source_pdf")
+        if upload is not None and not hasattr(upload, "read"):
+            raise HTTPException(status_code=400, detail="source_pdf must be a file upload")
+        if upload is not None and payload.get("source_file_id"):
+            raise HTTPException(status_code=400, detail="Use either source_file_id or source_pdf, not both")
+        return CreateTemplateRequest(**payload), upload
+
+    try:
+        data = await request.json()
+    except Exception as exc:
+        raise HTTPException(status_code=400, detail=f"Invalid request body: {exc}")
+    return CreateTemplateRequest(**data), None
+
+
+async def _upload_template_source_file(
+    ctx: ExamContext,
+    institute_id: str,
+    upload: UploadFile,
+) -> str:
+    file_bytes = await upload.read()
+    if not file_bytes:
+        raise HTTPException(status_code=400, detail="Uploaded PDF is empty")
+    if upload.content_type and upload.content_type != "application/pdf":
+        raise HTTPException(status_code=400, detail="Uploaded file must be a PDF")
+
+    service = SupabaseServiceRoleClient()
+    storage = StorageAdmin()
+
+    cabinet_name = SOURCE_CABINET_NAME
+    existing = _first(
+        service.supabase.table("file_cabinets")
+        .select("id")
+        .eq("user_id", ctx.user_id)
+        .eq("name", cabinet_name)
+        .limit(1)
+        .execute()
+    )
+    if existing:
+        cabinet_id = existing["id"]
+    else:
+        created_cabinet = _first(
+            service.supabase.table("file_cabinets")
+            .insert({"user_id": ctx.user_id, "name": cabinet_name})
+            .execute()
+        )
+        if not created_cabinet:
+            raise HTTPException(status_code=500, detail="Failed to create upload cabinet")
+        cabinet_id = created_cabinet["id"]
+
+    file_id = str(uuid.uuid4())
+    safe_name = os.path.basename(upload.filename or "template.pdf")
+    # Use the shared users bucket (exists on all envs). Per-institute private buckets
+    # (cc.institutes.<id>.private) are a future multi-tenant provisioning concern and are NOT
+    # created on dev .94 — using one here failed with "Bucket not found". The institute is already
+    # namespaced in the storage path + enforced by RLS on the files row.
+    bucket = SOURCE_BUCKET_FALLBACK
+    storage_path = f"exam-marker/{institute_id or 'noinst'}/{cabinet_id}/{file_id}/{safe_name}"
+
+    try:
+        storage.upload_file(bucket, storage_path, file_bytes, "application/pdf", upsert=True)
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"Storage upload failed: {exc}")
+
+    inserted = _first(
+        service.supabase.table("files").insert(
+            {
+                "id": file_id,
+                "cabinet_id": cabinet_id,
+                "name": safe_name,
+                "path": storage_path,
+                "bucket": bucket,
+                "mime_type": "application/pdf",
+                "uploaded_by": ctx.user_id,
+                "size_bytes": len(file_bytes),
+                "source": "classroomcopilot-web",
+                "is_directory": False,
+                "relative_path": safe_name,
+                "processing_status": "uploaded",
+            }
+        ).execute()
+    )
+    if not inserted:
+        raise HTTPException(status_code=500, detail="Failed to create file record")
+
+    return file_id
+
+
+# ─── templates ───────────────────────────────────────────────────────────────
+
+
+@router.post("/templates")
+async def create_template(
+    request: Request,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    body, upload = await _parse_create_template_request(request)
+    institute_id = ctx.resolve_institute(body.institute_id)
+
+    if body.exam_id and body.source_file_id:
+        raise HTTPException(status_code=400, detail="Use either exam_id or source_file_id, not both")
+
+    exam_code = body.exam_code
+    if body.exam_id and not exam_code:
+        exam_code = lookup_exam_code(body.exam_id)
+
+    source_file_id = body.source_file_id
+    if upload is not None:
+        source_file_id = await _upload_template_source_file(ctx, institute_id, upload)
+
+    row = {
+        "title": body.title,
+        "subject": body.subject,
+        "exam_id": body.exam_id,
+        "exam_code": exam_code,
+        "source_file_id": source_file_id,
+        "page_count": body.page_count,
+        "institute_id": institute_id,
+        "teacher_id": ctx.user_id,
+        "status": "draft",
+    }
+    row = {k: v for k, v in row.items() if v is not None}
+
+    res = ctx.supabase.table("exam_templates").insert(row).execute()
+    created = _first(res)
+    if not created:
+        raise HTTPException(status_code=500, detail="Failed to create template")
+    logger.info(f"Exam template created: {created.get('id')} by {ctx.user_id}")
+    return created
+
+
+@router.get("/catalogue")
+async def list_catalogue_papers() -> Dict[str, Any]:
+    """Lightweight exam-board paper catalogue for the create dialog."""
+    try:
+        sb = SupabaseServiceRoleClient().supabase
+        res = (
+            sb.table("eb_exams")
+            .select("id, exam_code, spec_code, paper_code, tier, session, type_code, storage_loc")
+            .eq("type_code", "QP")
+            .order("exam_code")
+            .execute()
+        )
+        return {"papers": _rows(res)}
+    except Exception as exc:
+        raise HTTPException(status_code=502, detail=f"Could not load catalogue papers: {exc}")
+
+
+@router.get("/templates")
+async def list_templates(
+    include_archived: bool = False,
+    institute_id: Optional[str] = None,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    # RLS already scopes to the caller's institutes; the optional filter narrows within that.
+    q = ctx.supabase.table("exam_templates").select("*")
+    if institute_id:
+        q = q.eq("institute_id", institute_id)
+    if not include_archived:
+        q = q.neq("status", "archived")
+    res = q.order("updated_at", desc=True).execute()
+    return {"templates": _rows(res)}
+
+
+@router.get("/templates/{template_id}")
+async def get_template(
+    template_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    template = _fetch_template_or_404(ctx, template_id)
+    questions = _rows(
+        ctx.supabase.table("exam_questions").select("*").eq("template_id", template_id).order("order").execute()
+    )
+    response_areas = _rows(
+        ctx.supabase.table("exam_response_areas").select("*").eq("template_id", template_id).execute()
+    )
+    boundaries = _rows(
+        ctx.supabase.table("exam_boundaries").select("*").eq("template_id", template_id).execute()
+    )
+    return {
+        **template,
+        "questions": questions,
+        "response_areas": response_areas,
+        "boundaries": boundaries,
+    }
+
+
+@router.get("/templates/{template_id}/source-pdf")
+async def get_template_source_pdf(
+    template_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Response:
+    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")
+
+    return Response(content=pdf_bytes, media_type="application/pdf")
+
+
+@router.put("/templates/{template_id}")
+async def replace_template(
+    template_id: str,
+    body: TemplateReplaceRequest,
+    background_tasks: BackgroundTasks,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    """Full-replace canvas save (R5.2). Replaces questions/response_areas/boundaries wholesale.
+
+    Note: the delete-then-insert spans several PostgREST calls and is therefore not atomic;
+    acceptable for the small (~20-question) payloads this carries. A transactional RPC is a
+    later hardening step if concurrent canvas saves become a concern.
+    """
+    template = _fetch_template_or_404(ctx, template_id)
+    _require_owner(ctx, template)
+
+    # Data-loss guard: the wholesale question delete below cascades to mark_entries
+    # (mark_entries.question_id → exam_questions ON DELETE CASCADE). Refuse a structural
+    # full-replace once any marks have been recorded against this template's batches, so
+    # re-saving the setup canvas mid-marking can't silently wipe a teacher's marking work.
+    # (Mark-scheme tweaks use PATCH /questions/{id}, which is unaffected.)
+    if _template_has_recorded_marks(ctx, template_id):
+        raise HTTPException(
+            status_code=409,
+            detail="Template has recorded marks; structural full-replace is blocked. "
+                   "Edit questions individually via PATCH /questions/{id}.",
+        )
+
+    # Optional template-level metadata update alongside the canvas.
+    if body.meta:
+        updates = _template_meta_updates(body.meta)
+        if updates:
+            ctx.supabase.table("exam_templates").update(updates).eq("id", template_id).execute()
+
+    sb = ctx.supabase
+    # Clear existing children. Order matters: response_areas/boundaries reference questions, so
+    # remove them first (we delete by template_id rather than rely on cascade for predictability).
+    sb.table("exam_response_areas").delete().eq("template_id", template_id).execute()
+    sb.table("exam_boundaries").delete().eq("template_id", template_id).execute()
+    sb.table("exam_questions").delete().eq("template_id", template_id).execute()
+
+    # Re-insert, preserving client-supplied UUIDs (Neo4j join keys, spec §2).
+    if body.questions:
+        q_rows = []
+        for q in body.questions:
+            r = {
+                "template_id": template_id,
+                "parent_id": q.parent_id,
+                "label": q.label,
+                "order": q.order,
+                "max_marks": q.max_marks,
+                "answer_type": q.answer_type,
+                "mcq_options": q.mcq_options,
+                "mark_scheme": q.mark_scheme,
+                "is_container": q.is_container,
+                "spec_ref": q.spec_ref,
+                "bounds": q.bounds,   # drawn Part box (73); null for derived main questions
+                "page": q.page,
+            }
+            if q.id:
+                r["id"] = q.id
+            q_rows.append({k: v for k, v in r.items() if v is not None})
+        sb.table("exam_questions").insert(q_rows).execute()
+
+    if body.response_areas:
+        ra_rows = []
+        for ra in body.response_areas:
+            r = {
+                "template_id": template_id,
+                "question_id": ra.question_id,
+                "page": ra.page,
+                "bounds": ra.bounds,
+                "kind": ra.kind,
+                "response_form": ra.response_form,
+                "context_type": ra.context_type,   # 73: optional Context differentiation
+                "source": ra.source,
+                "confirmed": ra.confirmed,
+                "confidence": ra.confidence,
+            }
+            if ra.id:
+                r["id"] = ra.id
+            ra_rows.append({k: v for k, v in r.items() if v is not None})
+        sb.table("exam_response_areas").insert(ra_rows).execute()
+
+    if body.boundaries:
+        b_rows = []
+        for b in body.boundaries:
+            r = {
+                "template_id": template_id,
+                "question_id": b.question_id,
+                "label": b.label,
+                "page_index": b.page_index,
+                "y": b.y,
+                "bounds": b.bounds,
+                "source": b.source,
+                "confirmed": b.confirmed,
+            }
+            if b.id:
+                r["id"] = b.id
+            b_rows.append({k: v for k, v in r.items() if v is not None})
+        sb.table("exam_boundaries").insert(b_rows).execute()
+
+    logger.info(
+        f"Exam template {template_id} replaced: {len(body.questions)} questions, "
+        f"{len(body.response_areas)} regions, {len(body.boundaries)} boundaries"
+    )
+    # R3.5.4: a successful save enqueues a graph projection into cc.public.exams. BackgroundTasks
+    # is acceptable for Sprint 4 (durability via a real queue is a later step); failures are
+    # swallowed so the canvas save itself never fails on a graph hiccup.
+    background_tasks.add_task(project_template_safe, template_id)
+    return await get_template(template_id, ctx)
+
+
+@router.patch("/templates/{template_id}")
+async def patch_template_meta(
+    template_id: str,
+    body: UpdateTemplateMetaRequest,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    """Metadata-only template update.
+
+    Unlike PUT /templates/{id}, this never deletes/re-inserts questions, response areas or
+    boundaries, so it is safe after marking has started. RLS scopes the initial read and
+    owner-only writes are enforced explicitly like the structural save path.
+    """
+    updates = _template_meta_updates(body, include_explicit_nulls=True)
+    if not updates:
+        raise HTTPException(status_code=400, detail="No fields to update")
+
+    template = _fetch_template_or_404(ctx, template_id)
+    _require_owner(ctx, template)
+
+    res = ctx.supabase.table("exam_templates").update(updates).eq("id", template_id).execute()
+    updated = _first(res)
+    if not updated:
+        raise HTTPException(status_code=404, detail="Template not found")
+    return updated
+
+
+@router.delete("/templates/{template_id}")
+async def archive_template(
+    template_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    """Soft-delete: status='archived' (R5.2). Never hard-deletes a teacher's work."""
+    template = _fetch_template_or_404(ctx, template_id)
+    _require_owner(ctx, template)
+    ctx.supabase.table("exam_templates").update({"status": "archived"}).eq("id", template_id).execute()
+    return {"status": "archived", "id": template_id}
+
+
+@router.post("/templates/{template_id}/neo4j-sync")
+async def neo4j_sync(
+    template_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    """Manual graph-projection trigger (R5.3) for dev/backfill — runs synchronously and returns
+    counts. Auth/ownership is checked as-the-user; the projection itself uses service role
+    (R3.5.1, the documented graph-writer path)."""
+    template = _fetch_template_or_404(ctx, template_id)
+    _require_owner(ctx, template)
+    try:
+        counts = project_template(template_id)
+    except Exception as exc:
+        logger.error(f"Manual neo4j-sync failed for template {template_id}: {exc}")
+        raise HTTPException(status_code=502, detail=f"Projection failed: {exc}")
+    return {"status": "ok", "projection": counts}
+
+
+# ─── questions (granular edit path, R5.2) ────────────────────────────────────
+
+
+@router.patch("/questions/{question_id}")
+async def patch_question(
+    question_id: str,
+    body: PatchQuestionRequest,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    updates = {k: v for k, v in _model_dump(body).items() if v is not None}
+    if not updates:
+        raise HTTPException(status_code=400, detail="No fields to update")
+
+    # RLS (exam_questions_all) enforces that the question belongs to a template owned by the
+    # caller; an out-of-scope id updates zero rows → 404, so no explicit pre-fetch is needed.
+    res = ctx.supabase.table("exam_questions").update(updates).eq("id", question_id).execute()
+    updated = _first(res)
+    if not updated:
+        raise HTTPException(status_code=404, detail="Question not found")
+    return updated

run/initialization/init_exam_graph.py

@@ -0,0 +1,128 @@
+"""
+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).
+
+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.
+"""
+import uuid
+from typing import Dict, Any
+
+from modules.database.tools.neo4j_driver_tools import get_driver
+
+EXAM_DB = "cc.public.exams"
+NS = uuid.UUID("00000000-0000-0000-0000-00000000e8a1")  # stable namespace for deterministic uuids
+
+BOARD = {"code": "AQA", "name": "AQA"}
+SPEC = {
+    "spec_code": "AQA-PHYS-8463",
+    "exam_board_code": "AQA",
+    "subject_code": "PHYS",
+    "award_code": "GCSE",
+    "title": "AQA GCSE Physics (8463)",
+}
+# Real AQA GCSE Physics (8463) top-level topics (ref = topic number).
+SPEC_POINTS = [
+    ("4.1", "Energy"),
+    ("4.2", "Electricity"),
+    ("4.3", "Particle model of matter"),
+    ("4.4", "Atomic structure"),
+    ("4.5", "Forces"),
+    ("4.6", "Waves"),
+    ("4.7", "Magnetism and electromagnetism"),
+    ("4.8", "Space physics"),
+]
+
+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",
+    "CREATE CONSTRAINT specpoint_uid     IF NOT EXISTS FOR (n:SpecPoint)     REQUIRE n.uuid_string IS UNIQUE",
+    "CREATE CONSTRAINT exampaper_uid     IF NOT EXISTS FOR (n:ExamPaper)     REQUIRE n.uuid_string IS UNIQUE",
+    "CREATE CONSTRAINT question_uid      IF NOT EXISTS FOR (n:Question)      REQUIRE n.uuid_string IS UNIQUE",
+    "CREATE CONSTRAINT part_uid          IF NOT EXISTS FOR (n:Part)          REQUIRE n.uuid_string IS UNIQUE",
+    "CREATE CONSTRAINT region_uid        IF NOT EXISTS FOR (n:Region)        REQUIRE n.uuid_string IS UNIQUE",
+    "CREATE CONSTRAINT spec_code_unique  IF NOT EXISTS FOR (n:Specification) REQUIRE n.spec_code   IS UNIQUE",
+    "CREATE CONSTRAINT exam_code_unique  IF NOT EXISTS FOR (n:ExamPaper)     REQUIRE n.exam_code   IS UNIQUE",
+    "CREATE CONSTRAINT board_code_unique IF NOT EXISTS FOR (n:ExamBoard)     REQUIRE n.code        IS UNIQUE",
+]
+
+
+def _uid(*parts: str) -> str:
+    return str(uuid.uuid5(NS, ":".join(parts)))
+
+
+def init() -> Dict[str, Any]:
+    driver = get_driver()
+    result: Dict[str, Any] = {"db": EXAM_DB, "constraints": 0, "spec_points": 0}
+
+    # 1. database
+    with driver.session(database="system") as s:
+        s.run(f"CREATE DATABASE `{EXAM_DB}` IF NOT EXISTS").consume()
+    # wait for availability
+    import time
+    for _ in range(30):
+        with driver.session(database="system") as s:
+            st = s.run("SHOW DATABASE $n YIELD currentStatus RETURN currentStatus", n=EXAM_DB).single()
+        if st and st["currentStatus"] == "online":
+            break
+        time.sleep(1)
+
+    with driver.session(database=EXAM_DB) as s:
+        # 2. constraints
+        for c in CONSTRAINTS:
+            s.run(c).consume()
+            result["constraints"] += 1
+
+        # 3. board + spec
+        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()
+        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']}",
+        ).consume()
+
+        # 4. spec points
+        for ref, desc in SPEC_POINTS:
+            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}",
+            ).consume()
+            result["spec_points"] += 1
+
+        counts = s.run(
+            "MATCH (b:ExamBoard) WITH count(b) AS boards "
+            "MATCH (sp:Specification) WITH boards, count(sp) AS specs "
+            "MATCH (p:SpecPoint) RETURN boards, specs, count(p) AS spec_points"
+        ).single()
+        result["verify"] = dict(counts) if counts else {}
+
+    return result
+
+
+if __name__ == "__main__":
+    import json
+    print(json.dumps(init(), indent=2, default=str))

run/initialization/seed_cohort_9p_ph1.py

@@ -0,0 +1,218 @@
+"""
+seed_cohort_9p_ph1.py — Markable cohort for exam-marker testing.
+
+Creates N student accounts and enrols them ALL into a single class (default the
+Greenfield Year 9 Physics class `9P/Ph1`), so there is a real cohort to mark.
+
+Why: the canonical timetable seeds enrol "one student per year-group band"
+(seed_greenfield_timetable.py), so every class has <=1 student — too few for a
+results table / per-question stats. This seeder fills one class to a usable size.
+
+Mechanics (identical paths to the canonical seeds — nothing bespoke server-side):
+  - auth user:   POST {SUPABASE_URL}/auth/v1/admin/users
+  - profile:     upsert public.profiles  (school_id = institute)
+  - membership:  upsert public.institute_memberships  (role 'student')
+  - enrolment:   POST {API_BASE_URL}/database/timetable/classes/{class_id}/students
+                 (as a school_admin; class_students upsert → idempotent)
+
+Idempotent: re-running skips existing auth users and upserts everything else.
+
+Env required:  SUPABASE_URL, SERVICE_ROLE_KEY   (API_BASE_URL defaults to api-dev)
+Optional env:  COHORT_COUNT, COHORT_CLASS_CODE, COHORT_INSTITUTE_ID,
+               SEED_STUDENT_PASSWORD, SEED_SCHOOL_ADMIN_PASSWORD
+
+Run (dev):
+    SUPABASE_URL=... SERVICE_ROLE_KEY=... API_BASE_URL=http://192.168.0.64:18000 \
+    python3 -c "from run.initialization.seed_cohort_9p_ph1 import seed; seed()"
+"""
+import os
+import time
+import requests
+from typing import Dict, Any, List, Optional
+
+# Greenfield Academy (the school actually populated on dev .94)
+GREENFIELD_ID = os.getenv("COHORT_INSTITUTE_ID", "a1b2c3d4-e5f6-7890-abcd-ef1234567890")
+GREENFIELD_DOMAIN = "greenfieldacademy.test"
+GREENFIELD_ADMIN_EMAIL = f"admin@{GREENFIELD_DOMAIN}"
+
+CLASS_CODE = os.getenv("COHORT_CLASS_CODE", "9P/Ph1")
+COHORT_COUNT = int(os.getenv("COHORT_COUNT", "10"))
+
+# Realistic-ish names so the results table doesn't read "Pupil 01..10".
+COHORT_NAMES = [
+    ("Amelia", "Clarke"), ("Noah", "Bennett"), ("Olivia", "Foster"), ("Leo", "Hughes"),
+    ("Ava", "Patel"), ("Jacob", "Reid"), ("Mia", "Turner"), ("Harry", "Ellis"),
+    ("Isla", "Morgan"), ("Oscar", "Khan"), ("Freya", "Walsh"), ("Theo", "Ndlovu"),
+]
+
+DEFAULT_STUDENT_PASSWORD = "Student@Cc2025!"
+DEFAULT_SCHOOL_ADMIN_PASSWORD = "Admin@Cc2025!"
+
+
+def _ctx() -> Dict[str, str]:
+    return {
+        "supa_url": os.environ["SUPABASE_URL"].rstrip("/"),
+        "service_key": os.environ["SERVICE_ROLE_KEY"],
+        "api_base": os.environ.get("API_BASE_URL", "http://192.168.0.64:18000").rstrip("/"),
+        "student_pw": os.getenv("SEED_STUDENT_PASSWORD", DEFAULT_STUDENT_PASSWORD),
+        "admin_pw": os.getenv("SEED_SCHOOL_ADMIN_PASSWORD", DEFAULT_SCHOOL_ADMIN_PASSWORD),
+    }
+
+
+def _sb_headers(ctx: Dict[str, str]) -> Dict[str, str]:
+    return {
+        "apikey": ctx["service_key"],
+        "Authorization": f"Bearer {ctx['service_key']}",
+        "Content-Type": "application/json",
+    }
+
+
+def _sign_in(ctx: Dict[str, str], email: str, password: str) -> str:
+    r = requests.post(
+        f"{ctx['supa_url']}/auth/v1/token?grant_type=password",
+        headers={"apikey": ctx["service_key"], "Content-Type": "application/json"},
+        json={"email": email, "password": password},
+    )
+    r.raise_for_status()
+    return r.json()["access_token"]
+
+
+def _resolve_class_id(ctx: Dict[str, str]) -> Optional[str]:
+    r = requests.get(
+        f"{ctx['supa_url']}/rest/v1/classes",
+        headers=_sb_headers(ctx),
+        params={"class_code": f"eq.{CLASS_CODE}",
+                "institute_id": f"eq.{GREENFIELD_ID}",
+                "select": "id,name", "limit": "1"},
+    )
+    data = r.json() if r.ok else []
+    return data[0]["id"] if data else None
+
+
+def _existing_auth_users(ctx: Dict[str, str]) -> Dict[str, str]:
+    r = requests.get(
+        f"{ctx['supa_url']}/auth/v1/admin/users",
+        headers=_sb_headers(ctx), params={"per_page": 200},
+    )
+    r.raise_for_status()
+    return {u["email"]: u["id"] for u in r.json().get("users", [])}
+
+
+def _create_auth_user(ctx: Dict[str, str], spec: Dict) -> Optional[str]:
+    r = requests.post(
+        f"{ctx['supa_url']}/auth/v1/admin/users",
+        headers=_sb_headers(ctx),
+        json={
+            "email": spec["email"], "password": ctx["student_pw"], "email_confirm": True,
+            "user_metadata": {
+                "username": spec["username"], "full_name": spec["full_name"],
+                "display_name": spec["display_name"], "user_type": "student",
+            },
+        },
+    )
+    if r.status_code in (200, 201):
+        return r.json()["id"]
+    return None
+
+
+def _upsert(ctx: Dict[str, str], table: str, row: Dict, on_conflict: str) -> bool:
+    h = {**_sb_headers(ctx), "Prefer": "resolution=merge-duplicates,return=minimal"}
+    r = requests.post(f"{ctx['supa_url']}/rest/v1/{table}",
+                      headers=h, json=row, params={"on_conflict": on_conflict})
+    return r.ok
+
+
+def _cohort_specs() -> List[Dict]:
+    specs = []
+    for i in range(1, COHORT_COUNT + 1):
+        first, last = COHORT_NAMES[(i - 1) % len(COHORT_NAMES)]
+        prefix = f"cohort{i:02d}"
+        specs.append({
+            "email": f"{prefix}@{GREENFIELD_DOMAIN}",
+            "username": f"{prefix}.{GREENFIELD_DOMAIN.replace('.', '_')}",
+            "full_name": f"{first} {last}",
+            "display_name": first,
+        })
+    return specs
+
+
+def seed(count: Optional[int] = None) -> Dict[str, Any]:
+    global COHORT_COUNT
+    if count is not None:
+        COHORT_COUNT = count
+    ctx = _ctx()
+    results: Dict[str, Any] = {"class_code": CLASS_CODE, "requested": COHORT_COUNT,
+                              "created": 0, "reused": 0, "enrolled": 0, "errors": []}
+
+    print(f"COHORT SEED → {CLASS_CODE} @ {GREENFIELD_DOMAIN} (target {COHORT_COUNT} students)")
+
+    class_id = _resolve_class_id(ctx)
+    if not class_id:
+        results["errors"].append(f"class {CLASS_CODE} not found for institute {GREENFIELD_ID}")
+        print(f"  ✗ {results['errors'][-1]}")
+        return results
+    print(f"  class_id = {class_id}")
+
+    existing = _existing_auth_users(ctx)
+    specs = _cohort_specs()
+
+    # 1) accounts: auth user + profile + membership
+    uids: Dict[str, str] = {}
+    for spec in specs:
+        email = spec["email"]
+        uid = existing.get(email)
+        if uid:
+            results["reused"] += 1
+        else:
+            uid = _create_auth_user(ctx, spec)
+            if not uid:
+                results["errors"].append(f"create auth user {email}")
+                print(f"  ✗ create {email}")
+                continue
+            results["created"] += 1
+            time.sleep(0.15)
+        uids[email] = uid
+        ok_p = _upsert(ctx, "profiles", {
+            "id": uid, "email": email, "user_type": "student",
+            "username": spec["username"], "full_name": spec["full_name"],
+            "display_name": spec["display_name"], "school_id": GREENFIELD_ID,
+            "neo4j_sync_status": "pending",
+        }, on_conflict="id")
+        ok_m = _upsert(ctx, "institute_memberships", {
+            "profile_id": uid, "institute_id": GREENFIELD_ID, "role": "student", "metadata": {},
+        }, on_conflict="profile_id,institute_id")
+        if not (ok_p and ok_m):
+            results["errors"].append(f"profile/membership {email} (p={ok_p} m={ok_m})")
+
+    # 2) enrol all into the class (via API, as school admin)
+    admin_token = _sign_in(ctx, GREENFIELD_ADMIN_EMAIL, ctx["admin_pw"])
+    for spec in specs:
+        uid = uids.get(spec["email"])
+        if not uid:
+            continue
+        r = requests.post(
+            f"{ctx['api_base']}/database/timetable/classes/{class_id}/students",
+            headers={"Authorization": f"Bearer {admin_token}", "Content-Type": "application/json"},
+            json={"student_id": uid},
+        )
+        body = {}
+        try:
+            body = r.json()
+        except Exception:
+            pass
+        if r.ok and (body.get("status") == "ok" or body.get("row") or body.get("id")):
+            results["enrolled"] += 1
+            print(f"  ✓ {spec['email'].split('@')[0]} → {CLASS_CODE}")
+        else:
+            results["errors"].append(f"enrol {spec['email']}: {r.status_code} {str(body)[:120]}")
+            print(f"  ✗ enrol {spec['email']}: {r.status_code}")
+        time.sleep(0.1)
+
+    print(f"\nDONE: created {results['created']}, reused {results['reused']}, "
+          f"enrolled {results['enrolled']}/{COHORT_COUNT}, errors {len(results['errors'])}")
+    return results
+
+
+if __name__ == "__main__":
+    import json
+    print(json.dumps(seed(), indent=2, default=str))

run/initialization/seed_curriculum.py

@@ -52,6 +52,18 @@ SPECIFICATIONS = [
         "storage_loc": "cc.public.snapshots/curriculum/aqa/physics/8203_spec.pdf",
         "doc_type": "pdf",
     },
+    # AQA GCSE Physics 8463 (standalone) — the real spec for the exam-marker test paper
+    # (AQA Physics Paper 1H 2022). Spec graph: cc.public.exams Specification AQA-PHYS-8463.
+    {
+        "spec_code": "AQA-PHYS-8463",
+        "exam_board_code": "AQA",
+        "award_code": "8463",
+        "subject_code": "PHYSICS",
+        "first_teach": "2016",
+        "spec_ver": "1.0",
+        "storage_loc": "cc.examboards/aqa/physics/8463/8463_spec.pdf",  # placeholder (no file yet)
+        "doc_type": "pdf",
+    },
     # Edexcel Maths
     {
         "spec_code": "EDX-MATH-1MA1",
@@ -102,6 +114,12 @@ SPECIFICATIONS = [
 # Realistic exam paper references linked to specifications.
 
 EXAMS = [
+    # AQA GCSE Physics 8463/1 Higher — the exam-marker test paper (real PDF uploaded to
+    # cc.examboards). Join key for cc.public.exams ExamPaper.exam_code.
+    {"exam_code": "AQA-PHYS-8463-1H-22-JUN", "spec_code": "AQA-PHYS-8463", "paper_code": "8463/1",
+     "tier": "higher", "session": "June", "type_code": "QP",
+     "storage_loc": "cc.examboards/aqa/physics/8463/AQA-PHYS-8463-1H-22-JUN.pdf"},
+
     # AQA Physics 8201/1 (Foundation)
     {"exam_code": "AQA-PHYS-8201-1-23-JUN", "spec_code": "AQA-PHYS-8201", "paper_code": "8201/1",
      "tier": "foundation", "session": "June", "type_code": "QP"},

run/routers.py

@@ -40,6 +40,7 @@ from routers.transcribe.canvas_events import router as canvas_events_router
 from routers.transcribe.keywords import router as keywords_router
 from routers.me.bootstrap_router import router as me_bootstrap_router
 from routers import tlsync_token as tlsync_token_router
+from routers.exam import router as exam_router
 
 def register_routes(app: FastAPI):
     logger.info("Starting to register routes...")
@@ -134,6 +135,9 @@ def register_routes(app: FastAPI):
     # TLSync auth token route
     app.include_router(tlsync_token_router.router, prefix="/api/tlsync", tags=["TLSync"])
 
+    # Exam-marker Routes (as-user Supabase, RLS-enforced; spec §4)
+    app.include_router(exam_router, prefix="/api/exam", tags=["Exam"])
+
     # Transcription Routes (CIS Phase 1)
     app.include_router(sessions_router, prefix="/transcribe", tags=["Transcription Sessions"])
     app.include_router(canvas_events_router, prefix="/transcribe", tags=["Transcription Canvas Events"])

tests/test_dev_stack.py

@@ -55,15 +55,19 @@ def test_dev_api_health_endpoint_is_healthy():
     assert payload['services']['redis']['database'] == 0
 
 
+# NOTE: these are >= baselines, not exact counts. The greenfield seed produces this floor;
+# additive exam-marker fixtures (S4-4 cohort adds ~10 students/memberships; ad-hoc classes) push
+# the live .94 counts above it. Exact == froze a snapshot that any new fixture breaks, while >=
+# still catches a broken or missing seed.
 def test_supabase_dev_seed_core_counts():
-    assert _rest_count('profiles') == 21
+    assert _rest_count('profiles') >= 21
-    assert _rest_count('institute_memberships') == 21
+    assert _rest_count('institute_memberships') >= 21
-    assert _rest_count('institutes') == 2
+    assert _rest_count('institutes') >= 2
 
 
 def test_supabase_dev_seed_timetable_counts():
-    assert _rest_count('classes') == 17
+    assert _rest_count('classes') >= 17
-    assert _rest_count('taught_lessons') == 1462
+    assert _rest_count('taught_lessons') >= 1462
 
 
 def test_runtime_identity_does_not_expose_secret_values():

tests/test_exam_batches.py

@@ -0,0 +1,301 @@
+"""Tests for /api/exam/batches, /marks, /scans (card S4-6).
+
+FakeSupabase emulates RLS by pre-filtering the visible store slice (same approach as
+test_exam_templates). Service-role helpers (name resolution, storage) are monkeypatched; live
+as-user RLS is covered by the .94 smoke.
+"""
+import io
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+import routers.exam.batches as batches_mod
+from routers.exam.batches import router
+from routers.exam.dependencies import ExamContext
+
+
+TEACHER = "00000000-0000-0000-0000-000000000001"
+INST_A = "10000000-0000-0000-0000-000000000001"
+TPL = "t-1"
+CLASS = "c-1"
+
+
+class FakeResult:
+    def __init__(self, data):
+        self.data = data
+
+
+class FakeQuery:
+    def __init__(self, store, table):
+        self.store = store
+        self.table = table
+        self.rows = list(store.get(table, []))
+        self._filters = []
+        self._op = None
+        self._payload = None
+        self._limit = None
+
+    def select(self, *_a, **_k):
+        self._op = "select"; return self
+
+    def insert(self, payload):
+        self._op = "insert"; self._payload = payload; return self
+
+    def update(self, payload):
+        self._op = "update"; self._payload = payload; return self
+
+    def upsert(self, payload):
+        self._op = "upsert"; self._payload = payload; return self
+
+    def delete(self):
+        self._op = "delete"; return self
+
+    def eq(self, k, v):
+        self._filters.append(("eq", k, v)); self.rows = [r for r in self.rows if r.get(k) == v]; return self
+
+    def neq(self, k, v):
+        self._filters.append(("neq", k, v)); self.rows = [r for r in self.rows if r.get(k) != v]; return self
+
+    def in_(self, k, vals):
+        vals = set(vals); self._filters.append(("in", k, vals)); self.rows = [r for r in self.rows if r.get(k) in vals]; return self
+
+    def order(self, *_a, **_k):
+        return self
+
+    def limit(self, n):
+        self._limit = n; return self
+
+    def _match(self, row):
+        for op, k, v in self._filters:
+            if op == "eq" and row.get(k) != v:
+                return False
+            if op == "neq" and row.get(k) == v:
+                return False
+            if op == "in" and row.get(k) not in v:
+                return False
+        return True
+
+    def execute(self):
+        backing = self.store.setdefault(self.table, [])
+        if self._op in ("insert", "upsert"):
+            payloads = self._payload if isinstance(self._payload, list) else [self._payload]
+            out = []
+            for p in payloads:
+                row = dict(p)
+                if self._op == "upsert" and row.get("id") is not None:
+                    existing = next((r for r in backing if r.get("id") == row["id"]), None)
+                    if existing:
+                        existing.update(row); out.append(existing); continue
+                row.setdefault("id", f"gen-{self.table}-{len(backing)}")
+                backing.append(row); out.append(row)
+            return FakeResult(out)
+        if self._op == "update":
+            out = []
+            for r in backing:
+                if self._match(r):
+                    r.update(self._payload); out.append(r)
+            return FakeResult(out)
+        if self._op == "delete":
+            self.store[self.table] = [r for r in backing if not self._match(r)]
+            return FakeResult([r for r in backing if self._match(r)])
+        rows = self.rows[: self._limit] if self._limit is not None else self.rows
+        return FakeResult(rows)
+
+
+class FakeSupabase:
+    def __init__(self, store):
+        self.store = store
+
+    def table(self, name):
+        return FakeQuery(self.store, name)
+
+
+def make_client(store, user_id=TEACHER, institute_ids=(INST_A,)):
+    app = FastAPI()
+    app.include_router(router, prefix="/api/exam")
+    from routers.exam.dependencies import get_exam_context
+    app.dependency_overrides[get_exam_context] = lambda: ExamContext(user_id, "tok", FakeSupabase(store), list(institute_ids))
+    return TestClient(app)
+
+
+def base_store(**extra):
+    store = {"exam_templates": [{"id": TPL, "institute_id": INST_A, "teacher_id": TEACHER, "status": "draft"}]}
+    store.update(extra)
+    return store
+
+
+# ─── batches ───────────────────────────────────────────────────────────────
+
+def test_create_batch_no_class():
+    store = base_store()
+    c = make_client(store)
+    r = c.post("/api/exam/batches", json={"template_id": TPL, "title": "Mock 1"})
+    assert r.status_code == 200
+    b = r.json()
+    assert b["teacher_id"] == TEACHER and b["institute_id"] == INST_A
+    assert b["status"] == "open" and b["submission_count"] == 0
+
+
+def test_create_batch_template_404():
+    c = make_client(base_store())
+    assert c.post("/api/exam/batches", json={"template_id": "nope"}).status_code == 404
+
+
+def test_create_batch_seeds_roster_as_absent(monkeypatch):
+    monkeypatch.setattr(batches_mod, "resolve_student_names",
+                        lambda ids: {sid: f"Name {sid}" for sid in ids})
+    store = base_store(class_students=[
+        {"class_id": CLASS, "student_id": "s1", "status": "active"},
+        {"class_id": CLASS, "student_id": "s2", "status": "active"},
+        {"class_id": CLASS, "student_id": "s3", "status": "inactive"},  # excluded
+    ])
+    c = make_client(store)
+    r = c.post("/api/exam/batches", json={"template_id": TPL, "class_id": CLASS})
+    assert r.status_code == 200
+    assert r.json()["submission_count"] == 2
+    subs = store["student_submissions"]
+    assert {s["student_id"] for s in subs} == {"s1", "s2"}
+    assert all(s["status"] == "absent" for s in subs)
+    assert all(s["student_name"].startswith("Name ") for s in subs)
+
+
+def test_list_batches_excludes_archived():
+    store = base_store(marking_batches=[
+        {"id": "b1", "template_id": TPL, "institute_id": INST_A, "teacher_id": TEACHER, "status": "open"},
+        {"id": "b2", "template_id": TPL, "institute_id": INST_A, "teacher_id": TEACHER, "status": "archived"},
+    ])
+    c = make_client(store)
+    ids = {b["id"] for b in c.get("/api/exam/batches").json()["batches"]}
+    assert ids == {"b1"}
+
+
+# ─── queue / results / csv (A7) ──────────────────────────────────────────────
+
+def _batch_with_cohort():
+    return base_store(
+        marking_batches=[{"id": "b1", "template_id": TPL, "institute_id": INST_A, "teacher_id": TEACHER, "status": "open"}],
+        exam_questions=[
+            {"id": "q1", "template_id": TPL, "label": "01", "max_marks": 3, "order": 0},
+            {"id": "q2", "template_id": TPL, "label": "02", "max_marks": 5, "order": 1},
+        ],
+        student_submissions=[
+            {"id": "sub1", "batch_id": "b1", "student_id": "s1", "student_name": "Alice", "status": "complete"},
+            {"id": "sub2", "batch_id": "b1", "student_id": "s2", "student_name": "Bob", "status": "absent"},
+        ],
+        mark_entries=[
+            {"id": "m1", "batch_id": "b1", "submission_id": "sub1", "question_id": "q1", "awarded_marks": 2},
+            {"id": "m2", "batch_id": "b1", "submission_id": "sub1", "question_id": "q2", "awarded_marks": 4},
+        ],
+    )
+
+
+def test_queue_progress_counts():
+    c = make_client(_batch_with_cohort())
+    body = c.get("/api/exam/batches/b1/queue").json()
+    assert body["progress"]["total"] == 2
+    assert body["progress"]["absent"] == 1 and body["progress"]["complete"] == 1
+    counts = {s["id"]: s["mark_entry_count"] for s in body["submissions"]}
+    assert counts == {"sub1": 2, "sub2": 0}
+
+
+def test_results_includes_absent_with_blank(monkeypatch):
+    c = make_client(_batch_with_cohort())
+    body = c.get("/api/exam/batches/b1/results").json()
+    by_id = {r["submission_id"]: r for r in body["results"]}
+    assert by_id["sub1"]["total"] == 6
+    assert by_id["sub2"]["total"] is None  # absent → blank total (A7)
+    assert set(by_id["sub2"]["marks"].values()) == {None}
+    assert {r["student_name"] for r in body["results"]} == {"Alice", "Bob"}  # absent NOT dropped
+
+
+def test_csv_includes_absent_row():
+    c = make_client(_batch_with_cohort())
+    text = c.get("/api/exam/batches/b1/csv").text
+    lines = [l for l in text.strip().splitlines() if l]
+    assert lines[0].split(",")[:3] == ["student_name", "student_id", "status"]
+    assert "01" in lines[0] and "02" in lines[0]  # question labels as columns
+    assert any(l.startswith("Bob,") and ",absent," in l for l in lines)  # absent present
+    assert len(lines) == 3  # header + 2 students (incl. absent)
+
+
+# ─── marks ───────────────────────────────────────────────────────────────────
+
+def test_upsert_mark_derives_batch_and_roundtrips():
+    store = _batch_with_cohort()
+    c = make_client(store)
+    r = c.put("/api/exam/marks/mk-1", json={"submission_id": "sub1", "question_id": "q1", "awarded_marks": 3})
+    assert r.status_code == 200
+    row = r.json()
+    assert row["batch_id"] == "b1" and row["awarded_marks"] == 3 and row["id"] == "mk-1"
+    # upsert again → same id updated, not duplicated
+    c.put("/api/exam/marks/mk-1", json={"submission_id": "sub1", "question_id": "q1", "awarded_marks": 1})
+    assert sum(1 for m in store["mark_entries"] if m["id"] == "mk-1") == 1
+
+
+def test_upsert_mark_flips_absent_submission_to_marking():
+    store = _batch_with_cohort()  # sub2 starts 'absent'
+    c = make_client(store)
+    c.put("/api/exam/marks/mk-2", json={"submission_id": "sub2", "question_id": "q1", "awarded_marks": 2})
+    sub2 = next(s for s in store["student_submissions"] if s["id"] == "sub2")
+    assert sub2["status"] == "marking"
+    # results now show a real total for the (formerly absent) marked student
+    res = {r["submission_id"]: r for r in c.get("/api/exam/batches/b1/results").json()["results"]}
+    assert res["sub2"]["total"] == 2
+
+
+def test_upsert_mark_submission_404():
+    c = make_client(_batch_with_cohort())
+    assert c.put("/api/exam/marks/mk-x", json={"submission_id": "nope", "question_id": "q1", "awarded_marks": 1}).status_code == 404
+
+
+# ─── scans (E3 guards) ───────────────────────────────────────────────────────
+
+def _batch_store():
+    return base_store(marking_batches=[{"id": "b1", "template_id": TPL, "institute_id": INST_A, "teacher_id": TEACHER, "status": "open"}],
+                      student_submissions=[{"id": "sub1", "batch_id": "b1", "student_id": "s1", "status": "absent"}])
+
+
+def test_scan_rejects_non_pdf_mime():
+    c = make_client(_batch_store())
+    r = c.post("/api/exam/batches/b1/scans", files={"file": ("x.png", b"\x89PNG", "image/png")}, data={"matching_method": "manual"})
+    assert r.status_code == 415
+
+
+def test_scan_rejects_spoofed_pdf():
+    c = make_client(_batch_store())
+    r = c.post("/api/exam/batches/b1/scans", files={"file": ("x.pdf", b"not really a pdf", "application/pdf")}, data={"matching_method": "manual"})
+    assert r.status_code == 415  # magic-byte sniff
+
+
+def test_scan_rejects_oversize(monkeypatch):
+    monkeypatch.setattr(batches_mod, "MAX_SCAN_BYTES", 8)
+    c = make_client(_batch_store())
+    r = c.post("/api/exam/batches/b1/scans", files={"file": ("x.pdf", b"%PDF-" + b"0" * 100, "application/pdf")}, data={"matching_method": "manual"})
+    assert r.status_code == 413
+
+
+class _FakeStorage:
+    def upload_file(self, *a, **k):
+        return None
+
+
+def test_scan_manual_match_happy(monkeypatch):
+    monkeypatch.setattr(batches_mod, "StorageAdmin", _FakeStorage)
+    store = _batch_store()
+    c = make_client(store)
+    r = c.post("/api/exam/batches/b1/scans",
+               files={"file": ("x.pdf", b"%PDF-1.7 minimal", "application/pdf")},
+               data={"matching_method": "manual", "student_id": "s1"})
+    assert r.status_code == 200
+    assert r.json()["status"] == "matched"
+    assert store["student_submissions"][0]["status"] == "matched"
+    assert store["student_submissions"][0]["scan_url"].startswith("exam-submissions/b1/")
+
+
+def test_scan_denied_for_non_owner(monkeypatch):
+    monkeypatch.setattr(batches_mod, "StorageAdmin", _FakeStorage)
+    store = _batch_store()
+    c = make_client(store, user_id="someone-else")
+    r = c.post("/api/exam/batches/b1/scans", files={"file": ("x.pdf", b"%PDF-1.7", "application/pdf")}, data={"matching_method": "manual"})
+    assert r.status_code == 403

tests/test_exam_templates.py

@@ -0,0 +1,445 @@
+"""Tests for the /api/exam/templates router (card S4-5).
+
+Mirrors the FakeSupabase + dependency_overrides pattern from test_me_bootstrap.py. The
+ExamContext dependency is overridden with an in-memory fake, so these tests exercise the
+router's auth/ownership/institute logic without a live Supabase — the as-user RLS itself is
+verified separately against .94 (see the evidence note).
+"""
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+import routers.exam.templates as templates_mod
+from routers.exam.templates import router
+from routers.exam.dependencies import ExamContext, get_exam_context
+
+
+TEACHER = "00000000-0000-0000-0000-000000000001"
+OTHER_TEACHER = "00000000-0000-0000-0000-000000000002"
+INST_A = "10000000-0000-0000-0000-000000000001"
+INST_B = "10000000-0000-0000-0000-000000000002"
+
+
+@pytest.fixture(autouse=True)
+def _stub_projection(monkeypatch):
+    """Record projection scheduling and never touch Neo4j/service-role in unit tests."""
+    calls = []
+    monkeypatch.setattr(templates_mod, "project_template_safe", lambda tid: calls.append(tid))
+    monkeypatch.setattr(templates_mod, "project_template", lambda tid: {"exam_code": "X", "questions": 1})
+    return calls
+
+
+# ─── in-memory fake supabase ─────────────────────────────────────────────────
+
+class FakeResult:
+    def __init__(self, data):
+        self.data = data
+
+
+class FakeQuery:
+    """Models the subset of the supabase-py builder the router uses, against a row list.
+
+    Crucially it emulates RLS: the backing store is pre-filtered to the rows the caller can
+    see, so cross-institute / non-owner access naturally reads back empty (→ 404)."""
+
+    def __init__(self, store, table):
+        self.store = store
+        self.table = table
+        self.rows = list(store.get(table, []))
+        self._filters = []
+        self._op = None
+        self._payload = None
+        self._limit = None
+
+    def select(self, *_a, **_k):
+        self._op = "select"
+        return self
+
+    def insert(self, payload):
+        self._op = "insert"
+        self._payload = payload
+        return self
+
+    def update(self, payload):
+        self._op = "update"
+        self._payload = payload
+        return self
+
+    def delete(self):
+        self._op = "delete"
+        return self
+
+    def eq(self, key, value):
+        self._filters.append(("eq", key, value))
+        self.rows = [r for r in self.rows if r.get(key) == value]
+        return self
+
+    def neq(self, key, value):
+        self._filters.append(("neq", 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._filters.append(("in", key, values))
+        self.rows = [r for r in self.rows if r.get(key) in values]
+        return self
+
+    def order(self, *_a, **_k):
+        return self
+
+    def limit(self, n):
+        self._limit = n
+        return self
+
+    def _matches(self, row):
+        for op, key, value in self._filters:
+            if op == "eq" and row.get(key) != value:
+                return False
+            if op == "neq" and row.get(key) == value:
+                return False
+        return True
+
+    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)
+        if self._op == "update":
+            updated = []
+            for row in backing:
+                if self._matches(row):
+                    row.update(self._payload)
+                    updated.append(row)
+            return FakeResult(updated)
+        if self._op == "delete":
+            kept = [r for r in backing if not self._matches(r)]
+            removed = [r for r in backing if self._matches(r)]
+            self.store[self.table] = kept
+            return FakeResult(removed)
+        # select
+        rows = self.rows[: self._limit] if self._limit is not None else self.rows
+        return FakeResult(rows)
+
+
+class FakeSupabase:
+    def __init__(self, store):
+        self.store = store
+
+    def table(self, name):
+        return FakeQuery(self.store, name)
+
+
+class _FakeStorageAdmin:
+    def upload_file(self, *args, **kwargs):
+        return None
+
+    def download_file(self, bucket_id, file_path):
+        return b"%PDF-1.7 fake"
+
+
+class _FakeServiceRoleClient:
+    def __init__(self, store):
+        self.supabase = FakeSupabase(store)
+
+
+def make_client(user_id=TEACHER, institute_ids=(INST_A,), store=None):
+    store = store if store is not None else {}
+    app = FastAPI()
+    app.include_router(router, prefix="/api/exam")
+
+    def _ctx():
+        return ExamContext(user_id, "fake-token", FakeSupabase(store), list(institute_ids))
+
+    app.dependency_overrides[get_exam_context] = _ctx
+    return TestClient(app), store
+
+
+# ─── tests ───────────────────────────────────────────────────────────────────
+
+def test_requires_auth_when_not_overridden():
+    app = FastAPI()
+    app.include_router(router, prefix="/api/exam")
+    # No dependency override → real SupabaseBearer runs and rejects the missing token.
+    resp = TestClient(app).get("/api/exam/templates")
+    assert resp.status_code in (401, 403)  # unauthenticated, not processed
+
+
+def test_create_template_sets_owner_and_institute():
+    client, store = make_client()
+    resp = client.post("/api/exam/templates", json={"title": "AQA Physics 1H", "subject": "Physics"})
+    assert resp.status_code == 200
+    row = resp.json()
+    assert row["title"] == "AQA Physics 1H"
+    assert row["teacher_id"] == TEACHER
+    assert row["institute_id"] == INST_A
+    assert row["status"] == "draft"
+
+
+def test_create_template_accepts_uploaded_source_pdf(monkeypatch):
+    store = {}
+    client, store = make_client(store=store)
+    monkeypatch.setattr(templates_mod, "StorageAdmin", _FakeStorageAdmin)
+    monkeypatch.setattr(templates_mod, "SupabaseServiceRoleClient", lambda: _FakeServiceRoleClient(store))
+
+    resp = client.post(
+        "/api/exam/templates",
+        data={"title": "AQA Physics 1H", "subject": "Physics"},
+        files={"source_pdf": ("paper.pdf", b"%PDF-1.7 test", "application/pdf")},
+    )
+    assert resp.status_code == 200
+    row = resp.json()
+    assert row["source_file_id"] is not None
+    assert store["files"][0]["id"] == row["source_file_id"]
+    assert store["files"][0]["uploaded_by"] == TEACHER
+
+
+def test_get_template_source_pdf_from_uploaded_file(monkeypatch):
+    store = {
+        "exam_templates": [{
+            "id": "t1",
+            "title": "p",
+            "status": "draft",
+            "institute_id": INST_A,
+            "teacher_id": TEACHER,
+            "source_file_id": "f1",
+        }],
+        "files": [{"id": "f1", "bucket": "cc.users", "path": "exam-marker/cab1/f1/paper.pdf", "name": "paper.pdf"}],
+    }
+    client, _ = make_client(store=store)
+    monkeypatch.setattr(templates_mod, "StorageAdmin", _FakeStorageAdmin)
+    # The download resolves the files row via service role (sidesteps the broken cabinet_memberships
+    # RLS recursion) — mock it to the same fake store, like the upload test does.
+    monkeypatch.setattr(templates_mod, "SupabaseServiceRoleClient", lambda: _FakeServiceRoleClient(store))
+    resp = client.get("/api/exam/templates/t1/source-pdf")
+    assert resp.status_code == 200
+    assert resp.headers["content-type"].startswith("application/pdf")
+    assert resp.content.startswith(b"%PDF-1.7")
+
+
+def test_create_template_rejects_foreign_institute():
+    client, _ = make_client(institute_ids=(INST_A,))
+    resp = client.post("/api/exam/templates", json={"title": "X", "institute_id": INST_B})
+    assert resp.status_code == 403
+
+
+def test_create_template_requires_institute_when_ambiguous():
+    client, _ = make_client(institute_ids=(INST_A, INST_B))
+    resp = client.post("/api/exam/templates", json={"title": "X"})
+    assert resp.status_code == 400
+
+
+def test_list_excludes_archived_by_default():
+    store = {
+        "exam_templates": [
+            {"id": "t1", "title": "live", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER},
+            {"id": "t2", "title": "gone", "status": "archived", "institute_id": INST_A, "teacher_id": TEACHER},
+        ]
+    }
+    client, _ = make_client(store=store)
+    titles = [t["title"] for t in client.get("/api/exam/templates").json()["templates"]]
+    assert titles == ["live"]
+    all_titles = {t["title"] for t in client.get("/api/exam/templates?include_archived=true").json()["templates"]}
+    assert all_titles == {"live", "gone"}
+
+
+def test_get_template_bundles_children():
+    store = {
+        "exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}],
+        "exam_questions": [{"id": "q1", "template_id": "t1", "label": "01", "order": 0}],
+        "exam_response_areas": [{"id": "r1", "template_id": "t1", "question_id": "q1", "page": 1}],
+        "exam_boundaries": [{"id": "b1", "template_id": "t1", "page_index": 0, "y": 10}],
+    }
+    client, _ = make_client(store=store)
+    body = client.get("/api/exam/templates/t1").json()
+    assert len(body["questions"]) == 1
+    assert len(body["response_areas"]) == 1
+    assert len(body["boundaries"]) == 1
+
+
+def test_get_other_institute_template_is_404():
+    # RLS emulation: a template the caller can't see isn't in their visible store slice.
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_B, "teacher_id": OTHER_TEACHER}]}
+    client, _ = make_client(institute_ids=(INST_A,), store=store)
+    # The fake store doesn't model institute filtering on read, so simulate the RLS-hidden row
+    # by querying an id the caller's store doesn't contain.
+    assert client.get("/api/exam/templates/does-not-exist").status_code == 404
+
+
+def test_put_replace_persists_children_with_client_ids():
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}]}
+    client, store = make_client(store=store)
+    payload = {
+        "questions": [{"id": "q-uuid-1", "label": "01.1", "order": 0, "max_marks": 3}],
+        "response_areas": [{"id": "r-uuid-1", "question_id": "q-uuid-1", "page": 1, "bounds": {"x": 1}, "kind": "response"}],
+        "boundaries": [{"id": "b-uuid-1", "page_index": 0, "y": 12.5}],
+    }
+    resp = client.put("/api/exam/templates/t1", json=payload)
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["questions"][0]["id"] == "q-uuid-1"  # client UUID preserved (Neo4j join key)
+    assert body["response_areas"][0]["id"] == "r-uuid-1"
+    assert body["boundaries"][0]["id"] == "b-uuid-1"
+
+
+def test_put_persists_region_kinds_and_part_geometry():
+    # S4-9 taxonomy: Part box geometry on the question; new region kinds + context_type.
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}]}
+    client, store = make_client(store=store)
+    resp = client.put("/api/exam/templates/t1", json={
+        "questions": [
+            {"id": "q1", "label": "01", "order": 0, "is_container": True},
+            {"id": "p1", "parent_id": "q1", "label": "01.1", "order": 0, "max_marks": 3,
+             "bounds": {"x": 1, "y": 2, "w": 3, "h": 4}, "page": 1},
+        ],
+        "response_areas": [
+            {"id": "r1", "question_id": "p1", "page": 1, "bounds": {"x": 1}, "kind": "response", "response_form": "lines"},
+            {"id": "c1", "question_id": "p1", "page": 1, "bounds": {"x": 1}, "kind": "context", "context_type": "data_table"},
+            {"id": "qn1", "question_id": "p1", "page": 1, "bounds": {"x": 1}, "kind": "question_number"},
+            {"id": "m1", "question_id": "p1", "page": 1, "bounds": {"x": 1}, "kind": "mark_area"},
+            {"id": "f1", "question_id": "p1", "page": 1, "bounds": {"x": 1}, "kind": "furniture"},
+        ],
+    })
+    assert resp.status_code == 200
+    part = next(q for q in store["exam_questions"] if q["id"] == "p1")
+    assert part["bounds"] == {"x": 1, "y": 2, "w": 3, "h": 4} and part["page"] == 1
+    ras = {r["id"]: r for r in store["exam_response_areas"]}
+    assert {ras["r1"]["kind"], ras["c1"]["kind"], ras["qn1"]["kind"], ras["m1"]["kind"], ras["f1"]["kind"]} == \
+        {"response", "context", "question_number", "mark_area", "furniture"}
+    assert ras["c1"]["context_type"] == "data_table"
+
+
+def test_put_replace_clears_previous_children():
+    store = {
+        "exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}],
+        "exam_questions": [{"id": "old", "template_id": "t1", "label": "stale", "order": 0}],
+    }
+    client, store = make_client(store=store)
+    client.put("/api/exam/templates/t1", json={"questions": [{"id": "new", "label": "fresh", "order": 0}]})
+    ids = {q["id"] for q in store["exam_questions"]}
+    assert ids == {"new"}  # old row replaced, not appended
+
+
+def test_put_replace_blocked_when_marks_recorded():
+    # Re-saving the structure after marking began would cascade-delete mark_entries → guard 409.
+    store = {
+        "exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}],
+        "marking_batches": [{"id": "b1", "template_id": "t1", "teacher_id": TEACHER, "institute_id": INST_A}],
+        "mark_entries": [{"id": "m1", "batch_id": "b1", "submission_id": "s1", "question_id": "q1", "awarded_marks": 2}],
+        "exam_questions": [{"id": "q1", "template_id": "t1", "label": "01", "order": 0}],
+    }
+    client, store = make_client(store=store)
+    r = client.put("/api/exam/templates/t1", json={"questions": [{"id": "q2", "label": "new", "order": 0}]})
+    assert r.status_code == 409
+    # original question untouched (no destructive delete happened)
+    assert {q["id"] for q in store["exam_questions"]} == {"q1"}
+
+
+def test_put_replace_allowed_when_batch_has_no_marks():
+    store = {
+        "exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}],
+        "marking_batches": [{"id": "b1", "template_id": "t1", "teacher_id": TEACHER, "institute_id": INST_A}],
+        "mark_entries": [],
+    }
+    client, _ = make_client(store=store)
+    assert client.put("/api/exam/templates/t1", json={"questions": [{"id": "q2", "label": "new", "order": 0}]}).status_code == 200
+
+
+def test_put_replace_denied_for_non_owner():
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": OTHER_TEACHER}]}
+    # Caller is a colleague in the same institute (can read), but not the owner → 403.
+    client, _ = make_client(user_id=TEACHER, institute_ids=(INST_A,), store=store)
+    resp = client.put("/api/exam/templates/t1", json={"questions": []})
+    assert resp.status_code == 403
+
+
+def test_patch_template_meta_does_not_replace_children_when_marks_recorded():
+    store = {
+        "exam_templates": [{"id": "t1", "title": "old", "subject": "Physics", "page_count": 12, "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}],
+        "exam_questions": [{"id": "q1", "template_id": "t1", "label": "01", "order": 0}],
+        "marking_batches": [{"id": "b1", "template_id": "t1", "teacher_id": TEACHER, "institute_id": INST_A}],
+        "mark_entries": [{"id": "m1", "batch_id": "b1", "submission_id": "s1", "question_id": "q1", "awarded_marks": 2}],
+    }
+    client, store = make_client(store=store)
+    resp = client.patch("/api/exam/templates/t1", json={"title": "renamed", "subject": None, "page_count": 13})
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["title"] == "renamed"
+    assert body["subject"] is None
+    assert body["page_count"] == 13
+    assert {q["id"] for q in store["exam_questions"]} == {"q1"}
+    assert {m["id"] for m in store["mark_entries"]} == {"m1"}
+
+
+def test_patch_template_meta_denied_for_non_owner():
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": OTHER_TEACHER}]}
+    client, _ = make_client(user_id=TEACHER, institute_ids=(INST_A,), store=store)
+    assert client.patch("/api/exam/templates/t1", json={"title": "nope"}).status_code == 403
+
+
+def test_patch_template_meta_empty_body_is_400():
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}]}
+    client, _ = make_client(store=store)
+    assert client.patch("/api/exam/templates/t1", json={}).status_code == 400
+
+
+def test_archive_soft_deletes():
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}]}
+    client, store = make_client(store=store)
+    resp = client.delete("/api/exam/templates/t1")
+    assert resp.status_code == 200
+    assert store["exam_templates"][0]["status"] == "archived"  # not hard-deleted
+
+
+def test_patch_question_updates_fields():
+    store = {"exam_questions": [{"id": "q1", "template_id": "t1", "label": "01", "max_marks": 0}]}
+    client, store = make_client(store=store)
+    resp = client.patch("/api/exam/questions/q1", json={"max_marks": 5, "spec_ref": "8.1.2"})
+    assert resp.status_code == 200
+    assert resp.json()["max_marks"] == 5
+    assert store["exam_questions"][0]["spec_ref"] == "8.1.2"
+
+
+def test_patch_question_missing_is_404():
+    client, _ = make_client(store={"exam_questions": []})
+    assert client.patch("/api/exam/questions/nope", json={"max_marks": 1}).status_code == 404
+
+
+def test_patch_question_empty_body_is_400():
+    store = {"exam_questions": [{"id": "q1", "template_id": "t1", "label": "01"}]}
+    client, _ = make_client(store=store)
+    assert client.patch("/api/exam/questions/q1", json={}).status_code == 400
+
+
+# ─── Neo4j projection (S4-7) ─────────────────────────────────────────────────
+
+def test_put_schedules_projection(_stub_projection):
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}]}
+    client, _ = make_client(store=store)
+    client.put("/api/exam/templates/t1", json={"questions": []})
+    assert _stub_projection == ["t1"]  # projection enqueued for the saved template
+
+
+def test_neo4j_sync_owner_runs():
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": TEACHER}]}
+    client, _ = make_client(store=store)
+    r = client.post("/api/exam/templates/t1/neo4j-sync")
+    assert r.status_code == 200
+    assert r.json()["projection"]["exam_code"] == "X"
+
+
+def test_neo4j_sync_non_owner_403():
+    store = {"exam_templates": [{"id": "t1", "title": "p", "status": "draft", "institute_id": INST_A, "teacher_id": OTHER_TEACHER}]}
+    client, _ = make_client(user_id=TEACHER, institute_ids=(INST_A,), store=store)
+    assert client.post("/api/exam/templates/t1/neo4j-sync").status_code == 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

tests/test_me_bootstrap.py

@@ -122,11 +122,11 @@ def test_supabase_client_for_user_uses_access_token_authorization(monkeypatch):
 
     assert anon.access_token == "user-token"
     assert captured["url"] == "http://supabase.test"
+    # apikey is supplied via the `key` positional arg (supabase-py sets the apikey header from it).
+    # options.headers must carry ONLY the per-user Authorization override — adding apikey here too
+    # produces a duplicate apikey header that Kong rejects ("Duplicate API key found").
     assert captured["key"] == "anon-key"
-    assert captured["options_kwargs"]["headers"] == {
+    assert captured["options_kwargs"]["headers"] == {"Authorization": "Bearer user-token"}
-        "apikey": "anon-key",
-        "Authorization": "Bearer user-token",
-    }
 
 
 def test_no_school_bootstrap_requires_school_membership_but_allows_canvas():

tests/test_p0_api_security.py

@@ -26,8 +26,10 @@ def test_supabase_anon_for_user_sets_user_authorization_header(monkeypatch):
 
     client_module.SupabaseAnonClient.for_user('Bearer user-jwt')
 
+    # apikey comes from the `key` arg (supabase-py sets the apikey header); options.headers must
+    # carry only the user Authorization override. A second apikey here → Kong "Duplicate API key".
     assert captured['key'] == 'anon-key'
-    assert captured['options'].headers['apikey'] == 'anon-key'
+    assert 'apikey' not in captured['options'].headers
     assert captured['options'].headers['Authorization'] == 'Bearer user-jwt'