[F007] choreographer: cell-level unchanged-PR re-submit loop-stopper for submit_up

The root loop-stopper (F016) was root-only; a weak cell PM could re-submit
the unchanged cell->root PR after a pr_fail and loop awaiting_pr_review ->
pr_fail forever (the cell analogue of the 2026-06-27 root loop).

pr_fail stamps the assembled PR's head SHA into notes_structured.pr_review
.head_sha for cell AND root gate tasks alike (the capture is gate-verb-
level, not root-level), so the same structural refusal applies to submit_up:
if the cell PR's current head SHA equals the SHA the last pr_fail recorded,
no new dev work landed on the cell branch -> the diff is byte-identical ->
refuse, do not re-open the gate. Different SHA -> branch advanced -> allow.

- _submit_up_unchanged_pr_guard mirrors _submit_root_unchanged_pr_guard
  (cell-PM remediation: re-delegate to the dev + wait for re-assembly),
  wired into submit_up after _submit_up_guard passes.
- Renamed shared _current_root_pr_head_sha -> _current_pr_head_sha (both
  guards use it; the lookup was never root-specific).
- Every ambiguous case FAILS OPEN (no prior fail, no recorded sha, no
  pr_number, no resolvable project, git/closed-PR None) — only the exact-
  unchanged case is hard-blocked.

TDD red->green; ruff + mypy clean; F007+F016 guard suites green (15 passed).

roboco/services/gateway/choreographer/_impl.py

@@ -5237,6 +5237,12 @@ class Choreographer:
         # + subtasks-terminal + branch-present. None of these are modelled by
         # the spec yet — keep them in the verb body.
         guard = await self._submit_up_guard(pm_agent_id, task_id, t, notes)
