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

@@ -13,6 +13,7 @@ join keys (spec §2).
 from __future__ import annotations
 
 import json
+import math
 import os
 import tempfile
 import time
@@ -342,6 +343,13 @@ def _pdf_has_text_layer(pdf_bytes: bytes) -> bool:
             pass
 
 
+# Canvas page width the frontend renders each PDF page at (app src/utils/exam-canvas/model.ts
+# PAGE_WIDTH). All auto-map canvas coords are emitted in this 780-wide, proportional-height space.
+CANVAS_PAGE_WIDTH = 780.0
+# Response/answer-region detector (api/services/docling/regions.py) renders at 144 DPI = 2 px / PDF point.
+REGIONS_PX_PER_PT = 2.0
+
+
 def _pdf_page_geometry(pdf_bytes: bytes) -> List[Dict[str, float]]:
     with tempfile.NamedTemporaryFile(prefix="cc-auto-map-geom-", suffix=".pdf", delete=False) as fh:
         fh.write(pdf_bytes)
@@ -355,14 +363,23 @@ def _pdf_page_geometry(pdf_bytes: bytes) -> List[Dict[str, float]]:
             for page in doc:
                 media = page.mediabox
                 crop = page.cropbox
-                rendered_w = float(crop.width or page.rect.width or 595.0)
+                page_pt_w = float(crop.width or page.rect.width or 1.0)
-                rendered_h = float(crop.height or page.rect.height or 842.0)
+                page_pt_h = float(crop.height or page.rect.height or 1.0)
+                # Emit canvas coords in the FRONTEND render space: the app draws each page at
+                # CANVAS_PAGE_WIDTH (app model.ts PAGE_WIDTH=780) with proportional height and stacks
+                # pages by those heights. Previously rendered_w/h were left in PDF points (~595x842),
+                # so every shape landed shrunk (~0.76x) and shifted up-left on the 780-wide canvas.
+                rendered_w = CANVAS_PAGE_WIDTH
+                # Mirror the app's canvas.height = Math.ceil(viewport.height) EXACTLY (pdfLoader.ts),
+                # so page_top accumulates identically. Using the raw float drifts ~1px/page, compounding
+                # to a visible upward shift on later pages of long papers (~36px over 40 pages).
+                rendered_h = float(math.ceil(CANVAS_PAGE_WIDTH * page_pt_h / page_pt_w))
                 pages.append({
                     "media_x0": float(media.x0),
                     "crop_x0": float(crop.x0),
                     "crop_y0": float(crop.y0),
-                    "page_pt_w": float(crop.width or page.rect.width or 1),
+                    "page_pt_w": page_pt_w,
-                    "page_pt_h": float(crop.height or page.rect.height or 1),
+                    "page_pt_h": page_pt_h,
                     "rendered_w": rendered_w,
                     "rendered_h": rendered_h,
                     "page_top": page_top,
@@ -384,11 +401,12 @@ def _pdf_page_geometry(pdf_bytes: bytes) -> List[Dict[str, float]]:
 def _page_geom(pages: List[Dict[str, float]], page_number: int) -> Dict[str, float]:
     if 1 <= page_number <= len(pages):
         return pages[page_number - 1]
+    _fallback_h = float(math.ceil(CANVAS_PAGE_WIDTH * 842.0 / 595.0))
     return {
         "media_x0": 0.0, "crop_x0": 0.0, "crop_y0": 0.0,
         "page_pt_w": 595.0, "page_pt_h": 842.0,
-        "rendered_w": 595.0, "rendered_h": 842.0,
+        "rendered_w": CANVAS_PAGE_WIDTH, "rendered_h": _fallback_h,
-        "page_top": (page_number - 1) * 842.0,
+        "page_top": (page_number - 1) * _fallback_h,
     }
 
 
@@ -397,12 +415,16 @@ def _box_to_canvas(box: Optional[Dict[str, Any]], page_number: int, pages: List[
         return None
     g = _page_geom(pages, page_number)
     if box.get("coord_origin") == "TOPLEFT" and {"x", "y", "w", "h"}.issubset(box):
-        scale = 0.5 if box.get("unit") == "px" else 1.0
+        # Scale the box into the 780-wide canvas space. px boxes (opencv/gemma regions) are in
+        # rendered-image px at REGIONS_PX_PER_PT px/point; TOPLEFT point boxes are 1 px/point.
+        px_per_pt = REGIONS_PX_PER_PT if box.get("unit") == "px" else 1.0
+        sx = g["rendered_w"] / (g["page_pt_w"] * px_per_pt)
+        sy = g["rendered_h"] / (g["page_pt_h"] * px_per_pt)
         return {
-            "x": round(float(box["x"]) * scale, 2),
+            "x": round(float(box["x"]) * sx, 2),
-            "y": round(g["page_top"] + float(box["y"]) * scale, 2),
+            "y": round(g["page_top"] + float(box["y"]) * sy, 2),
-            "w": round(float(box["w"]) * scale, 2),
+            "w": round(float(box["w"]) * sx, 2),
-            "h": round(float(box["h"]) * scale, 2),
+            "h": round(float(box["h"]) * sy, 2),
         }
     if not {"l", "t", "r", "b"}.issubset(box):
         return None
@@ -540,6 +562,23 @@ def _map_first_pass_to_rows(template_id: str, first_pass: Dict[str, Any], pdf_by
             response_form = _response_form_from_region_type(region.get("region_type"))
             if response_form:
                 response_areas.append({"id": _ai_id(template_id, "region", page_index, idx), "template_id": template_id, "question_id": first_part_by_page.get(page_index, default_qid), "page": page_index + 1, "bounds": bounds, "kind": "response", "response_form": response_form, "source": "ai", "confirmed": False, "confidence": _safe_confidence(region.get("confidence")), "derivation": region.get("detection_method") or "opencv-response-region"})
+    # Integrity guard: every response_area/boundary question_id must reference an inserted question
+    # (FK exam_response_areas/exam_boundaries -> exam_questions). On papers where band detection yields
+    # few/no questions but opencv/gemma still emit regions, those regions point at the synthetic
+    # default_qid which was never inserted. Ensure that fallback container question exists and reattach
+    # any orphan child rows to it, so persistence can't violate the FK.
+    qid_set = {q["id"] for q in questions}
+    orphans = [r for r in (response_areas + boundaries) if r.get("question_id") not in qid_set]
+    if orphans:
+        if default_qid not in qid_set:
+            questions.insert(0, {"id": default_qid, "template_id": template_id, "label": "Unassigned",
+                                 "order": 0, "max_marks": 0, "is_container": True, "source": "ai",
+                                 "confirmed": False, "confidence": 0.5,
+                                 "derivation": "auto-map-fallback-container"})
+            qid_set.add(default_qid)
+        for r in orphans:
+            r["question_id"] = default_qid
+
     return {"questions": questions, "response_areas": response_areas, "boundaries": boundaries, "layout": layout}
 
 

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():