From e269e67f279f4446e052138c70cf226e3978c389 Mon Sep 17 00:00:00 2001
From: CC Worker <cc-worker@local>
Date: Sat, 6 Jun 2026 19:15:35 +0000
Subject: [PATCH] fix(exam): block destructive template PUT once marks recorded
 (review #1)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

PUT full-replace deletes exam_questions, and mark_entries.question_id cascades
ON DELETE — so re-saving the setup canvas after marking began would silently
wipe recorded marks. Guard: 409 if any mark_entry exists for the template's
batches. Mark-scheme edits (PATCH /questions/{id}) are unaffected.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 routers/exam/templates.py    | 26 ++++++++++++++++++++++++++
 tests/test_exam_templates.py | 31 +++++++++++++++++++++++++++++++
 2 files changed, 57 insertions(+)

diff --git a/routers/exam/templates.py b/routers/exam/templates.py
index d4ebfee..3417d16 100644
--- a/routers/exam/templates.py
+++ b/routers/exam/templates.py
@@ -61,6 +61,20 @@ def _require_owner(ctx: ExamContext, template: Dict[str, Any]) -> None:
         raise HTTPException(status_code=403, detail="Only the template owner can modify it")
 
 
+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)
+
+
 # ─── templates ───────────────────────────────────────────────────────────────
 
 @router.post("/templates")
@@ -150,6 +164,18 @@ async def replace_template(
     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 = {k: v for k, v in body.meta.dict().items() if v is not None}
diff --git a/tests/test_exam_templates.py b/tests/test_exam_templates.py
index 96bfb60..2ef22ce 100644
--- a/tests/test_exam_templates.py
+++ b/tests/test_exam_templates.py
@@ -79,6 +79,12 @@ class FakeQuery:
         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
 
@@ -239,6 +245,31 @@ def test_put_replace_clears_previous_children():
     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.