Delegation detail-fidelity + PM-loop hardening (#541)

* feat(gateway): delegation detail-fidelity — details survive hand-off, both directions

Details thinned out at every delegation hop: a PM child task mapped to no
parent criterion was legal (coverage only surfaced at submit_up, after the
whole wave ran — a 12-subtask docs tree grew through 8 review rounds that
way, one child titled 'docs page and route wrapper' shipping only the
page), and QA could pass work on a gestalt read (a 4-scene video brief
shipped 3 scenes past every gate because the features existed only in
prose). Three chokepoint gates:

- delegate (down): every child must declare covers_parent_criteria
  resolving against the parent's real acceptance criteria — no mapping or
  an unresolvable ref rejects naming every offending child and the valid
  criteria; the success envelope carries parent_ac_coverage
  {covered, uncovered} so a wave-planning PM sees remaining gaps in the
  same turn. Full coverage stays enforced at submit_up (waves stay legal).
- pass_review (up): mandatory criteria_verified — one {criterion,
  evidence} entry per task AC, matched by the findings ledger's
  id-or-exact-text matcher, evidence soup-checked and capped; rejects
  naming the unverified criteria; entries render deterministically into
  qa_notes as '[AC] <criterion> — verified: <evidence>' lines. The old
  count-only ac_verdicts gate is superseded (arg kept for back-compat).
- video briefs (structured detail at origination): an enumerable feature
  list (release highlights, or input_props.highlights carried onto a
  reject re-author) becomes its own scene acceptance criterion, bounded to
  the AC caps; a re-author without highlights carries the
  feedback-addressed criterion instead.

Extracted findings.py's criterion matcher into shared unmatched_criteria /
uncovered_acceptance_criteria instead of duplicating it; criteria_verified
joins the WAF free-text exclusion set like findings/issues.

* fix(gateway): break the block/unblock wedge — four hardening fixes from the live PM loop

A cell task looped fe-pm/main-pm block/unblock for hours (10 cycles, 43
spawns): a transient GitHub API error resolving CI became an unwaivable
blocker finding whose own fix text said no code change was required, the
submit freshness guard then demanded a commit no finding called for,
escalate_up auto-blocked, and main-pm's correct recovery plan 422'd on
the approach length cap, degrading it to a bare unblock. Four fixes:

- pr_pass CI-unresolvable refusal is now explicitly transient-worded:
  retry pr_pass shortly, do NOT pr_fail over a CI-status lookup error —
  a platform blip is not a code finding
- submit freshness guard grants ONE unchanged-head resubmission per
  head sha when the findings ledger has zero open rows (all addressed
  without code changes) — stamped via the resubmit_unchanged_head
  marker so the same head can never loop a second time
- unblock carries a flip breaker: block_flip_count marker, and at the
  third flip a one-shot CEO notification flags the task as structurally
  wedged (unblock itself still succeeds — the breaker signals, it does
  not wedge recovery)
- i_will_plan's approach cap truncates at 800 chars instead of
  rejecting — an over-detailed plan must never cost the PM its turn

---------

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

tests/unit/gateway/test_choreographer_delegate_guards.py

@@ -137,6 +137,7 @@ async def test_delegate_allows_when_parent_in_progress_and_owned() -> None:
         assigned_to=pm_id,
         team="backend",
         quick_context="Decomposition planned; cells implement their slice next.",
+        acceptance_criteria=[],
     )
     new_task = MagicMock(id=uuid4())
     task_svc = AsyncMock()
@@ -194,6 +195,7 @@ async def test_delegate_allows_when_subtask_cap_within_soft_zone() -> None:
         assigned_to=pm_id,
         team="backend",
         quick_context="Decomposition planned; cells implement their slice next.",
+        acceptance_criteria=[],
     )
     many = [MagicMock(id=uuid4()) for _ in range(10)]
     new_task = MagicMock(id=uuid4())
@@ -223,6 +225,7 @@ async def test_delegate_allows_at_zero_subtasks() -> None:
         assigned_to=pm_id,
         team="backend",
         quick_context="Decomposition planned; cells implement their slice next.",
+        acceptance_criteria=[],
     )
     new_task = MagicMock(id=uuid4())
     task_svc = AsyncMock()
@@ -313,6 +316,7 @@ async def test_delegate_past_max_depth_returns_invalid_state_not_500() -> None:
         assigned_to=pm_id,
         team="backend",
         quick_context="Decomposition planned; cells implement their slice next.",
+        acceptance_criteria=[],
     )
     depth_msg = (
         "Task hierarchy would exceed MAX_TASK_DEPTH=4. Create this work as a "

tests/unit/gateway/test_choreographer_pm_extras.py

@@ -675,6 +675,8 @@ async def test_delegate_main_pm_to_cell_pm_creates_subtask() -> None:
         status="in_progress",
         assigned_to=main_pm_id,
         quick_context="Decomposition planned; cells implement their slice next.",
+        # This test is about role/chain wiring, not AC coverage.
+        acceptance_criteria=[],
     )
     new_task = MagicMock(id=uuid4())
     task_svc = AsyncMock()
@@ -719,6 +721,8 @@ async def test_delegate_cell_pm_to_team_dev_creates_subtask() -> None:
         assigned_to=cell_pm_id,
         team="backend",
         quick_context="Decomposition planned; cells implement their slice next.",
+        # This test is about role/chain wiring, not AC coverage.
+        acceptance_criteria=[],
     )
     new_task = MagicMock(id=uuid4())
     task_svc = AsyncMock()
@@ -1185,6 +1189,9 @@ async def test_delegate_main_pm_to_cell_pm_accepts_planning_subtask() -> None:
         status="in_progress",
         assigned_to=main_pm_id,
         quick_context="Decomposition planned; cells implement their slice next.",
