FX-3: project image-only papers to Neo4j after auto-map

Only the born-digital fast path enqueued project_template_safe; the async OCR
job (_run_auto_map_job) — where image-only papers, R3's primary target, are
routed — never projected, so those papers never reached the graph after
auto-map. Project at the end of the async job too.

Adds test_auto_map_ocr_path_projects_to_neo4j.

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

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.
-        # 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]
+        payload = _dedupe_rows_by_id(rows.get(key) or [])
         if payload:
             sb.table(table).insert(payload).execute()
 
@@ -640,6 +634,9 @@ def _run_auto_map_job(job_id: str, ctx: ExamContext, template_id: str, pdf_bytes
     _set_auto_map_status(job_id, {"status": "running", "template_id": template_id})
     try:
         rows = _run_auto_map_merge(ctx, template_id, pdf_bytes, source_label)
+        # Project to Neo4j like the born-digital fast path does — otherwise image-only papers (R3's
+        # primary target, routed here because they need OCR) never reach the graph after auto-map.
+        project_template_safe(template_id)
         _set_auto_map_status(job_id, {"status": "completed", "template_id": template_id, "counts": {k: len(v) for k, v in rows.items()}})
     except Exception as exc:
         logger.exception(f"auto-map job failed for template {template_id}: {exc}")

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)
@@ -739,4 +718,16 @@ def test_auto_map_ocr_returns_job_id_and_status_completes(monkeypatch):
     body = status.json()
     assert body["status"] == "completed"
     assert body["counts"]["questions"] >= 2
+
+
+def test_auto_map_ocr_path_projects_to_neo4j(monkeypatch, _stub_projection):
+    # Image-only papers route through the async OCR job; regression: that path must project to Neo4j
+    # like the born-digital fast path, or the graph is never built for the primary target.
+    store = _template_with_source()
+    client, store = make_client(store=store)
+    _patch_auto_map(monkeypatch, store, fast=False)
+    resp = client.post("/api/exam/templates/t1/auto-map")
+    assert resp.status_code == 202
+    # the BackgroundTask runs after the response under TestClient
+    assert "t1" in _stub_projection
     assert body["template"]["layout"]