From e202ce397d69fd4bd0869d63e8ee2f47c1c878f1 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sat, 27 Jun 2026 19:52:01 +0200 Subject: [PATCH] Fix: Make main_pm + task_type=code impossible --- agents/prompts/roles/prompter.md | 6 + roboco/foundation/policy/batch.py | 26 ++ roboco/foundation/policy/content/models.py | 18 +- roboco/foundation/policy/lifecycle.py | 31 +- .../services/gateway/choreographer/_impl.py | 15 +- .../services/gateway/choreographer/pr_gate.py | 86 +++- .../gateway/choreographer/pr_review.py | 102 ++++- roboco/services/prompter.py | 27 +- roboco/services/task.py | 168 ++++++++ tests/foundation/test_lifecycle_spec.py | 28 ++ .../policy/content/test_content_models.py | 43 ++ tests/unit/foundation/policy/test_batch.py | 26 ++ .../unit/gateway/test_pr_gate_notifies_pm.py | 232 ++++++++++ .../gateway/test_pr_gate_records_verdict.py | 65 +++ .../test_pr_review_hand_format_guard.py | 183 ++++++++ .../unit/services/test_main_pm_code_guard.py | 406 ++++++++++++++++++ tests/unit/services/test_prompter.py | 10 + 17 files changed, 1439 insertions(+), 33 deletions(-) create mode 100644 tests/unit/gateway/test_pr_gate_notifies_pm.py create mode 100644 tests/unit/gateway/test_pr_review_hand_format_guard.py create mode 100644 tests/unit/services/test_main_pm_code_guard.py diff --git a/agents/prompts/roles/prompter.md b/agents/prompts/roles/prompter.md index f360be60..4a223ba1 100644 --- a/agents/prompts/roles/prompter.md +++ b/agents/prompts/roles/prompter.md @@ -89,6 +89,12 @@ When you are scoped to a **MegaTask**, the CEO wants several distinct tasks work Over-declaring a surface is safe (the worst case is a task waits a little); under-declaring is not. You do **not** compute the order yourself — declare each surface honestly and the analyzer derives the waves. Present all the tasks in prose first (a short paragraph each), then call `propose_batch` once. If the conversation changes the set, call it again with the full updated batch. +**Each draft in a MegaTask becomes a Main-PM coordination root-subtask** — the Main PM coordinates it and delegates the actual code to the cells; the Main PM never writes the code itself. So draft each root-subtask as the coordination it is, not as code the Main PM will implement: + +- `task_type`: `"planning"` (the system coerces `code`→`planning` for a Main-PM root anyway, but draft it correctly — a Main PM task is never `code`). +- `acceptance_criteria`: **coordination-level**, not code-level. Write criteria the Main PM can satisfy by delegating and assembling — e.g. *"the chart-first Metrics refactor is delegated to fe-pm and lands on a cell PR"*, *"the cell→root PR is assembled and passes the in-path review gate"*, *"all cell subtasks are terminal and the root→master PR is merged"*. Do **not** write code-level criteria on the root — specific file paths (`frontend/src/components/timeseries-chart.tsx`), "lint/build clean", exact APIs — those belong on the **cell/dev subtasks** the Main PM delegates to, not on the root. A root carrying code-level ACs is the structural mismatch behind the 2026-06-27 meltdown (the gate reviewed code the Main PM couldn't fix → an infinite re-submit loop). +- `the_work` still names the per-cell breakdown (which cell does what) — that's the delegation plan the Main PM executes; it's correct here because it *is* the coordination spec. + ## What happens after you call `propose_draft` A draft card appears for the human with three choices: **Keep chatting**, **Board review & Start**, or **Approve & Start**. **Choosing is the human's action, not yours** — you cannot create, start, or route the task. If they pick **Board review & Start**, it becomes a pending task owned by the Board (Product Owner + Head of Marketing) to review first; if they pick **Approve & Start**, it becomes a pending task that goes straight to the Main PM to delegate to the cells. Either way, your job ends the moment you call `propose_draft`. Do not say you'll "kick it off", "send it to the PM chain", or route it anywhere — you have no such ability, and which path it takes is the human's choice on the card. diff --git a/roboco/foundation/policy/batch.py b/roboco/foundation/policy/batch.py index c898bba2..00adc30e 100644 --- a/roboco/foundation/policy/batch.py +++ b/roboco/foundation/policy/batch.py @@ -15,6 +15,8 @@ between sites. Inputs are typed ``object | None`` because callers pass either OR from __future__ import annotations +from roboco.models.base import TaskType, Team + def is_batch_umbrella( *, batch_id: object | None, parent_task_id: object | None @@ -97,3 +99,27 @@ def is_valid_batch_shape( return targets == 0 # root-subtask: exactly one target return targets == 1 + + +def main_pm_cannot_own_code(*, team: object | None, task_type: object | None) -> bool: + """True when a Main-PM-owned task must NOT be ``code``. + + A Main PM *coordinates* — it plans and delegates execution to the cells; it + has no verb to write code. A ``task_type=code`` task owned by ``main_pm`` is + the structural mismatch behind the 2026-06-27 MegaTask meltdown: the git/PR/ + review layer treated the Main-PM root-subtask as a code root (branch + PR + + ``submit_root`` + the ``pr_review`` gate) while the ownership/dispatch layer + treated it as coordination, and a code-style ``pr_fail`` landed on a + coordinator with no code verb → an infinite re-submit loop. The root's code + ACs were never decomposed to children because the roll-up gate is inert + unless a child declares ``parent_ac_refs``. + + This predicate is the single invariant every layer consults so ``main_pm`` + + ``code`` can never coexist: the intake coercion (``create_task_from_draft`` + retypes ``code``→``planning``), the ``TaskService.create`` backstop reject, + the reassign/escalation diversion, and the claim guard. Accepts either ORM + enum members or their ``.value`` strings (callers pass both shapes). + """ + team_value = str(getattr(team, "value", team)) + type_value = str(getattr(task_type, "value", task_type)) + return team_value == Team.MAIN_PM.value and type_value == TaskType.CODE.value diff --git a/roboco/foundation/policy/content/models.py b/roboco/foundation/policy/content/models.py index bbbb1894..93074e65 100644 --- a/roboco/foundation/policy/content/models.py +++ b/roboco/foundation/policy/content/models.py @@ -143,10 +143,19 @@ class AcVerdict(_Base): class PrReviewContent(_Content): - """A PR-review comment / reviewer verdict.""" + """A PR-review comment / reviewer verdict. + + ``findings`` carries the *structured* review (file/line/severity/expected/ + actual) used by the inbound external-PR path. The in-path gate instead + fails on free-text ``issues`` — those land in the additive ``issues`` slot + rather than being flattened into ``summary`` alone, so a reader of + ``notes_structured.pr_review`` (or the derived ``pr_reviewer_notes`` mirror) + gets the concrete change-requests either way. + """ summary: str findings: list[Finding] = Field(default_factory=list) + issues: list[str] = Field(default_factory=list) verdict: Verdict @field_validator("findings", mode="before") @@ -154,6 +163,11 @@ class PrReviewContent(_Content): def _coerce_findings(cls, v: Any) -> Any: return coerce_to_list(v) + @field_validator("issues", mode="before") + @classmethod + def _coerce_issues(cls, v: Any) -> Any: + return coerce_to_list(v) + @field_validator("summary") @classmethod def _nontrivial_summary(cls, v: str) -> str: @@ -174,6 +188,8 @@ class PrReviewContent(_Content): f"| {f.expected} → {f.actual} |" ) parts.append("## Findings\n" + "\n".join(rows)) + if self.issues: + parts.append(_section("Issues", _bullets(self.issues))) parts.append(_section("Verdict", self.verdict.value.replace("_", " "))) return _join(parts) diff --git a/roboco/foundation/policy/lifecycle.py b/roboco/foundation/policy/lifecycle.py index 5c27f47a..0b6167ae 100644 --- a/roboco/foundation/policy/lifecycle.py +++ b/roboco/foundation/policy/lifecycle.py @@ -30,7 +30,7 @@ from typing import TYPE_CHECKING, Any, Literal # Re-exported here so callers can import `Role` from this module alongside # the lifecycle tables that depend on it. New consumers may also import # from `roboco.foundation.identity` directly. -from roboco.foundation.identity import Role +from roboco.foundation.identity import Role, Team if TYPE_CHECKING: from collections.abc import Callable @@ -779,6 +779,27 @@ def _next_hint_dev_revise(_t: Any) -> str: return "idle - dev will revise and re-submit" +def _next_hint_pr_fail(t: Any) -> str: + # Steer a pr_fail verdict by owner shape. A Main-PM branch-bearing root is + # an assembled cell→root / root→master PR — coordination, not the Main PM's + # own code. The rejection is about the cells' merged code, which the Main PM + # cannot fix directly (no code verb); it must re-delegate the fixes to the + # owning cell PM(s) and wait for re-assembly. Re-submitting the unchanged + # root is the 2026-06-27 infinite pr_fail loop. A cell/dev task is revised + # in place by its dev, so keep the dev-revise hint there. + team = getattr(t, "team", None) + team_value = str(getattr(team, "value", team)) + branch = bool(getattr(t, "branch_name", None)) + if team_value == Team.MAIN_PM.value and branch: + return ( + "assembled cell work failed review — re-delegate the fixes to the" + " owning cell PM(s) via delegate(...), then wait for the cell" + " subtasks to complete and the PR to be re-assembled; do NOT" + " re-submit the root until then" + ) + return "idle - dev will revise and re-submit" + + def _next_hint_doc_after_claim(_t: Any) -> str: return ( "write docs in your workspace, commit them, then call" @@ -1152,7 +1173,7 @@ _INTENT_VERBS: dict[str, IntentSpec] = { composes=("pr_fail",), extra_preconditions=(), side_effects=(), - next_hint=_next_hint_dev_revise, + next_hint=_next_hint_pr_fail, ), # Phase 3: documenter verbs "claim_doc_task": IntentSpec( @@ -1244,8 +1265,10 @@ _INTENT_VERBS: dict[str, IntentSpec] = { "Main PM opens the root→master PR and moves the root task to" " awaiting_pr_review for the main reviewer (the root analogue of the" " cell PM's submit_up). After pr_pass, call complete to escalate to" - " the CEO. Only for code roots; branchless coordination roots skip" - " the gate and complete directly." + " the CEO. For branch-bearing roots (a Main-PM root-subtask assembles" + " the cells' merged work); branchless coordination roots skip the" + " gate and complete directly. The gate is branch-keyed, not" + " task_type-keyed — a Main-PM root is planning-typed, never code." ), composes=("submit_for_review",), extra_preconditions=(), diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index 86748024..26c16d7a 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -5780,12 +5780,15 @@ class Choreographer: main_pm_agent_id, root_task_id ), ) - # A code root must pass the in-path PR-review gate first: submit_root - # opens the root→master PR and moves it in_progress → awaiting_pr_review, - # then the main reviewer pr_passes it to awaiting_pm_review. So complete - # accepts only awaiting_pm_review for a code root. A branchless - # coordination root (product fan-out, no repo/PR) skips the gate, so it - # may still be walked from in_progress here. + # A branch-bearing root must pass the in-path PR-review gate first: + # submit_root opens the root→master PR and moves it in_progress → + # awaiting_pr_review, then the main reviewer pr_passes it to + # awaiting_pm_review. So complete accepts only awaiting_pm_review for a + # branch-bearing root (a Main-PM root-subtask is planning-typed, never + # code, but it still assembles the cells' merged work into a real PR). + # A branchless coordination root (product fan-out, no repo/PR) skips the + # gate, so it may still be walked from in_progress here. The split is + # branch-keyed, not task_type-keyed. root_is_branchless = not bool(t.branch_name) allowed_statuses = ( ("awaiting_pm_review", "in_progress") diff --git a/roboco/services/gateway/choreographer/pr_gate.py b/roboco/services/gateway/choreographer/pr_gate.py index 819094b0..58855b98 100644 --- a/roboco/services/gateway/choreographer/pr_gate.py +++ b/roboco/services/gateway/choreographer/pr_gate.py @@ -245,7 +245,7 @@ class PRGateMixin(_Base): # 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) + self._record_gate_verdict(t, verb, notes, issues=issues) runner = self._verb_runner() try: t = await runner.run_intent(verb, t, agent, spec_ctx) @@ -265,6 +265,49 @@ class PRGateMixin(_Base): # transition — a GitHub failure must not roll back the gate decision. reviewer_slug = getattr(agent, "slug", None) or role_str await self._post_gate_review_to_pr(t, verb, reviewer_slug, notes) + # Deliver the change-requests to the owner that now has to act on them + # — the cell PM the runner just re-assigned via _revision_pm_for_task. + # The reviewer posts the verdict on the PR itself but that never reaches + # any PM-readable channel (no a2a, and _briefing_for / build_task_handoff + # read neither pr_reviewer_notes nor notes_structured.pr_review). Without + # this the owning PM respawned into needs_revision saw a generic "needs + # revision" with zero concrete issues, concluded nothing to rework, and + # re-submitted the same PR — an infinite pr_fail loop (live on + # 9980d0a0 / PR #138). Mirrors QA's fail_review a2a to the dev (qa.py:671). + # Best-effort: the transition already committed, so a delivery failure + # must not roll the verdict back or 500 the reviewer. + if verb == "pr_fail" and t.assigned_to is not None: + # A Main-PM branch-bearing root is an assembled cell→root / root→master + # PR — coordination, not the Main PM's own code. The rejection is + # about the cells' merged code, which the Main PM cannot fix directly + # (no code verb). Steer the a2a body to re-delegate + wait for + # re-assembly so the PM doesn't re-submit the unchanged root (the + # 2026-06-27 infinite pr_fail loop). The Envelope ``next`` hint makes + # the same steer via _next_hint_pr_fail. + team = getattr(t, "team", None) + team_value = str(getattr(team, "value", team)) + is_main_pm_root = team_value == spec_module.Team.MAIN_PM.value and bool( + getattr(t, "branch_name", None) + ) + steer = ( + " Assembled cell work failed — re-delegate the fixes to the" + " owning cell PM(s) and wait for re-assembly; do NOT re-submit" + " the root." + if is_main_pm_root + else "" + ) + try: + await self.a2a.send( + from_agent=reviewer_agent_id, + to_agent=t.assigned_to, + skill="code_review", + task_id=task_id, + body=f"PR review needs changes. {notes}{steer}", + ) + except Exception: + logger.exception( + "pr_fail a2a to owning PM failed", task_id=str(task_id) + ) return Envelope.ok( status=str(t.status), task_id=str(task_id), @@ -302,27 +345,50 @@ class PRGateMixin(_Base): ) return None - def _record_gate_verdict(self, t: Any, verb: str, notes: str) -> None: + def _record_gate_verdict( + self, t: Any, verb: str, notes: str, issues: tuple[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. + failed) so it can never contradict the transition. For ``pr_fail`` the + free-text ``issues`` land in the structured ``issues`` slot (not the + format-enforced ``findings`` list, which needs file/severity/expected/ + actual) so a reader of ``notes_structured.pr_review`` — or the owning + 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. """ 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}, + if verb == "pr_fail" and issues: + # The free-text issues render under their own ``## Issues`` section + # (render_markdown). Baking them into ``summary`` too duplicated each + # issue on the Task Details "PR Reviewer Notes" card (once under + # ## Summary, once under ## Issues). The summary is a substantive + # non-issues sentence; ``notes`` (with the issues) still drives the + # GitHub PR post and the a2a to the owning PM — those are raw text, + # not rendered through render_markdown, so no duplication there. + summary = ( + f"In-path PR-review gate requested changes - " + f"{len(issues)} issue(s) listed below." ) + else: + summary = notes + payload: dict[str, Any] = { + "summary": summary, + "findings": [], + "verdict": verdict, + } + if issues: + payload["issues"] = list(issues) + try: + apply_structured_note(t, "pr_review", payload) except ContentValidationError: logger.warning( "gate verdict note skipped (invalid content)", diff --git a/roboco/services/gateway/choreographer/pr_review.py b/roboco/services/gateway/choreographer/pr_review.py index f98da999..a0220e96 100644 --- a/roboco/services/gateway/choreographer/pr_review.py +++ b/roboco/services/gateway/choreographer/pr_review.py @@ -160,6 +160,23 @@ class PRReviewerMixin(_Base): apply_structured_note(t, "pr_review", structured) return structured.render_markdown() + @staticmethod + def _is_hand_formatted_verdict(body: str) -> bool: + """True when a free-text ``body`` carries verdict/section markdown headers + the system would otherwise generate — i.e. the reviewer hand-formatted a + verdict into ``body`` instead of passing structured ``findings``. + + Matches the section headers the canonical renderer emits (``## Findings``) + plus the ones a hand-formatter reaches for (``## Summary`` / ``## Issues`` + / ``## Verdict``). A real one-paragraph summary does not contain ``## `` + headers, so the prose word "summary" never trips this. + """ + lowered = (body or "").lower() + return any( + header in lowered + for header in ("## summary", "## issues", "## verdict", "## findings") + ) + async def _post_review_side_effects( self, t: Any, @@ -219,11 +236,15 @@ class PRReviewerMixin(_Base): if isinstance(pre, Envelope): return pre agent, role_str, briefing, spec_ctx = pre - # Refuse a verdict that contradicts the findings BEFORE anything is - # recorded or posted to the contributor's PR (e.g. a forgotten - # event='APPROVE' that defaults to a blocking REQUEST_CHANGES with no - # findings cited). - conflict = await self._verdict_consistency_gate( + # Content gates BEFORE anything is recorded or posted to the + # contributor's PR: (1) refuse a verdict that contradicts the findings + # (a forgotten event='APPROVE' defaulting to a blocking REQUEST_CHANGES + # with no findings); (2) refuse a hand-formatted verdict body with no + # findings (the tool contract is "body = a one-paragraph summary; the + # system GENERATES the comment from structured findings — do not + # hand-format"). Folded into one helper so neither slips through and the + # verb body stays under the return-count lint ceiling. + rejection = await self._post_pr_review_content_gates( t, reviewer_agent_id, task_id, @@ -231,9 +252,10 @@ class PRReviewerMixin(_Base): briefing, event=event, findings=findings, + body=body, ) - if conflict is not None: - return conflict + if rejection is not None: + return rejection slug = await self._project_slug_for(t) pr_number = t.pr_number post_body = self._resolve_post_body(t, body, findings, event) @@ -351,6 +373,72 @@ class PRReviewerMixin(_Base): verb="post_pr_review", ) + async def _post_pr_review_content_gates( + self, + t: Any, + reviewer_agent_id: UUID, + task_id: UUID, + role_str: str, + briefing: dict[str, Any], + *, + event: str, + findings: list[dict[str, Any]] | None, + body: str, + ) -> Envelope | None: + """Pre-side-effect content gates for ``post_pr_review``: verdict + consistency, then the no-hand-formatted-body guard. Returns the first + rejection ``Envelope`` or ``None`` to proceed. + + The hand-format guard: the tool contract is "``body`` = a one-paragraph + summary; the system GENERATES the GitHub comment from structured + findings — do not hand-format it in ``body``". Nothing enforced that, so + a reviewer could pass ``findings=[]`` and dump a self-formatted + ``## Summary`` / ``## Issues`` / ``## Verdict`` blob into ``body``, + which ``_resolve_post_body`` posts verbatim (the renderer emits + ``## Findings``, never ``## Issues`` — so a ``## Issues`` section on the + PR is proof the body was hand-formatted). Observed live: a duplicated, + self-redundant hand-formatted verdict posted to a contributor's PR. + Refuse it and point the reviewer at the structured path. Scoped to empty + findings: with structured findings the system generates the comment, so + a header-shaped word in the summary is harmless; a genuine plain-note + ``COMMENT`` (no verdict headers) is still allowed. + """ + conflict = await self._verdict_consistency_gate( + t, + reviewer_agent_id, + task_id, + role_str, + briefing, + event=event, + findings=findings, + ) + if conflict is not None: + return conflict + if not findings and self._is_hand_formatted_verdict(body): + return await self._emit_rejection( + Envelope.invalid_state( + message=( + "post_pr_review body is hand-formatted as a verdict — " + "pass structured findings instead" + ), + remediate=( + "do not hand-format the review. Pass a one-paragraph " + "summary in `body` plus structured " + "`findings=[{file, line?, severity " + "(blocker|major|minor|nit), expected, actual}, ...]`; the " + "system generates the GitHub comment (summary + findings " + "table + verdict). event='REQUEST_CHANGES' requires >=1 " + "finding; a bare event='COMMENT' with no findings is for a " + "plain note, not a verdict" + ), + context_briefing=briefing, + ).with_introspection(task=t, role=role_str), + agent_id=reviewer_agent_id, + task_id=task_id, + verb="post_pr_review", + ) + return None + async def _resolve_role( self, t: Any, diff --git a/roboco/services/prompter.py b/roboco/services/prompter.py index 2d8d5ff0..d234b3e9 100644 --- a/roboco/services/prompter.py +++ b/roboco/services/prompter.py @@ -21,7 +21,7 @@ from sqlalchemy import select from roboco.db.tables import AgentTable, TaskTable from roboco.foundation.identity import CELL_TEAMS -from roboco.foundation.policy.batch import is_batch_umbrella +from roboco.foundation.policy.batch import is_batch_umbrella, main_pm_cannot_own_code from roboco.foundation.policy.content.validators import coerce_str_list from roboco.foundation.policy.sequencing.models import DraftSurface, SequencePlan from roboco.models.base import ( @@ -304,6 +304,23 @@ class PrompterService: default_lead=_lead, ) + # A Main PM coordinates — it never owns a code task. A main_pm + code + # draft is the structural mismatch behind the 2026-06-27 MegaTask + # meltdown (the git/PR/review layer treated the root as code while the + # ownership layer treated it as coordination → pr_fail loop). Intake + # coerces code -> planning here so the combo can never persist; a + # root-subtask / umbrella / single-task main_pm route is a coordination + # root whose code ACs live on the delegated cell/dev leaves. The + # TaskService.create backstop rejects main_pm + code for non-intake + # create paths (the HTTP route). + if main_pm_cannot_own_code(team=team, task_type=task_type): + self.log.info( + "Main-PM intake task coerced code->planning", + team=str(getattr(team, "value", team)), + title=_text(draft_data.get("title")) or "", + ) + task_type = TaskType.PLANNING + req = TaskCreateRequest( title=draft_data["title"], description=draft_data["description"], @@ -990,9 +1007,9 @@ def _compose_umbrella_draft( The umbrella targets neither project nor product (it is branchless) and carries no collision surface of its own — it exists to group the batch, hold the wave plan in its description for review, and be the one board-review / - CEO-approve / Main-PM-coordinate unit. ``task_type=code`` mirrors the existing - product coordination roots created from intake (the branch/PR exemption comes - from the batch identity, not the type). + CEO-approve / Main-PM-coordinate unit. It is a Main-PM coordination root, so + ``task_type=planning`` (a Main PM coordinates, it does not execute code); the + branch/PR exemption comes from the batch identity, not the type. """ item_titles = [ _text(d.get("title")) or f"Task {i + 1}" for i, d in enumerate(drafts) @@ -1014,7 +1031,7 @@ def _compose_umbrella_draft( "acceptance_criteria": [ "Every root-subtask in the MegaTask is completed and merged.", ], - "task_type": TaskType.CODE.value, + "task_type": TaskType.PLANNING.value, "nature": TaskNature.TECHNICAL.value, "estimated_complexity": Complexity.HIGH.value, "priority": 1, diff --git a/roboco/services/task.py b/roboco/services/task.py index 53e771e2..13934b81 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -39,6 +39,7 @@ from roboco.foundation.policy.batch import ( is_batch_umbrella, is_branchless_coordination, is_valid_batch_shape, + main_pm_cannot_own_code, ) from roboco.foundation.policy.content import markers from roboco.foundation.policy.content.validators import ContentValidationError @@ -195,6 +196,20 @@ def _board_cannot_own(task: TaskTable) -> bool: ) +def _task_type_is_code(task_type: Any) -> bool: + """True when ``task_type`` is ``TaskType.CODE`` (enum member or its value). + + Robust to the two shapes SQLAlchemy hands back: the ``TaskType`` enum or + its raw ``"code"`` string (the latter on detached/partially-hydrated rows). + Used by the Main-PM claim guard, which keys on the task's *type* (a Main PM + cannot execute code) rather than the team+type combo the create / reassign + guards use — claiming is owning, and a Main PM claiming a code task is a + mismatch regardless of the task's team. + """ + value = task_type.value if isinstance(task_type, TaskType) else task_type + return str(value) == TaskType.CODE.value + + _PM_OWNED_CELL_TASK_TYPES: frozenset[str] = frozenset( { TaskType.PLANNING.value, @@ -823,6 +838,23 @@ class TaskService(BaseService): if req.parent_task_id: await self._validate_parent_depth(req.parent_task_id) + # Impossibility backstop: a Main PM coordinates — it never owns a code + # task. ``main_pm`` + ``code`` on the same task is the structural + # mismatch behind the 2026-06-27 MegaTask meltdown (a root-subtask the + # git/PR/review layer treated as code while ownership treated it as + # coordination — never reconciled, pr_fail looped). Intake + # (create_task_from_draft) coerces code→planning, so this fires only on + # a non-intake create (the HTTP route / a direct internal create) that + # tries to persist the forbidden combo. + if main_pm_cannot_own_code(team=req.team, task_type=req.task_type): + raise ValidationError( + "MAIN_PM_NO_CODE: A Main PM task coordinates — it does not" + " execute code. Re-draft as `planning` with coordination-level" + " acceptance criteria, or target a cell so a developer owns the" + " code.", + field="task_type", + ) + # Stable per-criterion ids (1:1 with acceptance_criteria) so children can # reference specific parent criteria; generated here when not supplied. ac_ids = req.acceptance_criteria_ids or [ @@ -4728,6 +4760,24 @@ class TaskService(BaseService): reason=reason, ) return + # Impossibility backstop: a Main-PM target must never receive (back) a + # main_pm + code task — a coordinator with no code verb cannot fix the + # code, so escalating it to Main PM perpetuates the mismatch (the + # 2026-06-27 meltdown shape). Scoped to the team+type combo (NOT a broad + # code+main-pm-target rule) so a legacy main_pm+code task can still be + # escalated to a cell dev — the correct remediation. The combo is + # uncreatable going forward (create backstop + intake coercion), so this + # is a backstop for legacy / direct-ORM-write tasks. + if main_pm_cannot_own_code( + team=task.team, task_type=task.task_type + ) and await self._is_main_pm_agent(target_agent_id): + await self._release_code_task_to_pool( + task=task, + escalator_slug=escalator_slug, + blocked_target_slug=target_slug, + reason=reason, + ) + return if task.assigned_to and not task.blocker_raised_by: task.blocker_raised_by = cast("Any", task.assigned_to) # Capture before mutating: the audit row must record the real prior @@ -4983,6 +5033,20 @@ class TaskService(BaseService): # off the board — reflect the new owner. Team.MAIN_PM is a valid non-cell # team and does not affect dispatch (which routes by assignee, not team). task.team = cast("Any", Team.MAIN_PM) + # A board-approved task handed to Main PM must not stay `code` — a Main + # PM coordinates, it never owns a code task (the 2026-06-27 meltdown was + # a main_pm + code root). A board-routed PROJECT code task reaches here + # with team=cell and task_type=code (intake only coerces main_pm-team + # drafts); once team is flipped to MAIN_PM the combo would persist. + # Retype code→planning so it's a planning-typed coordination root the + # Main PM delegates to the cells. (Intake coerces this too; this is the + # board-review backstop.) + if main_pm_cannot_own_code(team=task.team, task_type=task.task_type): + self.log.info( + "approve_and_start retyped main-pm code task to planning", + task_id=str(task_id), + ) + task.task_type = cast("Any", TaskType.PLANNING) if notes: # Coordination metadata, not a human handoff — store as a marker so @@ -6128,6 +6192,28 @@ class TaskService(BaseService): if task.status == TaskStatus.AWAITING_PM_REVIEW: return await self._claim_review_state(task_id, claim_agent_id) + # Impossibility backstop (C8 — execution states only): a Main PM + # coordinates — it never claims a CODE task to execute (claiming here + # is owning through the lifecycle, NOT delegating; delegation uses the + # `delegate` verb, not claim). Scoped to run only AFTER the + # awaiting_pm_review review-claim above returned, so the legitimate + # Main-PM review/merge path is untouched. The effective claimant is the + # reassign target when ``allow_reassign`` (claim on behalf of), else the + # caller. The dispatcher never offers code tasks to main-pm (code→dev), + # so this is a backstop for a rogue / on-behalf-of-main-pm claim. + if allow_reassign: + claimant_is_main_pm = await self._is_main_pm_agent(claim_agent_id) + else: + claimant_is_main_pm = agent.role == AgentRole.MAIN_PM + if claimant_is_main_pm and _task_type_is_code(task.task_type): + raise UnauthorizedError( + action="claim", + reason=( + "MAIN_PM_NO_CODE: A Main PM coordinates — it does not own a" + " code task. Leave it for a developer; delegate instead." + ), + ) + claimed = await self.claim( task_id, claim_agent_id, allow_reassign=allow_reassign ) @@ -7079,6 +7165,46 @@ class TaskService(BaseService): ) return task + async def _maybe_divert_main_pm_code_reassign( + self, task: TaskTable, task_id: UUID, new_assignee: UUID | None + ) -> TaskTable | None: + """Divert a main_pm+code → Main-PM reassign to the pool, or None. + + Twin of ``_maybe_divert_board_advisory_reassign`` for the + impossibility invariant: a Main-PM target must never receive (back) a + ``main_pm`` + ``code`` task — a coordinator with no code verb cannot + fix the code, so re-handing it to Main PM perpetuates the mismatch + (the 2026-06-27 meltdown shape). Scoped to the team+type combo (NOT a + broad code+main-pm-target rule) so a legacy main_pm+code task can still + be reassigned to a cell dev — the correct remediation. The combo is + uncreatable going forward (create backstop + intake coercion + + approve_and_start retype), so this is a backstop for legacy / + direct-ORM-write tasks. + + Returns the refreshed task when it diverted, or ``None`` when no + diversion applies and the caller should proceed with the normal handoff. + """ + if not ( + new_assignee is not None + and main_pm_cannot_own_code(team=task.team, task_type=task.task_type) + and await self._is_main_pm_agent(new_assignee) + ): + return None + await self._divert_owned_task_to_pool( + task, + note=( + "\n\n[REASSIGN REDIRECTED] attempted to hand this main_pm + code" + " task to a Main PM that cannot own code. Released to the pool" + " for a role-matched claim instead." + ), + ) + self.log.info( + "main_pm + code task reassign to a Main PM diverted to pool", + task_id=str(task_id), + refused_assignee=str(new_assignee), + ) + return task + async def reassign( self, task_id: UUID, new_assignee: UUID | None ) -> TaskTable | None: @@ -7101,6 +7227,14 @@ class TaskService(BaseService): diverted = await self._maybe_divert_board_advisory_reassign( task, task_id, new_assignee ) + if diverted is not None: + return diverted + # main_pm + code → Main-PM hand-off is diverted to the pool (see + # `_maybe_divert_main_pm_code_reassign`); otherwise proceed with the + # normal handoff + cell-PM redirect. + diverted = await self._maybe_divert_main_pm_code_reassign( + task, task_id, new_assignee + ) if diverted is not None: return diverted # Invariant backstop: cell-team planning/research/administrative children @@ -7257,6 +7391,27 @@ class TaskService(BaseService): refused_assignee=str(new_assignee), ) return task + # Impossibility backstop (mirrors `reassign`): an active main_pm + code + # claim must not be handed to a Main PM — divert to the pool. Scoped to + # the team+type combo so a legacy main_pm+code task can still be + # reassign-claimed by a cell dev (the correct remediation). + if main_pm_cannot_own_code( + team=task.team, task_type=task.task_type + ) and await self._is_main_pm_agent(new_assignee): + await self._divert_owned_task_to_pool( + task, + note=( + "\n\n[REASSIGN REDIRECTED] attempted to hand this active" + " main_pm + code task to a Main PM that cannot own code." + " Released to the pool for a role-matched claim instead." + ), + ) + self.log.info( + "Active main_pm + code task reassign to a Main PM diverted", + task_id=str(task_id), + refused_assignee=str(new_assignee), + ) + return task # Invariant backstop: an active claim on a cell-team planning/research/ # administrative task must land on the cell PM, not main-pm. redirect = await self._resolve_cell_pm_redirect(task, new_assignee) @@ -7697,6 +7852,19 @@ class TaskService(BaseService): role = result.scalar_one_or_none() return role in _BOARD_ADVISORY_ROLES + async def _is_main_pm_agent(self, agent_id: UUID) -> bool: + """True if ``agent_id`` is the Main PM role (the coordinator with no code verb). + + Twin of ``_is_board_advisory_agent`` for the impossibility invariant — + consulted by the escalation / reassign / claim guards that divert or + refuse a ``main_pm`` + ``code`` hand-off to a Main-PM target. + """ + result = await self.session.execute( + select(AgentTable.role).where(AgentTable.id == agent_id) + ) + role = result.scalar_one_or_none() + return role == AgentRole.MAIN_PM + async def _divert_owned_task_to_pool(self, task: TaskTable, *, note: str) -> None: """Clear ownership and return ``task`` to PENDING for a role-matched claim. diff --git a/tests/foundation/test_lifecycle_spec.py b/tests/foundation/test_lifecycle_spec.py index 5010a072..3961fd9a 100644 --- a/tests/foundation/test_lifecycle_spec.py +++ b/tests/foundation/test_lifecycle_spec.py @@ -711,6 +711,34 @@ def test_run_all_validators_raises_on_unknown_intent_action( _validate.run_all_lifecycle_validators() +def test_next_hint_pr_fail_main_pm_root_steers_to_redelegate() -> None: + """A ``pr_fail`` on a Main-PM branch-bearing root must steer the Main PM to + re-delegate the fixes, NOT re-submit the unchanged root. The root is an + assembled cell→root / root→master PR — coordination, not the Main PM's own + code — so re-submitting it is the 2026-06-27 infinite ``pr_fail`` loop.""" + t = SimpleNamespace(team=spec.Team.MAIN_PM, branch_name="feature/main_pm/c80e19ff") + hint = _INTENT_VERBS["pr_fail"].next_hint(t) + assert "re-delegate" in hint + assert "do NOT re-submit" in hint + + +def test_next_hint_pr_fail_cell_dev_keeps_dev_revise() -> None: + """A cell / dev task is revised in place by its dev, so ``pr_fail`` keeps the + dev-revise hint (the cell→root PR carries that dev's own code).""" + t = SimpleNamespace(team=spec.Team.BACKEND, branch_name="feature/backend/abc12345") + hint = _INTENT_VERBS["pr_fail"].next_hint(t) + assert hint == "idle - dev will revise and re-submit" + + +def test_next_hint_pr_fail_branchless_main_pm_keeps_dev_revise() -> None: + """A branchless Main-PM umbrella (no ``branch_name``) assembles no PR of its + own, so the gate never lands a ``pr_fail`` on it — but defensively it keeps + the dev-revise hint rather than the re-delegate steer.""" + t = SimpleNamespace(team=spec.Team.MAIN_PM, branch_name=None) + hint = _INTENT_VERBS["pr_fail"].next_hint(t) + assert hint == "idle - dev will revise and re-submit" + + def test_unmigrated_is_pinned() -> None: """The known-debt set; remove an entry once that consumer is migrated.""" assert ( diff --git a/tests/unit/foundation/policy/content/test_content_models.py b/tests/unit/foundation/policy/content/test_content_models.py index 1665cbc8..199855b1 100644 --- a/tests/unit/foundation/policy/content/test_content_models.py +++ b/tests/unit/foundation/policy/content/test_content_models.py @@ -206,6 +206,49 @@ def test_findings_single_dict_coerced_to_list() -> None: assert len(c.findings) == 1 +def test_pr_review_issues_carry_free_text_change_requests() -> None: + # The in-path gate fails on free-text issues (not structured Finding + # objects, which require file/severity/expected/actual). Those issues now + # land in the additive `issues` slot instead of being flattened into the + # summary string alone — so a reader of notes_structured.pr_review gets the + # concrete change-requests, and the rendered TEXT mirror gains an Issues + # section. + c = validate_content( + "pr_review", + { + "summary": "PR review needs changes before this can merge.", + "verdict": "changes_requested", + "issues": ["seam mismatch on the rebase path", "docs lag the diff"], + }, + ) + assert isinstance(c, PrReviewContent) + assert c.issues == ["seam mismatch on the rebase path", "docs lag the diff"] + rendered = c.render_markdown() + assert "## Issues" in rendered + assert "seam mismatch on the rebase path" in rendered + assert "docs lag the diff" in rendered + + +def test_pr_review_issues_default_empty_and_single_scalar_coerced() -> None: + c = validate_content( + "pr_review", + {"summary": "Clean PR, no free-text issues to raise.", "verdict": "approved"}, + ) + assert isinstance(c, PrReviewContent) + assert c.issues == [] + assert "## Issues" not in c.render_markdown() + + coerced = validate_content( + "pr_review", + { + "summary": "One free-text issue passed as a bare string here.", + "verdict": "changes_requested", + "issues": "lone issue string", + }, + ) + assert coerced.issues == ["lone issue string"] + + def test_where_to_look_single_string_coerced() -> None: c = validate_content( "resumption", diff --git a/tests/unit/foundation/policy/test_batch.py b/tests/unit/foundation/policy/test_batch.py index d7ab48fd..f0987465 100644 --- a/tests/unit/foundation/policy/test_batch.py +++ b/tests/unit/foundation/policy/test_batch.py @@ -10,7 +10,9 @@ from roboco.foundation.policy.batch import ( is_batch_umbrella, is_branchless_coordination, is_valid_batch_shape, + main_pm_cannot_own_code, ) +from roboco.models.base import TaskType, Team def test_umbrella_is_batch_id_set_and_top_level() -> None: @@ -163,3 +165,27 @@ def test_valid_batch_shape_denies_cell_map_alongside_another_target() -> None: product_id=uuid4(), has_cell_projects=True, ) + + +def test_main_pm_cannot_own_code_predicate() -> None: + """``main_pm`` + ``code`` must never coexist — the single invariant behind + the intake coercion, the create backstop, the reassign/escalation diversion, + and the claim guard. Accepts ORM enums or their .value strings.""" + # The forbidden combo, in both enum and string form. + assert main_pm_cannot_own_code(team=Team.MAIN_PM, task_type=TaskType.CODE) + assert main_pm_cannot_own_code( + team=Team.MAIN_PM.value, task_type=TaskType.CODE.value + ) + # A Main PM coordinating (planning / research / etc.) is fine. + assert not main_pm_cannot_own_code(team=Team.MAIN_PM, task_type=TaskType.PLANNING) + assert not main_pm_cannot_own_code(team=Team.MAIN_PM, task_type=TaskType.RESEARCH) + assert not main_pm_cannot_own_code( + team=Team.MAIN_PM, task_type=TaskType.DOCUMENTATION + ) + # Code owned by any non-main_pm team (a cell dev, the board pre-approval) is fine. + assert not main_pm_cannot_own_code(team=Team.BACKEND, task_type=TaskType.CODE) + assert not main_pm_cannot_own_code(team=Team.FRONTEND, task_type=TaskType.CODE) + assert not main_pm_cannot_own_code(team=Team.BOARD, task_type=TaskType.CODE) + # Missing team or type cannot satisfy the invariant. + assert not main_pm_cannot_own_code(team=None, task_type=TaskType.CODE) + assert not main_pm_cannot_own_code(team=Team.MAIN_PM, task_type=None) diff --git a/tests/unit/gateway/test_pr_gate_notifies_pm.py b/tests/unit/gateway/test_pr_gate_notifies_pm.py new file mode 100644 index 00000000..44594606 --- /dev/null +++ b/tests/unit/gateway/test_pr_gate_notifies_pm.py @@ -0,0 +1,232 @@ +"""pr_fail delivers its change-requests to the owning PM, not just to GitHub. + +The in-path ``pr_fail`` gate persists the reviewer's verdict to +``notes_structured.pr_review`` and posts it on the assembled PR — but historically +never pushed the concrete issues to any channel the owning PM reads (no a2a, +and ``_briefing_for`` / ``build_task_handoff`` read neither ``pr_reviewer_notes`` +nor ``notes_structured.pr_review``). So the cell PM respawned into +``needs_revision`` saw a generic "needs revision" with zero actionable issues, +concluded there was nothing to rework, and re-submitted the same PR — an +infinite ``pr_fail`` loop (observed live on coordination root 9980d0a0 / PR #138). + +The fix mirrors QA's ``fail_review`` a2a to the dev (qa.py:671-678): on +``pr_fail`` the gate now sends an a2a to the owner the runner just re-assigned +(the cell PM, via ``_revision_pm_for_task``) carrying the issues, so the PM +"knows" and can ``delegate`` a rework subtask or action the change-requests +directly. ``pr_pass`` is unaffected. +""" + +from __future__ import annotations + +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 + + +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)) + + +def _stub_gate_path( + c: Choreographer, + *, + reviewer_id: Any, + t_before: Any, + t_after: Any, +) -> None: + """Drive ``_gate_decision`` past preflight/tracing/post and into the new + a2a step without exercising the heavy ownership/tracing logic (those have + their own tests). The runner rebinding to ``t_after`` is what simulates the + PM re-assignment the real ``_revision_pm_for_task`` performs. + """ + agent = MagicMock(role="pr_reviewer", slug="be-pr-reviewer") + c._gate_preflight = AsyncMock( # type: ignore[method-assign] + return_value=( + t_before, + agent, + "pr_reviewer", + {}, + spec_module.Context(actor_id=reviewer_id), + ) + ) + c._gate_tracing = AsyncMock(return_value=None) # type: ignore[method-assign] + c._record_gate_verdict = MagicMock() # type: ignore[method-assign] + c._post_gate_review_to_pr = AsyncMock() # type: ignore[method-assign] + runner = MagicMock() + runner.run_intent = AsyncMock(return_value=t_after) + c._verb_runner = MagicMock(return_value=runner) # type: ignore[method-assign] + + +@pytest.mark.asyncio +async def test_pr_fail_notifies_reassigned_owning_pm() -> None: + reviewer_id = uuid4() + pm_id = uuid4() + task_id = uuid4() + parent_id = uuid4() + t_before = MagicMock( + id=task_id, + assigned_to=reviewer_id, # owned by the reviewer until the runner runs + pr_number=138, + parent_task_id=parent_id, + status="awaiting_pr_review", + ) + # The runner reassigns the task to the owning PM (needs_revision owner). + t_after = MagicMock( + id=task_id, + assigned_to=pm_id, + pr_number=138, + parent_task_id=parent_id, + status="needs_revision", + ) + + c = _make_choreographer() + _stub_gate_path(c, reviewer_id=reviewer_id, t_before=t_before, t_after=t_after) + + await c.pr_fail(reviewer_id, task_id, ["seam mismatch", "docs lag the diff"]) + + c.a2a.send.assert_awaited_once() + kwargs = c.a2a.send.await_args.kwargs + assert kwargs["from_agent"] == reviewer_id + assert kwargs["to_agent"] == pm_id + assert kwargs["skill"] == "code_review" + assert kwargs["task_id"] == task_id + body = kwargs["body"] + assert "PR review needs changes." in body + assert "seam mismatch" in body + assert "docs lag the diff" in body + + +@pytest.mark.asyncio +async def test_pr_pass_does_not_notify_anyone() -> None: + 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() + _stub_gate_path(c, reviewer_id=reviewer_id, t_before=t_before, t_after=t_after) + + await c.pr_pass(reviewer_id, task_id, "Assembled root scope is clean and covered.") + + c.a2a.send.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_pr_fail_skips_a2a_when_no_assignee() -> None: + """If the runner left the task unassigned, there's nobody to notify — must + not raise and must not crash on ``a2a.send(None)``.""" + reviewer_id = uuid4() + task_id = uuid4() + t_before = MagicMock( + id=task_id, + assigned_to=reviewer_id, + pr_number=9, + parent_task_id=uuid4(), + status="awaiting_pr_review", + ) + t_after = MagicMock( + id=task_id, assigned_to=None, pr_number=9, status="needs_revision" + ) + + c = _make_choreographer() + _stub_gate_path(c, reviewer_id=reviewer_id, t_before=t_before, t_after=t_after) + + env = await c.pr_fail(reviewer_id, task_id, ["one concrete issue here"]) + c.a2a.send.assert_not_awaited() + assert env.status == "needs_revision" + + +@pytest.mark.asyncio +async def test_pr_fail_a2a_for_main_pm_root_steers_to_redelegate() -> None: + """A Main-PM branch-bearing root is an assembled cell→root / root→master PR — + coordination, not the Main PM's own code. The ``pr_fail`` a2a body must steer + the Main PM to re-delegate the fixes and NOT re-submit the unchanged root + (the 2026-06-27 infinite ``pr_fail`` loop), while still carrying the concrete + issues. ``_revision_pm_for_task`` returns main-pm for a non-cell team, so the + recipient stays the Main PM — correct, since the Main PM re-delegates.""" + reviewer_id = uuid4() + main_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=main_pm_id, + pr_number=139, + parent_task_id=uuid4(), + status="needs_revision", + ) + # Team is read off the runner-rebound task. ``getattr(team, "value", team)`` + # must yield ``"main_pm"``; a MagicMock team would not, so set the attribute + # explicitly. branch_name set => an assembled root, not a branchless umbrella. + t_after.team = spec_module.Team.MAIN_PM + t_after.branch_name = "feature/main_pm/c80e19ff" + + c = _make_choreographer() + _stub_gate_path(c, reviewer_id=reviewer_id, t_before=t_before, t_after=t_after) + + await c.pr_fail(reviewer_id, task_id, ["duplicate TimeseriesChart export"]) + + c.a2a.send.assert_awaited_once() + kwargs = c.a2a.send.await_args.kwargs + assert kwargs["to_agent"] == main_pm_id + body = kwargs["body"] + assert "PR review needs changes." in body + assert "duplicate TimeseriesChart export" in body + assert "re-delegate" in body + assert "do NOT re-submit" in body + + +@pytest.mark.asyncio +async def test_pr_fail_a2a_failure_is_swallowed() -> None: + """The gate transition already committed; an a2a delivery failure must not + roll back the verdict or 500 the reviewer (same posture as the PR-post step, + and the inverse of the cell_pm_complete None-deref crash).""" + reviewer_id = uuid4() + pm_id = uuid4() + task_id = uuid4() + t_before = MagicMock( + id=task_id, + assigned_to=reviewer_id, + pr_number=7, + parent_task_id=uuid4(), + status="awaiting_pr_review", + ) + t_after = MagicMock( + id=task_id, assigned_to=pm_id, pr_number=7, status="needs_revision" + ) + + c = _make_choreographer() + _stub_gate_path(c, reviewer_id=reviewer_id, t_before=t_before, t_after=t_after) + c.a2a.send = AsyncMock(side_effect=RuntimeError("db hiccup")) # type: ignore[method-assign] + + env = await c.pr_fail(reviewer_id, task_id, ["a concrete actionable issue"]) + # Verdict still landed — the owning PM is in needs_revision. + assert env.status == "needs_revision" diff --git a/tests/unit/gateway/test_pr_gate_records_verdict.py b/tests/unit/gateway/test_pr_gate_records_verdict.py index 5cea3dd4..a777610f 100644 --- a/tests/unit/gateway/test_pr_gate_records_verdict.py +++ b/tests/unit/gateway/test_pr_gate_records_verdict.py @@ -89,3 +89,68 @@ def test_record_gate_verdict_swallows_invalid_note() -> None: # 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" + + +def test_pr_fail_stores_issues_structurally_not_summary_only() -> None: + """pr_fail's free-text issues must persist into the structured ``issues`` + slot (so a reader of notes_structured.pr_review gets the concrete + change-requests), not be flattened into the summary string alone.""" + c = _make_choreographer() + t = _TaskWithNoNotes() + c._record_gate_verdict( + t, + "pr_fail", + "Issues:\n- seam mismatch\n- docs lag the diff", + issues=("seam mismatch", "docs lag the diff"), + ) + slot = t.notes_structured["pr_review"] + assert slot["verdict"] == "failed" + assert slot["issues"] == ["seam mismatch", "docs lag the diff"] + # The derived TEXT mirror surfaces them too, so pr_reviewer_notes carries + # the concrete change-requests for any future reader. + assert "seam mismatch" in t.pr_reviewer_notes + assert "docs lag the diff" in t.pr_reviewer_notes + + +def test_pr_fail_summary_does_not_duplicate_issues() -> None: + """pr_fail's issues must render only under ## Issues, not also baked into + ## Summary — otherwise the Task Details "PR Reviewer Notes" card shows each + issue twice (once under Summary, once under Issues). The summary is a + substantive non-issues sentence; the structured ``issues`` slot carries the + change-requests. ``notes`` (with the issues) still drives the GitHub PR post + and the a2a to the owning PM — those are raw text, not rendered through + ``render_markdown``, so no duplication there.""" + c = _make_choreographer() + t = _TaskWithNoNotes() + c._record_gate_verdict( + t, + "pr_fail", + "Issues:\n- seam mismatch\n- docs lag the diff", + issues=("seam mismatch", "docs lag the diff"), + ) + slot = t.notes_structured["pr_review"] + assert slot["verdict"] == "failed" + # Issues live in the structured issues slot... + assert slot["issues"] == ["seam mismatch", "docs lag the diff"] + # ...NOT baked into the summary. + assert "seam mismatch" not in slot["summary"] + assert "docs lag the diff" not in slot["summary"] + # The rendered TEXT mirror surfaces each issue (under ## Issues)... + assert "seam mismatch" in t.pr_reviewer_notes + assert "docs lag the diff" in t.pr_reviewer_notes + # ...with both section headers present, and the summary is the substantive + # sentence (not the issues-joined string). + assert "## Summary" in t.pr_reviewer_notes + assert "## Issues" in t.pr_reviewer_notes + assert "requested changes" in t.pr_reviewer_notes + + +def test_pr_pass_leaves_issues_slot_empty() -> None: + c = _make_choreographer() + t = _TaskWithNoNotes() + c._record_gate_verdict( + t, "pr_pass", "Assembled root scope is clean; every criterion is covered." + ) + slot = t.notes_structured["pr_review"] + assert slot["verdict"] == "passed" + assert slot.get("issues", []) == [] diff --git a/tests/unit/gateway/test_pr_review_hand_format_guard.py b/tests/unit/gateway/test_pr_review_hand_format_guard.py new file mode 100644 index 00000000..38e8fd63 --- /dev/null +++ b/tests/unit/gateway/test_pr_review_hand_format_guard.py @@ -0,0 +1,183 @@ +"""post_pr_review refuses a hand-formatted verdict body with no findings. + +The tool's contract (flow_server.post_pr_review docstring) is explicit: ``body`` +is a one-paragraph summary; when ``findings`` are given the GitHub comment is +GENERATED in the RoboCo format (summary + findings table + verdict) — "do not +hand-format it in body". Nothing enforced that, so a reviewer could pass +``findings=[]`` and dump a self-formatted ``## Summary`` / ``## Issues`` / +``## Verdict`` markdown blob into ``body`` — which the system posts verbatim +(``_resolve_post_body`` returns ``body`` as-is when there are no findings). The +deployed renderer emits ``## Findings`` (never ``## Issues``), so a ``## Issues`` +section on the PR is proof the agent hand-formatted. Observed live: the reviewer +posted a body that listed the issues under BOTH ``## Summary`` and ``## Issues`` +and then repeated the entire block twice — a duplicated, self-redundant +hand-formatted blob the contributor sees. + +The guard: when ``findings`` is empty AND ``body`` carries verdict/section +markdown headers, reject with ``invalid_state`` and a remediation that points the +reviewer at the structured-findings path. The system never posts a hand-formatted +verdict, so the duplication cannot recur. A clean plain-note ``COMMENT`` with no +findings is still allowed (only verdict-shaped bodies are blocked), and a review +with structured findings is unaffected (the system generates the comment). +""" + +from __future__ import annotations + +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 + + +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)) + + +def _stub_post_path(c: Choreographer, *, reviewer_id: Any, t: Any) -> None: + """Drive ``post_pr_review`` past preflight + the verdict-consistency gate so + the hand-format guard is the thing under test. The runner / side-effects are + stubbed so a passing case does not hit GitHub or the DB transition.""" + agent = MagicMock(role="pr_reviewer", slug="be-pr-reviewer") + c._post_pr_review_preflight = AsyncMock( # type: ignore[method-assign] + return_value=( + agent, + "pr_reviewer", + {}, + spec_module.Context(actor_id=reviewer_id), + ) + ) + c._verdict_consistency_gate = AsyncMock(return_value=None) # type: ignore[method-assign] + c._project_slug_for = AsyncMock(return_value="proj") # type: ignore[method-assign] + c._resolve_post_body = MagicMock(return_value="generated body") # type: ignore[method-assign] + runner = MagicMock() + runner.run_intent = AsyncMock(return_value=t) + c._verb_runner = MagicMock(return_value=runner) # type: ignore[method-assign] + c._post_review_side_effects = AsyncMock() # type: ignore[method-assign] + + +def _task() -> Any: + return MagicMock( + id=uuid4(), + pr_number=200, + status="in_progress", + notes_structured=None, + pr_reviewer_notes="", + ) + + +@pytest.mark.asyncio +async def test_hand_formatted_verdict_body_with_no_findings_is_rejected() -> None: + reviewer_id = uuid4() + task_id = uuid4() + c = _make_choreographer() + _stub_post_path(c, reviewer_id=reviewer_id, t=_task()) + + hand_formatted = ( + "## Summary\nIssues:\n- [BLOCKER] seam mismatch\n" + "## Issues\n- [BLOCKER] seam mismatch\n## Verdict\nfailed" + ) + env = await c.post_pr_review(reviewer_id, task_id, hand_formatted, "COMMENT") + + body = env.as_dict() + assert body["error"] == "invalid_state" + assert "hand-formatted" in body["message"].lower() + assert "findings" in body["remediate"].lower() + # Nothing posted / transitioned — the guard fired before any side effect. + c.git.post_pr_review.assert_not_awaited() + c._verb_runner().run_intent.assert_not_awaited() # type: ignore[union-attr] + + +@pytest.mark.asyncio +async def test_hand_formatted_body_rejected_even_for_request_changes() -> None: + # pr_review_conflict already blocks REQUEST_CHANGES + no findings, but the + # guard is independent of event — a hand-formatted verdict must never post, + # whatever event the agent picked. + reviewer_id = uuid4() + task_id = uuid4() + c = _make_choreographer() + _stub_post_path(c, reviewer_id=reviewer_id, t=_task()) + + env = await c.post_pr_review( + reviewer_id, task_id, "## Verdict\nfailed\nbad", "REQUEST_CHANGES" + ) + assert env.as_dict()["error"] == "invalid_state" + + +@pytest.mark.asyncio +async def test_clean_plain_note_with_no_findings_is_allowed() -> None: + # A genuine plain COMMENT note (no verdict headers, no findings) is a legit + # use of event=COMMENT — the guard must not block it. + reviewer_id = uuid4() + task_id = uuid4() + t = _task() + c = _make_choreographer() + _stub_post_path(c, reviewer_id=reviewer_id, t=t) + + env = await c.post_pr_review( + reviewer_id, + task_id, + "Left a note for the contributor: the CI flake on job X is tracked separately.", + "COMMENT", + ) + body = env.as_dict() + assert body.get("error") is None + assert body["status"] == "in_progress" + + +@pytest.mark.asyncio +async def test_structured_findings_with_summary_body_is_allowed() -> None: + # With structured findings the system generates the comment; the guard + # (scoped to empty findings) does not fire even if the summary body happens + # to contain a header-shaped word. + reviewer_id = uuid4() + task_id = uuid4() + t = _task() + c = _make_choreographer() + _stub_post_path(c, reviewer_id=reviewer_id, t=t) + + env = await c.post_pr_review( + reviewer_id, + task_id, + "The 422 path is unguarded.", + "REQUEST_CHANGES", + findings=[ + { + "file": "roboco/services/git.py", + "line": 42, + "severity": "blocker", + "expected": "retry as COMMENT", + "actual": "raises", + } + ], + ) + body = env.as_dict() + assert body.get("error") is None + + +@pytest.mark.asyncio +async def test_envelope_invalid_state_has_introspection_role() -> None: + # The rejection must carry role introspection like the other gates. + reviewer_id = uuid4() + task_id = uuid4() + c = _make_choreographer() + _stub_post_path(c, reviewer_id=reviewer_id, t=_task()) + + env = await c.post_pr_review( + reviewer_id, task_id, "## Summary\nstuff\n## Verdict\nfailed", "COMMENT" + ) + assert env.error == "invalid_state" + # with_introspection(role="pr_reviewer") populated the introspection fields. + assert env.current_state is not None + assert env.valid_next_verbs is not None diff --git a/tests/unit/services/test_main_pm_code_guard.py b/tests/unit/services/test_main_pm_code_guard.py new file mode 100644 index 00000000..d0922dc8 --- /dev/null +++ b/tests/unit/services/test_main_pm_code_guard.py @@ -0,0 +1,406 @@ +"""``main_pm`` + ``code`` must never coexist — the impossibility guard. + +The 2026-06-27 MegaTask meltdown traced to a Main-PM-owned root-subtask that +was ``task_type=code``: the git/PR/review layer treated it as a code root +(branch + PR + ``submit_root`` + the in-path ``pr_review`` gate + complete), +while the ownership/dispatch layer treated it as coordination (owned / +claimed / submitted / completed by the Main PM). The two layers never +reconciled — a ``pr_fail`` on the assembled code landed on a coordinator with +no code verb, who re-submitted the unchanged root → infinite loop. The CEO's +fix: *"It should be impossible to draft a task where main pm and task type +code COEXIST."* + +This suite pins the layered backstop that closes the invariant at every site +where the combo can be created or re-handed: + +* ``TaskService.create`` rejects ``main_pm`` + ``code`` (the HTTP-route / + direct-create backstop; intake coerces ``code``→``planning``). +* ``approve_and_start`` retypes a board-routed code task to ``planning`` when + it hands it to Main PM (the board→main-pm handoff is where a project code + task would otherwise become ``main_pm`` + ``code``). +* ``apply_escalation`` / ``reassign`` / ``reassign_active_claim`` divert a + ``main_pm`` + ``code`` task handed to a Main-PM target to the pool — but + STILL allow reassigning it to a cell dev (the correct remediation). +* ``claim_task_for_agent`` rejects a Main-PM agent claiming a CODE task in an + execution state (claiming = owning through the lifecycle, not delegating), + while leaving the ``awaiting_pm_review`` review-claim path untouched (C8). + +The single predicate every layer consults is +``roboco.foundation.policy.batch.main_pm_cannot_own_code`` (accepts ORM enums +or their ``.value`` strings); these tests exercise the service-layer sites +that call it. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.models.base import ( + AgentRole, + Complexity, + TaskNature, + TaskStatus, + TaskType, + Team, +) +from roboco.models.task import TaskCreateRequest +from roboco.services.base import UnauthorizedError, ValidationError +from roboco.services.task import TaskService + + +def _bind(svc: TaskService, name: str, value: object) -> None: + object.__setattr__(svc, name, value) + + +def _service() -> TaskService: + session = MagicMock() + session.flush = AsyncMock() + session.commit = AsyncMock() + return TaskService(session) + + +def _task( + *, + team: Team, + task_type: TaskType, + status: TaskStatus = TaskStatus.PENDING, + assigned_to: object = None, + board_review_complete: bool = True, +) -> MagicMock: + return MagicMock( + id=uuid4(), + team=team, + task_type=task_type, + status=status, + assigned_to=assigned_to, + claimed_by=assigned_to, + active_claimant_id=assigned_to, + blocker_raised_by=None, + board_review_complete=board_review_complete, + dev_notes="", + ) + + +def _agent(role: AgentRole = AgentRole.DEVELOPER) -> MagicMock: + return MagicMock(role=role, agent_id=uuid4()) + + +def _perms() -> MagicMock: + p = MagicMock() + p.can_perform_task_action = MagicMock(return_value=True) + return p + + +# --------------------------------------------------------------------------- +# Site 1 — TaskService.create backstop (invariant #1: the combo on the task) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_create_rejects_main_pm_plus_code() -> None: + svc = TaskService( + MagicMock(add=MagicMock(), flush=AsyncMock(), execute=AsyncMock()) + ) + req = TaskCreateRequest( + title="rogue main-pm code root", + description="should not persist", + acceptance_criteria=["ship it"], + team=Team.MAIN_PM, + created_by=uuid4(), + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.MEDIUM, + project_id=uuid4(), + ) + with pytest.raises(ValidationError, match="MAIN_PM"): + await svc.create(req) + + +@pytest.mark.asyncio +async def test_create_allows_main_pm_plus_planning() -> None: + # The backstop is the team+type combo, not main_pm alone — a Main-PM + # coordination root typed planning is the canonical shape, not a mismatch. + svc = TaskService( + MagicMock(add=MagicMock(), flush=AsyncMock(), execute=AsyncMock()) + ) + req = TaskCreateRequest( + title="main-pm coordination root", + description="fine", + acceptance_criteria=["coordinate the cells"], + team=Team.MAIN_PM, + created_by=uuid4(), + task_type=TaskType.PLANNING, + nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.MEDIUM, + project_id=uuid4(), + ) + task = await svc.create(req) + assert task.task_type == TaskType.PLANNING + + +# --------------------------------------------------------------------------- +# Site 2 — approve_and_start retype (board→main-pm handoff) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_approve_and_start_retypes_code_to_planning_for_main_pm( + monkeypatch: pytest.MonkeyPatch, +) -> None: + svc = _service() + main_pm_agent = MagicMock(id=uuid4(), slug="main-pm", role=AgentRole.MAIN_PM) + agent_svc = MagicMock() + agent_svc.get_by_slug = AsyncMock(return_value=main_pm_agent) + monkeypatch.setattr( + "roboco.services.agent.get_agent_service", lambda _session: agent_svc + ) + task = _task( + team=Team.BOARD, + task_type=TaskType.CODE, + status=TaskStatus.PENDING, + board_review_complete=True, + ) + _bind(svc, "get", AsyncMock(return_value=task)) + _bind(svc, "_activate_batch_root_subtasks", AsyncMock()) + _bind(svc, "_emit_task_event", AsyncMock()) + + await svc.approve_and_start(task.id) + + # Handed to Main PM (team set) AND retyped off code in the same write. + assert task.team == Team.MAIN_PM + assert task.task_type == TaskType.PLANNING + + +@pytest.mark.asyncio +async def test_approve_and_start_leaves_planning_untouched( + monkeypatch: pytest.MonkeyPatch, +) -> None: + svc = _service() + main_pm_agent = MagicMock(id=uuid4(), slug="main-pm", role=AgentRole.MAIN_PM) + agent_svc = MagicMock() + agent_svc.get_by_slug = AsyncMock(return_value=main_pm_agent) + monkeypatch.setattr( + "roboco.services.agent.get_agent_service", lambda _session: agent_svc + ) + task = _task( + team=Team.BOARD, + task_type=TaskType.PLANNING, + status=TaskStatus.PENDING, + board_review_complete=True, + ) + _bind(svc, "get", AsyncMock(return_value=task)) + _bind(svc, "_activate_batch_root_subtasks", AsyncMock()) + _bind(svc, "_emit_task_event", AsyncMock()) + + await svc.approve_and_start(task.id) + + assert task.team == Team.MAIN_PM + assert task.task_type == TaskType.PLANNING + + +# --------------------------------------------------------------------------- +# Site 3 — apply_escalation: divert a main_pm+code task → main-pm target to pool +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_apply_escalation_diverts_main_pm_code_to_main_pm_target() -> None: + svc = _service() + main_pm_target = uuid4() + task = _task( + team=Team.MAIN_PM, + task_type=TaskType.CODE, + status=TaskStatus.IN_PROGRESS, + assigned_to=uuid4(), + ) + _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False)) + _bind(svc, "_is_main_pm_agent", AsyncMock(return_value=True)) + release_mock = AsyncMock() + _bind(svc, "_release_code_task_to_pool", release_mock) + + await svc.apply_escalation( + task=task, + target_agent_id=main_pm_target, + escalator_slug="cell-pm", + target_slug="main-pm", + reason="blocked", + ) + + release_mock.assert_awaited_once() + assert task.assigned_to != main_pm_target + + +@pytest.mark.asyncio +async def test_apply_escalation_does_not_divert_main_pm_code_to_cell_dev() -> None: + # The correct remediation for a legacy main_pm+code task is to reassign it + # to a cell dev who can actually fix the code — the guard must NOT block + # that, only block re-handing it BACK to a Main-PM target. + svc = _service() + cell_dev_target = uuid4() + task = _task( + team=Team.MAIN_PM, + task_type=TaskType.CODE, + status=TaskStatus.IN_PROGRESS, + assigned_to=uuid4(), + ) + _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False)) + _bind(svc, "_is_main_pm_agent", AsyncMock(return_value=False)) + release_mock = AsyncMock() + _bind(svc, "_release_code_task_to_pool", release_mock) + _bind(svc, "_emit_task_event", AsyncMock()) + + await svc.apply_escalation( + task=task, + target_agent_id=cell_dev_target, + escalator_slug="main-pm", + target_slug="be-dev-1", + reason="reassign to a dev to fix", + ) + + # Not diverted — proceeds to the normal escalation path. + release_mock.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# Site 4 — reassign / reassign_active_claim diversion +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_reassign_diverts_main_pm_code_to_main_pm_target() -> None: + svc = _service() + main_pm_target = uuid4() + task = _task(team=Team.MAIN_PM, task_type=TaskType.CODE, assigned_to=uuid4()) + _bind(svc, "get", AsyncMock(return_value=task)) + _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False)) + _bind(svc, "_is_main_pm_agent", AsyncMock(return_value=True)) + diverted = AsyncMock() + _bind(svc, "_divert_owned_task_to_pool", diverted) + # The cell-PM redirect must NOT be reached when the main-pm-code divert fires. + _bind( + svc, + "_resolve_cell_pm_redirect", + AsyncMock( + side_effect=AssertionError( + "cell-PM redirect must not run when main-pm-code divert fires" + ) + ), + ) + + result = await svc.reassign(task.id, main_pm_target) + + assert result is task + diverted.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_reassign_active_claim_diverts_main_pm_code_to_main_pm_target() -> None: + svc = _service() + main_pm_target = uuid4() + task = _task( + team=Team.MAIN_PM, + task_type=TaskType.CODE, + status=TaskStatus.IN_PROGRESS, + assigned_to=uuid4(), + ) + _bind(svc, "get", AsyncMock(return_value=task)) + _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False)) + _bind(svc, "_is_main_pm_agent", AsyncMock(return_value=True)) + diverted = AsyncMock() + _bind(svc, "_divert_owned_task_to_pool", diverted) + _bind( + svc, + "_resolve_cell_pm_redirect", + AsyncMock( + side_effect=AssertionError( + "cell-PM redirect must not run when main-pm-code divert fires" + ) + ), + ) + + result = await svc.reassign_active_claim(task.id, main_pm_target) + + assert result is task + diverted.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# Site 5 — claim_task_for_agent: main-pm claiming code (execution state only) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_claim_rejects_main_pm_claiming_code_in_execution_state( + monkeypatch: pytest.MonkeyPatch, +) -> None: + svc = TaskService.__new__(TaskService) + svc.session = AsyncMock() + task = _task(team=Team.BACKEND, task_type=TaskType.CODE, status=TaskStatus.PENDING) + monkeypatch.setattr(svc, "_load_task_or_raise", AsyncMock(return_value=task)) + plain_claim = AsyncMock(return_value=task) + monkeypatch.setattr(svc, "claim", plain_claim) + + with pytest.raises(UnauthorizedError, match="MAIN_PM_NO_CODE"): + await svc.claim_task_for_agent( + task.id, _agent(role=AgentRole.MAIN_PM), _perms(), claim_target_slug=None + ) + + plain_claim.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_claim_allows_main_pm_claiming_planning_in_execution_state( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # A Main PM coordinates planning roots — claiming a planning task is the + # legitimate coordination path, not a mismatch. + svc = TaskService.__new__(TaskService) + svc.session = AsyncMock() + task = MagicMock( + team=Team.MAIN_PM, + task_type=TaskType.PLANNING, + status=TaskStatus.PENDING, + ) + monkeypatch.setattr(svc, "_load_task_or_raise", AsyncMock(return_value=task)) + claimed = MagicMock() + plain_claim = AsyncMock(return_value=claimed) + monkeypatch.setattr(svc, "claim", plain_claim) + + result = await svc.claim_task_for_agent( + task.id, _agent(role=AgentRole.MAIN_PM), _perms(), claim_target_slug=None + ) + + assert result is claimed + plain_claim.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_claim_does_not_reject_main_pm_code_in_awaiting_pm_review( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # C8: awaiting_pm_review is a REVIEW state, not an execution state. A + # Main PM legitimately claims it to complete/merge — the code guard must + # not fire there (it runs only after the review-state early return). + svc = TaskService.__new__(TaskService) + svc.session = AsyncMock() + task = MagicMock( + team=Team.MAIN_PM, + task_type=TaskType.CODE, + status=TaskStatus.AWAITING_PM_REVIEW, + ) + monkeypatch.setattr(svc, "_load_task_or_raise", AsyncMock(return_value=task)) + review_claim = AsyncMock(return_value=task) + monkeypatch.setattr(svc, "_claim_review_state", review_claim) + plain_claim = AsyncMock( + side_effect=AssertionError("transitioning claim must not run for review state") + ) + monkeypatch.setattr(svc, "claim", plain_claim) + + result = await svc.claim_task_for_agent( + task.id, _agent(role=AgentRole.MAIN_PM), _perms(), claim_target_slug=None + ) + + assert result is task + review_claim.assert_awaited_once() diff --git a/tests/unit/services/test_prompter.py b/tests/unit/services/test_prompter.py index 5f2def8b..81754335 100644 --- a/tests/unit/services/test_prompter.py +++ b/tests/unit/services/test_prompter.py @@ -469,6 +469,9 @@ async def test_confirm_live_draft_product_routes_to_main_pm(db_session: Any) -> assert row.team == Team.MAIN_PM assert row.product_id == product_id assert row.project_id is None + # A Main-PM coordination root is never code — intake coerces code->planning + # so main_pm + code can never coexist (the 2026-06-27 meltdown shape). + assert row.task_type == TaskType.PLANNING # ============================================================================= @@ -556,6 +559,8 @@ async def test_confirm_live_batch_builds_umbrella_and_sequenced_subtasks( assert umbrella.team == Team.MAIN_PM assert umbrella.status == TaskStatus.PENDING assert umbrella.branch_name is None # branchless + # A Main-PM coordination root is never code — the umbrella is planning-typed. + assert umbrella.task_type == TaskType.PLANNING a, b, c = [await db_session.get(TaskTable, UUID(sid)) for sid in ids] for sub in (a, b, c): @@ -563,6 +568,11 @@ async def test_confirm_live_batch_builds_umbrella_and_sequenced_subtasks( assert sub.batch_id == umbrella.batch_id assert sub.team == Team.MAIN_PM assert sub.status == TaskStatus.PENDING + # Each root-subtask is a Main-PM coordination root: code->planning coerced + # at intake so main_pm + code can never coexist (the 2026-06-27 meltdown + # shape). It still gets its own branch + PR + submit_root + pr_review gate + # — the gate is branch-keyed, not task_type-keyed. + assert sub.task_type == TaskType.PLANNING assert a.project_id == project1 assert b.project_id == project1 assert c.project_id == project2