+        # This test is about the code-vs-planning task_type guard, not AC
+        # coverage.
+        acceptance_criteria=[],
     )
     new_task = MagicMock(id=uuid4())
     task_svc = AsyncMock()

tests/unit/gateway/test_choreographer_qa.py

@@ -320,6 +320,94 @@ async def test_pass_review_succeeds_and_transitions() -> None:
     a2a_svc.send.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_pass_review_rejects_without_criteria_verified_when_acs_present() -> None:
+    """A task with real acceptance criteria demands criteria_verified — a
+    gestalt "looks good" notes string alone is no longer enough."""
+    qa_id = uuid4()
+    task_id = uuid4()
+    t = _qa_owned_task(
+        task_id, qa_id, acceptance_criteria=["returns 200", "includes timestamp"]
+    )
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.agent_for.return_value = _qa_agent_mock(qa_id)
+    journal_svc = AsyncMock()
+    journal_svc.has_learning_for_task.return_value = True
+    deps = _make_deps(task=task_svc, journal=journal_svc)
+    c = Choreographer(deps)
+
+    notes = "x" * 100
+    env = await c.pass_review(qa_id, task_id, notes=notes)
+    body = env.as_dict()
+    assert body["error"] == "invalid_state", body
+    assert "returns 200" in body["message"]
+    assert "includes timestamp" in body["message"]
+
+
+@pytest.mark.asyncio
+async def test_pass_review_renders_criteria_verified_into_notes() -> None:
+    """Happy path: every AC matched + evidenced renders '[AC] ...' lines into
+    the persisted qa_notes and the transition still fires."""
+    qa_id = uuid4()
+    task_id = uuid4()
+    t = _qa_owned_task(
+        task_id, qa_id, acceptance_criteria=["returns 200", "includes timestamp"]
+    )
+    after = MagicMock(
+        id=task_id,
+        status="awaiting_documentation",
+        assigned_to=qa_id,
+        team="backend",
+        pr_url="https://x/pr/8",
+        qa_evidence_inspected=True,
+    )
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.agent_for.return_value = _qa_agent_mock(qa_id)
+    task_svc.qa_pass.return_value = after
+    task_svc.documenter_for_team.return_value = MagicMock(id=uuid4())
+    task_svc.session = MagicMock()
+    task_svc.session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
+    _stub_empty_ledger(task_svc.session)
+    journal_svc = AsyncMock()
+    journal_svc.has_learning_for_task.return_value = True
+    a2a_svc = AsyncMock()
+    deps = _make_deps(task=task_svc, journal=journal_svc, a2a=a2a_svc)
+    c = Choreographer(deps)
+
+    notes = (
+        "Reviewed PR carefully. Rendered every scene and checked each frame "
+        "against the brief before approving."
+    )
+    env = await c.pass_review(
+        qa_id,
+        task_id,
+        notes=notes,
+        criteria_verified=[
+            {"criterion": "returns 200", "evidence": "test_healthz asserts 200"},
+            {
+                "criterion": "includes timestamp",
+                "evidence": "frame diff shows ts field at README.md line 12",
+            },
+        ],
+    )
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_documentation"
+    task_svc.qa_pass.assert_awaited_once()
+    persisted_notes = task_svc.qa_pass.call_args.args[2]
+    assert "[AC] returns 200 — verified: test_healthz asserts 200" in persisted_notes
+    assert (
+        "[AC] includes timestamp — verified: frame diff shows ts field at "
+        "README.md line 12" in persisted_notes
+    )
+
+
 @pytest.mark.asyncio
 async def test_pass_review_not_assigned_returns_not_authorized() -> None:
     qa_id = uuid4()

tests/unit/gateway/test_delegate_ac_coverage_gate.py

