roboco/tests/unit/gateway/test_choreographer_claim_guards.py

"""Gate Set A: claim-time guards restored from pre-gateway _helpers.py:124-204.

Predicates ported into Choreographer claim verbs:
- 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.
"""

from __future__ import annotations

from datetime import UTC, datetime
from typing import Any
from unittest.mock import AsyncMock, MagicMock
from uuid import uuid4

import pytest
from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
from roboco.services.gateway.claim_guards import paused_tasks_guard

# #172: a developer fresh claim must carry a substantive step checklist.
# Inert on re-entry/error/non-dev paths, so safe to pass everywhere.
_STEPS = [
    {
        "title": "Implement the change",
        "description": (
            "edit the target file, add tests, run them, and stage the "
            "change for commit on the task branch"
        ),
    }
]


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)
    # VerbRunner uses task.session.begin_nested() as a savepoint context
    # manager. AsyncMock auto-attributes any access (so hasattr always
    # returns True); we always overwrite session to a MagicMock with the
    # correct async-context-manager protocol.
    task = base["task"]
    task.session = MagicMock()
    task.session.begin_nested = MagicMock(
        return_value=MagicMock(
            __aenter__=AsyncMock(return_value=None),
            __aexit__=AsyncMock(return_value=False),
        )
    )
    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 = []
    # C8: default-fresh journal:decision so PM-decision gate passes.
    # Tests that exercise the gate boundary stub their own value.
    # The check matches MagicMock and AsyncMock (the two default sentinel
    # types pytest's unittest.mock leaves on un-stubbed return_values).
    _ldef = base["journal"].latest_decision_at.return_value
    if type(_ldef).__name__ in ("MagicMock", "AsyncMock"):
        base["journal"].latest_decision_at.return_value = datetime.now(UTC)
    return ChoreographerDeps(**base)


def _task_svc_with(
    target: MagicMock,
    *,
    role: str = "developer",
    agent_id: object | None = None,
    lookups: dict[str, list[MagicMock]] | None = None,
) -> AsyncMock:
    """Build a task service mock primed with the active-task and sibling lookups.

    `agent_id` (when supplied) is used as the GatewayAgentView's id so that
    runner-driven calls like ``task.claim(task.id, agent.id)`` line up
    with the test's assert_awaited_with(target_id, agent_id).

    `lookups` carries the optional in_progress / paused / siblings lists
    (defaulting empty). One bag avoids ruff PLR0913 on the helper sig.
    """
    lookups = lookups or {}
    task_svc = AsyncMock()
    task_svc.get.return_value = target
    task_svc.agent_for.return_value = MagicMock(
        id=agent_id, role=role, team="backend", slug=None
    )
    task_svc.list_in_progress_for_agent.return_value = lookups.get("in_progress", [])
    task_svc.list_paused_for_agent.return_value = lookups.get("paused", [])
    task_svc.get_subtasks.return_value = lookups.get("siblings", [])
    return task_svc


# ---------------------------------------------------------------------------
# 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, lookups={"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", steps=_STEPS)
    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, agent_id=agent_id)
    # 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, steps=_STEPS)
    assert env.error is None
    task_svc.start.assert_awaited_once_with(task_id, agent_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, lookups={"paused": [paused]})
    deps = _make_deps(task=task_svc)
    c = Choreographer(deps)

    env = await c.i_will_work_on(agent_id, target_id, plan="x", steps=_STEPS)
    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", steps=_STEPS)
    body = env.as_dict()
    assert body["error"] == "not_authorized"
    # Spec produces "role 'cell_pm' may not call 'i_will_work_on'".
    assert "cell_pm" in body["message"]
    assert "i_will_work_on" in body["message"]
    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", steps=_STEPS)
    body = env.as_dict()
    assert body["error"] == "not_authorized"


