[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.

roboco/foundation/policy/content/models.py

@@ -157,6 +157,14 @@ class PrReviewContent(_Content):
     findings: list[Finding] = Field(default_factory=list)
     issues: list[str] = Field(default_factory=list)
     verdict: Verdict
+    # The head SHA of the assembled PR at the moment a ``pr_fail`` landed —
+    # captured so the next ``submit_root`` can structurally refuse to re-submit
+    # the unchanged root (the 2026-06-27 infinite ``pr_fail`` re-submit loop: a
+    # weak coordinator re-submitted PR #139 with no new cell work on the root
+    # branch, so the reviewed diff was byte-identical and the gate failed
+    # again). ``None`` for a ``pr_pass`` verdict and for verdicts recorded before
+    # this field existed. Optional + JSON column, so no migration.
+    head_sha: str | None = None
 
     @field_validator("findings", mode="before")
     @classmethod

roboco/services/gateway/choreographer/_impl.py

@@ -14,7 +14,7 @@ from __future__ import annotations
 import contextlib
 from dataclasses import dataclass
 from datetime import UTC, datetime
-from typing import Any, ClassVar, cast
+from typing import TYPE_CHECKING, Any, ClassVar, cast
 from uuid import UUID
 
 import structlog
@@ -56,6 +56,13 @@ from roboco.services.gateway.remediation import (
 
 logger = structlog.get_logger()
 
+if TYPE_CHECKING:
+    # The composed ``Choreographer`` resolves mixin helpers (``_project_slug_for``,
+    # etc.) via MRO, but ``_LegacyChoreographer`` itself does not inherit
+    # ``ChoreographerHelpers`` — so mypy can't see those names on ``self`` here.
+    # The cast below reaches the typed view the mixins use (``_Base`` pattern).
+    from roboco.services.gateway.choreographer._protocol import ChoreographerHelpers
+
 # Minimum character length enforced on rich_plan["approach"] by the PM
 # sub-tasks gate. Must match the Pydantic min_length on
 # IWillPlanRequest.approach. Raised 20→150: plans were vague
@@ -5671,6 +5678,78 @@ class Choreographer:
             )
         return None
 
+    async def _submit_root_unchanged_pr_guard(
+        self, t: Any, briefing: dict[str, Any]
+    ) -> Envelope | None:
+        """Refuse to re-submit a root PR whose diff is unchanged since the last fail.
+
+        The hard loop-stopper for the 2026-06-27 ``pr_fail`` re-submit loop. A
+        prior ``pr_fail`` stamped the assembled PR's head SHA into
+        ``notes_structured.pr_review.head_sha`` (see ``_capture_pr_head_sha`` /
+        ``_record_gate_verdict``). On the next ``submit_root`` this looks up the
+        PR's CURRENT head SHA and compares. Equal ⇒ no new cell work landed on
+        the root branch since the fail ⇒ the diff the reviewer would see is
+        byte-identical to the one just rejected ⇒ refuse, so no model can loop
+        itself back into ``awaiting_pr_review``. Different ⇒ the branch advanced
+        ⇒ allow. Returns ``None`` (proceed) on every ambiguous case so the gate
+        FAILS OPEN: no prior ``pr_fail`` verdict, no recorded ``head_sha`` (e.g.
+        a verdict written before this field existed), no ``pr_number``, no
+        resolvable project, or a git/closed-PR lookup that returns ``None``.
+        Only the exact-unchanged case is hard-blocked; the rest fall through to
+        the reviewer, who can still ``pr_fail`` if the diff is bad.
+        """
+        pr_review = (getattr(t, "notes_structured", None) or {}).get("pr_review") or {}
+        if pr_review.get("verdict") != "failed":
+            return None
+        recorded = pr_review.get("head_sha")
+        if not recorded:
+            return None
+        current = await self._current_root_pr_head_sha(t)
+        if current is None or current != recorded:
+            return None
+        return Envelope.invalid_state(
+            message=(
+                "the assembled root PR is unchanged since the last pr_fail"
+                f" (head {current[:7]}). No new cell work has landed on the"
+                " root branch, so re-submitting would re-open the exact diff"
+                " the reviewer just rejected and loop straight back to"
+                " awaiting_pr_review."
+            ),
+            remediate=(
+                "re-delegate the fixes to the owning cell PM(s) via"
+                " delegate(...) and wait for the cell subtasks to complete"
+                " and the root branch to be re-assembled. Do NOT call"
+                " submit_root again until new cell work has advanced the"
+                " root branch HEAD."
+            ),
+            context_briefing=briefing,
+        )
+
+    async def _current_root_pr_head_sha(self, t: Any) -> str | None:
+        """Best-effort current head SHA of the task's assembled PR (fail-open).
+
+        The lookup the unchanged-PR gate compares against. Returns ``None`` on
+        every ambiguous case (no ``pr_number``, no resolvable project slug, a
+        git error, or a closed/missing PR) so the gate fails open rather than
+        wedging the PM — only the exact-unchanged case is hard-blocked.
+        """
+        pr_number = getattr(t, "pr_number", None)
+        if not pr_number:
+            return None
+        try:
+            # ``_LegacyChoreographer`` doesn't inherit ``ChoreographerHelpers``,
+            # so cast to the typed view the mixins use to reach the shared
+            # ``_project_slug_for`` resolver (it delegates to the module-level
+            # ``resolve_task_project_slug``). Same resolver the pr_fail capture
+            # path uses, so one stub controls both gate paths.
+            slug = await cast("ChoreographerHelpers", self)._project_slug_for(t)
+            if not slug:
+                return None
+            sha = await self.git.get_pr_head_sha(slug, int(pr_number))
+            return sha if isinstance(sha, str) else None
+        except Exception:
+            return None
+
     async def submit_root(
         self, main_pm_agent_id: UUID, task_id: UUID, notes: str
     ) -> Envelope:
@@ -5725,6 +5804,21 @@ class Choreographer:
         # journal:decision + subtasks-terminal + branch-present. submit_root's
         # tracing requirements mirror submit_up's, so the shared guard applies.
         guard = await self._submit_up_guard(main_pm_agent_id, task_id, t, notes)
+        # Hard unchanged-PR gate (the 2026-06-27 pr_fail re-submit loop-stopper).
+        # A hint (the pr_fail a2a / next-hint steer "do NOT re-submit") is ignored
+        # by a weak coordinator (minimax-m3 re-submitted PR #139 byte-identical),
+        # so this refuses the re-submit STRUCTURALLY: if the last pr_fail stamped
+        # the assembled PR's head SHA and the current PR head SHA is the same, no
+        # new cell work has landed on the root branch since the fail — re-running
+        # submit_root would re-open / re-push the exact diff the reviewer just
+        # rejected, looping straight back to awaiting_pr_review → pr_fail. Force
+        # the PM to re-delegate the fixes and wait for re-assembly instead.
+        # Fail-open on ANY ambiguity (no prior fail, no recorded sha, no
+        # resolvable project, git error, closed/missing PR) — only the
+        # exact-unchanged case is hard-blocked; everything else proceeds and
+        # relies on the reviewer to re-fail if the diff is still bad.
+        if guard is None:
+            guard = await self._submit_root_unchanged_pr_guard(t, briefing)
         if guard is not None:
             guard.with_introspection(task=t, role=role_str)
             return await self._emit_rejection(

roboco/services/gateway/choreographer/pr_gate.py

@@ -244,8 +244,17 @@ class PRGateMixin(_Base):
         # it is persisted by the same commit (mirrors post_pr_review). This is
         # what keeps notes_structured.pr_review in lock-step with the decision —
         # a later pr_fail overwrites an earlier pr_pass verdict instead of
-        # leaving a stale "passed" on a task that was just sent back.
-        self._record_gate_verdict(t, verb, notes, issues=issues)
+        # leaving a stale "passed" on a task that was just sent back. On pr_fail
+        # also stamp the assembled PR's head SHA so the next submit_root can
+        # structurally refuse to re-submit the unchanged root (the 2026-06-27
+        # infinite pr_fail re-submit loop). Best-effort: a capture failure
+        # (no token, no PR, GitHub error) leaves head_sha absent and the
+        # submit_root gate fails open rather than wedging the PM.
+        if verb == "pr_fail":
+            head_sha = await self._capture_pr_head_sha(t)
+            self._record_gate_verdict(t, verb, notes, issues=issues, head_sha=head_sha)
+        else:
+            self._record_gate_verdict(t, verb, notes, issues=issues)
         runner = self._verb_runner()
         try:
             t = await runner.run_intent(verb, t, agent, spec_ctx)
@@ -346,7 +355,13 @@ class PRGateMixin(_Base):
         return None
 
     def _record_gate_verdict(
-        self, t: Any, verb: str, notes: str, issues: tuple[str, ...] = ()
+        self,
+        t: Any,
+        verb: str,
+        notes: str,
+        issues: tuple[str, ...] = (),
+        *,
+        head_sha: str | None = None,
     ) -> None:
         """Persist the gate verdict as the canonical ``pr_review`` note.
 
@@ -361,6 +376,12 @@ class PRGateMixin(_Base):
         PM's briefing that mirrors it — gets the concrete change-requests.
         Best-effort: content validation (e.g. a too-short summary) must never
         roll back the gate, so a malformed payload is logged and skipped.
+
+        On ``pr_fail`` the assembled PR's head SHA is stamped into the slot
+        (``head_sha``) so the next ``submit_root`` can structurally refuse to
+        re-submit the unchanged root — the 2026-06-27 infinite ``pr_fail``
+        re-submit loop. ``None`` (the default) leaves it absent, which the
+        ``submit_root`` gate treats as fail-open.
         """
         from roboco.foundation.policy.content import ContentValidationError
         from roboco.services.content_notes import apply_structured_note
@@ -387,6 +408,8 @@ class PRGateMixin(_Base):
         }
         if issues:
             payload["issues"] = list(issues)
+        if verb == "pr_fail" and head_sha:
+            payload["head_sha"] = head_sha
         try:
             apply_structured_note(t, "pr_review", payload)
         except ContentValidationError:
@@ -396,6 +419,40 @@ class PRGateMixin(_Base):
                 task_id=str(getattr(t, "id", "")),
             )
 
