[F114] single-claimant guard on pr_gate_claim

pr_gate_claim delegated straight to _qa_or_doc_claim, which overwrites
claimed_by / active_claimant_id with no single-claimant check. Two
reviewers race-claiming the same awaiting_pr_review task would
last-write-wins overwrite the first claim, and the first reviewer's
subsequent pr_pass / pr_fail would actor-mismatch against the new owner
(wasting a review cycle). The orchestrator's gate dispatcher already
prevents double-reviewer-dispatch in normal flow (one task -> one team
-> one reviewer + is_agent_active + per-tick spawned set), so the race
is only reachable via direct concurrent API calls (defense-in-depth).

Add a role-aware single-claimant guard in pr_gate_claim: lock the row
FOR UPDATE (serialize concurrent claims, mirroring the dev claim path),
then refuse only when the task is already actively claimed by a
DIFFERENT PR-reviewer. The gate task is owned by the PM at entry
(submit_for_review does not clear ownership, unlike submit_for_qa), so
the guard must distinguish a PM/dev owner — which the first reviewer
legitimately overclaims — from a competing reviewer claim; checking the
existing claimant's role (pr_reviewer) does exactly that. A re-claim by
the same reviewer is idempotent (skipped by the != check). The gateway
claim_gate_review handler already maps a None return to a clean
invalid_state envelope ('it may already be claimed; give_me_work for
the next'), so no gateway change is needed.

TDD: 3 integration tests in test_task_service_basics.py — reject a second
reviewer race-claim (returns None, first claim intact), allow the first
reviewer when the PM owns the root (regression guard for the
PM-owns-at-entry model), idempotent re-claim by the same reviewer.
Confirmed the reject test RED first (race-claim succeeded, overwriting
reviewer1).

tests/integration/test_task_service_basics.py

@@ -1417,6 +1417,133 @@ async def test_doc_claim_returns_none_for_missing(task_setup: dict) -> None:
     assert await svc.doc_claim(doc_agent_id=uuid4(), task_id=uuid4()) is None
 
 
+# ---------------------------------------------------------------------------
+# pr_gate_claim — single-claimant guard (F114)
+# ---------------------------------------------------------------------------
+
+
+def _reviewer(name: str) -> AgentTable:
+    return AgentTable(
+        id=uuid4(),
+        name=name,
+        slug=f"pr-reviewer-{uuid4().hex[:8]}",
+        role=AgentRole.PR_REVIEWER,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="reviewer",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+
+
+def _pm(name: str) -> AgentTable:
+    return AgentTable(
+        id=uuid4(),
+        name=name,
+        slug=f"be-pm-{uuid4().hex[:8]}",
+        role=AgentRole.CELL_PM,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="pm",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+
+
+async def _gate_task(task_setup: dict, db_session: AsyncSession) -> Any:
+    """A task sitting in awaiting_pr_review, owned by ``owner``."""
+    svc = task_setup["svc"]
+    task = await svc.create(_req(task_setup))
+    task.status = TaskStatus.AWAITING_PR_REVIEW
+    await db_session.flush()
+    return task
+
+
+@pytest.mark.asyncio
+async def test_pr_gate_claim_rejects_second_reviewer_race(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """F114: a second PR-reviewer race-claiming a gate task already claimed by a
+    reviewer must be refused (last-write-wins would otherwise overwrite the
+    first reviewer's claim and the first reviewer's pr_pass/pr_fail would
+    actor-mismatch)."""
+    svc = task_setup["svc"]
+    reviewer1 = _reviewer("R1")
+    reviewer2 = _reviewer("R2")
+    db_session.add_all([reviewer1, reviewer2])
+    await db_session.flush()
+    task = await _gate_task(task_setup, db_session)
+    # reviewer1 already claimed the gate task.
+    task.active_claimant_id = reviewer1.id
+    task.claimed_by = reviewer1.id
+    task.assigned_to = reviewer1.id
+    await db_session.flush()
+
+    rejected = await svc.pr_gate_claim(reviewer2.id, task.id)
+    # Refused, not silently stolen.
+    assert rejected is None
+    await db_session.refresh(task)
+    # The first reviewer's claim is intact (NOT overwritten by reviewer2).
+    assert task.active_claimant_id == reviewer1.id
+    assert task.claimed_by == reviewer1.id
+    assert task.assigned_to == reviewer1.id
+
+
+@pytest.mark.asyncio
+async def test_pr_gate_claim_allows_first_reviewer_when_pm_owns_root(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """F114 regression guard: the gate task is owned by the PM at entry
+    (submit_for_review does not clear ownership), so the FIRST reviewer must
+    still be allowed to claim — the guard only rejects a competing REVIEWER
+    claim, not the PM owner."""
+    svc = task_setup["svc"]
+    pm = _pm("PM")
+    reviewer = _reviewer("R")
+    db_session.add_all([pm, reviewer])
+    await db_session.flush()
+    task = await _gate_task(task_setup, db_session)
+    # The PM owns the coordination root at gate entry.
+    task.active_claimant_id = pm.id
+    task.claimed_by = pm.id
+    task.assigned_to = pm.id
+    await db_session.flush()
+
+    claimed = await svc.pr_gate_claim(reviewer.id, task.id)
+    assert claimed is not None
+    await db_session.refresh(task)
+    # The reviewer took over the gate (the legitimate overclaim).
+    assert task.active_claimant_id == reviewer.id
+    assert task.claimed_by == reviewer.id
+    assert task.assigned_to == reviewer.id
+
+
+@pytest.mark.asyncio
+async def test_pr_gate_claim_idempotent_for_same_reviewer(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """F114: a reviewer re-claiming its OWN gate claim is idempotent (allowed),
+    not rejected — the guard only refuses a DIFFERENT reviewer."""
+    svc = task_setup["svc"]
+    reviewer = _reviewer("R")
+    db_session.add(reviewer)
+    await db_session.flush()
+    task = await _gate_task(task_setup, db_session)
+    task.active_claimant_id = reviewer.id
+    task.claimed_by = reviewer.id
+    task.assigned_to = reviewer.id
+    await db_session.flush()
+
+    claimed = await svc.pr_gate_claim(reviewer.id, task.id)
+    assert claimed is not None
+    await db_session.refresh(task)
+    assert task.active_claimant_id == reviewer.id
+
+
 # ---------------------------------------------------------------------------
 # qa_pass / qa_fail / cell_pm_complete (404 paths)
 # ---------------------------------------------------------------------------