@pytest.mark.asyncio
async def test_cell_pm_can_plan_code_typed_parent_via_i_will_plan() -> None:
    """Rule change (2026-05-08): the PM-cannot-execute-code guard belongs
    on `i_will_work_on` (the EXECUTION verb), not on `i_will_plan` (the
    PLANNING verb). PMs decompose code-typed parent tasks into
    developer-claimable subtasks all the time; that's planning, not
    executing. Pre-fix this rejection deadlocked every code-typed parent
    in the smoke test (see PRE_GATEWAY_LIFECYCLE.md and the 2026-05-08
    audit-log analysis).
    """
    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="Decompose into 2 dev subtasks.",
        rich_plan={
            "approach": (
                "Split the code-typed parent into two developer-claimable "
                "subtasks: one for API implementation, one for test coverage. "
                "Sequenced so the API lands first; QA reviews each PR after "
                "it opens, documentation follows, then complete and submit "
                "up. No cross-cell dependencies for this slice."
            ),
            "sub_tasks": [
                {
                    "title": "API subtask",
                    "description": (
                        "be-dev-1 implements the endpoint with tests, commits "
                        "with the task-id prefix, opens the leaf PR for QA."
                    ),
                },
            ],
        },
    )
    body = env.as_dict()
    # The PM-cannot-execute-code rejection must NOT fire on i_will_plan.
    assert body.get("error") != "not_authorized", (
        f"i_will_plan was rejected for a code-typed parent; envelope: {body}"
    )


