feat(gateway): restore Gate Set A claim-time guards

Ports five pre-gateway predicates that were dropped when the gateway
displaced the MCP claim handler. Predicates restored from
roboco/mcp/tasks/handlers/_helpers.py:124-204 and
roboco/mcp/tasks/handlers/claim.py:121-180 at commit 254cc93:

- SEQUENCE_ORDER_VIOLATION: a sibling task with sequence < N must be
  in completed/cancelled before sibling N can be claimed.
- ALREADY_ACTIVE: agent cannot claim while owning an in_progress /
  claimed / verifying task other than the one being resumed.
- PAUSED_TASKS_EXIST: agent cannot claim while paused tasks exist.
- PM_CANNOT_EXECUTE_CODE: cell_pm/main_pm cannot claim task_type=code.
- ROLE_TYPED_CLAIM: developer claim is restricted to
  code/research/design; qa/documenter must use claim_review /
  claim_doc_task.

All five guards run inside Choreographer._run_claim_guards before
i_will_work_on / i_will_plan / claim_review / claim_doc_task mutate
state. Skip flags isolate guards that don't apply to a verb (e.g.,
PM-code skipped on QA verb, role-typed skipped on PM verb).

The guards live in roboco/services/gateway/claim_guards.py so
choreographer.py stays focused on orchestration.

Existing tests updated to provide the new mock primings; the
permissive auto-mock behavior they relied on no longer applies.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

roboco/services/gateway/choreographer.py

@@ -16,6 +16,13 @@ from typing import Any
 from uuid import UUID
 
 from roboco.config import settings
+from roboco.services.gateway.claim_guards import (
+    already_active_guard,
+    paused_tasks_guard,
+    pm_cannot_execute_code_guard,
+    role_typed_claim_guard,
+    sibling_sequence_guard,
+)
 from roboco.services.gateway.envelope import Envelope
 from roboco.services.gateway.evidence_builder import (
     BriefingInputs,
@@ -162,6 +169,64 @@ class Choreographer:
         )
         return build_context_briefing(inputs)
 
+    async def _run_claim_guards(
+        self,
+        *,
+        agent_id: UUID,
+        task: Any,
+        skip_role_typed: bool = False,
+        skip_pm_code: bool = False,
+        skip_sequence: bool = False,
+    ) -> Envelope | None:
+        """Run claim-time guards (Gate Set A). Returns rejection or None.
+
+        Pre-gateway location: _helpers.py:124-204 + claim.py:121-180.
+
+        Optional skip flags isolate guards that don't apply to a given verb:
+        - skip_role_typed: i_will_plan/claim_review/claim_doc_task have their
+          own role checks; only i_will_work_on uses role_typed_claim_guard.
+        - skip_pm_code: claim_review/claim_doc_task call sites cannot be PMs
+          to begin with; pm_cannot_execute_code is meaningless there.
+        - skip_sequence: some verbs (resumption of an already-claimed task)
+          do not need to re-validate sibling order.
+        """
+        agent = await self.task.agent_for(agent_id)
+        role = agent.role if agent is not None else "developer"
+        task_type = str(getattr(task, "task_type", "code") or "code")
+
+        if not skip_pm_code and (
+            guard := pm_cannot_execute_code_guard(role, task_type)
+        ):
+            return guard
+        if not skip_role_typed and (
+            guard := role_typed_claim_guard(role, task_type)
+        ):
+            return guard
+        in_progress = await self.task.list_in_progress_for_agent(agent_id)
+        if guard := already_active_guard(in_progress, task.id):
+            return guard
+        paused = await self.task.list_paused_for_agent(agent_id)
+        if guard := paused_tasks_guard(paused):
+            return guard
+        if not skip_sequence:
+            siblings = await self._fetch_siblings(task)
+            if guard := sibling_sequence_guard(task, siblings):
+                return guard
+        return None
+
+    async def _fetch_siblings(self, task: Any) -> list[Any]:
+        """Fetch sibling tasks for the sequence-order guard.
+
+        Returns ``[]`` when the task has no parent (root task) so the
+        guard short-circuits. Otherwise returns the parent's subtasks via
+        ``TaskService.get_subtasks``.
+        """
+        parent_id = getattr(task, "parent_task_id", None)
+        if parent_id is None:
+            return []
+        siblings: list[Any] = await self.task.get_subtasks(parent_id)
+        return siblings
+
     async def i_will_work_on(
         self, agent_id: UUID, task_id: UUID, plan: str | None = None
     ) -> Envelope:
