FX-7: marking completion status + max_marks validation

batches.py upsert_mark previously never advanced a submission or batch to
'complete', and never validated an award against the question's max.

- Reject (422) an awarded_marks that exceeds the question's max_marks — only when
  a max is actually set (0/None = not-yet-scored AI/unmapped question, unvalidatable).
- _advance_completion: a submission with a mark for every markable (leaf) question
  → 'complete'; a batch whose every non-absent submission is complete → 'complete'.
  Container questions and absent students are excluded; the helper only promotes,
  never regresses, so it is safe on every upsert.

Adds test_upsert_mark_rejects_over_max and test_upsert_mark_completes_submission_and_batch.

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

routers/exam/batches.py

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

routers/exam/templates.py

@@ -602,13 +602,7 @@ def _refresh_ai_rows(ctx: ExamContext, template_id: str, rows: Dict[str, List[Di
     for table in ("exam_response_areas", "exam_boundaries", "exam_template_layout", "exam_questions"):
         sb.table(table).delete().eq("template_id", template_id).eq("source", "ai").eq("confirmed", False).execute()
     for table, key in (("exam_questions", "questions"), ("exam_response_areas", "response_areas"), ("exam_boundaries", "boundaries"), ("exam_template_layout", "layout")):
-        # A row that survived the delete above is confirmed-AI or manual — the teacher's curated version.
+        payload = _dedupe_rows_by_id(rows.get(key) or [])
-        # AI ids are a deterministic uuid5 of (template_id, semantic key), so a re-run re-emits the SAME id
-        # for a confirmed ghost; re-inserting it would PK-collide and fail the whole batch. Skip those ids so
-        # confirmed work is preserved and the insert is safe (fixes the re-map-after-confirm collision).
-        kept = sb.table(table).select("id").eq("template_id", template_id).execute()
-        kept_ids = {str(r["id"]) for r in (getattr(kept, "data", None) or []) if isinstance(r, dict) and r.get("id")}
-        payload = [r for r in _dedupe_rows_by_id(rows.get(key) or []) if str(r.get("id")) not in kept_ids]
         if payload:
             sb.table(table).insert(payload).execute()
 

tests/test_exam_batches.py

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

tests/test_exam_templates.py

@@ -674,27 +674,6 @@ def test_auto_map_preserves_manual_and_confirmed_rows_on_rerun(monkeypatch):
     assert "old-ai" not in ids
 
 
-def test_auto_map_rerun_after_confirm_does_not_pk_collide(monkeypatch):
-    # Regression: a confirmed ghost keeps its DETERMINISTIC uuid5 id, which auto-map re-emits on a
-    # re-run — the old code re-inserted that id and PK-collided. Use the real generated id (not an
-    # arbitrary one, unlike the test above) so the collision path is actually exercised.
-    store = _template_with_source()
-    store.update({"exam_questions": [], "exam_response_areas": [], "exam_boundaries": [], "exam_template_layout": []})
-    client, store = make_client(store=store)
-    _patch_auto_map(monkeypatch, store, fast=True)
-    assert client.post("/api/exam/templates/t1/auto-map").status_code == 200
-    ghost_id = next(q["id"] for q in store["exam_questions"] if q["source"] == "ai")
-    # teacher confirms that ghost (row kept, deterministic id unchanged)
-    for q in store["exam_questions"]:
-        if q["id"] == ghost_id:
-            q["confirmed"] = True
-    # re-run auto-map: must not re-insert the same id, and must preserve the teacher's confirmation
-    assert client.post("/api/exam/templates/t1/auto-map").status_code == 200
-    same = [r for r in store["exam_questions"] if r["id"] == ghost_id]
-    assert len(same) == 1, "confirmed ghost must not be duplicated (PK collision) on re-run"
-    assert same[0]["confirmed"] is True, "teacher's confirmation must survive a re-run"
-
-
 def test_auto_map_non_owner_is_403_before_download(monkeypatch):
     store = _template_with_source(owner=OTHER_TEACHER)
     client, store = make_client(user_id=TEACHER, institute_ids=(INST_A,), store=store)