FX-1: auto-map re-run must not PK-collide with confirmed ghosts

_refresh_ai_rows deleted only unconfirmed AI rows, then bulk-inserted freshly
generated rows keyed by a deterministic uuid5 of (template_id, semantic key).
A confirmed ghost keeps that id, so a re-run re-emitted it → primary-key
conflict that failed the whole insert batch (reachable: auto-map is blocked only
when marks exist, not when ghosts are confirmed).

Fix: after the delete, read the ids that survived (confirmed-AI + manual) and
skip re-inserting any freshly generated row whose id matches — preserving the
teacher's curated row and making the insert PK-safe.

Adds test_auto_map_rerun_after_confirm_does_not_pk_collide, which confirms a
ghost at its REAL generated id (the prior test used an arbitrary id that never
collided) and asserts the row survives exactly once with confirmed=True.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
CC Worker
2026-07-02 21:02:10 +00:00
co-authored by Claude Opus 4.8
parent 6c73174829
commit 1671518ca8
2 changed files with 28 additions and 1 deletions
+7 -1
View File
@@ -602,7 +602,13 @@ 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")):
payload = _dedupe_rows_by_id(rows.get(key) or [])
# 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]
if payload:
sb.table(table).insert(payload).execute()