@@ -0,0 +1,213 @@
+"""Decomposition-coverage gate: delegate rejects a child that doesn't map to
+the parent's acceptance criteria, and a successful delegate reports the
+parent's remaining coverage gaps in evidence.
+
+Live failure this closes (see CLAUDE.md "delegate" section): unmapped
+children were only ever caught late, at submit_up's roll-up gate — after a
+whole wave of subtasks had already run. Moving the mapping check to delegate
+time surfaces "child covers nothing" / "ref doesn't match a real criterion"
+before any subtask is created.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.choreographer import (
+    Choreographer,
+    ChoreographerDeps,
+    DelegateInputs,
+)
+
+
+def _make_deps(**overrides: Any) -> ChoreographerDeps:
+    base: dict[str, Any] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    base.update(overrides)
+    repo = base["evidence_repo"]
+    for method in (
+        "list_unread_a2a",
+        "list_unread_mentions",
+        "list_pending_notifications",
+        "task_metadata_gaps",
+        "recent_team_activity",
+        "blockers_in_lane",
+        "journal_highlights_for_task",
+    ):
+        getattr(repo, method).return_value = []
+    _ldef = base["journal"].latest_decision_at.return_value
+    if type(_ldef).__name__ in ("MagicMock", "AsyncMock"):
+        base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    return ChoreographerDeps(**base)
+
+
+def _parent_with_criteria(pm_id: Any) -> MagicMock:
+    return MagicMock(
+        id=uuid4(),
+        project_id=uuid4(),
+        status="in_progress",
+        assigned_to=pm_id,
+        team="backend",
+        quick_context="Decomposition planned; cells implement their slice next.",
+        acceptance_criteria=["Criterion A", "Criterion B"],
+        acceptance_criteria_ids=["id-a", "id-b"],
+    )
+
+
+def _inputs(**kw: Any) -> DelegateInputs:
+    base: dict[str, Any] = {
+        "title": "Implement endpoint",
+        "description": "Add /v1/foo endpoint with tests",
+        "assigned_to": "be-dev-1",
+        "team": "backend",
+        "task_type": "code",
+        "nature": "technical",
+        "acceptance_criteria": ["GET /v1/foo returns 200 with body"],
+        "intends_to_touch": ["backend/api/routers/foo.py"],
+    }
+    base.update(kw)
+    return DelegateInputs(**base)
+
+
+@pytest.mark.asyncio
+async def test_delegate_rejects_child_with_no_mapping() -> None:
+    """A parent with real ACs rejects a child that maps to none of them."""
+    pm_id = uuid4()
+    parent = _parent_with_criteria(pm_id)
+    task_svc = AsyncMock()
+    task_svc.get.return_value = parent
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    task_svc.get_subtasks.return_value = []
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.delegate(pm_id, parent.id, _inputs(title="Orphan slice"))
+    body = env.as_dict()
+    assert body["error"] == "invalid_state", body
+    assert "Orphan slice" in body["message"]
+    assert "no covers_parent_criteria" in body["message"]
+    assert "Criterion A" in body["remediate"] and "Criterion B" in body["remediate"]
+    task_svc.create_subtask.assert_not_awaited()
+    task_svc.unknown_ac_refs.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_delegate_rejects_unresolvable_ref_lists_valid_criteria() -> None:
+    """A ref matching neither a criterion id nor its exact text is rejected,
+    with the parent's real criteria named so the PM can pick a valid one."""
+    pm_id = uuid4()
+    parent = _parent_with_criteria(pm_id)
+    task_svc = AsyncMock()
+    task_svc.get.return_value = parent
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    task_svc.get_subtasks.return_value = []
+    task_svc.unknown_ac_refs = MagicMock(return_value=["bogus-ref"])
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.delegate(
+        pm_id,
+        parent.id,
+        _inputs(title="Endpoint slice", covers_parent_criteria=["bogus-ref"]),
+    )
+    body = env.as_dict()
+    assert body["error"] == "invalid_state", body
+    assert "Endpoint slice" in body["message"]
+    assert "bogus-ref" in body["message"]
+    assert "Criterion A" in body["remediate"] and "Criterion B" in body["remediate"]
+    task_svc.create_subtask.assert_not_awaited()
+    task_svc.unknown_ac_refs.assert_called_once_with(parent, ["bogus-ref"])
+
+
+@pytest.mark.asyncio
+async def test_delegate_rejects_multiple_unresolvable_refs_in_one_envelope() -> None:
+    """Every unresolvable ref is named in the one rejection, not just the first."""
+    pm_id = uuid4()
+    parent = _parent_with_criteria(pm_id)
+    task_svc = AsyncMock()
+    task_svc.get.return_value = parent
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    task_svc.get_subtasks.return_value = []
+    task_svc.unknown_ac_refs = MagicMock(return_value=["bogus-one", "bogus-two"])
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.delegate(
+        pm_id,
+        parent.id,
+        _inputs(covers_parent_criteria=["bogus-one", "bogus-two"]),
+    )
+    body = env.as_dict()
+    assert body["error"] == "invalid_state", body
+    assert "bogus-one" in body["message"]
+    assert "bogus-two" in body["message"]
+    task_svc.create_subtask.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_delegate_success_evidence_carries_covered_and_uncovered() -> None:
+    """A resolvable mapping creates the subtask and reports the parent's
+    coverage split in evidence, using the same primitive submit_up checks."""
+    pm_id = uuid4()
+    parent = _parent_with_criteria(pm_id)
+    new_task = MagicMock(id=uuid4())
+    task_svc = AsyncMock()
+    task_svc.get.return_value = parent
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    task_svc.get_subtasks.return_value = []
+    task_svc.unknown_ac_refs = MagicMock(return_value=[])
+    task_svc.create_subtask.return_value = new_task
+    task_svc.uncovered_parent_acceptance_criteria.return_value = ["Criterion B"]
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.delegate(pm_id, parent.id, _inputs(covers_parent_criteria=["id-a"]))
+    body = env.as_dict()
+    assert body["error"] is None, body
+    assert body["status"] == "created"
+    assert body["evidence"]["parent_ac_coverage"] == {
+        "covered": ["Criterion A"],
+        "uncovered": ["Criterion B"],
+    }
+
+
+@pytest.mark.asyncio
+async def test_delegate_wave_leaving_acs_uncovered_still_succeeds() -> None:
+    """No full-coverage hard gate at delegate: a wave may leave criteria for a
+    later delegate call — the child is still created, gaps just get listed."""
+    pm_id = uuid4()
+    parent = _parent_with_criteria(pm_id)
+    parent.acceptance_criteria = ["Criterion A", "Criterion B", "Criterion C"]
+    parent.acceptance_criteria_ids = ["id-a", "id-b", "id-c"]
+    new_task = MagicMock(id=uuid4())
+    task_svc = AsyncMock()
+    task_svc.get.return_value = parent
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    task_svc.get_subtasks.return_value = []
+    task_svc.unknown_ac_refs = MagicMock(return_value=[])
+    task_svc.create_subtask.return_value = new_task
+    task_svc.uncovered_parent_acceptance_criteria.return_value = [
+        "Criterion B",
+        "Criterion C",
+    ]
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.delegate(pm_id, parent.id, _inputs(covers_parent_criteria=["id-a"]))
+    body = env.as_dict()
+    assert body["error"] is None, body
+    coverage = body["evidence"]["parent_ac_coverage"]
+    assert coverage["covered"] == ["Criterion A"]
+    assert coverage["uncovered"] == ["Criterion B", "Criterion C"]
+    task_svc.create_subtask.assert_awaited_once()