@@ -173,10 +238,15 @@ class Choreographer:
         briefing = await self._briefing_for(agent_id, task_id)
 
         if status == "needs_revision":
+            # Resumption after QA rejection: agent already owned the task,
+            # role-typed claim already passed at original claim time.
             if t.assigned_to != agent_id:
                 t = await self.task.claim(agent_id, task_id)
             t = await self.task.start(agent_id, task_id)
         elif status == "pending":
+            # Fresh claim — run all claim-time gates BEFORE mutating state.
+            if guard := await self._run_claim_guards(agent_id=agent_id, task=t):
+                return self._with_briefing(guard, briefing)
             if t.assigned_to is None or t.assigned_to != agent_id:
                 t = await self.task.claim(agent_id, task_id)
             if not t.plan and not plan:
@@ -193,6 +263,13 @@ class Choreographer:
                 t = await self.task.set_plan(task_id, plan)
             t = await self.task.start(agent_id, task_id)
         elif status == "claimed" and t.assigned_to == agent_id:
+            # Resumption: skip sibling-sequence (already passed at claim).
+            # Still enforce already_active/paused so concurrent claims fail.
+            guard = await self._run_claim_guards(
+                agent_id=agent_id, task=t, skip_sequence=True
+            )
+            if guard:
+                return self._with_briefing(guard, briefing)
             t = await self.task.start(agent_id, task_id)
         else:
             return Envelope.invalid_state(
@@ -211,6 +288,12 @@ class Choreographer:
             context_briefing=briefing,
         )
 
+    @staticmethod
+    def _with_briefing(env: Envelope, briefing: dict[str, Any]) -> Envelope:
+        """Attach a context_briefing to an Envelope (mutate-and-return helper)."""
+        env.context_briefing = briefing
+        return env
+
     async def i_have_committed(self, agent_id: UUID, message: str) -> Envelope:
         """Record that the dev made a commit; auto-creates progress entry."""
         t = await self.task.get_active_task_for_agent(agent_id)
@@ -453,6 +536,23 @@ class Choreographer:
                 remediate="call give_me_work() to find an actionable QA task",
                 context_briefing=await self._briefing_for(qa_agent_id, task_id),
             )
+
+        # Gate Set A: ALREADY_ACTIVE / PAUSED_TASKS_EXIST guard QA from
+        # juggling reviews while their previous in_progress task is open.
+        # role_typed/PM-code skipped: QA verb only ever fires for QA role.
+        # sequence skipped: QA reviews are siblings on a different axis.
+        guard = await self._run_claim_guards(
+            agent_id=qa_agent_id,
+            task=t,
+            skip_role_typed=True,
+            skip_pm_code=True,
+            skip_sequence=True,
+        )
+        if guard:
+            return self._with_briefing(
+                guard, await self._briefing_for(qa_agent_id, task_id)
+            )
+
         t = await self.task.qa_claim(qa_agent_id, task_id)
 
         # Auto-mark evidence as inspected — we surface it inline in this response
@@ -636,6 +736,22 @@ class Choreographer:
                 remediate="call give_me_work() to find an actionable doc task",
                 context_briefing=await self._briefing_for(doc_agent_id, task_id),
             )
+
+        # Gate Set A: ALREADY_ACTIVE / PAUSED_TASKS_EXIST. The doc verb only
+        # ever fires for documenter role; PM-code and role-typed claim guards
+        # are skipped. Sequence guard is also irrelevant here.
+        guard = await self._run_claim_guards(
+            agent_id=doc_agent_id,
+            task=t,
+            skip_role_typed=True,
+            skip_pm_code=True,
+            skip_sequence=True,
+        )
+        if guard:
+            return self._with_briefing(
+                guard, await self._briefing_for(doc_agent_id, task_id)
+            )
+
         t = await self.task.doc_claim(doc_agent_id, task_id)
         files_changed: list[str] = []
         if t.work_session_id:
@@ -727,18 +843,10 @@ class Choreographer:
             context_briefing=await self._briefing_for(doc_agent_id, task_id),
         )
 