+    async def _capture_pr_head_sha(self, t: Any) -> str | None:
+        """Best-effort capture of the assembled PR's head SHA at ``pr_fail`` time.
+
+        Resolves the project slug via ``_project_slug_for`` (handles a
+        Main-PM root that carries only a ``product_id``) and asks
+        ``git.get_pr_head_sha``. Returns ``None`` on ANY failure (no PR number,
+        no resolvable project, git error) so the ``submit_root`` unchanged-PR
+        gate fails open rather than wedging the PM — only the exact-unchanged
+        case is hard-blocked.
+        """
+        pr_number = getattr(t, "pr_number", None)
+        if not pr_number:
+            return None
+        try:
+            slug = await self._project_slug_for(t)
+        except Exception:
+            logger.exception(
+                "pr_fail head-sha capture: slug resolve failed",
+                task_id=str(getattr(t, "id", "")),
+            )
+            return None
+        if not slug:
+            return None
+        try:
+            sha = await self.git.get_pr_head_sha(slug, int(pr_number))
+            return sha if isinstance(sha, str) else None
+        except Exception:
+            logger.exception(
+                "pr_fail head-sha capture: git lookup failed",
+                task_id=str(getattr(t, "id", "")),
+                pr=pr_number,
+            )
+            return None
+
     async def _post_gate_review_to_pr(
         self, t: Any, verb: str, reviewer_slug: str, notes: str
     ) -> None:

