fix(run-hardening): let PMs re-claim needs_revision coordination roots

The lifecycle spec (CLAIM_RULES) grants CELL_PM/MAIN_PM claim of NEEDS_REVISION so a rejected coordination root (pr_fail / qa_fail / ceo_reject) can be re-claimed via i_will_plan and re-delegated. The runtime mapping _ROLE_CLAIM_STATUSES omitted it for PMs, so the spec gate allowed i_will_plan on a needs_revision root while the composed claim() rejected it -> returned None -> INVALID_STATE: the PM could neither plan nor idle its own rejected root and respawn-looped (observed live on cell root 0e49e04e, ~143 INVALID_STATE rejections across 11 PM sessions; the tail of the 2026-06-24/25 run).

Add NEEDS_REVISION to the cell_pm/main_pm runtime claim statuses, and a parity test locking _ROLE_CLAIM_STATUSES to lifecycle.CLAIM_RULES so the two can't drift again.

roboco/services/task.py

@@ -75,8 +75,23 @@ _UUID_HYPHEN_COUNT = 4  # Number of hyphens in a UUID
 _ROLE_CLAIM_STATUSES: dict[str, set[TaskStatus]] = {
     "qa": {TaskStatus.PENDING, TaskStatus.AWAITING_QA},
     "documenter": {TaskStatus.PENDING, TaskStatus.AWAITING_DOCUMENTATION},
-    "cell_pm": {TaskStatus.PENDING, TaskStatus.AWAITING_PM_REVIEW},
+    # PMs re-claim NEEDS_REVISION to recover a rejected coordination/assembled
-    "main_pm": {TaskStatus.PENDING, TaskStatus.AWAITING_PM_REVIEW},
+    # task (pr_fail / qa_fail / ceo_reject lands it there): the spec
+    # (lifecycle.CLAIM_RULES) grants it, and the runtime must match or the spec
+    # gate passes i_will_plan on a needs_revision root while the composed
+    # claim() returns None -> INVALID_STATE -> the PM respawn-loops on its own
+    # rejected root, unable to plan or idle it. Parity is locked by
+    # tests/unit/services/test_pm_claim_needs_revision.py.
+    "cell_pm": {
+        TaskStatus.PENDING,
+        TaskStatus.NEEDS_REVISION,
+        TaskStatus.AWAITING_PM_REVIEW,
+    },
+    "main_pm": {
+        TaskStatus.PENDING,
+        TaskStatus.NEEDS_REVISION,
+        TaskStatus.AWAITING_PM_REVIEW,
+    },
 }
 
 

tests/unit/services/test_pm_claim_needs_revision.py

@@ -0,0 +1,48 @@
+"""Runtime claim-status parity: PMs may re-claim a NEEDS_REVISION coordination task.
+
+The lifecycle spec (``lifecycle.CLAIM_RULES``) lets ``CELL_PM`` / ``MAIN_PM``
+claim ``NEEDS_REVISION`` so a rejected coordination / assembled task (pr_fail,
+qa_fail, ceo_reject) can be re-claimed via ``i_will_plan`` and re-delegated.
+
+The runtime claim path (``TaskService.claim`` ->
+``_get_valid_claim_statuses`` -> ``_ROLE_CLAIM_STATUSES``) must honour the same
+authority. Otherwise the spec gate *allows* ``i_will_plan`` on a
+``needs_revision`` root, but the composed ``claim()`` inside the verb returns
+``None`` (source status not in the runtime mapping) -> the verb runner raises
+``INVALID_STATE`` -> the PM can neither plan nor idle its own rejected root and
+respawn-loops on it (observed live 2026-06-25 on the ``0e49e04e`` cell root,
+~143 INVALID_STATE rejections across 11 PM sessions).
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+from roboco.foundation.policy import lifecycle as spec
+from roboco.models.base import TaskStatus
+from roboco.services.task import _default_claim_statuses, _get_valid_claim_statuses
+
+
+@pytest.mark.parametrize("role", ["cell_pm", "main_pm"])
+def test_pm_runtime_claim_statuses_include_needs_revision(role: str) -> None:
+    agent = SimpleNamespace(role=role)
+    assert TaskStatus.NEEDS_REVISION in _get_valid_claim_statuses(
+        agent, allow_reassign=False
+    )
+    assert TaskStatus.NEEDS_REVISION in _default_claim_statuses(role)
+
+
+@pytest.mark.parametrize("role", [spec.Role.CELL_PM, spec.Role.MAIN_PM])
+def test_runtime_pm_claim_mapping_covers_spec_claim_rules(role: spec.Role) -> None:
+    """The runtime mapping must cover every status the spec grants the role.
+
+    Guards against the spec (CLAIM_RULES) and the runtime mapping
+    (_ROLE_CLAIM_STATUSES) drifting apart again — the parity invariant.
+    """
+    runtime_values = {s.value for s in _default_claim_statuses(role.value)}
+    for status in spec.CLAIM_RULES[role]:
+        assert status.value in runtime_values, (
+            f"runtime claim mapping for {role.value} is missing spec-allowed "
+            f"status '{status.value}'"
+        )