feat(content): PR reviewer writes its own slot, stops clobbering QA/Dev

roboco/foundation/policy/content/models.py

@@ -27,7 +27,6 @@ from pydantic import (
     ValidationError,
     ValidationInfo,
     field_validator,
-    model_validator,
 )
 
 from roboco.foundation.identity import CELL_TEAMS, Team
@@ -160,16 +159,6 @@ class PrReviewContent(_Content):
     def _nontrivial_summary(cls, v: str) -> str:
         return reject_trivial(v, field="summary", min_chars=_SUMMARY_MIN)
 
-    @model_validator(mode="after")
-    def _findings_required_for_negative(self) -> PrReviewContent:
-        if self.verdict in (Verdict.CHANGES_REQUESTED, Verdict.FAILED) and not (
-            self.findings
-        ):
-            raise ValueError(
-                "findings must be non-empty when verdict is changes_requested or failed"
-            )
-        return self
-
     def render_markdown(self) -> str:
         parts = [_section("Summary", self.summary)]
         if self.findings:

roboco/services/task.py

@@ -353,6 +353,7 @@ PR_REVIEW_SOURCES = ("external_pr", "internal_pr")
 # Approve-&-Starts it; the loop itself never starts/approves/merges it.
 SELF_HEAL_SOURCE = "self_heal"
 
+
 def extract_self_heal_fingerprint(task: Any) -> str | None:
     """The self-heal dedupe fingerprint from a task's markers, or None.
 
@@ -799,11 +800,7 @@ class TaskService(BaseService):
                 TaskTable.confirmed_by_human.is_(False),
             )
         )
-        return [
+        return [t for t in result.scalars().all() if not markers.is_dismissed(t)]
-            t
-            for t in result.scalars().all()
-            if not markers.is_dismissed(t)
-        ]
 
     async def list_external_pr_reviews(self) -> list[TaskTable]:
         """Live external-PR reviews for the panel: in-flight PLUS awaiting-decision.
@@ -830,11 +827,7 @@ class TaskService(BaseService):
                 ),
             )
         )
-        return [
+        return [t for t in result.scalars().all() if not markers.is_dismissed(t)]
-            t
-            for t in result.scalars().all()
-            if not markers.is_dismissed(t)
-        ]
 
     async def dismiss_external_pr_review(self, task_id: UUID) -> TaskTable | None:
         """CEO declines to act on a reviewed external PR — drop it from the queue.
@@ -905,8 +898,7 @@ class TaskService(BaseService):
             return None
         if task.status != TaskStatus.IN_PROGRESS:
             return None
-        if notes:
+        self._record_pr_review(task, summary=notes, verdict="changes_requested")
-            task.qa_notes = notes
         reviewer_id = to_python_uuid(task.claimed_by) or reviewer_agent_id
         task.assigned_to = None
         task.claimed_by = None
@@ -3642,6 +3634,35 @@ class TaskService(BaseService):
             return
         markers.set_documenter(task, task.assigned_to)
 
+    @staticmethod
+    def _record_pr_review(
+        task: TaskTable,
+        *,
+        summary: str | None,
+        verdict: str,
+        issues: list[str] | None = None,
+    ) -> None:
+        """Record a PR-reviewer verdict in the reviewer's OWN slot.
+
+        Routes through the content chokepoint (``pr_reviewer_notes`` mirror +
+        ``notes_structured["pr_review"]``) so a review never overwrites
+        ``qa_notes`` / ``dev_notes``. Structured per-line findings arrive via the
+        reviewer verb; until then the free-text summary + issues are captured.
+        Best-effort: a too-trivial review body must never block the transition.
+        """
+        body = (summary or "").strip()
+        if issues:
+            bullets = "\n".join(f"- {i}" for i in issues if i and i.strip())
+            body = f"{body}\n\n{bullets}".strip() if body else bullets
+        if not body:
+            return
+        try:
+            apply_structured_note(
+                task, "pr_review", {"summary": body, "verdict": verdict}
+            )
+        except ContentValidationError:
+            task.pr_reviewer_notes = _append_capped(task.pr_reviewer_notes, body)
+
     async def _resolve_pm_for_review(self, task: TaskTable) -> UUID | None:
         """Walk up the parent chain to find the PM who owns this work.
 
