fix(pr-gate): land gate verdict on product-scoped PRs + persist it to notes

Two in-path PR-review-gate bugs surfaced reviewing the guard-core-app recovery
roots (PR #107 / root fead4372):

1. No verdict comment reached the PR. _project_slug_for returned None whenever
   project_id was None — but a Main-PM coordination root (the only task a
   root->master PR ever sits on) carries just a product_id (the cell->repo map),
   so _post_gate_review_to_pr resolved a None slug and silently no-op'd. It now
   falls through to the product's first distinct project (mirrors
   GitService._project_for_task), so the gate verdict actually lands on the PR.

2. The task's PR-reviewer notes contradicted the transition. pr_pass / pr_fail
   only threaded their notes through the tracing-gate shim and posted to GitHub;
   nothing wrote notes_structured.pr_review. A root passed once and later failed
   kept showing verdict=passed while the real transition was pr_fail. The gate
   now authors the canonical pr_review note on every decision (pr_pass -> passed,
   pr_fail -> failed), best-effort so a malformed note never rolls back the gate.

Adds unit tests for the product-slug fallback and the verdict persistence.

roboco/services/gateway/choreographer/pr_gate.py

@@ -240,6 +240,12 @@ class PRGateMixin(_Base):
             )
             if blocked is not None:
                 return blocked
+        # Author the canonical pr_review verdict note BEFORE the transition so
+        # 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)
         runner = self._verb_runner()
         try:
             t = await runner.run_intent(verb, t, agent, spec_ctx)
@@ -296,6 +302,34 @@ class PRGateMixin(_Base):
                 )
         return None
 
+    def _record_gate_verdict(self, t: Any, verb: str, notes: str) -> None:
+        """Persist the gate verdict as the canonical ``pr_review`` note.
+
+        The tracing gate only threads ``notes`` through a throwaway shim, so
+        nothing wrote the task's structured PR-reviewer slot — a task passed
+        once and later failed kept showing the stale ``verdict: passed``. This
+        authors the slot on every decision (``pr_pass`` → passed, ``pr_fail`` →
+        failed) so it can never contradict the transition. Best-effort: content
+        validation (e.g. a too-short summary) must never roll back the gate, so a
+        malformed payload is logged and skipped rather than raised.
+        """
+        from roboco.foundation.policy.content import ContentValidationError
+        from roboco.services.content_notes import apply_structured_note
+
+        verdict = "passed" if verb == "pr_pass" else "failed"
+        try:
+            apply_structured_note(
+                t,
+                "pr_review",
+                {"summary": notes, "findings": [], "verdict": verdict},
+            )
+        except ContentValidationError:
+            logger.warning(
+                "gate verdict note skipped (invalid content)",
+                verb=verb,
+                task_id=str(getattr(t, "id", "")),
+            )
+
     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

@@ -403,10 +403,34 @@ class PRReviewerMixin(_Base):
         )
 
     async def _project_slug_for(self, t: Any) -> str | None:
-        """Resolve the project slug for a task (for read-only PR API calls)."""
+        """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.
+        """
+        from uuid import UUID
+
         from roboco.services.project import get_project_service
 