-    async def i_will_plan(
+    async def _i_will_plan_preflight(
-        self, pm_agent_id: UUID, task_id: UUID, plan: str
+        self, pm_agent_id: UUID, task_id: UUID, t: Any, plan: str
-    ) -> Envelope:
+    ) -> Envelope | None:
-        """PM mirror of i_will_work_on for parent tasks.
+        """Run i_will_plan's role / status / plan / claim guards. None = pass."""
-
-        Transitions a pending task owned (or claimable by) this PM into
-        in_progress with the supplied plan. Required before a PM can call
-        ``delegate`` to spawn subtasks.
-        """
-        t = await self.task.get(task_id)
-        if t is None:
-            return Envelope.not_found(message=f"task {task_id} not found")
         agent = await self.task.agent_for(pm_agent_id)
         if agent is None or agent.role not in ("cell_pm", "main_pm"):
             return Envelope.not_authorized(
@@ -761,6 +869,34 @@ class Choreographer:
                 ),
                 context_briefing=await self._briefing_for(pm_agent_id, task_id),
             )
+        # Gate Set A: PM_CANNOT_EXECUTE_CODE — cell_pm/main_pm can only plan
+        # non-code tasks. role_typed_claim_guard is skipped here because
+        # i_will_plan only services PM roles, which fall into the PM-code
+        # branch. ALREADY_ACTIVE/PAUSED still apply.
+        guard = await self._run_claim_guards(
+            agent_id=pm_agent_id, task=t, skip_role_typed=True
+        )
+        if guard:
+            return self._with_briefing(
+                guard, await self._briefing_for(pm_agent_id, task_id)
+            )
+        return None
+
+    async def i_will_plan(
+        self, pm_agent_id: UUID, task_id: UUID, plan: str
+    ) -> Envelope:
+        """PM mirror of i_will_work_on for parent tasks.
+
+        Transitions a pending task owned (or claimable by) this PM into
+        in_progress with the supplied plan. Required before a PM can call
+        ``delegate`` to spawn subtasks.
+        """
+        t = await self.task.get(task_id)
+        if t is None:
+            return Envelope.not_found(message=f"task {task_id} not found")
+        rejection = await self._i_will_plan_preflight(pm_agent_id, task_id, t, plan)
+        if rejection is not None:
+            return rejection
 
         if t.assigned_to is None or t.assigned_to != pm_agent_id:
             t = await self.task.claim(pm_agent_id, task_id)

roboco/services/gateway/claim_guards.py

@@ -0,0 +1,183 @@
+"""Claim-time predicates restored from pre-gateway _helpers.py:124-204.
+
+These guards run BEFORE any task-status mutation in the claim verbs
+(``i_will_work_on``, ``i_will_plan``, ``claim_review``, ``claim_doc_task``).
+Each predicate returns a rejection ``Envelope`` if it fires; ``None`` if it
+passes. The first non-None return short-circuits the claim.
+
+Pre-gateway location at commit 0c3d15a:
+    roboco/mcp/tasks/handlers/_helpers.py:124-204
+    roboco/mcp/tasks/handlers/claim.py:121-180  (sibling sequence)
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from roboco.services.gateway.envelope import Envelope
+
+if TYPE_CHECKING:
+    from uuid import UUID
+
+# Roles that may NOT claim a code task — pre-gateway _helpers.py:181-204.
+_PM_ROLES: frozenset[str] = frozenset({"cell_pm", "main_pm"})
+
+# Statuses that count as "still actively worked" — pre-gateway
+# _helpers.py:check_blocking_tasks 134-152.
+_ACTIVE_BLOCKING_STATUSES: frozenset[str] = frozenset(
+    {"claimed", "in_progress", "verifying"}
+)
+
+# Terminal statuses that satisfy the sibling-sequence check —
+# pre-gateway claim.py:153.
+_TERMINAL_STATUSES: frozenset[str] = frozenset({"completed", "cancelled"})
+
+# Allowed task_types per role for the claim verb. Mirrors the role-typed
+# claim policy: developers do code-like work; QA reviews; documenters
+# document.  PMs cannot claim code (see pm_cannot_execute_code).
+_ROLE_TASK_TYPE_ALLOW: dict[str, frozenset[str]] = {
+    "developer": frozenset(
+        {"code", "research", "design"}
+    ),
+    "qa": frozenset(),  # QA never enters via i_will_work_on
+    "documenter": frozenset(),  # Doc never enters via i_will_work_on
+}
+
+
+def already_active_guard(
+    in_progress_tasks: list[Any], target_task_id: UUID
+) -> Envelope | None:
+    """Refuse claim if agent has any in_progress task other than this one.
+
+    Pre-gateway: _helpers.py:check_blocking_tasks 134-152.
+    """
+    blocking = [
+        t
+        for t in in_progress_tasks
+        if str(t.status) in _ACTIVE_BLOCKING_STATUSES and t.id != target_task_id
+    ]
+    if not blocking:
+        return None
+    blocker = blocking[0]
+    return Envelope.invalid_state(
+        message=(
+            f"You have a {blocker.status} task ({blocker.id}); "
+            "finish or pause it before claiming new work."
+        ),
+        remediate=(
+            f"finish or pause {blocker.id} first via i_am_done(...) "
+            "or i_am_idle()"
+        ),
+    )
+
+
+def paused_tasks_guard(paused_tasks: list[Any]) -> Envelope | None:
+    """Refuse claim if agent has any paused tasks.
+
+    Pre-gateway: _helpers.py:check_paused_tasks 154-165.
+    """
+    if not paused_tasks:
+        return None
+    paused = paused_tasks[0]
+    return Envelope.invalid_state(
+        message=(
+            f"You have {len(paused_tasks)} paused task(s); resume before "
+            "claiming new work."
+        ),
+        remediate=(
+            f"resume {paused.id} (call i_will_work_on again) before starting "
+            "new work"
+        ),
+    )
+
+
+def pm_cannot_execute_code_guard(role: str, task_type: str) -> Envelope | None:
+    """Refuse cell_pm/main_pm from claiming a code task.
+
+    Pre-gateway: _helpers.py:_guard_pm_from_code_tasks 181-204.
+    """
+    if role not in _PM_ROLES:
+        return None
+    if task_type != "code":
+        return None
+    nice_role = role.replace("_", " ").title()
+    return Envelope.not_authorized(
+        message=(
+            f"{nice_role} cannot claim code tasks. "
+            "PMs coordinate, never execute code."
+        ),
+        remediate=(
+            "PMs coordinate, never execute code. Delegate this to a "
+            "developer in your cell via delegate(parent_task_id, "
+            "title=..., description=..., assigned_to='be-dev-1', "
+            "team='backend')."
+        ),
+    )
+
+
+def role_typed_claim_guard(role: str, task_type: str) -> Envelope | None:
+    """Refuse cross-role claim attempts (developer claiming doc/qa, etc).
+
+    Pre-gateway: _helpers.py:_CLAIMABLE_STATUSES + the per-role status mapping
+    at lines 144-150 plus the implicit task_type cohesion.  The pre-gateway
+    code routed by status; here we route by ``task_type`` because the verbs
+    already split by status (claim_review, claim_doc_task vs i_will_work_on).
+
+    Only runs for non-PM roles; PMs route through pm_cannot_execute_code_guard
+    and i_will_plan instead.
+    """
+    if role in _PM_ROLES:
+        return None
+    if role not in _ROLE_TASK_TYPE_ALLOW:
+        # Unknown roles default to developer-like — silently allowed; the
+        # service-layer enforcement catches misuse downstream.
+        return None
+    allowed = _ROLE_TASK_TYPE_ALLOW[role]
+    if task_type in allowed:
+        return None
+    return Envelope.not_authorized(
+        message=(
+            f"role {role!r} cannot claim a {task_type!r} task via "
+            "i_will_work_on"
+        ),
+        remediate=(
+            "developer claims code/research/design; qa uses claim_review; "
+            "documenter uses claim_doc_task"
+        ),
+    )
+
+
+def sibling_sequence_guard(
+    target_task: Any, siblings: list[Any]
+) -> Envelope | None:
+    """Refuse claim if any earlier-sequence sibling is non-terminal.
+
+    Pre-gateway: claim.py:_validate_sibling_sequence 121-180.
+
+    A task with sequence=N is blocked while any sibling with sequence<N is
+    not in (completed, cancelled). Tasks without a parent_task_id (root) or
+    sequence==0 (first in line) are always allowed.
+    """
+    parent_id = getattr(target_task, "parent_task_id", None)
+    if parent_id is None:
+        return None
+    my_sequence = getattr(target_task, "sequence", 0) or 0
+    if my_sequence == 0:
+        return None
+    for sib in siblings:
+        if sib.id == target_task.id:
+            continue
+        sib_seq = getattr(sib, "sequence", 0) or 0
+        sib_status = str(getattr(sib, "status", ""))
+        if sib_seq < my_sequence and sib_status not in _TERMINAL_STATUSES:
+            return Envelope.invalid_state(
+                message=(
+                    f"sequence {my_sequence} blocked: earlier sibling "
+                    f"{sib.id} (sequence {sib_seq}) is in {sib_status}"
+                ),
+                remediate=(
+                    f"wait for sibling {sib.id} (sequence {sib_seq}) to "
+                    "reach completed/cancelled before claiming this task"
+                ),
+            )
+    return None

