[fix] submit_root: hard unchanged-PR gate stops the pr_fail re-submit loop

The 2026-06-27 infinite pr_fail loop: a Main-PM root (PR #139) was pr_fail'd,
routed to needs_revision, and re-submitted byte-identical → awaiting_pr_review
→ pr_fail again, forever. The prior hint/a2a steer was ignored by the weak
coordinator model — hints don't stop a model that won't read them. A HARD gate
refuses the re-submit when the assembled root PR's head SHA is unchanged since
the last pr_fail (no new cell work → identical diff); a different SHA ⇒ the
branch advanced ⇒ allow. Every ambiguous case fails open (no prior fail, no
recorded SHA, no pr_number, unresolvable slug, git error, closed PR) — only the
exact-unchanged case is hard-blocked.

- content/models: PrReviewContent.head_sha (optional; JSON col → no migration).
- git: get_pr_head_sha (GitHub pulls API; None on any failure → fail-open).
- pr_gate: pr_fail captures head_sha into the verdict record; pr_pass does not.
- _impl: submit_root runs _submit_root_unchanged_pr_guard after _submit_up_guard;
  _current_root_pr_head_sha resolves slug + current SHA (fail-open).
- pr_review: extract module-level resolve_task_project_slug, shared by the mixin
  and the gate helper (_LegacyChoreographer reaches it via cast to the
  ChoreographerHelpers typed view — it doesn't inherit the helpers mixin).
- tests: test_submit_root_unchanged_pr_guard (11 — refuse/allow/6 fail-open/3
  capture-side, mypy-clean via cc:Any spy idiom, zero type:ignore) +
  test_pr_gate_notifies_pm capture-path stub.

tests/unit/gateway/test_pr_gate_notifies_pm.py

@@ -63,6 +63,14 @@ def _stub_gate_path(
         )
     )
     c._gate_tracing = AsyncMock(return_value=None)  # type: ignore[method-assign]
+    # These tests exercise the pr_fail a2a / notify path, not the head-sha
+    # capture (which has its own suite in test_submit_root_unchanged_pr_guard).
+    # Stub the capture so it does not walk the mock session into un-awaited
+    # coroutines; the verdict still lands via the _record_gate_verdict spy.
+    # Alias to ``Any`` so this addition needs no type:ignore (mypy doesn't flag
+    # attribute assignment on ``Any``; avoids ruff B010's no-setattr rule too).
+    cc: Any = c
+    cc._capture_pr_head_sha = AsyncMock(return_value=None)
     c._record_gate_verdict = MagicMock()  # type: ignore[method-assign]
     c._post_gate_review_to_pr = AsyncMock()  # type: ignore[method-assign]
     runner = MagicMock()

tests/unit/gateway/test_submit_root_unchanged_pr_guard.py

@@ -0,0 +1,406 @@
+"""The hard ``submit_root`` unchanged-PR gate — the pr_fail re-submit loop-stopper.
+
+The 2026-06-27 infinite ``pr_fail`` loop: a Main-PM-owned root (S1
+"chart-first Metrics", PR #139) was ``pr_fail``'d for a real code defect, routed
+to ``needs_revision``, the Main PM re-claimed + re-delegated nothing, and
+re-submitted the **unchanged** root → ``awaiting_pr_review`` → ``pr_fail`` again,
+forever. The prior fixes (the ``pr_fail`` a2a steer + the ``next_hint`` "do NOT
+re-submit") are *hints* — a weak coordinator (minimax-m3:cloud) ignored them and
+re-submitted PR #139 byte-identical. Hints do not stop a model that won't read
+them; only a structural refusal does.
+
+This gate refuses the re-submit when the assembled root PR's head SHA is
+unchanged since the last ``pr_fail`` (no new cell work landed on the root
+branch). ``pr_fail`` stamps that SHA into ``notes_structured.pr_review.head_sha``
+(``_capture_pr_head_sha`` + ``_record_gate_verdict``); ``submit_root`` reads it
+back and compares against the PR's current head SHA. Equal ⇒ refuse; different
+⇒ the branch advanced ⇒ allow. Every ambiguous case FAILS OPEN (no prior fail,
+no recorded SHA, no PR number, no resolvable project, git/closed-PR lookup
+returns ``None``) — only the exact-unchanged case is hard-blocked.
+"""
+
+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 import lifecycle as spec_module
+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 _resubmit_root(
+    *,
+    notes_structured: dict[str, Any] | None,
+    pr_number: int | None = 139,
+) -> tuple[Choreographer, Any, Any]:
+    """A Main-PM root re-submitted from ``in_progress`` after a ``pr_fail``.
+
+    Mirrors the live c80e19ff / PR #139 re-submit: the root is back in
+    ``in_progress`` (re-claimed out of ``needs_revision``), carries the prior
+    ``pr_fail`` verdict in ``notes_structured.pr_review``, and the PR is still
+    open. The ``_submit_up_guard`` preflight is satisfied (owned, journal
+    decision, subtasks terminal, branch present, notes long enough) so the
+    unchanged-PR gate is the thing under test.
+    """
+    main_pm_id = uuid4()
+    root_task_id = uuid4()
+    in_prog = MagicMock(
+        id=root_task_id,
+        status="in_progress",
+        assigned_to=main_pm_id,
+        pr_number=pr_number,
+        branch_name="feature/main_pm/c80e19ff",
+        parent_task_id=None,
+        batch_id=None,
+        team="main_pm",
+        notes_structured=notes_structured,
+    )
+    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="main_pm", team="main_pm")
+    task_svc.session.begin_nested = MagicMock(
+        return_value=MagicMock(__aenter__=AsyncMock(), __aexit__=AsyncMock())
+    )
+    c = Choreographer(_make_deps(task=task_svc, git=AsyncMock()))
+    # Real _project_slug_for would walk a mock session into a MagicMock slug; the
+    # gate under test needs a real string slug + a controllable head SHA. Alias to
+    # ``Any`` so mypy doesn't flag the method-spy assignment (no type:ignore owed).
+    cc: Any = c
+    cc._project_slug_for = AsyncMock(return_value="proj-slug")
+    return c, main_pm_id, root_task_id
+
+
+# ---------------------------------------------------------------------------
+# The hard block — refuse the byte-identical re-submit
+# ---------------------------------------------------------------------------
+
+
+@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."""
+    c, main_pm_id, root_task_id = _resubmit_root(
+        notes_structured={
+            "pr_review": {"verdict": "failed", "head_sha": SHA_OLD, "summary": "..."}
+        }
+    )
+    c.git.get_pr_head_sha = AsyncMock(return_value=SHA_OLD)
+
+    env = await c.submit_root(
+        main_pm_id, root_task_id, notes="re-submitting the root after the fix"
+    )
+
+    assert env.error is not None, env.as_dict()
+    assert env.error == "invalid_state"
+    assert "unchanged" in (env.message or "").lower()
+    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()
+
+
+@pytest.mark.asyncio
+async def test_submit_root_allows_after_root_branch_advanced() -> None:
+    """A different current head SHA ⇒ cell work landed on the root branch ⇒
+    the diff changed ⇒ allow the re-submit into the gate."""
+    c, main_pm_id, root_task_id = _resubmit_root(
+        notes_structured={
+            "pr_review": {"verdict": "failed", "head_sha": SHA_OLD, "summary": "..."}
+        }
+    )
+    c.git.get_pr_head_sha = AsyncMock(return_value=SHA_NEW)
+
+    env = await c.submit_root(
+        main_pm_id, root_task_id, notes="re-submitting after the cell re-assembly"
+    )
+
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"
+    c.task.submit_for_review.assert_awaited_once()
+
+
+# ---------------------------------------------------------------------------
+# Fail-open — ambiguous cases proceed and rely on the reviewer to re-fail
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_submit_root_fail_open_when_no_prior_pr_fail_verdict() -> None:
+    """No pr_review (first submit) or a passed verdict ⇒ nothing to compare ⇒
+    allow."""
+    c, main_pm_id, root_task_id = _resubmit_root(notes_structured=None)
+    env = await c.submit_root(
+        main_pm_id, root_task_id, notes="first root submit; nothing to compare yet"
+    )
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"
+
+
+@pytest.mark.asyncio
+async def test_submit_root_fail_open_when_prior_fail_recorded_no_head_sha() -> None:
+    """A pr_fail verdict written before this field existed has no ``head_sha`` ⇒
+    cannot compare ⇒ allow (fail open, not wedge)."""
+    c, main_pm_id, root_task_id = _resubmit_root(
+        notes_structured={"pr_review": {"verdict": "failed", "summary": "..."}}
+    )
+    env = await c.submit_root(
+        main_pm_id, root_task_id, notes="re-submit; no recorded sha to compare"
+    )
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"
+
+
+@pytest.mark.asyncio
+async def test_submit_root_fail_open_when_git_lookup_returns_none() -> None:
+    """A closed/missing PR or a git error returns ``None`` ⇒ ambiguous ⇒ allow
+    (the reviewer can still pr_fail if the diff is bad)."""
+    c, main_pm_id, root_task_id = _resubmit_root(
+        notes_structured={
+            "pr_review": {"verdict": "failed", "head_sha": SHA_OLD, "summary": "..."}
+        }
+    )
+    c.git.get_pr_head_sha = AsyncMock(return_value=None)
+    env = await c.submit_root(
+        main_pm_id, root_task_id, notes="re-submit; the prior PR was closed or missing"
+    )
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"
+
+
+@pytest.mark.asyncio
+async def test_submit_root_fail_open_when_no_pr_number() -> None:
+    """A root with no ``pr_number`` has nothing to look up ⇒ allow."""
+    c, main_pm_id, root_task_id = _resubmit_root(
+        notes_structured={
+            "pr_review": {"verdict": "failed", "head_sha": SHA_OLD, "summary": "..."}
+        },
+        pr_number=None,
+    )
+    env = await c.submit_root(
+        main_pm_id, root_task_id, notes="re-submit; this root has no pr number"
+    )
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"
+
+
+@pytest.mark.asyncio
+async def test_submit_root_fail_open_when_slug_unresolvable() -> None:
+    """No resolvable project slug (a product-only root the product service can't
+    expand) ⇒ can't query git ⇒ allow."""
+    c, main_pm_id, root_task_id = _resubmit_root(
+        notes_structured={
+            "pr_review": {"verdict": "failed", "head_sha": SHA_OLD, "summary": "..."}
+        }
+    )
+    cc: Any = c
+    cc._project_slug_for = AsyncMock(return_value=None)
+    c.git.get_pr_head_sha = AsyncMock(return_value=SHA_OLD)
+    env = await c.submit_root(
+        main_pm_id, root_task_id, notes="re-submit; project slug unresolvable here"
+    )
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"
+
+
+@pytest.mark.asyncio
+async def test_submit_root_fail_open_when_git_lookup_raises() -> None:
+    """A git lookup that raises must not 500 the PM — the gate swallows it and
+    proceeds (fail open)."""
+    c, main_pm_id, root_task_id = _resubmit_root(
+        notes_structured={
+            "pr_review": {"verdict": "failed", "head_sha": SHA_OLD, "summary": "..."}
+        }
+    )
+    c.git.get_pr_head_sha = AsyncMock(side_effect=RuntimeError("boom"))
+    env = await c.submit_root(
+        main_pm_id, root_task_id, notes="re-submit; git head-sha lookup raised an error"
+    )
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"
+
+
+# ---------------------------------------------------------------------------
+# The capture side — pr_fail stamps the head SHA into notes_structured
+# ---------------------------------------------------------------------------
+
+
+def _make_choreographer_for_gate() -> Choreographer:
+    """A choreographer wired to drive ``_gate_decision`` past preflight/tracing
+    and into the verdict-record step without exercising the heavy ownership
+    logic (those have their own tests). Mirrors test_pr_gate_notifies_pm."""
+    base: dict[str, Any] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    return Choreographer(ChoreographerDeps(**base))
+
+
+def _stub_gate_path(
+    c: Choreographer, *, reviewer_id: Any, t_before: Any, t_after: Any
+) -> MagicMock:
+    """Drive ``_gate_decision`` past preflight/tracing/post and into the verdict
+    record step. Returns the ``_record_gate_verdict`` spy so callers can assert
+    on the recorded kwargs. The ``cc: Any`` alias is the method-spy idiom: mypy
+    doesn't flag attribute assignment on ``Any`` (no method-assign / no
+    attr-defined), so no ``type: ignore`` is owed and ruff's B010 (no ``setattr``
+    with a constant) is sidestepped too.
+    """
+    cc: Any = c
+    agent = MagicMock(role="pr_reviewer", slug="be-pr-reviewer")
+    cc._gate_preflight = AsyncMock(
+        return_value=(
+            t_before,
+            agent,
+            "pr_reviewer",
+            {},
+            spec_module.Context(actor_id=reviewer_id),
+        )
+    )
+    cc._gate_tracing = AsyncMock(return_value=None)
+    # Spy on the verdict record so we can assert the head_sha kwarg without
+    # running the real apply_structured_note (which needs a real ORM task).
+    record_spy = MagicMock()
+    cc._record_gate_verdict = record_spy
+    cc._post_gate_review_to_pr = AsyncMock()
+    runner = MagicMock()
+    runner.run_intent = AsyncMock(return_value=t_after)
+    cc._verb_runner = MagicMock(return_value=runner)
+    return record_spy
+
+
+@pytest.mark.asyncio
+async def test_pr_fail_captures_head_sha_into_verdict() -> None:
+    """pr_fail resolves the PR's head SHA and threads it into the verdict record
+    so the next submit_root can compare against it."""
+    reviewer_id = uuid4()
+    pm_id = uuid4()
+    task_id = uuid4()
+    t_before = MagicMock(
+        id=task_id,
+        assigned_to=reviewer_id,
+        pr_number=139,
+        parent_task_id=uuid4(),
+        status="awaiting_pr_review",
+    )
+    t_after = MagicMock(
+        id=task_id,
+        assigned_to=pm_id,
+        pr_number=139,
+        parent_task_id=uuid4(),
+        status="needs_revision",
+    )
+
+    c = _make_choreographer_for_gate()
+    record_spy = _stub_gate_path(
+        c, reviewer_id=reviewer_id, t_before=t_before, t_after=t_after
+    )
+    cc: Any = c
+    cc._project_slug_for = AsyncMock(return_value="proj-slug")
+    c.git.get_pr_head_sha = AsyncMock(return_value=SHA_OLD)
+
+    await c.pr_fail(reviewer_id, task_id, ["duplicate TimeseriesChart export"])
+
+    record_spy.assert_called_once()
+    kwargs = record_spy.call_args.kwargs
+    assert kwargs["head_sha"] == SHA_OLD
+
+
+@pytest.mark.asyncio
+async def test_pr_fail_capture_best_effort_when_git_raises() -> None:
+    """A git head-sha lookup that raises must not crash the gate — head_sha
+    falls back to None (the submit_root gate then fails open) and the
+    transition still proceeds to needs_revision."""
+    reviewer_id = uuid4()
+    pm_id = uuid4()
+    task_id = uuid4()
+    t_before = MagicMock(
+        id=task_id,
+        assigned_to=reviewer_id,
+        pr_number=139,
+        parent_task_id=uuid4(),
+        status="awaiting_pr_review",
+    )
+    t_after = MagicMock(
+        id=task_id, assigned_to=pm_id, pr_number=139, status="needs_revision"
+    )
+
+    c = _make_choreographer_for_gate()
+    record_spy = _stub_gate_path(
+        c, reviewer_id=reviewer_id, t_before=t_before, t_after=t_after
+    )
+    cc: Any = c
+    cc._project_slug_for = AsyncMock(return_value="proj-slug")
+    c.git.get_pr_head_sha = AsyncMock(side_effect=RuntimeError("github 503"))
+
+    env = await c.pr_fail(reviewer_id, task_id, ["a concrete issue"])
+
+    assert env.status == "needs_revision"
+    record_spy.assert_called_once()
+    assert record_spy.call_args.kwargs["head_sha"] is None
+
+
+@pytest.mark.asyncio
+async def test_pr_pass_does_not_capture_head_sha() -> None:
+    """Only pr_fail stamps a head SHA — pr_pass must not (there is no loop to
+    guard against a pass)."""
+    reviewer_id = uuid4()
+    pm_id = uuid4()
+    task_id = uuid4()
+    t_before = MagicMock(
+        id=task_id,
+        assigned_to=reviewer_id,
+        pr_number=42,
+        parent_task_id=uuid4(),
+        status="awaiting_pr_review",
+    )
+    t_after = MagicMock(
+        id=task_id, assigned_to=pm_id, pr_number=42, status="awaiting_pm_review"
+    )
+
+    c = _make_choreographer_for_gate()
+    record_spy = _stub_gate_path(
+        c, reviewer_id=reviewer_id, t_before=t_before, t_after=t_after
+    )
+    cc: Any = c
+    cc._project_slug_for = AsyncMock(return_value="proj-slug")
+    c.git.get_pr_head_sha = AsyncMock(return_value=SHA_OLD)
+
+    await c.pr_pass(reviewer_id, task_id, "Assembled root scope is clean.")
+
+    record_spy.assert_called_once()
+    # pr_pass path never calls _capture_pr_head_sha, so head_sha is absent
+    # from the kwargs (the default None is not passed).
+    assert "head_sha" not in record_spy.call_args.kwargs