feat(gateway): root-owned acceptance criteria via declare_coverage (#357)

The coverage gates had no vocabulary for criteria only the root itself
can satisfy (the supersede PR from feature/main_pm/*, closing the
contributor's PR): once a Main PM declared coverage for the legitimate
cell criteria, the idle gate demanded a cell for the impossible ones
too, so they got pushed into a cell task and the cell PM (correctly)
escalated. declare_coverage now accepts the PM's own task: self-declared
criteria count as claimed for the idle gate and satisfied for the
roll-up (the roll-up actor is their owner by construction), surface as
claimed_by=root in the briefing, and both PM prompts say to never hand
a cell a criterion it cannot satisfy inside its own cell.

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

tests/unit/gateway/test_choreographer_pm.py

@@ -1320,3 +1320,65 @@ async def test_declare_coverage_then_submit_up_gate_passes() -> None:
         pm_id, parent_id, context_phrase="bubbling up"
     )
     assert gate_env is None
+
+
+# ---------------------------------------------------------------------------
+# declare_coverage — root-owned self-declare
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_declare_coverage_self_declare_accepted_for_owning_pm() -> None:
+    """Root-owned mode: task_id is the PM's OWN root (assigned_to ==
+    pm_agent_id) — parent is child by identity (the self-declare signal), no
+    separate parent fetch, criteria validate against the root's own ACs."""
+    pm_id = uuid4()
+    root_id = uuid4()
+    root = MagicMock(
+        id=root_id,
+        assigned_to=pm_id,
+        parent_task_id=None,
+        acceptance_criteria=["crit a", "crit b"],
+        acceptance_criteria_ids=["id-a", "id-b"],
+    )
+    task_svc = AsyncMock()
+    task_svc.get.return_value = root
+    task_svc.agent_for.return_value = MagicMock(role="main_pm", team="board")
+    task_svc.unknown_ac_refs = MagicMock(return_value=[])
+    task_svc.add_parent_ac_refs.return_value = root
+    task_svc.uncovered_parent_acceptance_criteria.return_value = []
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.declare_coverage(pm_id, root_id, ["id-a"])
+    assert env.error is None, env.as_dict()
+    task_svc.add_parent_ac_refs.assert_awaited_once_with(
+        root_id, ["id-a"], declared_by=pm_id
+    )
+    # Self-declare targets its own uncovered-set, never a parent_task_id.
+    task_svc.uncovered_parent_acceptance_criteria.assert_awaited_once_with(root_id)
+
+
+@pytest.mark.asyncio
+async def test_declare_coverage_self_declare_rejected_for_non_owner() -> None:
+    """A PM who is not the assignee of the target root cannot self-declare on
+    it — falls through to the ordinary child-declare path, which requires a
+    parent_task_id; a root has none, so it is rejected."""
+    pm_id, owner_id = uuid4(), uuid4()
+    root_id = uuid4()
+    root = MagicMock(
+        id=root_id,
+        assigned_to=owner_id,
+        parent_task_id=None,
+        acceptance_criteria=["crit a"],
+        acceptance_criteria_ids=["id-a"],
+    )
+    task_svc = AsyncMock()
+    task_svc.get.return_value = root
+    task_svc.agent_for.return_value = MagicMock(role="main_pm", team="board")
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.declare_coverage(pm_id, root_id, ["id-a"])
+    assert env.error == "invalid_state"
+    task_svc.add_parent_ac_refs.assert_not_awaited()

tests/unit/services/test_task.py

@@ -1113,6 +1113,7 @@ async def test_parent_ac_coverage_normalizes_text_refs() -> None:
     parent = _build_task(
         acceptance_criteria=["crit a", "crit b"],
         acceptance_criteria_ids=["id-a", "id-b"],
+        parent_ac_refs=[],
     )
     svc = _svc_with_children(parent, [(TaskStatus.COMPLETED, ["crit a"])])
     cov = await svc.parent_ac_coverage(parent.id)
@@ -1121,6 +1122,7 @@ async def test_parent_ac_coverage_normalizes_text_refs() -> None:
         "text": "crit a",
         "claimed": True,
         "verified": True,
+        "claimed_by": "child",
     }
 
 
@@ -1132,6 +1134,7 @@ async def test_parent_ac_coverage_maps_claimed_and_verified() -> None:
     parent = _build_task(
         acceptance_criteria=["crit a", "crit b", "crit c"],
         acceptance_criteria_ids=["id-a", "id-b", "id-c"],
+        parent_ac_refs=[],
     )
     svc = _svc_with_children(
         parent,
@@ -1141,9 +1144,27 @@ async def test_parent_ac_coverage_maps_claimed_and_verified() -> None:
         ],
     )
     assert await svc.parent_ac_coverage(parent.id) == [
-        {"id": "id-a", "text": "crit a", "claimed": True, "verified": True},
-        {"id": "id-b", "text": "crit b", "claimed": True, "verified": False},
-        {"id": "id-c", "text": "crit c", "claimed": False, "verified": False},
+        {
+            "id": "id-a",
+            "text": "crit a",
+            "claimed": True,
+            "verified": True,
+            "claimed_by": "child",
+        },
+        {
+            "id": "id-b",
+            "text": "crit b",
+            "claimed": True,
+            "verified": False,
+            "claimed_by": "child",
+        },
+        {
+            "id": "id-c",
+            "text": "crit c",
+            "claimed": False,
+            "verified": False,
+            "claimed_by": None,
+        },
     ]
 
 
@@ -1156,6 +1177,85 @@ async def test_parent_ac_coverage_empty_without_ac_ids() -> None:
     assert await svc.parent_ac_coverage(parent.id) == []
 
 
+@pytest.mark.asyncio
+async def test_parent_ac_coverage_marks_root_owned_claimed_by() -> None:
+    # A criterion the parent declared on ITS OWN parent_ac_refs (root-owned,
+    # via declare_coverage(task_id=<own root>, ...)) reads claimed_by="root";
+    # one claimed only via a child reads "child"; untouched is None.
+    parent = _build_task(
+        acceptance_criteria=["crit a", "crit b", "crit c"],
+        acceptance_criteria_ids=["id-a", "id-b", "id-c"],
+        parent_ac_refs=["id-a"],
+    )
+    svc = _svc_with_children(parent, [(TaskStatus.COMPLETED, ["id-b"])])
+    assert await svc.parent_ac_coverage(parent.id) == [
+        {
+            "id": "id-a",
+            "text": "crit a",
+            "claimed": True,
+            "verified": True,
+            "claimed_by": "root",
+        },
+        {
+            "id": "id-b",
+            "text": "crit b",
+            "claimed": True,
+            "verified": True,
+            "claimed_by": "child",
+        },
+        {
+            "id": "id-c",
+            "text": "crit c",
+            "claimed": False,
+            "verified": False,
+            "claimed_by": None,
+        },
+    ]
+
+
+@pytest.mark.asyncio
+async def test_uncovered_parent_acs_root_owned_satisfied_without_child() -> None:
+    # Root-owned refs are satisfied unconditionally -- no child, no COMPLETED
+    # status to wait on. The root's own machinery (PR-supersede, closing a
+    # contributor PR) does the work at/after submit, not a cell.
+    parent = _build_task(
+        acceptance_criteria=["crit a", "crit b"],
+        acceptance_criteria_ids=["id-a", "id-b"],
+        parent_ac_refs=["id-a"],
+    )
+    svc = _svc_with_children(parent, [])
+    assert await svc.uncovered_parent_acceptance_criteria(parent.id) == ["crit b"]
+
+
+@pytest.mark.asyncio
+async def test_unclaimed_parent_acs_root_owned_counts_as_claimed() -> None:
+    parent = _build_task(
+        acceptance_criteria=["crit a", "crit b"],
+        acceptance_criteria_ids=["id-a", "id-b"],
+        parent_ac_refs=["id-a"],
+    )
+    svc = _svc_with_children(parent, [])
+    assert await svc.unclaimed_parent_acceptance_criteria(parent.id) == ["crit b"]
+
+
+@pytest.mark.asyncio
+async def test_production_replay_mixed_child_and_root_owned_coverage() -> None:
+    # The live pattern: a Main-PM root with 5 ACs -- 3 satisfied by a
+    # completed cell subtask, 2 that only the root's own machinery can
+    # satisfy (PR-supersede, closing a contributor PR) and are declared
+    # root-owned instead of pushed into the cell's acceptance_criteria. Both
+    # the idle gate (unclaimed) and the roll-up gate (uncovered) must pass.
+    ids = [f"id-{i}" for i in range(5)]
+    parent = _build_task(
+        acceptance_criteria=[f"crit {i}" for i in range(5)],
+        acceptance_criteria_ids=ids,
+        parent_ac_refs=["id-3", "id-4"],
+    )
+    svc = _svc_with_children(parent, [(TaskStatus.COMPLETED, ["id-0", "id-1", "id-2"])])
+    assert await svc.unclaimed_parent_acceptance_criteria(parent.id) == []
+    assert await svc.uncovered_parent_acceptance_criteria(parent.id) == []
+
+
 @pytest.mark.asyncio
 async def test_unclaimed_parent_acs_inert_without_declared_coverage() -> None:
     # The decomposition floor is opt-in: with no child declaring parent_ac_refs