@@ -6704,8 +6725,7 @@ class TaskService(BaseService):
                 claimed_by=str(task.claimed_by),
             )
         captured = to_python_uuid(task.claimed_by)
-        if notes:
+        self._record_pr_review(task, summary=notes, verdict="passed")
-            task.qa_notes = _append_capped(task.qa_notes, "[PR REVIEW]\n" + notes)
         task.assigned_to = None
         task.claimed_by = None
         task.active_claimant_id = cast("Any", None)
@@ -6747,11 +6767,7 @@ class TaskService(BaseService):
                 claimed_by=str(task.claimed_by),
             )
         captured = to_python_uuid(task.claimed_by)
-        if issues:
+        self._record_pr_review(task, summary=notes, verdict="failed", issues=issues)
-            issue_block = "[PR REVIEW ISSUES]\n" + "\n".join(f"- {i}" for i in issues)
-            task.dev_notes = _append_capped(task.dev_notes, issue_block)
-        if notes:
-            task.qa_notes = _append_capped(task.qa_notes, "[PR REVIEW]\n" + notes)
         # Hand the failed assembled task to its PM to revise (cell PM for a cell
         # team, Main PM for the root); the revision dispatcher re-spawns whoever
         # owns a needs_revision task. Fall back to unassigned if no PM resolves.

tests/unit/foundation/policy/content/test_content_models.py

@@ -95,16 +95,19 @@ def test_pr_review_trivial_summary_rejected() -> None:
         validate_content("pr_review", {"summary": "wip", "verdict": "approved"})
 
 
-def test_pr_review_negative_verdict_requires_findings() -> None:
+def test_pr_review_negative_verdict_allows_summary_only() -> None:
-    with pytest.raises(ContentValidationError):
+    # A reviewer can fail on a summary alone (e.g. "CI is red"); findings are
-        validate_content(
+    # format-enforced (file/line/expected/actual) when present, not mandatory.
-            "pr_review",
+    c = validate_content(
-            {
+        "pr_review",
-                "summary": "Looks broken but no detail given here.",
+        {
-                "verdict": "failed",
+            "summary": "CI is red on this PR; the failing job blocks merge.",
-                "findings": [],
+            "verdict": "failed",
-            },
+            "findings": [],
-        )
+        },
+    )
+    assert isinstance(c, PrReviewContent)
+    assert c.verdict.value == "failed"
 
 
 def test_pr_review_approved_allows_empty_findings() -> None:

tests/unit/services/test_pr_review_db.py

@@ -247,12 +247,15 @@ async def test_pr_review_claim_and_complete(db_session: AsyncSession) -> None:
     # Re-claiming a non-pending task is a no-op.
     assert await svc.pr_review_claim(reviewer_id, task_id) is None
 
-    # Complete: in_progress -> completed, claim cleared, notes recorded.
+    # Complete: in_progress -> completed, claim cleared, review in its OWN slot.
     done = await svc.complete_review(reviewer_id, task_id, notes="Posted review.")
     await db_session.flush()
     assert done is not None
     assert done.status == TaskStatus.COMPLETED
-    assert done.qa_notes == "Posted review."
+    # Reviewer content lands in pr_reviewer_notes (not qa_notes) and is structured.
+    assert "Posted review." in (done.pr_reviewer_notes or "")
+    assert done.qa_notes is None
+    assert (done.notes_structured or {}).get("pr_review", {}).get("verdict")
     assert done.claimed_by is None
 
     # Re-completing a completed task is a no-op.