fix(lifecycle): let a PM recover its rejected coordination task from needs_revision

The in-path PR-review gate created a deadlock: when an assembled cell→root /
root→master PR fails the gate (pr_fail) — or qa_fail / ceo_reject fires — the
PM-owned coordination task lands in needs_revision, which was developer-claim-
only. So the task had no actor and no exit but cancel, and the cell PM escalated
in a loop (8KB of [ESCALATED] dev_notes on one task). Pre-gate, the PM simply
re-delegated from in_progress; the gate routed the failure through the PM's own
task instead.

Add NEEDS_REVISION to the CELL_PM / MAIN_PM claim rules so the PM re-claims via
i_will_plan, revises the plan, and re-delegates the fixes — pr_fail/qa_fail
already reassign the failed task to its owning PM and the revision dispatcher
re-spawns it; the claim rule was the only missing piece.

Scope is by give_me_work routing (offers only the caller's own assigned tasks),
the same mechanism that scopes a developer's leaf-revision — NOT a gateway-only
ownership gate, which would violate the spec=gateway parity invariant and can't
use task_type anyway (main-PM coordination roots can be code-typed). Regenerates
panel/lib/lifecycle.json.

panel/lib/lifecycle.json

@@ -2,6 +2,7 @@
   "claim_rules": {
     "auditor": [],
     "cell_pm": [
+      "needs_revision",
       "pending"
     ],
     "ceo": [],
@@ -15,6 +16,7 @@
     ],
     "head_marketing": [],
     "main_pm": [
+      "needs_revision",
       "pending"
     ],
     "pr_reviewer": [

roboco/foundation/policy/lifecycle.py

@@ -666,8 +666,20 @@ CLAIM_RULES: dict[Role, frozenset[Status]] = {
     Role.DEVELOPER: frozenset({Status.PENDING, Status.NEEDS_REVISION}),
     Role.QA: frozenset({Status.AWAITING_QA}),
     Role.DOCUMENTER: frozenset({Status.PENDING, Status.AWAITING_DOCUMENTATION}),
-    Role.CELL_PM: frozenset({Status.PENDING}),
-    Role.MAIN_PM: frozenset({Status.PENDING}),
+    # PMs re-claim NEEDS_REVISION to recover a rejected coordination task: when
+    # an assembled cell→root / root→master PR fails the in-path gate (pr_fail),
+    # QA fails a planning task, or the CEO rejects (ceo_reject), the task lands
+    # in NEEDS_REVISION. Before this, that state was developer-claim-only, so a
+    # PM-owned coordination task had NO actor and no exit but cancel — the cell
+    # PM escalated in a loop (the in-path PR-review gate introduced this path;
+    # pre-gate, the PM re-delegated from in_progress). The PM now re-claims via
+    # i_will_plan, revises the plan, and re-delegates the fixes. pr_fail/qa_fail
+    # reassign the task to its owning PM, so this is scoped by the SAME mechanism
+    # that scopes a developer's leaf-revision: give_me_work only ever offers an
+    # agent its own assigned tasks. (A per-instance ownership gate at the gateway
+    # would diverge from this spec — the parity invariant forbids that.)
+    Role.CELL_PM: frozenset({Status.PENDING, Status.NEEDS_REVISION}),
+    Role.MAIN_PM: frozenset({Status.PENDING, Status.NEEDS_REVISION}),
     Role.PRODUCT_OWNER: frozenset(),
     Role.HEAD_MARKETING: frozenset(),
     Role.AUDITOR: frozenset(),

roboco/services/gateway/choreographer/_impl.py

@@ -1067,6 +1067,11 @@ class Choreographer:
         Per-role claim authority (CLAIM_RULES) is enforced inside
         spec.can_invoke_action when action == "claim", called by
         can_invoke_intent, so no separate spec.can_claim call is needed.
+        A PM recovering a NEEDS_REVISION coordination task is scoped by
+        give_me_work routing (it only ever offers an agent its OWN assigned
+        tasks), exactly as developer leaf-revisions are — not by a gateway-only
+        ownership gate, which would diverge from the spec (the parity invariant
+        requires gateway authorization == spec authorization).
         """
         t, briefing, role_str = ctx.task, ctx.briefing, ctx.role_str
         verb_name = ctx.verb_name

tests/foundation/test_lifecycle_spec.py

@@ -406,7 +406,10 @@ def test_qa_pass_self_review_blocks() -> None:
 def test_claim_rules_match_pre_gateway_table() -> None:
     """PERMISSIONS.md "What Each Role Can Claim From" — exact match.
 
-    PMs claim from PENDING only; BACKLOG → PENDING is a separate `activate`
+    PMs claim from PENDING and NEEDS_REVISION — the latter to recover a rejected
+    coordination task (pr_fail / qa_fail / ceo_reject) by re-planning and
+    re-delegating fixes (scoped by give_me_work routing, which offers only the
+    caller's own assigned tasks). BACKLOG → PENDING is a separate `activate`
     action (strict transitions; no implicit activate-on-claim).
     """
     assert spec.CLAIM_RULES[spec.Role.DEVELOPER] == frozenset(
@@ -416,8 +419,12 @@ def test_claim_rules_match_pre_gateway_table() -> None:
     assert spec.CLAIM_RULES[spec.Role.DOCUMENTER] == frozenset(
         {spec.Status.PENDING, spec.Status.AWAITING_DOCUMENTATION}
     )
-    assert spec.CLAIM_RULES[spec.Role.CELL_PM] == frozenset({spec.Status.PENDING})
-    assert spec.CLAIM_RULES[spec.Role.MAIN_PM] == frozenset({spec.Status.PENDING})
+    assert spec.CLAIM_RULES[spec.Role.CELL_PM] == frozenset(
+        {spec.Status.PENDING, spec.Status.NEEDS_REVISION}
+    )
+    assert spec.CLAIM_RULES[spec.Role.MAIN_PM] == frozenset(
+        {spec.Status.PENDING, spec.Status.NEEDS_REVISION}
+    )
 
 
 def test_team_rules_pin_team_for_seeded_agents() -> None:

tests/unit/gateway/test_needs_revision_recovery.py

@@ -0,0 +1,96 @@
+"""PM recovery of a rejected coordination task from needs_revision.
+
+The in-path PR-review gate (and qa_fail / ceo_reject) can land a PM-owned
+coordination/assembled task in ``needs_revision``. That state used to be
+developer-claim-only, so the task had no actor and no exit but ``cancel`` — the
+cell PM escalated in a loop. Now a PM re-claims its rejected task
+(``i_will_plan``), revises the plan, and re-delegates the fixes.
+
+Scope: ``pr_fail`` / ``qa_fail`` reassign the task to its owning PM, and
+``give_me_work`` only ever offers an agent its OWN assigned tasks — so a PM is
+never handed a developer's leaf, exactly as a developer is never handed a PM's
+coordination root. (The scope lives in routing, not a gateway-only ownership
+gate, which would break the spec=gateway parity invariant.)
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.foundation.policy import lifecycle as spec
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+# --------------------------------------------------------------------------- #
+# Lifecycle authority: PMs may now claim needs_revision
+# --------------------------------------------------------------------------- #
+
+
+def test_pm_can_claim_needs_revision() -> None:
+    task = SimpleNamespace(
+        status="needs_revision", task_type="planning", assigned_to=None
+    )
+    for role in (spec.Role.CELL_PM, spec.Role.MAIN_PM):
+        assert spec.can_invoke_action(role, "claim", task).allowed
+    # Developers still own leaf revisions; QA / documenter still cannot claim it.
+    assert spec.can_invoke_action(spec.Role.DEVELOPER, "claim", task).allowed
+    assert not spec.can_invoke_action(spec.Role.QA, "claim", task).allowed
+    assert not spec.can_invoke_action(spec.Role.DOCUMENTER, "claim", task).allowed
+
+
+def test_pm_claim_needs_revision_works_for_code_typed_root() -> None:
+    # Main-PM coordination roots can be code-typed, so the claim must NOT be
+    # task_type-gated — authority is status-based, scoped by routing.
+    task = SimpleNamespace(status="needs_revision", task_type="code", assigned_to=None)
+    assert spec.can_invoke_action(spec.Role.MAIN_PM, "claim", task).allowed
+
+
+# --------------------------------------------------------------------------- #
+# Routing scope: give_me_work offers a PM its OWN rejected coordination task
+# --------------------------------------------------------------------------- #
+
+
+def _make_deps(task_svc: AsyncMock) -> ChoreographerDeps:
+    repo = AsyncMock()
+    for m in (
+        "list_unread_a2a",
+        "list_unread_mentions",
+        "list_pending_notifications",
+        "task_metadata_gaps",
+        "recent_team_activity",
+        "blockers_in_lane",
+        "journal_highlights_for_task",
+        "company_goals",
+    ):
+        getattr(repo, m).return_value = []
+    return ChoreographerDeps(
+        task=task_svc,
+        work_session=AsyncMock(),
+        git=AsyncMock(),
+        a2a=AsyncMock(),
+        journal=AsyncMock(),
+        audit=AsyncMock(),
+        evidence_repo=repo,
+    )
+
+
+@pytest.mark.asyncio
+async def test_give_me_work_offers_pm_its_needs_revision_task() -> None:
+    pm_id, task_id = uuid4(), uuid4()
+    task = MagicMock(
+        id=task_id, status="needs_revision", team="backend", dependency_ids=[]
+    )
+    task_svc = AsyncMock()
+    task_svc.list_pending_for_agent.return_value = []
+    task_svc.list_assigned_for_agent.return_value = [task]
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    c = Choreographer(_make_deps(task_svc))
+
+    env = await c.give_me_work(pm_id)
+    body = env.as_dict()
+
+    assert body["task_id"] == str(task_id)
+    # The PM is told to re-plan (revise) the rejected task, not a dev verb.
+    assert "i_will_plan" in body["next"]