@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",
        rich_plan={
            "approach": (
                "Single-cell decomposition: backend handles the full scope; "
                "no frontend or ux work required for this planning task. "
                "be-dev-1 owns the change end to end; QA reviews after the "
                "PR opens, documentation follows, then be-pm completes and "
                "submits up. Strict sequencing, no cross-cell dependencies."
            ),
            "sub_tasks": [
                {
                    "title": "Backend planning slice",
                    "description": (
                        "scope the change, assign be-dev-1, who implements "
                        "with tests and opens the leaf PR for QA review."
                    ),
                }
            ],
        },
    )
    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.

    spec.CLAIM_RULES restricts DEVELOPER to PENDING/NEEDS_REVISION; an
    awaiting_qa task is reserved for QA. spec.can_claim surfaces this as
    not_authorized (status reserved for another role) which the verb
    relays via Envelope.from_decision. Pre-spec the verb body produced
    invalid_state from a custom else-branch; the spec-driven body now
    returns the more accurate not_authorized.
    """
    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, steps=_STEPS)
    body = env.as_dict()
    assert body["error"] == "not_authorized"
    assert "developer" in body["message"]
    assert "awaiting_qa" in body["message"]


@pytest.mark.asyncio
async def test_qa_cannot_claim_code_task_via_claim_review() -> None:
    """QA calling claim_review on PENDING task is rejected by claim-rules.

    QA's CLAIM_RULES is {AWAITING_QA}. PENDING is owned by dev/pm — so
    ``_check_claim_rules_narrow`` returns ``not_authorized`` (the
    "other_role_owns_status" branch). Pre-spec the verb body's status
    pre-check returned invalid_state; post-migration the spec gate
    drives the rejection kind.
    """
    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",
        quick_context=None,
    )
    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"] == "not_authorized"


@pytest.mark.asyncio
async def test_documenter_cannot_claim_code_task_via_claim_doc_task() -> None:
    """Documenter calling claim_doc_task on AWAITING_QA task is rejected by claim-rules.

    Documenter's CLAIM_RULES is {PENDING, AWAITING_DOCUMENTATION}. A
    documenter calling claim_doc_task on AWAITING_QA hits the
    "other_role_owns_status" branch and returns ``not_authorized``.
    Pre-spec the verb body returned invalid_state on the status check;
    post-migration the spec gate drives the rejection kind.
    """
    doc_id = uuid4()
    task_id = uuid4()
    target = MagicMock(
        id=task_id,
        status="awaiting_qa",
        plan=None,
        assigned_to=None,
        parent_task_id=None,
        sequence=0,
        task_type="code",
        team="backend",
        quick_context=None,
    )
    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"] == "not_authorized"


@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", steps=_STEPS)
    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", lookups={"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", lookups={"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()


# ---------------------------------------------------------------------------
# Coordinator exemption — a PM plans + delegates many roots in parallel, so the
# single-active-task guards (already_active / paused) must NOT gate it; only a
# real upstream sequence dependency may hold a PM's root back. (Developers stay
# blocked — see the A.2 / A.3 tests above.)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_main_pm_can_plan_second_root_despite_active_and_paused() -> None:
    """A main_pm may plan a new root while it already holds one in_progress and
    one paused root — the developer concurrency guards do not gate a coordinator.
    """
    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="main_pm",
    )
    claimed = MagicMock(
        id=task_id, status="claimed", plan=None, assigned_to=pm_id, task_type="code"
    )
    started = MagicMock(
        id=task_id,
        status="in_progress",
        plan={"text": "x"},
        assigned_to=pm_id,
        task_type="code",
    )
    # The coordinator already holds one in_progress root and one paused root —
    # both would trip the guards for a non-PM caller.
    other_active = MagicMock(id=uuid4(), status="in_progress")
    other_paused = MagicMock(id=uuid4(), status="paused")
    task_svc = _task_svc_with(
        target,
        role="main_pm",
        agent_id=pm_id,
        lookups={"in_progress": [other_active], "paused": [other_paused]},
    )
    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="route to backend + frontend cells",
        rich_plan={
            "approach": (
                "Route this root to the backend and frontend cells in parallel: "
                "be-pm owns the API contract, fe-pm consumes it. No cross-cell "
                "dependency for this slice, so both cells start at once."
            ),
            "sub_tasks": [
                {
                    "title": "Backend slice",
                    "description": (
                        "be-pm decomposes the API change and assigns be-dev-1, "
                        "who implements with tests and opens the leaf PR for QA."
                    ),
                }
            ],
        },
    )
    assert env.error is None, env.as_dict()
    task_svc.start.assert_awaited()


@pytest.mark.asyncio
async def test_main_pm_recovers_claimed_root_with_paused_sibling() -> None:
    """The live deadlock: a respawned main_pm re-enters i_will_plan on a stuck
    `claimed` root while another root is paused (i_am_idle auto-paused it). The
    paused guard must NOT block the coordinator's recovery — set_plan + start
    runs and the root reaches in_progress.
    """
    pm_id = uuid4()
    task_id = uuid4()
    claimed = MagicMock(
        id=task_id,
        status="claimed",
        plan=None,
        assigned_to=pm_id,
        parent_task_id=None,
        sequence=0,
        task_type="code",
        team="main_pm",
        branch_name="feature/main_pm/abc",
    )
    started = MagicMock(
        id=task_id,
        status="in_progress",
        plan={"text": "x"},
        assigned_to=pm_id,
        task_type="code",
    )
    other_paused = MagicMock(id=uuid4(), status="paused")
    task_svc = _task_svc_with(
        claimed, role="main_pm", agent_id=pm_id, lookups={"paused": [other_paused]}
    )
    task_svc.set_plan.return_value = started
    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="route this root to the backend + frontend cells"
    )
    assert env.error is None, env.as_dict()
    task_svc.start.assert_awaited_once_with(task_id, pm_id)


def test_paused_tasks_guard_excludes_target() -> None:
    """A paused task that IS the claim target must not self-block, mirroring
    already_active_guard's target exclusion (the 2026-06-14 self-deadlock)."""
    target_id = uuid4()
    other_id = uuid4()
    # Only the target itself is paused -> no block.
    assert paused_tasks_guard([MagicMock(id=target_id)], target_id) is None
    # A different paused task -> block, naming that task.
    env = paused_tasks_guard([MagicMock(id=other_id)], target_id)
    assert env is not None
    assert str(other_id) in env.as_dict()["remediate"]
    # Back-compat: with no target supplied, any paused task blocks.
    assert paused_tasks_guard([MagicMock(id=other_id)]) is not None