+        # F007: the cell-level unchanged-PR loop-stopper. Only consult it once
+        # the state guard above has passed (ownership/tracing/branch all OK) —
+        # mirroring submit_root, a prior preflight reject short-circuits before
+        # the head-sha comparison runs.
+        if guard is None:
+            guard = await self._submit_up_unchanged_pr_guard(t, briefing)
         if guard is not None:
             guard.with_introspection(task=t, role=role_str)
             return await self._emit_rejection(
@@ -5980,7 +5986,7 @@ class Choreographer:
         recorded = pr_review.get("head_sha")
         if not recorded:
             return None
-        current = await self._current_root_pr_head_sha(t)
+        current = await self._current_pr_head_sha(t)
         if current is None or current != recorded:
             return None
         return Envelope.invalid_state(
@@ -6001,13 +6007,16 @@ class Choreographer:
             context_briefing=briefing,
         )
 
-    async def _current_root_pr_head_sha(self, t: Any) -> str | None:
+    async def _current_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
+        The lookup both unchanged-PR gates (submit_root F016 + submit_up F007)
-        every ambiguous case (no ``pr_number``, no resolvable project slug, a
+        compare against — ``pr_fail`` stamps the head SHA for cell AND root
-        git error, or a closed/missing PR) so the gate fails open rather than
+        gate tasks alike (the capture is gate-verb-level, not root-level), so
-        wedging the PM — only the exact-unchanged case is hard-blocked.
+        one resolver serves both. 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:
@@ -6026,6 +6035,48 @@ class Choreographer:
         except Exception:
             return None
 
+    async def _submit_up_unchanged_pr_guard(
+        self, t: Any, briefing: dict[str, Any]
+    ) -> Envelope | None:
+        """F007 — the cell-level analogue of ``_submit_root_unchanged_pr_guard``.
+
+        The root loop-stopper was root-only; a weak cell PM could re-submit the
+        unchanged cell→root PR after a ``pr_fail`` and loop
+        ``awaiting_pr_review`` → ``pr_fail`` forever. ``pr_fail`` stamps the
+        assembled PR's head SHA into ``notes_structured.pr_review.head_sha``
+        for cell gate tasks too (the capture is gate-verb-level), so the same
+        structural refusal applies: if the cell PR's current head SHA equals
+        the SHA the last ``pr_fail`` recorded, no new dev work landed on the
+        cell branch ⇒ the diff is byte-identical ⇒ refuse. Every ambiguous case
+        FAILS OPEN (shared ``_current_pr_head_sha``) — only the exact-unchanged
+        case is hard-blocked; the rest fall through to the cell reviewer.
+        """
+        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_pr_head_sha(t)
+        if current is None or current != recorded:
+            return None
+        return Envelope.invalid_state(
+            message=(
+                "the assembled cell PR is unchanged since the last pr_fail"
+                f" (head {current[:7]}). No new dev work has landed on the"
+                " cell 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 developer(s) via"
+                " delegate(...) and wait for the dev subtasks to complete and"
+                " the cell branch to be re-assembled. Do NOT call submit_up"
+                " again until new dev work has advanced the cell branch HEAD."
+            ),
+            context_briefing=briefing,
+        )
+
     async def submit_root(
         self, main_pm_agent_id: UUID, task_id: UUID, notes: str
     ) -> Envelope:

tests/unit/gateway/test_submit_up_unchanged_pr_guard.py

@@ -0,0 +1,144 @@
+"""F007 — the unchanged-PR re-submit loop-stopper is root-only; ``submit_up``
+(cell→root) had no head_sha guard, so a weak cell PM could re-submit the
+unchanged cell PR and loop ``awaiting_pr_review`` → ``pr_fail`` forever
+(the cell-level analogue of the 2026-06-27 root loop F016 closes).
+
+``pr_fail`` stamps the assembled PR's head SHA into
+``notes_structured.pr_review.head_sha`` for BOTH cell and root gate tasks
+(``pr_gate._capture_pr_head_sha`` / ``_record_gate_verdict`` are
+gate-verb-level, not root-level). So the same structural refusal applies
+to ``submit_up``: if the cell PR's current head SHA equals the SHA the
+last ``pr_fail`` recorded, no new dev work landed on the cell branch ⇒
+the diff is byte-identical ⇒ refuse, do not re-open the gate. Every
+ambiguous case FAILS OPEN, identical to the root guard (shared
+``_current_pr_head_sha``).
+"""
+
+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
+
+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_cell(
+    *,
+    notes_structured: dict[str, Any] | None,
+    pr_number: int | None = 132,
+) -> tuple[Choreographer, Any, Any]:
+    """A cell-PM task re-submitted from ``in_progress`` after a ``pr_fail``.
+
+    Mirrors a live cell re-submit: the cell task is back in ``in_progress``
+    (re-claimed out of ``needs_revision``), carries the prior ``pr_fail``
+    verdict in ``notes_structured.pr_review``, and the cell→root PR is still
+    open. The ``_submit_up_guard`` preflight is satisfied (owned by the cell
+    PM, journal decision, subtasks terminal, branch present, notes long
+    enough) so the unchanged-PR gate is the thing under test.
+    """
+    cell_pm_id = uuid4()
+    cell_task_id = uuid4()
+    in_prog = MagicMock(
+        id=cell_task_id,
+        status="in_progress",
+        assigned_to=cell_pm_id,
+        pr_number=pr_number,
+        branch_name="feature/backend/cell-task",
+        parent_task_id=uuid4(),
+        batch_id=None,
+        team="backend",
+        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="cell_pm", team="backend")
+    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")
+    return c, cell_pm_id, cell_task_id
+
+
+@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."""
+    c, cell_pm_id, cell_task_id = _resubmit_cell(
+        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_up(
+        cell_pm_id, cell_task_id, notes="re-submitting the cell after the fix"
+    )
+
+    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()
+
+
+@pytest.mark.asyncio
+async def test_submit_up_allows_after_cell_branch_advanced() -> None:
+    """A different current head SHA ⇒ dev work landed on the cell branch ⇒ the
+    diff changed ⇒ allow the re-submit into the gate."""
+    c, cell_pm_id, cell_task_id = _resubmit_cell(
+        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_up(
+        cell_pm_id, cell_task_id, notes="re-submitting after the dev re-assembly"
+    )
+
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"
+    c.task.submit_for_review.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_submit_up_fail_open_when_no_prior_pr_fail_verdict() -> None:
+    """No pr_review (first submit) ⇒ nothing to compare ⇒ allow."""
+    c, cell_pm_id, cell_task_id = _resubmit_cell(notes_structured=None)
+    env = await c.submit_up(
+        cell_pm_id, cell_task_id, notes="first cell submit; nothing to compare yet"
+    )
+    assert env.error is None, env.as_dict()
+    assert env.status == "awaiting_pr_review"