tests/unit/gateway/test_delegate_incomplete_input.py

@@ -65,6 +65,8 @@ def _parent_in_progress(pm_id: Any) -> MagicMock:
         priority=2,
         # delegate obligates the PM's quick_context resumption section.
         quick_context="Decomposition planned; cells implement their slice next.",
+        # These tests are about task_completeness, not AC coverage.
+        acceptance_criteria=[],
     )
 
 

tests/unit/gateway/test_delegate_parent_lock.py

@@ -63,6 +63,8 @@ def _parent(pm_id: object) -> MagicMock:
         team="backend",
         # delegate obligates the PM's quick_context resumption section.
         quick_context="Decomposition planned; cells implement their slice next.",
+        # These tests are about the parent-lock ordering, not AC coverage.
+        acceptance_criteria=[],
     )
 
 

tests/unit/gateway/test_delegate_project_routing.py

@@ -51,6 +51,8 @@ def _parent(pm_id: Any, product_id: Any = None, project_id: Any = None) -> Magic
         team="backend",
         # delegate obligates the PM's quick_context resumption section.
         quick_context="Decomposition planned; cells implement their slice next.",
+        # These tests are about project routing, not AC coverage.
+        acceptance_criteria=[],
     )
 
 

tests/unit/gateway/test_pr_pass_ci_status_guard.py

@@ -4,12 +4,14 @@ Before this guard, ``pr_pass`` had no CI-status check at all — a reviewer coul
 pass an assembled PR whose CI was red, still running, or not yet scheduled.
 ``_ci_status_guard`` (wired into ``_pr_pass_blocked`` alongside the existing
 toolchain/conventions guards) reads ``GitService.get_pr_ci_status`` and blocks
-on failure/pending/pending_not_scheduled/error with reviewer-aware remediation
-(``pr_fail``, never ``i_am_blocked`` — a reviewer has no such verb). A project
-with no CI configured at all passes through cleanly, stamping the verdict note
-with why the guard did not block. ``pr_fail`` is unaffected by CI state
-entirely, and the separate inbound ``PRReviewerMixin`` surface
-(``claim_pr_review`` / ``post_pr_review``) never consults CI status at all.
+on failure/pending/pending_not_scheduled/error; only ``failure`` remediates
+via ``pr_fail`` (never ``i_am_blocked`` — a reviewer has no such verb) — the
+``error`` state is a transient GitHub API lookup failure and remediates via
+retry only, never ``pr_fail``. A project with no CI configured at all passes
+through cleanly, stamping the verdict note with why the guard did not block.
+``pr_fail`` is unaffected by CI state entirely, and the separate inbound
+``PRReviewerMixin`` surface (``claim_pr_review`` / ``post_pr_review``) never
+consults CI status at all.
 """
 
 from __future__ import annotations
@@ -162,6 +164,7 @@ async def test_pr_pass_blocked_on_github_api_error() -> None:
     assert env.error == "invalid_state"
     assert "GitHub API error" in (env.message or "")
     assert "retry" in (env.remediate or "").lower()
+    assert "do NOT pr_fail" in (env.remediate or "")
 
 
 @pytest.mark.asyncio

tests/unit/gateway/test_qa_criteria_verified.py

@@ -0,0 +1,175 @@
+"""pass_review requires a matched, evidenced verification per acceptance
+criterion — not just a count of arbitrary strings.
+
+Live failure this closes: QA passed a rendered video shipping 3 of the
+brief's 4 named scenes because nothing forced the reviewer to walk each
+acceptance criterion individually. Mirrors the test idiom in
+test_qa_ac_coverage.py, one level stricter: criteria_verified entries must
+each match a real AC (by id or exact text) and carry substantive evidence.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from roboco.services.gateway.choreographer import Choreographer
+
+_EVIDENCE_CAP = 500
+
+
+def _task(criteria: list[str], ids: list[str] | None = None) -> SimpleNamespace:
+    return SimpleNamespace(
+        acceptance_criteria=criteria,
+        acceptance_criteria_ids=ids or [],
+    )
+
+
+def test_no_criteria_imposes_no_requirement() -> None:
+    pairs, rej = Choreographer._validate_criteria_verified(_task([]), None)
+    assert pairs == []
+    assert rej is None
+
+
+def test_none_supplied_lists_every_criterion_verbatim() -> None:
+    criteria = [
+        "scene 1 renders",
+        "scene 2 renders",
+        "scene 3 renders",
+        "scene 4 renders",
+    ]
+    t = _task(criteria)
+    pairs, rej = Choreographer._validate_criteria_verified(t, None)
+    assert pairs == []
+    assert rej is not None
+    body = rej.as_dict()
+    assert body["error"] == "invalid_state", body
+    for crit in criteria:
+        assert crit in body["message"]
+
+
+def test_empty_list_is_treated_as_none_supplied() -> None:
+    t = _task(["a"])
+    pairs, rej = Choreographer._validate_criteria_verified(t, [])
+    assert pairs == []
+    assert rej is not None
+
+
+def test_partial_coverage_names_the_missing_criterion() -> None:
+    t = _task(["a", "b", "c"])
+    pairs, rej = Choreographer._validate_criteria_verified(
+        t,
+        [
+            {"criterion": "a", "evidence": "frame 1 shows a rendered"},
+            {"criterion": "b", "evidence": "frame 2 shows b rendered"},
+        ],
+    )
+    assert pairs == []
+    assert rej is not None
+    assert "c" in rej.as_dict()["message"]
+
+
+def test_unmatched_criterion_is_rejected_naming_valid_ones() -> None:
+    t = _task(["a", "b"])
+    pairs, rej = Choreographer._validate_criteria_verified(
+        t,
+        [
+            {"criterion": "a", "evidence": "frame 1 shows a rendered"},
+            {"criterion": "not-a-real-ac", "evidence": "frame 2 shows something"},
+        ],
+    )
+    assert pairs == []
+    assert rej is not None
+    body = rej.as_dict()
+    assert "not-a-real-ac" in body["message"]
+    assert "a" in body["remediate"] and "b" in body["remediate"]
+
+
+def test_missing_criterion_key_is_rejected() -> None:
+    t = _task(["a"])
+    pairs, rej = Choreographer._validate_criteria_verified(
+        t, [{"evidence": "frame 1 shows a rendered"}]
+    )
+    assert pairs == []
+    assert rej is not None
+
+
+def test_blank_evidence_is_rejected() -> None:
+    t = _task(["a"])
+    pairs, rej = Choreographer._validate_criteria_verified(
+        t, [{"criterion": "a", "evidence": "   "}]
+    )
+    assert pairs == []
+    assert rej is not None
+
+
+def test_soup_evidence_is_rejected() -> None:
+    t = _task(["a"])
+    pairs, rej = Choreographer._validate_criteria_verified(
+        t, [{"criterion": "a", "evidence": "wip"}]
+    )
+    assert pairs == []
+    assert rej is not None
+
+
+def test_overlong_evidence_is_rejected() -> None:
+    t = _task(["a"])
+    pairs, rej = Choreographer._validate_criteria_verified(
+        t, [{"criterion": "a", "evidence": "x" * (_EVIDENCE_CAP + 100)}]
+    )
+    assert pairs == []
+    assert rej is not None
+    assert str(_EVIDENCE_CAP) in rej.as_dict()["message"]
+
+
+def test_full_coverage_by_exact_text_passes() -> None:
+    t = _task(["a", "b"])
+    pairs, rej = Choreographer._validate_criteria_verified(
+        t,
+        [
+            {"criterion": "a", "evidence": "frame 1 shows a rendered fully"},
+            {"criterion": "b", "evidence": "frame 2 shows b rendered fully"},
+        ],
+    )
+    assert rej is None
+    assert pairs == [
+        ("a", "frame 1 shows a rendered fully"),
+        ("b", "frame 2 shows b rendered fully"),
+    ]
+
+
+def test_full_coverage_by_ac_id_passes() -> None:
+    t = _task(["scene renders"], ids=["AC-1"])
+    pairs, rej = Choreographer._validate_criteria_verified(
+        t, [{"criterion": "AC-1", "evidence": "rendered-frame path: out/frame3.png"}]
+    )
+    assert rej is None
+    assert pairs == [("AC-1", "rendered-frame path: out/frame3.png")]
+
+
+def test_extra_entries_beyond_the_ac_set_are_allowed() -> None:
+    t = _task(["a"])
+    pairs, rej = Choreographer._validate_criteria_verified(
+        t,
+        [{"criterion": "a", "evidence": "frame 1 shows a rendered fully"}],
+    )
+    assert rej is None
+    assert len(pairs) == 1
+
+
+def test_render_criteria_verified_matches_style() -> None:
+    lines = Choreographer._render_criteria_verified(
+        [("scene 1 renders", "frame 12 shows scene 1 fully")]
+    )
+    assert lines == ["[AC] scene 1 renders — verified: frame 12 shows scene 1 fully"]
+
+
+def test_merge_criteria_verified_into_notes() -> None:
+    merged = Choreographer._merge_criteria_verified_into_notes(
+        "base review", [("a", "evidence a"), ("b", "evidence b")]
+    )
+    assert "[AC] a — verified: evidence a" in merged
+    assert "[AC] b — verified: evidence b" in merged
+
+
+def test_merge_with_no_pairs_returns_notes_unchanged() -> None:
+    assert Choreographer._merge_criteria_verified_into_notes("base", []) == "base"

