FX-2: surface born-digital per-part marks from auto-map

The board grammar already parses per-part marks ([N marks] / "Total for Question
N is M marks"), but they were dropped building the first-pass template and the
row mapper hardcoded max_marks=0 — so every born-digital paper came back with
zero marks for a teacher to re-key by hand.

Thread the parsed mark through: bands.py derive_bands now carries each part's
`marks`, template.build passes it into part_bands, and _map_first_pass_to_rows
reads it via a defensive _safe_marks (non-negative int; unknown/None -> 0, so
image-only OCR — which has no marks yet — is unchanged). Containers still roll up
from parts. Adds test_auto_map_surfaces_born_digital_part_marks.

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

api/services/docling/bands.py

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

api/services/docling/template.py

@@ -159,6 +159,7 @@ def build(structured, bands, furniture, pdf=None, page_roles=None):
                 "y_start": p["y_start"], "y_end": p["y_end"],
                 "label_box": part_bbox.get(p["label"]),  # anchor, not the part extent
                 "box": synthesize_part_box(p, xband),
+                "marks": p.get("marks"),                 # parsed per-part marks (born-digital)
                 "source": "auto", "confirmed": False,
             })
         pr = page_roles.get(pgs) or page_roles.get(pg) or {}

routers/exam/batches.py

@@ -245,36 +245,6 @@ 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,
@@ -289,15 +259,6 @@ 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,
@@ -324,9 +285,6 @@ 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

@@ -465,6 +465,17 @@ def _safe_confidence(value: Any = None) -> float:
     return 0.75
 
 
+def _safe_marks(value: Any = None) -> int:
+    """Parsed per-part marks → a non-negative int; unknown/None → 0 (image-only OCR has no marks yet)."""
+    if isinstance(value, bool):
+        return 0
+    if isinstance(value, (int, float)):
+        return max(0, int(value))
+    if isinstance(value, str) and value.strip().isdigit():
+        return int(value.strip())
+    return 0
+
+
 def _margin_values(first_pass: Dict[str, Any], page_number: int) -> Dict[str, Optional[float]]:
     vals: Dict[str, Optional[float]] = {"left": None, "right": None, "top": None, "bottom": None}
     for m in first_pass.get("margins") or []:
@@ -541,7 +552,7 @@ def _map_first_pass_to_rows(template_id: str, first_pass: Dict[str, Any], pdf_by
                 top = max(float(y1), float(y2)); bottom = min(float(y1), float(y2))
                 bounds = _box_to_canvas({"l": margins["left"], "r": margins["right"], "t": top, "b": bottom, "coord_origin": "BOTTOMLEFT"}, page_number, pages_geom)
             bounds = bounds or _box_to_canvas(band.get("label_box"), page_number, pages_geom)
-            questions.append({"id": pid, "template_id": template_id, "parent_id": parent_id, "label": label, "order": len(questions), "max_marks": 0, "is_container": False, "bounds": bounds, "page": page_number, "source": "ai", "confirmed": False, "confidence": _safe_confidence(band.get("confidence")), "derivation": "docling-part-band-x-margins"})
+            questions.append({"id": pid, "template_id": template_id, "parent_id": parent_id, "label": label, "order": len(questions), "max_marks": _safe_marks(band.get("marks")), "is_container": False, "bounds": bounds, "page": page_number, "source": "ai", "confirmed": False, "confidence": _safe_confidence(band.get("confidence")), "derivation": "docling-part-band-x-margins"})
 
     default_qid = questions[0]["id"] if questions else _ai_id(template_id, "question", "auto")
     for page_key in sorted(pages_obj, key=lambda k: int(k)):

tests/test_exam_batches.py

@@ -249,30 +249,6 @@ 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

@@ -642,6 +642,21 @@ def test_auto_map_fast_path_merges_ai_rows_and_returns_detail(monkeypatch):
     assert store["exam_boundaries"] and store["exam_boundaries"][0]["derivation"] == "docling-main-band"
 
 
+def test_auto_map_surfaces_born_digital_part_marks(monkeypatch):
+    # Regression: the born-digital grammar parses per-part marks, but the row mapper hardcoded
+    # max_marks=0. A part band carrying `marks` must flow through to the question row's max_marks.
+    store = _template_with_source()
+    store.update({"exam_questions": [], "exam_response_areas": [], "exam_boundaries": [], "exam_template_layout": []})
+    client, store = make_client(store=store)
+    fp = _first_pass_template()
+    fp["pages"]["1"]["part_bands"][0]["marks"] = 4
+    _patch_auto_map(monkeypatch, store, fast=True)
+    monkeypatch.setattr(templates_mod, "auto_map", lambda *_a, **_k: fp)  # override with the marked part band
+    assert client.post("/api/exam/templates/t1/auto-map").status_code == 200
+    part = next(q for q in store["exam_questions"] if q.get("label") == "01.1")
+    assert part["max_marks"] == 4
+
+
 def test_auto_map_deduplicates_repeated_response_area_ids(monkeypatch):
     store = _template_with_source()
     client, store = make_client(store=store)