fix(tasks): reconcile acceptance_criteria_ids at the update chokepoint (#682)

Every post-create rewrite of acceptance_criteria (task PATCH route,
prompter update_live_draft / _patch_batch_child / update_live_batch)
routes through TaskService.update()'s generic field loop, which
overwrote the criteria without touching acceptance_criteria_ids —
leaving ids mismatched or empty, and an empty id list silently
disabled the parent-coverage gate entirely.

- New pure _reconcile_ac_ids: one id per new criterion; text-unchanged
  criteria keep their id (children and findings reference criteria by
  id or exact text — a blanket re-mint would orphan every live
  reference), new/reworded text mints fresh, dropped criteria drop
  theirs. create() now stamps through the same helper (explicitly
  supplied ids still win).
- update() derives acceptance_criteria_ids whenever acceptance_criteria
  is rewritten without an explicit id list.
- The parent-coverage gate self-heals a criteria-bearing row whose ids
  are empty/out-of-length (re-stamp in place) instead of returning
  early and silently waiving coverage for the whole subtree.

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

tests/integration/test_prompter_redraft.py

@@ -207,6 +207,22 @@ async def test_update_live_draft_main_pm_updates_and_hands_off(
     assert task.team == Team.MAIN_PM
 
 
+@pytest.mark.asyncio
+async def test_update_live_draft_reconciles_acceptance_criteria_ids(
+    redraft_setup: dict,
+) -> None:
+    # The wiping path: update_live_draft patches acceptance_criteria via the
+    # generic TaskService.update(), which used to leave acceptance_criteria_ids
+    # stale/mismatched against the revised criteria (the fe-pm delegate-loop
+    # incident). Must come out 1:1 with the new criteria after a redraft.
+    _N = 2
+    task = redraft_setup["mk"](True)
+    await redraft_setup["db"].flush()
+    await redraft_setup["svc"].update_live_draft(task.id, _DRAFT, route="main_pm")
+    assert len(task.acceptance_criteria_ids) == len(task.acceptance_criteria) == _N
+    assert len(set(task.acceptance_criteria_ids)) == _N
+
+
 @pytest.mark.asyncio
 async def test_update_live_draft_reboard_resets_flag(redraft_setup: dict) -> None:
     task = redraft_setup["mk"](True)

tests/integration/test_task_service_basics.py

@@ -514,6 +514,40 @@ async def test_update_returns_none_for_missing(task_setup: dict) -> None:
     assert await svc.update(uuid4(), title="x") is None
 
 
+@pytest.mark.asyncio
+async def test_update_acceptance_criteria_restamps_ids_1to1(
+    task_setup: dict,
+) -> None:
+    # fe-pm delegate-loop root cause: TaskService.update() rewrote
+    # acceptance_criteria without touching acceptance_criteria_ids, leaving a
+    # stale/mismatched id list. A criterion whose TEXT is unchanged must keep
+    # its id (children/findings reference it by id or exact text); a new one
+    # mints a fresh id; a dropped one drops its id.
+    _N = 3
+    svc = task_setup["svc"]
+    task = await svc.create(_req(task_setup, acceptance_criteria=["a", "b", "c"]))
+    old_ids = list(task.acceptance_criteria_ids)
+    updated = await svc.update(task.id, acceptance_criteria=["a", "c", "d"])
+    assert updated is not None
+    assert len(updated.acceptance_criteria_ids) == _N
+    assert updated.acceptance_criteria_ids[0] == old_ids[0]  # "a" unchanged
+    assert updated.acceptance_criteria_ids[1] == old_ids[2]  # "c" unchanged
+    assert updated.acceptance_criteria_ids[2] not in old_ids  # "d" is new
+    assert len(set(updated.acceptance_criteria_ids)) == _N
+
+
+@pytest.mark.asyncio
+async def test_update_without_acceptance_criteria_leaves_ids_untouched(
+    task_setup: dict,
+) -> None:
+    svc = task_setup["svc"]
+    task = await svc.create(_req(task_setup, acceptance_criteria=["a", "b"]))
+    old_ids = list(task.acceptance_criteria_ids)
+    updated = await svc.update(task.id, title="renamed")
+    assert updated is not None
+    assert list(updated.acceptance_criteria_ids) == old_ids
+
+
 @pytest.mark.asyncio
 async def test_add_progress(task_setup: dict) -> None:
     svc = task_setup["svc"]

tests/integration/test_task_service_transitions.py

@@ -1694,6 +1694,46 @@ async def test_production_replay_root_with_child_and_root_owned_coverage(
     assert await svc.uncovered_parent_acceptance_criteria(parent.id) == []
 
 
+@pytest.mark.asyncio
+async def test_parent_ac_edit_preserves_ids_children_still_resolve(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """A board-redraft-style parent AC edit that drops one criterion and adds
+    another must keep the id of every criterion whose TEXT is unchanged, so a
+    child's existing parent_ac_refs -- by id OR by exact text -- still
+    resolves after the rewrite instead of orphaning."""
+    svc = task_setup["svc"]
+    parent = await svc.create(
+        _req(task_setup, acceptance_criteria=["crit a", "crit b", "crit x"])
+    )
+    id_a, _id_b, id_x = parent.acceptance_criteria_ids
+    child_by_id = await svc.create(
+        _req(task_setup, parent_task_id=parent.id, parent_ac_refs=[id_a])
+    )
+    child_by_id.status = TaskStatus.COMPLETED
+    child_by_text = await svc.create(
+        _req(task_setup, parent_task_id=parent.id, parent_ac_refs=["crit x"])
+    )
+    child_by_text.status = TaskStatus.COMPLETED
+    await db_session.flush()
+
+    # "crit b" dropped, "crit new" added, "crit a"/"crit x" kept (reordered).
+    await svc.update(
+        parent.id,
+        acceptance_criteria=["crit x", "crit a", "crit new"],
+    )
+    await db_session.flush()
+
+    refreshed = await svc.get(parent.id)
+    assert refreshed is not None
+    assert set(refreshed.acceptance_criteria_ids[:2]) == {id_a, id_x}
+    assert refreshed.acceptance_criteria_ids[2] not in {id_a, id_x}
+
+    # Both the id-based and text-based refs still resolve -- only the
+    # brand-new criterion is uncovered.
+    assert await svc.uncovered_parent_acceptance_criteria(parent.id) == ["crit new"]
+
+
 # ---------------------------------------------------------------------------
 # _unblock_dependents
 # ---------------------------------------------------------------------------

tests/unit/services/test_task.py

@@ -35,6 +35,7 @@ from roboco.services.task import (
     GatewayAgentView,
     TaskService,
     _ceo_reject_finding_texts,
+    _reconcile_ac_ids,
     get_task_service,
 )
 from sqlalchemy import select
@@ -1214,12 +1215,49 @@ async def test_create_generates_ac_ids_and_carries_parent_ac_refs() -> None:
     assert list(task.parent_ac_refs) == ["parent-ac-1", "parent-ac-2"]
 
 
+def test_reconcile_ac_ids_preserves_new_and_drops() -> None:
+    # Unchanged text keeps its id (position may shift); a reworded/new entry
+    # mints a fresh one; a dropped criterion's id disappears with it.
+    _N = 3
+    ids = _reconcile_ac_ids(
+        old_criteria=["a", "b", "c"],
+        old_ids=["id-a", "id-b", "id-c"],
+        new_criteria=["a", "c", "d"],  # b dropped, a/c kept (reordered), d new
+    )
+    assert ids[0] == "id-a"
+    assert ids[1] == "id-c"
+    assert ids[2] not in {"id-a", "id-b", "id-c"}
+    assert len(ids) == len(set(ids)) == _N
+
+
+def test_reconcile_ac_ids_mints_fresh_when_nothing_to_preserve() -> None:
+    _N = 2
+    ids = _reconcile_ac_ids(old_criteria=[], old_ids=[], new_criteria=["x", "y"])
+    assert len(ids) == len(set(ids)) == _N
+
+
+def test_reconcile_ac_ids_duplicate_text_matched_in_order() -> None:
+    ids = _reconcile_ac_ids(
+        old_criteria=["dup", "dup"],
+        old_ids=["id-1", "id-2"],
+        new_criteria=["dup", "dup", "dup"],
+    )
+    assert ids[:2] == ["id-1", "id-2"]
+    assert ids[2] not in {"id-1", "id-2"}
+
+
 def _svc_with_children(parent: object, child_rows: list[tuple]) -> TaskService:
     """TaskService whose get() returns `parent` and whose execute() yields the
-    (status, parent_ac_refs) child rows the coverage primitive selects."""
+    (status, parent_ac_refs) child rows the coverage primitive selects.
+
+    ``flush`` is a real AsyncMock (not the unconfigured default) so the
+    empty/mismatched-ids self-heal in ``_parent_ac_ref_sets`` can await it.
+    """
     rows = MagicMock()
     rows.all.return_value = child_rows
-    svc = TaskService(MagicMock(execute=AsyncMock(return_value=rows)))
+    svc = TaskService(
+        MagicMock(execute=AsyncMock(return_value=rows), flush=AsyncMock())
+    )
     _bind(svc, "get", AsyncMock(return_value=parent))
     return svc
 
@@ -1362,12 +1400,48 @@ async def test_parent_ac_coverage_maps_claimed_and_verified() -> None:
 
 
 @pytest.mark.asyncio
-async def test_parent_ac_coverage_empty_without_ac_ids() -> None:
-    # No stable ids on the parent (e.g. created before the linkage) -> nothing to
-    # report; the digest stays absent rather than emitting bogus rows.
+async def test_parent_ac_coverage_self_heals_empty_ac_ids() -> None:
+    # fe-pm delegate-loop incident: a coordination root had real
+    # acceptance_criteria but zero acceptance_criteria_ids (an update rewrote
+    # criteria without reconciling ids) -> the digest used to stay [] forever,
+    # silently disabling the parent-coverage gate. It now self-heals by
+    # stamping fresh ids in place so the digest reports for real.
     parent = _build_task(acceptance_criteria=["a"], acceptance_criteria_ids=[])
     svc = _svc_with_children(parent, [(TaskStatus.IN_PROGRESS, ["id-a"])])
-    assert await svc.parent_ac_coverage(parent.id) == []
+    cov = await svc.parent_ac_coverage(parent.id)
+    assert len(cov) == 1
+    assert cov[0]["text"] == "a"
+    assert cov[0]["id"]  # freshly stamped, non-empty
+    assert list(parent.acceptance_criteria_ids) == [cov[0]["id"]]
+
+
+@pytest.mark.asyncio
+async def test_uncovered_parent_acs_self_heals_and_still_matches_text_refs() -> None:
+    # Same legacy shape as above, but proves the self-heal doesn't regress the
+    # id-or-text matching: a child that declared coverage by TEXT (the only
+    # option while the parent had no ids) still resolves through the
+    # freshly-stamped ids, and the gate is live again instead of permanently
+    # inert.
+    _N = 2
+    parent = _build_task(
+        acceptance_criteria=["crit a", "crit b"], acceptance_criteria_ids=[]
+    )
+    svc = _svc_with_children(parent, [(TaskStatus.COMPLETED, ["crit a"])])
+    assert await svc.uncovered_parent_acceptance_criteria(parent.id) == ["crit b"]
+    assert len(parent.acceptance_criteria_ids) == _N
+    assert len(set(parent.acceptance_criteria_ids)) == _N
+
+
+@pytest.mark.asyncio
+async def test_parent_ac_coverage_no_self_heal_when_ids_already_1to1() -> None:
+    # The common case (ids already match criteria 1:1) must not re-stamp —
+    # the self-heal only fires on an actual length mismatch.
+    parent = _build_task(
+        acceptance_criteria=["a", "b"], acceptance_criteria_ids=["id-a", "id-b"]
+    )
+    svc = _svc_with_children(parent, [(TaskStatus.COMPLETED, ["id-a", "id-b"])])
+    await svc.parent_ac_coverage(parent.id)
+    assert list(parent.acceptance_criteria_ids) == ["id-a", "id-b"]
 
 
 @pytest.mark.asyncio