From 5a5e2b5fa05c8a3e28ee98f7ffa3011e833e8523 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 22:46:05 +0200 Subject: [PATCH] [F114] single-claimant guard on pr_gate_claim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- roboco/services/task.py | 40 ++++++ tests/integration/test_task_service_basics.py | 127 ++++++++++++++++++ 2 files changed, 167 insertions(+) diff --git a/roboco/services/task.py b/roboco/services/task.py index 06d26018..731f84c1 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -7870,7 +7870,47 @@ class TaskService(BaseService): The in-path PR-review gate mirrors QA's claim_review: status stays at awaiting_pr_review while the reviewer inspects the assembled diff; pr_pass / pr_fail perform the transition. + + Single-claimant guard: two reviewers race-claiming the same gate task + would otherwise overwrite the first's claim (last-write-wins) and the + first reviewer's subsequent pr_pass / pr_fail would actor-mismatch + against the new owner — wasting a review cycle. The guard refuses a + second reviewer 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. It does this by checking the existing claimant's + ROLE: only a PR-reviewer active claimant is a competing review claim. + The row is locked ``FOR UPDATE`` so concurrent claim attempts serialize + at the DB level (the second claim sees the first's committed claim), + mirroring the dev ``claim`` path. A re-claim by the SAME reviewer is + idempotent (the ``existing != reviewer_agent_id`` check skips the + guard). """ + lock_result = await self.session.execute( + select(TaskTable) + .where(TaskTable.id == task_id) + .with_for_update(of=TaskTable) + ) + task = lock_result.scalar_one_or_none() + if task is None or task.status != TaskStatus.AWAITING_PR_REVIEW: + return None + existing = to_python_uuid(task.active_claimant_id) + if existing is not None and existing != reviewer_agent_id: + existing_agent = await self.agent_for(existing) + if ( + existing_agent is not None + and existing_agent.role == AgentRole.PR_REVIEWER.value + ): + self.log.warning( + "pr_gate_claim rejected - gate task already claimed by" + " another reviewer", + task_id=str(task_id), + existing_claimant=str(existing), + requesting_reviewer=str(reviewer_agent_id), + ) + return None return await self._qa_or_doc_claim( reviewer_agent_id, task_id, TaskStatus.AWAITING_PR_REVIEW ) diff --git a/tests/integration/test_task_service_basics.py b/tests/integration/test_task_service_basics.py index 8928ff81..65500095 100644 --- a/tests/integration/test_task_service_basics.py +++ b/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) # ---------------------------------------------------------------------------