tests/unit/gateway/test_resubmit_unchanged_head_exemption.py

@@ -0,0 +1,233 @@
+"""The one-shot resubmit exemption on the unchanged-PR loop-stopper.
+
+Live wedge: a cell task looped block/unblock for hours because the freshness
+guard (``_unchanged_pr_guard``, shared by ``submit_root``/``submit_up``)
+demands a new commit after ANY ``pr_fail`` — a structural deadlock when the
+rejection round's findings require no code change (e.g. a transient
+CI-lookup error ledgered as a finding, then waived/addressed). Once every
+ledger finding is resolved (``_open_finding_ids`` empty), ONE resubmission is
+exempted per head sha (``markers.resubmit_unchanged_head``); a second attempt
+at the SAME head still refuses. Covers both call sites: ``submit_root``
+(root) and ``submit_up`` (cell).
+
+Note on test style: the upstream ``FINDINGS_ADDRESSED`` tracing gate
+(``_check_submit_up_gates``, shared by both verbs) already refuses the whole
+verb with ``tracing_gap`` whenever findings are open — by the time
+``_unchanged_pr_guard`` runs, open findings are already impossible through the
+public ``submit_root``/``submit_up`` entrypoints. The "findings still open"
+and "ambiguous case short-circuits before the findings check" scenarios are
+therefore exercised by calling the guard method directly (defense-in-depth on
+the guard's own logic); the exemption-grant and one-shot-per-head scenarios
+drive the full verb end-to-end to prove the real wiring.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.foundation.policy.content import markers
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+SHA_OLD = "aaaa1111bbbb2222cccc3333dddd4444eeee5555"
+SHA_NEW = "9999888877776666555544443333222211110000"
+
+
+def _make_deps(**overrides: Any) -> ChoreographerDeps:
+    base: dict[str, Any] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    base.update(overrides)
+    base["journal"].has_decision_for_task.return_value = True
+    base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    base["journal"].has_reflect_for_task.return_value = True
+    return ChoreographerDeps(**base)
+
+
+def _unchanged_notes(head_sha: str = SHA_OLD) -> dict[str, Any]:
+    return {"pr_review": {"verdict": "failed", "head_sha": head_sha, "summary": "..."}}
+
+
+async def _call_guard(
+    c: Choreographer, kind: str, t: Any, briefing: dict[str, Any]
+) -> Any:
+    if kind == "root":
+        return await c._submit_root_unchanged_pr_guard(t, briefing)
+    return await c._submit_up_unchanged_pr_guard(t, briefing)
+
+
+# ---------------------------------------------------------------------------
+# Direct guard tests — isolate ``_unchanged_pr_guard`` from the upstream
+# FINDINGS_ADDRESSED gate, which already forbids open findings from ever
+# reaching here through the real submit_root/submit_up flow.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["root", "cell"])
+async def test_refused_when_findings_still_open(kind: str) -> None:
+    """Unchanged head + open findings -> refused (behavior unchanged)."""
+    c = Choreographer(_make_deps())
+    cc: Any = c
+    cc._current_pr_head_sha = AsyncMock(return_value=SHA_OLD)
+    cc._open_finding_ids = AsyncMock(return_value=("abcd1234",))
+    t = MagicMock(
+        id=uuid4(), notes_structured=_unchanged_notes(), orchestration_markers=None
+    )
+
+    env = await _call_guard(c, kind, t, {})
+
+    assert env is not None
+    assert env.error == "invalid_state", env.as_dict()
+    assert "unchanged" in (env.message or "").lower()
+    assert markers.get_resubmit_unchanged_head(t) is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["root", "cell"])
+async def test_allowed_when_head_advanced_before_findings_are_even_checked(
+    kind: str,
+) -> None:
+    """A different current head SHA -> allowed; the ambiguity check short-
+    circuits before the findings/exemption logic ever runs."""
+    c = Choreographer(_make_deps())
+    cc: Any = c
+    cc._current_pr_head_sha = AsyncMock(return_value=SHA_NEW)
+    open_findings_spy = AsyncMock(return_value=("abcd1234",))
+    cc._open_finding_ids = open_findings_spy
+    t = MagicMock(
+        id=uuid4(), notes_structured=_unchanged_notes(), orchestration_markers=None
+    )
+
+    env = await _call_guard(c, kind, t, {})
+
+    assert env is None
+    open_findings_spy.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# End-to-end tests — drive the real submit_root/submit_up verb to prove the
+# exemption's marker stamp actually lets the assembled PR proceed into the
+# gate. Zero open findings satisfies both the upstream FINDINGS_ADDRESSED gate
+# and the new exemption check, mirroring test_submit_root_unchanged_pr_guard.py
+# / test_submit_up_unchanged_pr_guard.py's fixture shape.
+# ---------------------------------------------------------------------------
+
+
+def _resubmit(
+    kind: str,
+    *,
+    notes_structured: dict[str, Any] | None,
+    pr_number: int | None = 139,
+) -> tuple[Choreographer, Any, Any]:
+    pm_id = uuid4()
+    task_id = uuid4()
+    if kind == "root":
+        role, team, parent = "main_pm", "main_pm", None
+        branch = "feature/main_pm/c80e19ff"
+    else:
+        role, team, parent = "cell_pm", "backend", uuid4()
+        branch = "feature/backend/cell-task"
+    in_prog = MagicMock(
+        id=task_id,
+        status="in_progress",
+        assigned_to=pm_id,
+        pr_number=pr_number,
+        branch_name=branch,
+        parent_task_id=parent,
+        batch_id=None,
+        team=team,
+        notes_structured=notes_structured,
+        orchestration_markers=None,
+    )
+    gated = MagicMock(**{**in_prog.__dict__, "status": "awaiting_pr_review"})
+    task_svc = AsyncMock()
+    task_svc.get.return_value = in_prog
+    task_svc.submit_for_review.return_value = gated
+    task_svc.all_subtasks_terminal.return_value = True
+    task_svc.uncovered_parent_acceptance_criteria.return_value = []
+    task_svc.agent_for.return_value = MagicMock(role=role, team=team)
+    task_svc.session.begin_nested = MagicMock(
+        return_value=MagicMock(__aenter__=AsyncMock(), __aexit__=AsyncMock())
+    )
+    c = Choreographer(_make_deps(task=task_svc, git=AsyncMock()))
+    cc: Any = c
+    cc._project_slug_for = AsyncMock(return_value="proj-slug")
+    # Zero open findings throughout — satisfies the upstream FINDINGS_ADDRESSED
+    # gate so the flow reaches the unchanged-PR guard under test.
+    cc._open_finding_ids = AsyncMock(return_value=())
+    return c, pm_id, task_id
+
+
+async def _call_submit(
+    c: Choreographer, kind: str, pm_id: Any, task_id: Any, notes: str
+) -> Any:
+    if kind == "root":
+        return await c.submit_root(pm_id, task_id, notes=notes)
+    return await c.submit_up(pm_id, task_id, notes=notes)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["root", "cell"])
+async def test_exemption_granted_when_no_open_findings_and_no_prior_marker(
+    kind: str,
+) -> None:
+    """Unchanged head + zero open findings + no marker -> allowed once, marker
+    stamped with the current head."""
+    c, pm_id, task_id = _resubmit(kind, notes_structured=_unchanged_notes())
+    c.git.get_pr_head_sha = AsyncMock(return_value=SHA_OLD)
+
+    env = await _call_submit(
+        c, kind, pm_id, task_id, "re-submitting; CI blip only, no code change needed"
+    )
+
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"
+    c.task.submit_for_review.assert_awaited_once()
+    t = await c.task.get(task_id)
+    assert markers.get_resubmit_unchanged_head(t) == SHA_OLD
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["root", "cell"])
+async def test_second_attempt_at_same_head_is_refused(kind: str) -> None:
+    """The exemption is one-shot: a second resubmit at the SAME head, still
+    with no open findings, refuses — the marker already recorded this head."""
+    c, pm_id, task_id = _resubmit(kind, notes_structured=_unchanged_notes())
+    c.git.get_pr_head_sha = AsyncMock(return_value=SHA_OLD)
+
+    first = await _call_submit(c, kind, pm_id, task_id, "first resubmit; CI blip only")
+    assert first.error is None, first.as_dict()
+
+    second = await _call_submit(
+        c, kind, pm_id, task_id, "second resubmit; still the same unchanged head"
+    )
+    assert second.error == "invalid_state", second.as_dict()
+    assert "unchanged" in (second.message or "").lower()
+    assert "already used" in (second.message or "").lower()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["root", "cell"])
+async def test_allowed_when_head_advanced(kind: str) -> None:
+    """A different current head SHA -> allowed (existing fail-open behavior),
+    end to end through the real verb."""
+    c, pm_id, task_id = _resubmit(kind, notes_structured=_unchanged_notes())
+    c.git.get_pr_head_sha = AsyncMock(return_value=SHA_NEW)
+
+    env = await _call_submit(
+        c, kind, pm_id, task_id, "resubmitting after the real fix landed"
+    )
+
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"
+    c.task.submit_for_review.assert_awaited_once()

tests/unit/gateway/test_submit_root_unchanged_pr_guard.py

@@ -105,8 +105,15 @@ def _resubmit_root(
 
 @pytest.mark.asyncio
 async def test_submit_root_refuses_unchanged_pr_after_pr_fail() -> None:
-    """The loop-stopper: prior pr_fail stamped head SHA X, the PR head is still
-    X (no new cell work on the root branch) → refuse, do not open the gate."""
+    """The loop-stopper still holds past the one-shot exemption: prior
+    pr_fail stamped head SHA X, the PR head is still X (no new cell work on
+    the root branch). The findings ledger here fail-opens to "nothing open"
+    (mock session, no real query) — the same signal ``_check_submit_up_gates``
+    upstream already reads for FINDINGS_ADDRESSED — so the first resubmit at
+    this head is the one-shot exemption (see
+    test_resubmit_unchanged_head_exemption.py for full exemption coverage);
+    a second resubmit at the SAME head refuses, so the loop still can't run
+    forever."""
     c, main_pm_id, root_task_id = _resubmit_root(
         notes_structured={
             "pr_review": {"verdict": "failed", "head_sha": SHA_OLD, "summary": "..."}
@@ -114,9 +121,14 @@ async def test_submit_root_refuses_unchanged_pr_after_pr_fail() -> None:
     )
     c.git.get_pr_head_sha = AsyncMock(return_value=SHA_OLD)
 
-    env = await c.submit_root(
+    first = await c.submit_root(
         main_pm_id, root_task_id, notes="re-submitting the root after the fix"
     )
+    assert first.error is None, first.as_dict()
+
+    env = await c.submit_root(
+        main_pm_id, root_task_id, notes="re-submitting again; still unchanged"
+    )
 
     assert env.error is not None, env.as_dict()
     assert env.error == "invalid_state"
@@ -124,8 +136,9 @@ async def test_submit_root_refuses_unchanged_pr_after_pr_fail() -> None:
     remediate = env.remediate or ""
     assert "re-delegate" in remediate
     assert "submit_root" in remediate
-    # The PR was NOT re-opened / re-pushed — the runner never ran.
-    c.task.submit_for_review.assert_not_awaited()
+    # The second attempt's PR was NOT re-opened / re-pushed — the runner ran
+    # only for the first (exempted) call.
+    c.task.submit_for_review.assert_awaited_once()
 
 
 @pytest.mark.asyncio

tests/unit/gateway/test_submit_up_unchanged_pr_guard.py

@@ -80,9 +80,15 @@ def _resubmit_cell(
 
 @pytest.mark.asyncio
 async def test_submit_up_refuses_unchanged_pr_after_pr_fail() -> None:
-    """The loop-stopper: prior pr_fail stamped head SHA X, the cell PR head is
-    still X (no new dev work on the cell branch) → refuse, do not re-open the
-    gate."""
+    """The loop-stopper still holds past the one-shot exemption: prior
+    pr_fail stamped head SHA X, the cell PR head is still X (no new dev work
+    on the cell branch). The findings ledger here fail-opens to "nothing
+    open" (mock session, no real query) — the same signal
+    ``_check_submit_up_gates`` upstream already reads for FINDINGS_ADDRESSED
+    — so the first resubmit at this head is the one-shot exemption (see
+    test_resubmit_unchanged_head_exemption.py for full exemption coverage);
+    a second resubmit at the SAME head refuses, so the loop still can't run
+    forever."""
     c, cell_pm_id, cell_task_id = _resubmit_cell(
         notes_structured={
             "pr_review": {"verdict": "failed", "head_sha": SHA_OLD, "summary": "..."}
@@ -90,17 +96,23 @@ async def test_submit_up_refuses_unchanged_pr_after_pr_fail() -> None:
     )
     c.git.get_pr_head_sha = AsyncMock(return_value=SHA_OLD)
 
-    env = await c.submit_up(
+    first = await c.submit_up(
         cell_pm_id, cell_task_id, notes="re-submitting the cell after the fix"
     )
+    assert first.error is None, first.as_dict()
+
+    env = await c.submit_up(
+        cell_pm_id, cell_task_id, notes="re-submitting again; still unchanged"
+    )
 
     assert env.error is not None, env.as_dict()
     assert env.error == "invalid_state", env.as_dict()
     assert "unchanged" in (env.message or "").lower()
     remediate = env.remediate or ""
     assert "submit_up" in remediate
-    # The cell PR was NOT re-opened / re-pushed — the runner never ran.
-    c.task.submit_for_review.assert_not_awaited()
+    # The second attempt's PR was NOT re-opened / re-pushed — the runner ran
+    # only for the first (exempted) call.
+    c.task.submit_for_review.assert_awaited_once()
 
 
 @pytest.mark.asyncio

tests/unit/gateway/test_unblock_flip_breaker.py

@@ -0,0 +1,151 @@
+"""The block/unblock flip-flop breaker.
+
+Live wedge: fe-pm's escalate_up auto-blocks a task and main_pm's unblock
+resolves it, repeat — 10 flips, 43 spawns, no forward progress, no cycle
+breaker. ``unblock`` now stamps a per-task flip counter
+(``markers.block_flip_count``) and, at exactly the 3rd flip, best-effort
+alerts the CEO once — the unblock itself always still succeeds.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.foundation.policy.content import markers
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+from roboco.services.notification import NotificationService
+
+# Named constants — ruff PLR2004 forbids magic-value comparisons.
+_TWO_FLIPS = 2
+_THREE_FLIPS = 3
+_FOUR_FLIPS = 4
+
+
+def _make_deps(**overrides: Any) -> ChoreographerDeps:
+    base: dict[str, Any] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    base.update(overrides)
+    base["journal"].has_decision_for_task.return_value = True
+    base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    return ChoreographerDeps(**base)
+
+
+def _flip_setup() -> tuple[Choreographer, Any, Any, Any]:
+    """A blocked task whose ``unblock_with_restore`` returns the SAME mock
+    object each call, so the flip-counter marker persists across repeated
+    unblock() calls the way it would on one real ORM row across requests.
+    """
+    pm_id = uuid4()
+    task_id = uuid4()
+    t = MagicMock(
+        id=task_id,
+        status="blocked",
+        pre_block_state="in_progress",
+        pre_block_assignee=uuid4(),
+        pre_block_metadata={},
+        dependency_ids=[],
+        orchestration_markers=None,
+    )
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.unblock_with_restore.return_value = t
+    task_svc.unmet_dependency_ids.return_value = []
+    c = Choreographer(_make_deps(task=task_svc))
+    return c, pm_id, task_id, t
+
+
+async def _unblock_once(c: Choreographer, pm_id: Any, task_id: Any, t: Any) -> Any:
+    """Re-block before each call — a fresh flip in the cycle."""
+    t.status = "blocked"
+    return await c.unblock(pm_id, task_id, "resolved upstream; restoring")
+
+
+@pytest.mark.asyncio
+async def test_first_and_second_unblock_do_not_notify() -> None:
+    c, pm_id, task_id, t = _flip_setup()
+    cc: Any = c
+    notify = AsyncMock()
+    cc._notify_ceo_block_flip = notify
+
+    for _ in range(2):
+        env = await _unblock_once(c, pm_id, task_id, t)
+        assert env.error is None, env.as_dict()
+
+    notify.assert_not_awaited()
+    assert markers.get_block_flip_count(t) == _TWO_FLIPS
+
+
+@pytest.mark.asyncio
+async def test_third_unblock_notifies_ceo_once() -> None:
+    c, pm_id, task_id, t = _flip_setup()
+    cc: Any = c
+    notify = AsyncMock()
+    cc._notify_ceo_block_flip = notify
+
+    for _ in range(3):
+        env = await _unblock_once(c, pm_id, task_id, t)
+        assert env.error is None, env.as_dict()
+
+    notify.assert_awaited_once_with(task_id, _THREE_FLIPS)
+    assert markers.is_block_flip_notified(t) is True
+
+
+@pytest.mark.asyncio
+async def test_fourth_unblock_does_not_renotify() -> None:
+    c, pm_id, task_id, t = _flip_setup()
+    cc: Any = c
+    notify = AsyncMock()
+    cc._notify_ceo_block_flip = notify
+
+    for _ in range(4):
+        env = await _unblock_once(c, pm_id, task_id, t)
+        assert env.error is None, env.as_dict()
+
+    notify.assert_awaited_once()
+    assert markers.get_block_flip_count(t) == _FOUR_FLIPS
+
+
+@pytest.mark.asyncio
+async def test_notification_failure_does_not_fail_unblock(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The real ``_notify_ceo_block_flip`` swallows a notify-service failure —
+    unblock still succeeds on the 3rd flip."""
+    c, pm_id, task_id, t = _flip_setup()
+    monkeypatch.setattr(
+        NotificationService,
+        "send_block_flip_notification",
+        AsyncMock(side_effect=RuntimeError("notification service down")),
+    )
+
+    env = None
+    for _ in range(3):
+        env = await _unblock_once(c, pm_id, task_id, t)
+        assert env.error is None, env.as_dict()
+
+    assert env is not None
+    assert env.error is None
+    assert markers.is_block_flip_notified(t) is True
+
+
+@pytest.mark.asyncio
+async def test_counter_persists_via_marker_across_calls() -> None:
+    c, pm_id, task_id, t = _flip_setup()
+    cc: Any = c
+    cc._notify_ceo_block_flip = AsyncMock()
+
+    await _unblock_once(c, pm_id, task_id, t)
+    assert markers.get_block_flip_count(t) == 1
+    await _unblock_once(c, pm_id, task_id, t)
+    assert markers.get_block_flip_count(t) == _TWO_FLIPS