-        if t.project_id is None:
+        project_service = get_project_service(self.task.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 None:
             return None
-        project = await get_project_service(self.task.session).get(t.project_id)
+        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

tests/unit/gateway/test_pr_gate_records_verdict.py

@@ -0,0 +1,80 @@
+"""The in-path gate persists its verdict to ``notes_structured.pr_review``.
+
+Without this, ``pr_pass`` / ``pr_fail`` only threaded their notes through the
+tracing-gate shim and posted to GitHub — they never wrote the task's structured
+PR-reviewer slot. So a task that was passed once and later failed kept showing
+the stale ``verdict: passed`` while the real transition was ``pr_fail`` →
+needs_revision (observed live on root fead4372 / PR #107). The gate now authors
+the canonical ``pr_review`` note on every decision so the slot can never
+contradict the transition.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+
+def _make_choreographer() -> Choreographer:
+    base: dict[str, Any] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    return Choreographer(ChoreographerDeps(**base))
+
+
+class _Task:
+    def __init__(self) -> None:
+        self.id = uuid4()
+        # A stale PASS verdict from an earlier pr_pass.
+        self.notes_structured: dict[str, Any] | None = {
+            "pr_review": {
+                "summary": "All seven acceptance criteria are satisfied.",
+                "findings": [],
+                "verdict": "passed",
+            }
+        }
+        self.pr_reviewer_notes = "stale"
+
+
+def test_pr_fail_overwrites_stale_passed_verdict() -> None:
+    c = _make_choreographer()
+    t = _Task()
+    c._record_gate_verdict(
+        t, "pr_fail", "Issues:\n- docs contradict AC1\n- README states wrong count"
+    )
+    assert t.notes_structured is not None
+    assert t.notes_structured["pr_review"]["verdict"] == "failed"
+    assert "docs contradict AC1" in t.notes_structured["pr_review"]["summary"]
+    # The derived TEXT mirror is regenerated too.
+    assert "failed" in t.pr_reviewer_notes.lower()
+
+
+def test_pr_pass_records_passed_verdict() -> None:
+    c = _make_choreographer()
+    t = _Task()
+    t.notes_structured = None
+    c._record_gate_verdict(
+        t, "pr_pass", "Assembled root scope is clean; every criterion is covered."
+    )
+    assert t.notes_structured is not None
+    assert t.notes_structured["pr_review"]["verdict"] == "passed"
+
+
+def test_record_gate_verdict_swallows_invalid_note() -> None:
+    """A too-short summary would fail content validation — it must never raise
+    (the transition already committed); the slot is just left untouched."""
+    c = _make_choreographer()
+    t = _Task()
+    c._record_gate_verdict(t, "pr_fail", "x")  # below the summary minimum
+    # No exception, and the stale slot is left as-is rather than corrupted.
+    assert t.notes_structured is not None
+    assert t.notes_structured["pr_review"]["verdict"] == "passed"

tests/unit/gateway/test_project_slug_for.py

@@ -0,0 +1,73 @@
+"""``_project_slug_for`` resolves a product-only coordination root's repo.
+
+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. Before the fallback, the in-path
+gate's PR-comment post and the external reviewer's diff fetch both resolved a
+``None`` slug and silently no-op'd — so the gate verdict never reached the PR
+(observed live: PR #107 / root fead4372 got pr_fail'd with no PR comment).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+
+def _make_choreographer() -> Choreographer:
+    task = AsyncMock()
+    task.session = MagicMock()
+    base: dict[str, Any] = {
+        "task": task,
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    return Choreographer(ChoreographerDeps(**base))
+
+
+@pytest.mark.asyncio
+async def test_slug_via_project_id_unchanged() -> None:
+    c = _make_choreographer()
+    proj = MagicMock(slug="be-proj")
+    with patch("roboco.services.project.get_project_service") as gps:
+        gps.return_value.get = AsyncMock(return_value=proj)
+        slug = await c._project_slug_for(MagicMock(project_id=uuid4(), product_id=None))
+    assert slug == "be-proj"
+
+
+@pytest.mark.asyncio
+async def test_slug_falls_back_to_product_when_no_project_id() -> None:
+    c = _make_choreographer()
+    proj = MagicMock(slug="gca-backend")
+    with (
+        patch("roboco.services.project.get_project_service") as gps,
+        patch("roboco.services.product.get_product_service") as gprod,
+    ):
+        gps.return_value.get = AsyncMock(return_value=proj)
+        gprod.return_value.distinct_project_ids = AsyncMock(return_value=[uuid4()])
+        slug = await c._project_slug_for(MagicMock(project_id=None, product_id=uuid4()))
+    assert slug == "gca-backend"
+
+
+@pytest.mark.asyncio
+async def test_slug_none_when_no_project_and_no_product() -> None:
+    c = _make_choreographer()
+    slug = await c._project_slug_for(MagicMock(project_id=None, product_id=None))
+    assert slug is None
+
+
+@pytest.mark.asyncio
+async def test_slug_none_when_product_has_no_projects() -> None:
+    c = _make_choreographer()
+    with patch("roboco.services.product.get_product_service") as gprod:
+        gprod.return_value.distinct_project_ids = AsyncMock(return_value=[])
+        slug = await c._project_slug_for(MagicMock(project_id=None, product_id=uuid4()))
+    assert slug is None