tests/unit/gateway/test_choreographer_claim_guards.py

@@ -0,0 +1,559 @@
+"""Gate Set A: claim-time guards restored from pre-gateway _helpers.py:124-204.
+
+Predicates ported into Choreographer claim verbs:
+- SEQUENCE_ORDER_VIOLATION   (earlier sibling must be terminal)
+- ALREADY_ACTIVE              (no claim while in_progress task is open)
+- PAUSED_TASKS_EXIST          (no claim while paused tasks exist)
+- PM_CANNOT_EXECUTE_CODE      (cell_pm/main_pm cannot claim task_type=code)
+- ROLE_TYPED_CLAIM            (developer/qa/documenter cannot cross-claim)
+
+These mirror pre-gateway gates at commit 0c3d15a, file
+roboco/mcp/tasks/handlers/_helpers.py lines 124-204 plus
+roboco/mcp/tasks/handlers/claim.py:121-180 for the sibling sequence check.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+
+def _make_deps(**overrides: Any) -> ChoreographerDeps:
+    base: dict[str, Any] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    base.update(overrides)
+    repo = base["evidence_repo"]
+    for method in (
+        "list_unread_a2a",
+        "list_unread_mentions",
+        "list_pending_notifications",
+        "task_metadata_gaps",
+        "recent_team_activity",
+        "blockers_in_lane",
+        "journal_highlights_for_task",
+    ):
+        getattr(repo, method).return_value = []
+    return ChoreographerDeps(**base)
+
+
+def _task_svc_with(
+    target: MagicMock,
+    *,
+    role: str = "developer",
+    in_progress: list[MagicMock] | None = None,
+    paused: list[MagicMock] | None = None,
+    siblings: list[MagicMock] | None = None,
+) -> AsyncMock:
+    """Build a task service mock primed with the active-task and sibling lookups."""
+    task_svc = AsyncMock()
+    task_svc.get.return_value = target
+    task_svc.agent_for.return_value = MagicMock(role=role, team="backend")
+    task_svc.list_in_progress_for_agent.return_value = in_progress or []
+    task_svc.list_paused_for_agent.return_value = paused or []
+    task_svc.get_subtasks.return_value = siblings or []
+    return task_svc
+
+
+# ---------------------------------------------------------------------------
+# A.1 SEQUENCE_ORDER_VIOLATION
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_i_will_work_on_blocks_when_earlier_sibling_open() -> None:
+    """Sequence=2 cannot be claimed while sequence=1 sibling is still open."""
+    agent_id = uuid4()
+    parent_id = uuid4()
+    target_id = uuid4()
+    earlier_id = uuid4()
+    target = MagicMock(
+        id=target_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=parent_id,
+        sequence=2,
+        task_type="code",
+        team="backend",
+    )
+    earlier = MagicMock(
+        id=earlier_id,
+        status="in_progress",
+        sequence=1,
+        title="Earlier sibling",
+    )
+    later = MagicMock(
+        id=target_id,
+        status="pending",
+        sequence=2,
+    )
+    task_svc = _task_svc_with(target, siblings=[earlier, later])
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(agent_id, target_id, plan="x")
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+    assert "sequence" in body["message"].lower()
+    assert str(earlier_id) in body["remediate"]
+    task_svc.claim.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_i_will_work_on_allows_when_earlier_sibling_terminal() -> None:
+    """Earlier siblings completed/cancelled do not block."""
+    agent_id = uuid4()
+    parent_id = uuid4()
+    target_id = uuid4()
+    target = MagicMock(
+        id=target_id,
+        status="pending",
+        plan={"x": 1},
+        assigned_to=None,
+        parent_task_id=parent_id,
+        sequence=2,
+        task_type="code",
+        team="backend",
+    )
+    earlier_done = MagicMock(id=uuid4(), status="completed", sequence=1)
+    earlier_cancelled = MagicMock(id=uuid4(), status="cancelled", sequence=0)
+    self_row = MagicMock(id=target_id, status="pending", sequence=2)
+    task_svc = _task_svc_with(
+        target, siblings=[earlier_done, earlier_cancelled, self_row]
+    )
+    task_svc.claim.return_value = MagicMock(
+        id=target_id,
+        status="claimed",
+        plan={"x": 1},
+        assigned_to=agent_id,
+        task_type="code",
+    )
+    task_svc.start.return_value = MagicMock(
+        id=target_id, status="in_progress", plan={"x": 1}, assigned_to=agent_id
+    )
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(agent_id, target_id)
+    assert env.error is None
+    task_svc.claim.assert_awaited_once_with(agent_id, target_id)
+
+
+@pytest.mark.asyncio
+async def test_root_task_no_sequence_check() -> None:
+    """Root tasks (no parent) skip the sequence check entirely."""
+    agent_id = uuid4()
+    target_id = uuid4()
+    target = MagicMock(
+        id=target_id,
+        status="pending",
+        plan={"x": 1},
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=5,
+        task_type="code",
+        team="backend",
+    )
+    task_svc = _task_svc_with(target)
+    task_svc.claim.return_value = MagicMock(
+        id=target_id, status="claimed", plan={"x": 1}, assigned_to=agent_id
+    )
+    task_svc.start.return_value = MagicMock(
+        id=target_id, status="in_progress", plan={"x": 1}, assigned_to=agent_id
+    )
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(agent_id, target_id)
+    assert env.error is None
+    # Sequence check should not have queried siblings on a root task
+    task_svc.get_subtasks.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# A.2 ALREADY_ACTIVE
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_i_will_work_on_blocks_when_agent_has_in_progress_task() -> None:
+    agent_id = uuid4()
+    target_id = uuid4()
+    other_id = uuid4()
+    target = MagicMock(
+        id=target_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+    )
+    in_progress = MagicMock(id=other_id, status="in_progress")
+    task_svc = _task_svc_with(target, in_progress=[in_progress])
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(agent_id, target_id, plan="x")
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+    assert str(other_id) in body["message"] or str(other_id) in body["remediate"]
+    assert "i_am_done" in body["remediate"] or "i_am_idle" in body["remediate"]
+    task_svc.claim.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_i_will_work_on_resumption_does_not_self_block() -> None:
+    """Resuming a claimed task already owned must not trigger ALREADY_ACTIVE."""
+    agent_id = uuid4()
+    task_id = uuid4()
+    claimed = MagicMock(
+        id=task_id,
+        status="claimed",
+        plan={"x": 1},
+        assigned_to=agent_id,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+        branch_name="feature/backend/abc",
+    )
+    started = MagicMock(
+        id=task_id, status="in_progress", plan={"x": 1}, assigned_to=agent_id
+    )
+    task_svc = _task_svc_with(target=claimed)
+    # Even if there's an in_progress task with the SAME id, that's the resumption itself
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.start.return_value = started
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(agent_id, task_id)
+    assert env.error is None
+    task_svc.start.assert_awaited_once_with(agent_id, task_id)
+
+
+# ---------------------------------------------------------------------------
+# A.3 PAUSED_TASKS_EXIST
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_i_will_work_on_blocks_when_agent_has_paused_task() -> None:
+    agent_id = uuid4()
+    target_id = uuid4()
+    paused_id = uuid4()
+    target = MagicMock(
+        id=target_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+    )
+    paused = MagicMock(id=paused_id, status="paused")
+    task_svc = _task_svc_with(target, paused=[paused])
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(agent_id, target_id, plan="x")
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+    assert str(paused_id) in body["remediate"]
+    assert "resume" in body["remediate"].lower()
+    task_svc.claim.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# A.4 PM_CANNOT_EXECUTE_CODE
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_cell_pm_cannot_claim_code_task_via_i_will_work_on() -> None:
+    pm_id = uuid4()
+    task_id = uuid4()
+    target = MagicMock(
+        id=task_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+    )
+    task_svc = _task_svc_with(target, role="cell_pm")
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(pm_id, task_id, plan="x")
+    body = env.as_dict()
+    assert body["error"] == "not_authorized"
+    assert "PM" in body["message"] or "code" in body["message"].lower()
+    assert "delegate" in body["remediate"].lower() or "developer" in body[
+        "remediate"
+    ].lower()
+    task_svc.claim.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_main_pm_cannot_claim_code_task_via_i_will_work_on() -> None:
+    pm_id = uuid4()
+    task_id = uuid4()
+    target = MagicMock(
+        id=task_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+    )
+    task_svc = _task_svc_with(target, role="main_pm")
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(pm_id, task_id, plan="x")
+    body = env.as_dict()
+    assert body["error"] == "not_authorized"
+
+
+@pytest.mark.asyncio
+async def test_cell_pm_cannot_claim_code_task_via_i_will_plan() -> None:
+    pm_id = uuid4()
+    task_id = uuid4()
+    target = MagicMock(
+        id=task_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+    )
+    task_svc = _task_svc_with(target, role="cell_pm")
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_plan(pm_id, task_id, plan="x")
+    body = env.as_dict()
+    assert body["error"] == "not_authorized"
+    assert "code" in body["message"].lower() or "execute" in body["message"].lower()
+
+
+@pytest.mark.asyncio
+async def test_pm_can_plan_non_code_parent() -> None:
+    pm_id = uuid4()
+    task_id = uuid4()
+    target = MagicMock(
+        id=task_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="planning",
+        team="backend",
+    )
+    claimed = MagicMock(
+        id=task_id, status="claimed", plan=None, assigned_to=pm_id, task_type="planning"
+    )
+    started = MagicMock(
+        id=task_id,
+        status="in_progress",
+        plan={"text": "x"},
+        assigned_to=pm_id,
+        task_type="planning",
+    )
+    task_svc = _task_svc_with(target, role="cell_pm")
+    task_svc.claim.return_value = claimed
+    task_svc.set_plan.return_value = claimed
+    task_svc.start.return_value = started
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_plan(pm_id, task_id, plan="break it down")
+    assert env.error is None
+
+
+# ---------------------------------------------------------------------------
+# A.5 ROLE_TYPED_CLAIM (cross-role rejection)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_developer_cannot_claim_qa_status_task() -> None:
+    """Dev calling i_will_work_on on awaiting_qa task gets explicit rejection."""
+    dev_id = uuid4()
+    task_id = uuid4()
+    target = MagicMock(
+        id=task_id,
+        status="awaiting_qa",
+        plan={"x": 1},
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+    )
+    task_svc = _task_svc_with(target, role="developer")
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(dev_id, task_id)
+    body = env.as_dict()
+    # Pre-existing path returns invalid_state with status complaint
+    assert body["error"] == "invalid_state"
+
+
+@pytest.mark.asyncio
+async def test_qa_cannot_claim_code_task_via_claim_review() -> None:
+    """QA calling claim_review on non-awaiting_qa task is rejected by status check."""
+    qa_id = uuid4()
+    task_id = uuid4()
+    target = MagicMock(
+        id=task_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+    )
+    task_svc = _task_svc_with(target, role="qa")
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.claim_review(qa_id, task_id)
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+
+
+@pytest.mark.asyncio
+async def test_documenter_cannot_claim_code_task_via_claim_doc_task() -> None:
+    """Documenter calling claim_doc_task on non-awaiting-doc task is rejected."""
+    doc_id = uuid4()
+    task_id = uuid4()
+    target = MagicMock(
+        id=task_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+    )
+    task_svc = _task_svc_with(target, role="documenter")
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.claim_doc_task(doc_id, task_id)
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+
+
+@pytest.mark.asyncio
+async def test_non_developer_role_cannot_claim_via_i_will_work_on() -> None:
+    """Even if status would allow, a documenter calling i_will_work_on on pending
+    code task is blocked by role-typed claim gate."""
+    doc_id = uuid4()
+    task_id = uuid4()
+    target = MagicMock(
+        id=task_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+    )
+    task_svc = _task_svc_with(target, role="documenter")
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_work_on(doc_id, task_id, plan="x")
+    body = env.as_dict()
+    # Role-typed claim refuses with not_authorized
+    assert body["error"] == "not_authorized"
+    task_svc.claim.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# Claim review (QA) — A.2/A.3 mirror
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_claim_review_blocks_when_qa_has_in_progress_task() -> None:
+    qa_id = uuid4()
+    task_id = uuid4()
+    other_id = uuid4()
+    target = MagicMock(
+        id=task_id,
+        status="awaiting_qa",
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+        work_session_id=uuid4(),
+        branch_name="feature/backend/abc",
+    )
+    in_progress = MagicMock(id=other_id, status="in_progress")
+    task_svc = _task_svc_with(target, role="qa", in_progress=[in_progress])
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.claim_review(qa_id, task_id)
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+    assert "i_am_done" in body["remediate"] or "i_am_idle" in body["remediate"]
+    task_svc.qa_claim.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_claim_doc_task_blocks_when_documenter_has_paused_task() -> None:
+    doc_id = uuid4()
+    task_id = uuid4()
+    paused_id = uuid4()
+    target = MagicMock(
+        id=task_id,
+        status="awaiting_documentation",
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+        team="backend",
+        work_session_id=uuid4(),
+        branch_name="feature/backend/abc",
+    )
+    paused = MagicMock(id=paused_id, status="paused")
+    task_svc = _task_svc_with(target, role="documenter", paused=[paused])
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.claim_doc_task(doc_id, task_id)
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+    assert "resume" in body["remediate"].lower()
+    task_svc.doc_claim.assert_not_awaited()

tests/unit/gateway/test_choreographer_dev.py

@@ -89,12 +89,24 @@ async def test_give_me_work_returns_idle_when_no_work() -> None:
 async def test_i_will_work_on_pending_with_plan() -> None:
     agent_id = uuid4()
     task_id = uuid4()
-    pending_task = MagicMock(id=task_id, status="pending", plan=None, assigned_to=None)
+    pending_task = MagicMock(
+        id=task_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
+    )
     in_progress_task = MagicMock(
         id=task_id, status="in_progress", plan={"text": "do x"}, assigned_to=agent_id
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = pending_task
+    task_svc.agent_for.return_value = MagicMock(role="developer", team="backend")
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.list_paused_for_agent.return_value = []
+    task_svc.get_subtasks.return_value = []
     task_svc.claim.return_value = MagicMock(
         id=task_id, status="claimed", plan=None, assigned_to=agent_id
     )
@@ -123,9 +135,16 @@ async def test_i_will_work_on_pending_no_plan_returns_tracing_gap() -> None:
         plan=None,
         assigned_to=None,
         description="task description",
+        parent_task_id=None,
+        sequence=0,
+        task_type="code",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = pending_task
+    task_svc.agent_for.return_value = MagicMock(role="developer", team="backend")
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.list_paused_for_agent.return_value = []
+    task_svc.get_subtasks.return_value = []
     task_svc.claim.return_value = MagicMock(
         id=task_id, status="claimed", plan=None, assigned_to=agent_id
     )

tests/unit/gateway/test_choreographer_doc.py

@@ -57,6 +57,9 @@ async def test_claim_doc_task_returns_evidence() -> None:
     after = MagicMock(**{**t.__dict__, "assigned_to": doc_id})
     task_svc = AsyncMock()
     task_svc.get.return_value = t
+    task_svc.agent_for.return_value = MagicMock(role="documenter", team="backend")
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.list_paused_for_agent.return_value = []
     task_svc.doc_claim.return_value = after
     work_svc = AsyncMock()
     work_svc.files_changed.return_value = ["README.md"]

tests/unit/gateway/test_choreographer_pm_extras.py

@@ -53,14 +53,34 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps:
 async def test_i_will_plan_claims_starts_and_sets_plan() -> None:
     pm_id = uuid4()
     task_id = uuid4()
-    pending = MagicMock(id=task_id, status="pending", plan=None, assigned_to=None)
+    pending = MagicMock(
-    claimed = MagicMock(id=task_id, status="claimed", plan=None, assigned_to=pm_id)
+        id=task_id,
+        status="pending",
+        plan=None,
+        assigned_to=None,
+        task_type="planning",
+        parent_task_id=None,
+        sequence=0,
+    )
+    claimed = MagicMock(
+        id=task_id,
+        status="claimed",
+        plan=None,
+        assigned_to=pm_id,
+        task_type="planning",
+    )
     started = MagicMock(
-        id=task_id, status="in_progress", plan={"text": "x"}, assigned_to=pm_id
+        id=task_id,
+        status="in_progress",
+        plan={"text": "x"},
+        assigned_to=pm_id,
+        task_type="planning",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = pending
     task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.list_paused_for_agent.return_value = []
     task_svc.claim.return_value = claimed
     task_svc.set_plan.return_value = claimed
     task_svc.start.return_value = started

tests/unit/gateway/test_choreographer_qa.py

@@ -65,6 +65,9 @@ async def test_claim_review_returns_evidence_inline() -> None:
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = t_initial
+    task_svc.agent_for.return_value = MagicMock(role="qa", team="backend")
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.list_paused_for_agent.return_value = []
     task_svc.qa_claim.return_value = t_claimed
     work_svc = AsyncMock()
     work_svc.files_changed.return_value = ["README.md"]
@@ -119,6 +122,9 @@ async def test_claim_review_marks_evidence_inspected() -> None:
     t_claimed = MagicMock(**{**t.__dict__, "assigned_to": qa_id})
     task_svc = AsyncMock()
     task_svc.get.return_value = t
+    task_svc.agent_for.return_value = MagicMock(role="qa", team="backend")
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.list_paused_for_agent.return_value = []
     task_svc.qa_claim.return_value = t_claimed
     git_svc = AsyncMock()
     git_svc.diff.return_value = ""