roboco/services/gateway/choreographer/pr_review.py

@@ -544,45 +544,57 @@ class PRReviewerMixin(_Base):
 
     async def _project_slug_for(self, t: Any) -> str | None:
         """Resolve the project slug for a task (read-only PR API calls + the
-        in-path gate's PR comment).
-
-        A normal task carries ``project_id``. A Main-PM coordination root — the
-        only task a root→master PR ever sits on — often carries just a
-        ``product_id`` (the cell→repo map) and no project of its own; its
-        ``feature/main_pm/{root}`` branch + PR live in the product's repo. Fall
-        through to the product's first distinct project (a monorepo product maps
-        every cell to one repo) so the gate verdict reaches the PR instead of
-        silently no-op'ing. Purely additive: a task WITH ``project_id`` resolves
-        exactly as before.
+        in-path gate's PR comment). Delegates to the module-level resolver so
+        the unchanged-PR gate in ``_impl.py`` reuses the EXACT same path (it
+        can't see this mixin method — ``_LegacyChoreographer`` does not inherit
+        ``ChoreographerHelpers``). See ``resolve_task_project_slug``.
         """
-        from uuid import UUID
+        return await resolve_task_project_slug(self.task.session, t)
 
-        from roboco.services.project import get_project_service
 
-        project_service = get_project_service(self.task.session)
-        if t.project_id is not None:
-            project = await project_service.get(t.project_id)
+async def resolve_task_project_slug(session: Any, t: Any) -> str | None:
+    """Resolve the project slug for a task (read-only PR API calls + the
+    in-path gate's PR comment). Module-level so it is shared by
+    ``PRReviewerMixin._project_slug_for`` and the unchanged-PR gate's
+    ``_current_root_pr_head_sha`` (in ``_impl.py``) — DRY, and the only way the
+    legacy choreographer class can reach the resolver without inheriting
+    ``ChoreographerHelpers``.
+
+    A normal task carries ``project_id``. A Main-PM coordination root — the
+    only task a root→master PR ever sits on — often carries just a
+    ``product_id`` (the cell→repo map) and no project of its own; its
+    ``feature/main_pm/{root}`` branch + PR live in the product's repo. Fall
+    through to the product's first distinct project (a monorepo product maps
+    every cell to one repo) so the gate verdict reaches the PR instead of
+    silently no-op'ing. Purely additive: a task WITH ``project_id`` resolves
+    exactly as before.
+    """
+    from uuid import UUID
+
+    from roboco.services.project import get_project_service
+
+    project_service = get_project_service(session)
+    if t.project_id is not None:
+        project = await project_service.get(t.project_id)
+        return project.slug if project is not None else None
+    product_id = getattr(t, "product_id", None)
+    if product_id is not None:
+        from roboco.services.product import get_product_service
+
+        product_service = get_product_service(session)
+        project_ids = await product_service.distinct_project_ids(UUID(str(product_id)))
+        if not project_ids:
+            return None
+        project = await project_service.get(project_ids[0])
+        return project.slug if project is not None else None
+    # Ad-hoc per-cell map root-subtask: mirror the product root's first-project
+    # resolution so the gate verdict reaches the PR in the mapped repo.
+    cell_map = getattr(t, "cell_projects", None) or []
+    seen: set[UUID] = set()
+    for mapping in sorted(cell_map, key=lambda m: m.team.value):
+        pid = UUID(str(mapping.project_id))
+        if pid not in seen:
+            seen.add(pid)
+            project = await project_service.get(pid)
             return project.slug if project is not None else None
