[sweep] lifecycle: 6 confirmed gaps fixed (cancel-ceo-gate, claim_pr_review gate, needs_team_match, valid_next_verbs narrowing, pr_reviewer unclaim, complete side_effect ordering)

roboco/foundation/policy/lifecycle.py

@@ -370,13 +370,19 @@ _STATUS_TRANSITIONS: tuple[StatusTransition, ...] = (
         "submit_pm_review",
         None,
     ),
-    # Cancel — PM/CEO can cancel from any non-terminal status
+    # Cancel — PM/CEO can cancel from any non-terminal status. A task sitting
+    # in the CEO approval queue is the CEO's decision to make: a Cell/Main PM
+    # cancelling it would bypass the human CEO gate (CLAUDE.md role table pins
+    # awaiting_ceo_approval -> cancelled as CEO-only), so that one source is
+    # gated to {CEO} while every other non-terminal source stays PM+CEO.
     *(
         StatusTransition(
             src,
             Status.CANCELLED,
             "cancel",
-            frozenset({Role.CELL_PM, Role.MAIN_PM, Role.CEO}),
+            frozenset({Role.CEO})
+            if src is Status.AWAITING_CEO_APPROVAL
+            else frozenset({Role.CELL_PM, Role.MAIN_PM, Role.CEO}),
         )
         for src in Status
         if src not in (Status.COMPLETED, Status.CANCELLED)
@@ -880,6 +886,12 @@ class Context:
     """
 
     actor_id: UUID | None = None
+    # The calling agent's team, for the needs_team_match rule. None means
+    # "caller did not supply it" — team-match is then enforced at the service
+    # layer (the historical sole enforcer), so the spec gate stays permissive
+    # and backward-compatible. Supplying it lets the spec gate close the gap
+    # for any consumer that trusts can_invoke_action as authoritative.
+    agent_team: str | None = None
     plan: str | dict[str, Any] | None = None
     has_journal_decision: bool = False
     has_journal_reflect: bool = False
@@ -974,6 +986,30 @@ PRECONDITION_NON_TERMINAL = Precondition(
 )
 
 
+def _p_external_review_pending(task: Any, _agent: Any, _ctx: Any) -> bool:
+    """True iff the task is PENDING — the only valid source for an inbound
+    external-PR review task (claim_pr_review). An awaiting_pr_review task is a
+    gate review, a distinct verb (claim_gate_review); letting claim_pr_review's
+    composed ``claim`` action's union source_statuses accept awaiting_pr_review
+    made the spec gate looser than the runtime and pointed the reviewer at the
+    wrong verb."""
+    status = getattr(task, "status", None)
+    value = status.value if isinstance(status, Status) else str(status)
+    return value == Status.PENDING.value
+
+
+PRECONDITION_EXTERNAL_REVIEW_STATE = Precondition(
+    key="external_review_state",
+    check=_p_external_review_pending,
+    remediate=(
+        "claim_pr_review is for an inbound external-PR task in pending only;"
+        " an assembled in-path gate review uses claim_gate_review"
+    ),
+    missing_token="external_review_state",
+    rejection_kind="invalid_state",
+)
+
+
 # The set of states from which a PR may be opened — the lifecycle-owned canon.
 # The HTTP PR-create path (GitService._assert_pr_create_allowed) and the gateway
 # ``open_pr`` intent must agree on this, so it lives here (the policy layer) as
@@ -1202,10 +1238,15 @@ _INTENT_VERBS: dict[str, IntentSpec] = {
     ),
     "unclaim": IntentSpec(
         name="unclaim",
-        allowed_roles=frozenset(_DEV_ROLES | _QA_ROLES | _DOC_ROLES | _PM_ROLES),
+        allowed_roles=frozenset(
+            _DEV_ROLES | _QA_ROLES | _DOC_ROLES | _PM_ROLES | {Role.PR_REVIEWER}
+        ),
         description=(
             "Voluntarily release a claim back to pending. The"
-            " work-in-progress branch is preserved."
+            " work-in-progress branch is preserved. A PR reviewer who claimed"
+            " an external review (in_progress) or a gate review"
+            " (awaiting_pr_review) and cannot finish releases the claim here"
+            " rather than wedging the lane until the stale-claim reaper."
         ),
         composes=(),  # special - cleared in service layer
         extra_preconditions=(),
@@ -1300,7 +1341,7 @@ _INTENT_VERBS: dict[str, IntentSpec] = {
             " pending -> claimed -> in_progress."
         ),
         composes=("claim", "start"),
-        extra_preconditions=(),
+        extra_preconditions=(PRECONDITION_EXTERNAL_REVIEW_STATE,),
         side_effects=(),
         next_hint=lambda _t: (
             "review the contributor's diff, then post_pr_review(task_id, ...)"
@@ -1385,10 +1426,14 @@ _INTENT_VERBS: dict[str, IntentSpec] = {
             "Cell PM merges the PR (leaf into the cell branch, or the gated"
             " cell→root PR into the root branch) + transitions to completed;"
             " Main PM escalates the root to the CEO (who merges root→master)."
+            " The merge runs BEFORE the complete transition: TaskService.complete"
+            " asserts the PR is already merged, so the choreographer verb body"
+            " (cell_pm_complete / main_pm_complete) owns the merge-first"
+            " ordering — no trailing pr_merge side_effect is declared here."
         ),
         composes=("complete",),
         extra_preconditions=(),
-        side_effects=("pr_merge",),
+        side_effects=(),
         next_hint=_next_hint_pm_complete,
     ),
     "escalate_up": IntentSpec(
@@ -1663,6 +1708,25 @@ def can_invoke_action(
     rejection = _check_self_review_and_preconditions(action, spec_action, task, ctx)
     if rejection is not None:
         return rejection
+    # Team-match rule. needs_team_match was historically a dead spec field —
+    # enforced only at the service layer, so a consumer trusting the spec gate
+    # alone let a backend dev claim a frontend task. When the caller supplies
+    # the agent's team via Context, enforce it here; absent, defer to the
+    # service layer (backward compatible).
+    if spec_action.needs_team_match and getattr(ctx, "agent_team", None) is not None:
+        task_team = getattr(task, "team", None)
+        if task_team is not None and ctx.agent_team != task_team:
+            return Decision.reject(
+                kind="not_authorized",
+                message=(
+                    f"team '{ctx.agent_team}' may not act on a"
+                    f" '{task_team}' task (team-based restriction)"
+                ),
+                remediate=(
+                    "this task belongs to another team; call give_me_work()"
+                    " to find a task in your own team"
+                ),
+            )
     if action == "claim":
         rejection = _check_claim_rules_narrow(role, task)
         if rejection is not None:
@@ -1773,6 +1837,18 @@ def valid_next_verbs(role: Role, task: Any) -> list[str]:
                 "invalid_state",
             ):
                 continue
+        elif name in ("claim_review", "claim_doc_task", "claim_gate_review"):
+            # Empty-compose claim verbs still gate on CLAIM_RULES (a status
+            # gate, not a Precondition). Mirroring can_invoke_intent, skip the
+            # verb when the role cannot claim from the task's current status —
+            # otherwise a QA reviewer on a non-awaiting_qa task is told
+            # claim_review is callable and wastes a turn on the rejection.
+            rejection = _check_claim_rules_narrow(role, task)
+            if rejection is not None and rejection.rejection_kind in (
+                "not_authorized",
+                "invalid_state",
+            ):
+                continue
         out.append(name)
     return sorted(out)
 

tests/foundation/test_lifecycle_spec.py

@@ -354,12 +354,20 @@ def test_status_transitions_role_constraints_match_canon() -> None:
     assert by_pair[
         (spec.Status.AWAITING_CEO_APPROVAL, spec.Status.NEEDS_REVISION, "ceo_reject")
     ] == frozenset({spec.Role.CEO})
-    # Cancel: PM + CEO from any non-terminal status
+    # Cancel: PM + CEO from any non-terminal status EXCEPT the CEO approval
+    # queue — cancelling a task the CEO is reviewing is the CEO's call, so
+    # awaiting_ceo_approval -> cancelled is gated to CEO only (a PM cancelling
+    # it would bypass the human CEO gate).
     cancel_constraint = frozenset({spec.Role.CELL_PM, spec.Role.MAIN_PM, spec.Role.CEO})
     for src in spec.Status:
         if src in (spec.Status.COMPLETED, spec.Status.CANCELLED):
             continue
-        assert by_pair[(src, spec.Status.CANCELLED, "cancel")] == cancel_constraint, (
+        expected = (
+            frozenset({spec.Role.CEO})
+            if src is spec.Status.AWAITING_CEO_APPROVAL
+            else cancel_constraint
+        )
+        assert by_pair[(src, spec.Status.CANCELLED, "cancel")] == expected, (
             f"cancel from {src.value} has wrong role_constraint"
         )
 
@@ -962,3 +970,105 @@ def test_claim_allows_developer_claiming_code_from_pending() -> None:
     enforced at create/delegate + i_will_work_on, not the claim gate)."""
     t = _claim_task(status="pending", task_type="code")
     assert spec.can_invoke_action(spec.Role.DEVELOPER, "claim", t).allowed
+
+
+# ---------------------------------------------------------------------------
+# Edge cases — logical-gap element sweep (2026-06-30)
+# ---------------------------------------------------------------------------
+
+
+def test_claim_pr_review_rejected_on_gate_task_points_to_claim_gate_review() -> None:
+    """claim_pr_review is for an inbound external-PR task in PENDING only. An
+    awaiting_pr_review gate task must be rejected (and remediation must point
+    the reviewer at claim_gate_review), not silently accepted by the spec gate."""
+    d = spec.can_invoke_intent(
+        spec.Role.PR_REVIEWER,
+        "claim_pr_review",
+        _stub_task(status="awaiting_pr_review"),
+    )
+    assert d.allowed is False
+    assert d.rejection_kind == "invalid_state"
+    assert "claim_gate_review" in (d.remediate or "")
+
+
+def test_claim_pr_review_allowed_on_pending_external_review() -> None:
+    """Green path: a pending external-PR review task is claimable."""
+    d = spec.can_invoke_intent(
+        spec.Role.PR_REVIEWER,
+        "claim_pr_review",
+        _stub_task(status="pending"),
+    )
+    assert d.allowed is True
+
+
+def test_needs_team_match_rejects_cross_team_claim_when_agent_team_supplied() -> None:
+    """needs_team_match was a dead spec field; when the caller supplies the
+    agent's team via Context, the spec gate must enforce it (a backend dev
+    cannot claim a frontend task)."""
+    d = spec.can_invoke_action(
+        spec.Role.DEVELOPER,
+        "claim",
+        _stub_task(status="pending", team="frontend"),
+        context=spec.Context(agent_team="backend"),
+    )
+    assert d.allowed is False
+    assert d.rejection_kind == "not_authorized"
+
+
+def test_needs_team_match_allows_same_team_claim() -> None:
+    d = spec.can_invoke_action(
+        spec.Role.DEVELOPER,
+        "claim",
+        _stub_task(status="pending", team="backend"),
+        context=spec.Context(agent_team="backend"),
+    )
+    assert d.allowed is True
+
+
+def test_needs_team_match_defers_when_agent_team_absent() -> None:
+    """Backward compat: without agent_team in Context, the spec gate stays
+    permissive (the service layer still enforces team-match)."""
+    d = spec.can_invoke_action(
+        spec.Role.DEVELOPER,
+        "claim",
+        _stub_task(status="pending", team="frontend"),
+    )
+    assert d.allowed is True
+
+
+def test_valid_next_verbs_omits_claim_review_when_qa_not_in_awaiting_qa() -> None:
+    """valid_next_verbs must apply claim-rule narrowing for empty-compose
+    claim verbs; a QA reviewer on a COMPLETED task must not be told
+    claim_review is callable."""
+    verbs = spec.valid_next_verbs(spec.Role.QA, _stub_task(status="completed"))
+    assert "claim_review" not in verbs
+
+
+def test_valid_next_verbs_includes_claim_review_for_qa_in_awaiting_qa() -> None:
+    verbs = spec.valid_next_verbs(spec.Role.QA, _stub_task(status="awaiting_qa"))
+    assert "claim_review" in verbs
+
+
+def test_pr_reviewer_has_unclaim_release_verb() -> None:
+    """A PR reviewer who cannot finish a review must have a self-release
+    verb (unclaim), not wedge the lane until the stale-claim reaper."""
+    assert "unclaim" in spec.intents_for_role(spec.Role.PR_REVIEWER)
+
+
+def test_unclaim_allowed_for_pr_reviewer() -> None:
+    d = spec.can_invoke_intent(
+        spec.Role.PR_REVIEWER,
+        "unclaim",
+        _stub_task(status="awaiting_pr_review"),
+    )
+    assert d.allowed is True
+
+
+def test_complete_intent_declares_no_inverted_pr_merge_side_effect() -> None:
+    """complete's IntentSpec must not declare a trailing pr_merge side_effect:
+    TaskService.complete asserts the PR is already merged, so the merge runs
+    FIRST (choreographer verb body owns the ordering). The spec must match
+    reality, not lie about a complete-then-merge composition."""
+    iv = spec._INTENT_VERBS["complete"]
+    assert iv.composes == ("complete",)
+    assert iv.side_effects == ()

tests/unit/enforcement/test_task_lifecycle.py

@@ -200,3 +200,34 @@ def test_sla_seconds_for_unknown_pair() -> None:
 
 def test_sla_seconds_for_no_role() -> None:
     assert sla_seconds_for(None, "in_progress") is None
+
+
+# ---------------------------------------------------------------------------
+# Cancel from the CEO approval queue is the CEO's call, not a PM's
+# ---------------------------------------------------------------------------
+
+
+def test_cancel_from_awaiting_ceo_is_ceo_only() -> None:
+    """A PM must not cancel a task the CEO is reviewing — that bypasses the
+    human CEO-approval gate. Only the CEO can cancel from awaiting_ceo_approval."""
+    assert can_agent_transition("awaiting_ceo_approval", "cancelled", "ceo") is True
+    assert (
+        can_agent_transition("awaiting_ceo_approval", "cancelled", "cell_pm") is False
+    )
+    assert (
+        can_agent_transition("awaiting_ceo_approval", "cancelled", "main_pm") is False
+    )
+
+
+def test_cancel_from_other_non_terminal_remains_pm_plus_ceo() -> None:
+    """The CEO-only narrowing is scoped to the approval queue; elsewhere a PM
+    may still cancel (unchanged behavior)."""
+    assert can_agent_transition("in_progress", "cancelled", "cell_pm") is True
+    assert can_agent_transition("awaiting_qa", "cancelled", "main_pm") is True
+
+
+def test_validate_cancel_from_awaiting_ceo_raises_for_pm() -> None:
+    with pytest.raises(TaskLifecycleError):
+        validate_task_transition("awaiting_ceo_approval", "cancelled", "cell_pm")
+    # CEO is allowed — no raise.
+    assert validate_task_transition("awaiting_ceo_approval", "cancelled", "ceo") is True