-        product_id = getattr(t, "product_id", None)
-        if product_id is not None:
-            from roboco.services.product import get_product_service
-
-            product_service = get_product_service(self.task.session)
-            project_ids = await product_service.distinct_project_ids(
-                UUID(str(product_id))
-            )
-            if not project_ids:
-                return None
-            project = await project_service.get(project_ids[0])
-            return project.slug if project is not None else None
-        # Ad-hoc per-cell map root-subtask: mirror the product root's first-project
-        # resolution so the gate verdict reaches the PR in the mapped repo.
-        cell_map = getattr(t, "cell_projects", None) or []
-        seen: set[UUID] = set()
-        for mapping in sorted(cell_map, key=lambda m: m.team.value):
-            pid = UUID(str(mapping.project_id))
-            if pid not in seen:
-                seen.add(pid)
-                project = await project_service.get(pid)
-                return project.slug if project is not None else None
-        return None
+    return None

roboco/services/git.py

@@ -2345,6 +2345,61 @@ class GitService(BaseService):
             return ""
         return resp.text
 
+    async def get_pr_head_sha(self, project_slug: str, pr_number: int) -> str | None:
+        """Fetch a PR's current head commit SHA READ-ONLY via the GitHub API.
+
+        Used by the hard ``submit_root`` gate to detect that the assembled root
+        PR is byte-identical to the one a prior ``pr_fail`` already rejected —
+        i.e. no new cell work landed on the root branch since the fail — so the
+        gate can structurally refuse the re-submit instead of looping a weak
+        coordinator back into the same failed review (the 2026-06-27
+        ``pr_fail`` re-submit loop on PR #139). The head SHA equals the branch
+        HEAD at PR-open time, so two PRs opened from an unchanged branch share a
+        head SHA — the comparison holds whether or not a new PR number was cut.
+        Returns ``None`` on a missing token / unparseable remote / GitHub error
+        / a closed-or-missing PR so the gate FAILS OPEN rather than wedging the
+        PM (only the exact-unchanged case is hard-blocked; ambiguous cases pass
+        through and rely on the reviewer to re-fail).
+        """
+        project = await get_project_service(self.session).get_by_slug(project_slug)
+        if project is None or not project.git_url:
+            return None
+        try:
+            owner, repo = self._parse_git_url(project.git_url)
+        except GitError:
+            return None
+        git_token = await self._token_for_project(project_slug)
+        if not git_token:
+            return None
+        api_base = settings.github_api_base_url.rstrip("/")
+        try:
+            async with httpx.AsyncClient(timeout=_default_git_timeout()) as client:
+                resp = await client.get(
+                    f"{api_base}/repos/{owner}/{repo}/pulls/{pr_number}",
+                    headers={
+                        "Authorization": f"Bearer {git_token}",
+                        "Accept": "application/vnd.github+json",
+                        "X-GitHub-Api-Version": "2022-11-28",
+                    },
+                )
+            if not resp.is_success:
+                self.log.warning(
+                    "get_pr_head_sha non-2xx",
+                    project=project_slug,
+                    pr=pr_number,
+                    status=resp.status_code,
+                )
+                return None
+            return str(resp.json()["head"]["sha"])
+        except (httpx.HTTPError, ValueError, KeyError, TypeError) as e:
+            self.log.warning(
+                "get_pr_head_sha request/parse failed",
+                project=project_slug,
+                pr=pr_number,
+                error=str(e),
+            )
+            return None
+
     async def update_pr_for_task(
         self,
         task_id: UUID,

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