diff --git a/roboco/api/routes/tasks.py b/roboco/api/routes/tasks.py index 4dde923b..786ee1e2 100644 --- a/roboco/api/routes/tasks.py +++ b/roboco/api/routes/tasks.py @@ -1615,6 +1615,44 @@ async def unblock_task( return task_to_response(task) +@router.post("/{task_id}/assign-review-pm", response_model=TaskResponse) +@guard_deco.rate_limit(requests=30, window=60) +async def assign_review_pm( + task_id: UUID, + db: DbSession, + agent: CurrentAgentContext, + permissions: PermissionServiceDep, +) -> TaskResponse: + """Recovery seam: (re)assign an awaiting_pm_review task to its owning PM. + + CLAIM_RULES deliberately excludes AWAITING_PM_REVIEW — no claim() edge + into it (the i_will_plan re-claim-loop fix) — so the orchestrator's + pm-review dispatchers call this instead of the claim route to place an + unassigned or stale-assigned review task with its real owner before + spawning it. Gated on the same ASSIGN permission that guards ownership + fields on the generic task PATCH. + """ + service = get_task_service(db) + task = await service.get(task_id) + if not task: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Task not found" + ) + if not permissions.can_perform_task_action(agent, TaskAction.ASSIGN, task.team): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Not authorized to assign this task", + ) + task = await service.assign_review_pm(task_id) + if not task: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Cannot assign review PM - task not awaiting_pm_review", + ) + await db.commit() + return task_to_response(task) + + @router.post("/{task_id}/pause", response_model=TaskResponse) @guard_deco.rate_limit(requests=30, window=60) async def pause_task( diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index c635950c..ea8743f6 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -11412,6 +11412,46 @@ Start by: logger.error("Claim task error", task_id=task_id, error=str(e)) return False + async def _ensure_review_pm_assigned( + self, client: httpx.AsyncClient, task: dict[str, Any] + ) -> str | None: + """POST assign-review-pm; return the resulting owner's agent slug. + + CLAIM_RULES has no claim() edge into AWAITING_PM_REVIEW (the + i_will_plan re-claim-loop fix), so an unassigned task (pr_pass + resolved no PM) or a stale one (an escalate/unblock(restore=True) + round trip restores status but not ownership — see the pr_pass + ownership-clearing fix) needs this instead of ``_claim_task_for_agent``. + + ``None`` on ANY rejection or transport error — this reports the + route's own outcome only and does NOT fall back to the task's own + (possibly stale) ``assigned_to``. A blind fallback here would let a + transient failure silently clobber a caller's independently-known- + correct default (``_maybe_spawn_pm_closure``'s team-resolved + ``pm_id`` — see ``_closure_review_pm``); callers that have no better + default than the task's own ``assigned_to`` (``_dispatch_pm_review_ + work`` — see ``_review_pm_slug``) apply that fallback themselves. + """ + try: + resp = await client.post( + f"{self._api_url}/tasks/{task['id']}/assign-review-pm" + ) + if resp.status_code == http_status.HTTP_200_OK: + assigned_to = resp.json().get("assigned_to") + return self._resolve_agent_slug(assigned_to) if assigned_to else None + logger.warning( + "assign-review-pm rejected", + task_id=task.get("id"), + status=resp.status_code, + ) + except Exception as e: + logger.error( + "assign-review-pm error", + task_id=task.get("id"), + error=str(e), + ) + return None + async def _fetch_tasks( self, client: httpx.AsyncClient, @@ -14396,6 +14436,8 @@ Start now: evidence(task_id="{task_id}") if handled: return + pm_id = await self._closure_review_pm(client, task, pm_id) + prompt = self._build_pm_closure_prompt( task, descendants, auto_submit_reason=auto_submit_reason ) @@ -14407,6 +14449,28 @@ Start now: evidence(task_id="{task_id}") spawned_by="_maybe_spawn_pm_closure", ) + async def _closure_review_pm( + self, client: httpx.AsyncClient, task: dict[str, Any], default_pm: str + ) -> str: + """The PM to spawn for a closure parent already at awaiting_pm_review. + + CLAIM_RULES has no claim() edge into this status, so pr_pass's/ + mark_pr_created's own PM resolution, or a stale escalate/ + unblock(restore=True) round trip, may have left assigned_to wrong + or unset — assign_review_pm corrects it. ``default_pm`` (the + caller's already-correct ``_closure_pm_for_team`` value) is kept on + ANY failure: ``_ensure_review_pm_assigned`` returns ``None`` on both + an unresolvable owner and a transient route/transport error, and + adopting a stale fallback there would spawn the wrong PM in exactly + the incident shape this fix targets (a task mis-assigned to main-pm + that should be be-pm). A no-op for any other status — ownership + there already came from the normal claim/delegate flow. + """ + if task.get("status") != "awaiting_pm_review": + return default_pm + resolved = await self._ensure_review_pm_assigned(client, task) + return resolved or default_pm + async def _dispatch_pm_closure_work(self, client: httpx.AsyncClient) -> None: """ Dispatch PM closure work - check parent tasks ready to close. @@ -15331,6 +15395,31 @@ Never `commit`, never write code, never run `git`. PMs coordinate. and getattr(sib, "status", None) not in terminal ) + async def _review_pm_slug( + self, client: httpx.AsyncClient, task: dict[str, Any] + ) -> str | None: + """Owning PM slug for an awaiting_pm_review task. + + Skips the assign-review-pm round trip (a row lock + internal HTTP + call) when the task's own ``assigned_to`` already matches the + team-resolved owner (the same ``_closure_pm_for_team`` mapping) — + every review task hitting that route on every tick even when + already correct is needless DB-lock pressure on a stack with + documented lock-contention incidents (#721/#726). The route stays + authoritative for the mismatch/unassigned case (it re-resolves + under its own row lock). Falls back to the task's own (possibly + stale) ``assigned_to`` when the route itself fails/rejects — unlike + ``_closure_review_pm``, there is no independently-known-better + default here to protect. + """ + expected = self._closure_pm_for_team(task.get("team")) + current = task.get("assigned_to") + current_slug = self._resolve_agent_slug(current) if current else None + if current_slug == expected: + return expected + resolved = await self._ensure_review_pm_assigned(client, task) + return resolved or current_slug + async def _dispatch_pm_review_work(self, client: httpx.AsyncClient) -> None: """ Dispatch PM review work to cell PMs or Main PM. @@ -15341,74 +15430,46 @@ Never `commit`, never write code, never run `git`. PMs coordinate. tasks = await self._fetch_tasks(client, "awaiting_pm_review") for task in tasks: - team = task.get("team") - assigned_to = task.get("assigned_to") - # Sequence-ordered merge: don't review/merge a leaf until its # earlier same-team siblings have landed, so they merge into the # shared cell branch in order instead of racing and wedging. if await self._blocked_by_earlier_sibling(task): continue - # If already assigned, check if that agent is running - if assigned_to: - assigned_slug = self._resolve_agent_slug(assigned_to) - # Human-only roles (CEO / prompter / secretary) are never - # containers — there is no reviewer agent to respawn. Leave - # the task for the human (the CEO approves via the panel). - # A stale/ex-human slug is also skipped: is_spawnable_agent_slug - # is False for it, so a renamed secretary slug can't slip past - # the layered guard to a doomed spawn (#49). Mirrors the - # spawn_agent human-role guard; a skip here keeps a mis-assigned - # human task from aborting this dispatcher's whole tick. - if not is_spawnable_agent_slug(assigned_slug): - continue - if self._is_agent_active(assigned_slug): - continue - # Loop guard: a review task that keeps re-surfacing without - # advancing (e.g. an unmergeable PR that re-blocks every cycle) - # must stop respawning the reviewer, else it burns tokens - # forever. The gate notifies the CEO once it trips. - if await self._pm_respawn_should_gate(assigned_slug, task): - continue - # Agent not running - spawn them to continue - await self.spawn_agent( - agent_id=assigned_slug, - task_id=task["id"], - initial_prompt=self._build_pm_review_prompt(task), - git_context=self._task_git_context(task), - spawned_by="_dispatch_pm_review_work", - ) - continue - - # Unassigned task - select PM based on team - # Cell tasks go to Cell PM, cross-cell/main_pm tasks go to Main PM - if team in ["backend", "frontend", "ux_ui"]: - pm_id = self._TEAM_PM_MAP.get(team, "be-pm") - else: - # main_pm, board, or no team → Main PM handles it - pm_id = "main-pm" - - if self._is_agent_active(pm_id): - continue - - # Claim the task for PM BEFORE spawning - if not await self._claim_task_for_agent(client, task["id"], pm_id): + assigned_slug = await self._review_pm_slug(client, task) + if not assigned_slug: logger.warning( - "Failed to claim awaiting_pm_review task for PM", + "Could not resolve/assign an owning PM for awaiting_pm_review task", task_id=task["id"], - agent_id=pm_id, ) continue + # Human-only roles (CEO / prompter / secretary) are never + # containers — there is no reviewer agent to respawn. Leave + # the task for the human (the CEO approves via the panel). + # A stale/ex-human slug is also skipped: is_spawnable_agent_slug + # is False for it, so a renamed secretary slug can't slip past + # the layered guard to a doomed spawn (#49). Mirrors the + # spawn_agent human-role guard; a skip here keeps a mis-assigned + # human task from aborting this dispatcher's whole tick. + if not is_spawnable_agent_slug(assigned_slug): + continue + if self._is_agent_active(assigned_slug): + continue + # Loop guard: a review task that keeps re-surfacing without + # advancing (e.g. an unmergeable PR that re-blocks every cycle) + # must stop respawning the reviewer, else it burns tokens + # forever. The gate notifies the CEO once it trips. + if await self._pm_respawn_should_gate(assigned_slug, task): + continue + # Agent not running - spawn them to continue await self.spawn_agent( - agent_id=pm_id, + agent_id=assigned_slug, task_id=task["id"], initial_prompt=self._build_pm_review_prompt(task), git_context=self._task_git_context(task), spawned_by="_dispatch_pm_review_work", ) - break async def _dispatch_marketing_work(self, client: httpx.AsyncClient) -> None: """ diff --git a/roboco/services/task.py b/roboco/services/task.py index 8a70e67f..312039c3 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -6646,20 +6646,33 @@ class TaskService(BaseService): # by submit_for_qa's handoff), so the explicit audit_agent_id from # the caller wins. captured_dev_id = audit_agent_id or to_python_uuid(task.claimed_by) + # Capture the prior claimant (the dev) before it's overwritten + # below, mirroring _maybe_advance_to_pm_review's own capture — + # this is the mirror-image entry into the same parallel-completion + # transition (pr_created arriving second instead of docs_complete). + prior_claimant = to_python_uuid(task.active_claimant_id) self._validate_and_set_status( task, TaskStatus.AWAITING_PM_REVIEW, "developer", audit_agent_id=captured_dev_id, ) - # Clear assignment so PM can claim the task for review - task.assigned_to = None - task.claimed_by = None + # Assign to the owning PM (walks the parent chain — see + # _resolve_pm_for_review) instead of clearing to None: CLAIM_RULES + # has no claim() edge into AWAITING_PM_REVIEW, so an unassigned + # task here has no way back to a PM (the pr_pass ownership- + # clearing wedge, same bug class, same fix). + owning_pm = await self._resolve_pm_for_review(task) + task.assigned_to = cast("Any", owning_pm) if owning_pm else None + task.claimed_by = cast("Any", owning_pm) if owning_pm else None + task.active_claimant_id = cast("Any", owning_pm) if owning_pm else None + await self._clear_agent_current_task(prior_claimant, task_id) self.log.info( "PR created, awaiting PM review", task_id=str(task_id), pr_number=pr_number, docs_complete=task.docs_complete, + routed_to_pm=str(owning_pm) if owning_pm else None, ) else: # PR created but docs not yet complete @@ -11301,9 +11314,15 @@ class TaskService(BaseService): ) -> TaskTable | None: """Reviewer passes the gate: awaiting_pr_review -> awaiting_pm_review. - Clears the claim so the PM-closure dispatcher routes the now-unassigned - task to the owning PM to merge — the same path a leaf takes after - docs_complete. Mirrors qa_pass. + Hands off to the owning PM (cell PM for a cell team, else the Main + PM — the same resolution ``pr_fail`` uses) instead of clearing + ownership: CLAIM_RULES deliberately excludes AWAITING_PM_REVIEW (no + claim() edge into it — the i_will_plan re-claim-loop fix), so an + unassigned task here has no way back to a PM and wedges the closure + dispatcher's complete() ownership guard (assigned_to != pm_agent_id) + forever. active_claimant_id still clears — the reviewer's claim on + THIS task ends here, and the PM's ownership is assigned_to/claimed_by + only (mirrors pr_fail exactly). """ task = await self.get(task_id) if task is None or task.status != TaskStatus.AWAITING_PR_REVIEW: @@ -11320,8 +11339,9 @@ class TaskService(BaseService): ) captured = to_python_uuid(task.claimed_by) self._record_pr_review(task, summary=notes, verdict="passed") - task.assigned_to = None - task.claimed_by = None + pm = await self._revision_pm_for_task(task) + task.assigned_to = cast("Any", pm.id) if pm is not None else None + task.claimed_by = cast("Any", pm.id) if pm is not None else None task.active_claimant_id = cast("Any", None) # The gate reviewer's claim on THIS task ends here — release the # fleet marker (mirrors _qa_or_doc_claim's own ACTIVE-marking). @@ -11333,7 +11353,58 @@ class TaskService(BaseService): audit_agent_id=captured, ) await self.session.flush() - self.log.info("Assembled PR passed review", task_id=str(task_id)) + self.log.info( + "Assembled PR passed review", + task_id=str(task_id), + routed_to_pm=str(pm.id) if pm is not None else None, + ) + return task + + async def assign_review_pm(self, task_id: UUID) -> TaskTable | None: + """Recovery seam: (re)assign an ``awaiting_pm_review`` task to its + owning PM. + + CLAIM_RULES deliberately excludes AWAITING_PM_REVIEW (no claim() + edge into it — the i_will_plan re-claim-loop fix), so the normal + claim() route can't place an unassigned review task with its real + owner, and a task that went through a block/escalate/ + unblock(restore=True) round trip can come back with status restored + but ownership stuck on the stale escalation target. Called by the + orchestrator's pm-review dispatchers right before spawning, using + the same team-based resolution ``pr_fail``/``pr_pass`` use. Also + sets ``active_claimant_id`` (unlike pr_pass/pr_fail, which leave it + to a subsequent claim() that AWAITING_PM_REVIEW has none of) so the + newly-assigned PM's own note()/commit() calls don't bounce off + ``_active_claim_violation`` before ever claiming — mirrors + ``_maybe_advance_to_pm_review``'s identical concern. A no-op outside + awaiting_pm_review, and when the resolved owner already matches. + """ + 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_PM_REVIEW: + return None + pm = await self._revision_pm_for_task(task) + pm_id = cast("Any", pm.id) if pm is not None else None + if to_python_uuid(task.assigned_to) == to_python_uuid(pm_id) and to_python_uuid( + task.active_claimant_id + ) == to_python_uuid(pm_id): + return task + prior_claimant = to_python_uuid(task.active_claimant_id) + task.assigned_to = pm_id + task.claimed_by = pm_id + task.active_claimant_id = pm_id + if prior_claimant is not None and prior_claimant != to_python_uuid(pm_id): + await self._clear_agent_current_task(prior_claimant, task_id) + await self.session.flush() + self.log.info( + "Reassigned awaiting_pm_review task to owning PM", + task_id=str(task_id), + pm_id=str(pm.id) if pm is not None else None, + ) return task async def pr_fail( diff --git a/tests/integration/test_lifecycle_real_db.py b/tests/integration/test_lifecycle_real_db.py index eb8da751..23fbde16 100644 --- a/tests/integration/test_lifecycle_real_db.py +++ b/tests/integration/test_lifecycle_real_db.py @@ -899,10 +899,22 @@ async def test_pr_review_gate_pass_path( assert reviewer_row.status == AgentStatus.ACTIVE assert reviewer_row.current_task_id == task.id + # Resolved via the same team-based query pr_pass itself uses (rather than + # assumed to be this fixture's own cell_pm_agent) — the shared/cumulative + # integration DB may carry other BACKEND/CELL_PM agents from earlier + # tests, and _agent_with_role_and_team has no ordering guarantee. + expected_pm = await svc.cell_pm_for_team(Team.BACKEND) + assert expected_pm is not None + passed = await svc.pr_pass(reviewer_id, task.id, notes="integration verified") assert passed is not None assert str(passed.status) == Status.AWAITING_PM_REVIEW.value - assert passed.assigned_to is None # cleared so the PM-closure dispatch routes + # Hands off to the owning cell PM (team-resolved) rather than clearing — + # AWAITING_PM_REVIEW has no claim() edge, so an unassigned task here has + # no way back to a PM. + assert passed.assigned_to == expected_pm.id + assert passed.claimed_by == expected_pm.id + assert passed.active_claimant_id is None # pr_pass releases the reviewer's fleet marker too. reviewer_row = await db_session.get(AgentTable, reviewer_id) assert reviewer_row is not None diff --git a/tests/integration/test_tasks_routes.py b/tests/integration/test_tasks_routes.py index 9419e49f..afefa267 100644 --- a/tests/integration/test_tasks_routes.py +++ b/tests/integration/test_tasks_routes.py @@ -797,6 +797,53 @@ async def test_unblock_unknown_returns_404(task_client: dict) -> None: ) +@pytest.mark.asyncio +async def test_assign_review_pm_unknown_returns_404(task_client: dict) -> None: + client = task_client["client"] + response = await client.post(f"/api/tasks/{uuid4()}/assign-review-pm", headers=_HDR) + assert response.status_code == HTTPStatus.NOT_FOUND + + +@pytest.mark.asyncio +async def test_assign_review_pm_places_owning_pm(task_client: dict) -> None: + """The orchestrator's recovery seam: a review task with no (or a stale) + owner is placed with the real team-resolved PM — CLAIM_RULES has no + claim() edge into AWAITING_PM_REVIEW for the normal route to do this. + + Resolves the expected PM via the real ``main_pm_agent()`` query rather + than assuming this fixture's own agent — the test DB is shared/cumulative + across the module, and "earliest-created" main_pm may be an older row + from an earlier test. + """ + setup = task_client + client = setup["client"] + task = _seed_task( + setup, + status=TaskStatus.AWAITING_PM_REVIEW, + team=Team.MAIN_PM, + assigned_to=None, + ) + await setup["db"].commit() + expected_pm = await TaskService(setup["db"]).main_pm_agent() + assert expected_pm is not None + + response = await client.post(f"/api/tasks/{task.id}/assign-review-pm", headers=_HDR) + assert response.status_code == HTTPStatus.OK + body = response.json() + assert body["assigned_to"] == str(expected_pm.id) + + +@pytest.mark.asyncio +async def test_assign_review_pm_rejects_non_review_status(task_client: dict) -> None: + setup = task_client + client = setup["client"] + task = _seed_task(setup, status=TaskStatus.IN_PROGRESS) + await setup["db"].commit() + + response = await client.post(f"/api/tasks/{task.id}/assign-review-pm", headers=_HDR) + assert response.status_code == HTTPStatus.BAD_REQUEST + + @pytest.mark.asyncio async def test_pause_unknown_returns_404(task_client: dict) -> None: client = task_client["client"] diff --git a/tests/unit/gateway/test_choreographer_pm.py b/tests/unit/gateway/test_choreographer_pm.py index fbf2e5bb..ea7bc8cf 100644 --- a/tests/unit/gateway/test_choreographer_pm.py +++ b/tests/unit/gateway/test_choreographer_pm.py @@ -495,6 +495,47 @@ async def test_cell_pm_complete_not_assigned_returns_not_authorized() -> None: assert env.as_dict()["error"] == "not_authorized" +@pytest.mark.asyncio +async def test_cell_pm_complete_passes_ownership_guard_after_pr_pass_handoff() -> None: + """#740 (d87e2d9b) removed the illegal awaiting_pm_review -> claimed + re-claim edge, which used to be the PM's only way back to ownership + after the PR gate — pr_pass cleared assigned_to to None, so the owning + PM's complete() dead-ended on this exact guard forever + (not_authorized). pr_pass now hands off to the resolved owning PM + instead (see TaskService.pr_pass), so a task in the real post-gate + shape — assigned_to == the calling PM, no subtasks — must clear this + guard rather than bounce.""" + pm_id = uuid4() + task_id = uuid4() + t = MagicMock( + id=task_id, + status="awaiting_pm_review", + assigned_to=pm_id, + pr_number=8, + branch_name="feature/backend/abc--def", + parent_task_id=None, + team="backend", + ) + after = MagicMock(**{**t.__dict__, "status": "completed"}) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.all_subtasks_terminal.return_value = True + task_svc.cell_pm_complete.return_value = after + git_svc = AsyncMock() + git_svc.is_pr_merged_for_task.return_value = False + git_svc.pr_merge.return_value = {"merged": True, "merge_commit_sha": "merge-abc"} + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = True + journal_svc.latest_decision_at.return_value = datetime.now(UTC) + journal_svc.has_reflect_for_task.return_value = True + deps = _make_deps(task=task_svc, git=git_svc, journal=journal_svc) + c = Choreographer(deps) + + env = await c.cell_pm_complete(pm_id, task_id, notes="reviewed and approved") + assert env.as_dict().get("error") != "not_authorized" + assert env.error is None + + @pytest.mark.asyncio async def test_cell_pm_complete_in_progress_steers_to_submit_up() -> None: """Mirror of the main-PM submit_root steer: a cell task still in_progress diff --git a/tests/unit/runtime/test_pm_review_assign_seam.py b/tests/unit/runtime/test_pm_review_assign_seam.py new file mode 100644 index 00000000..7b47c5e9 --- /dev/null +++ b/tests/unit/runtime/test_pm_review_assign_seam.py @@ -0,0 +1,333 @@ +"""assign-review-pm dispatch seam — the recovery half of the pr_pass +ownership-clearing fix. + +CLAIM_RULES has no claim() edge into AWAITING_PM_REVIEW (the i_will_plan +re-claim-loop fix, #740/d87e2d9b) — pr_pass hands off to the owning PM +directly now (``TaskService.pr_pass``), but an unassigned task from before +the fix, or a block/escalate/unblock(restore=True) round trip landing on a +stale owner, still needs correcting before the dispatcher spawns a PM that +can't pass its own ownership guard. + +``_ensure_review_pm_assigned`` is the raw route call — it reports the +route's own outcome only (``None`` on ANY rejection/transport error, never a +stale fallback baked in). Its two callers each own their own fallback +policy: ``_closure_review_pm`` (``_maybe_spawn_pm_closure``) keeps its +already-known-correct team-resolved default on any failure, since a stale +``assigned_to`` fallback there could clobber it and spawn the wrong PM; +``_review_pm_slug`` (``_dispatch_pm_review_work``) has no better default +than the task's own ``assigned_to`` and also pre-checks it to skip the row +lock + HTTP round trip when already correct. +""" + +from __future__ import annotations + +from typing import Any, cast +from unittest.mock import AsyncMock, MagicMock, patch +from uuid import uuid4 + +import pytest +from roboco.config import settings +from roboco.runtime.orchestrator import AgentOrchestrator + + +def _orch() -> AgentOrchestrator: + orch = object.__new__(AgentOrchestrator) + orch._instances = {} + return orch + + +def _orch_any() -> Any: + """Untyped handle for tests that stub methods via direct attribute + assignment (mypy's method-assign check only fires on the concrete + ``AgentOrchestrator`` type; ``patch.object``-based tests use ``_orch()`` + instead).""" + orch = object.__new__(AgentOrchestrator) + orch._instances = {} + return orch + + +def _review_task(**over: Any) -> dict[str, Any]: + base: dict[str, Any] = { + "id": str(uuid4()), + "status": "awaiting_pm_review", + "team": "backend", + "assigned_to": None, + } + base.update(over) + return base + + +def _client_with_response(status_code: int, body: dict[str, Any]) -> Any: + resp = MagicMock(status_code=status_code) + resp.json.return_value = body + client = MagicMock() + client.post = AsyncMock(return_value=resp) + return client + + +# --------------------------------------------------------------------------- +# _ensure_review_pm_assigned — the route call, no fallback baked in +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_ensure_review_pm_assigned_resolves_slug_on_success() -> None: + orch = _orch() + task = _review_task() + pm_uuid = str(uuid4()) + client = _client_with_response(200, {"assigned_to": pm_uuid}) + + with patch.object(orch, "_resolve_agent_slug", return_value="be-pm") as resolve: + result = await orch._ensure_review_pm_assigned(client, task) + + assert result == "be-pm" + client.post.assert_awaited_once_with( + f"{settings.internal_api_url}/tasks/{task['id']}/assign-review-pm" + ) + resolve.assert_called_once_with(pm_uuid) + + +@pytest.mark.asyncio +async def test_ensure_review_pm_assigned_none_when_endpoint_returns_no_owner() -> None: + """An unresolvable PM (assign_review_pm's own fallback) means no owner — + nothing to spawn this tick.""" + orch = _orch() + task = _review_task() + client = _client_with_response(200, {"assigned_to": None}) + + result = await orch._ensure_review_pm_assigned(client, task) + + assert result is None + + +@pytest.mark.asyncio +async def test_ensure_review_pm_assigned_none_on_rejection_no_stale_fallback() -> None: + """A non-200 must report failure cleanly — NOT fall back to the task's + own (possibly stale) assigned_to. A blind fallback here is exactly what + let a transient failure clobber _closure_review_pm's already-correct + default in the pre-fix version of this seam.""" + orch = _orch() + stale_pm = str(uuid4()) + task = _review_task(assigned_to=stale_pm) + client = _client_with_response(500, {}) + + with patch.object(orch, "_resolve_agent_slug") as resolve: + result = await orch._ensure_review_pm_assigned(client, task) + + assert result is None + resolve.assert_not_called() + + +@pytest.mark.asyncio +async def test_ensure_review_pm_assigned_none_on_transport_error() -> None: + orch = _orch() + task = _review_task(assigned_to=str(uuid4())) + client = MagicMock() + client.post = AsyncMock(side_effect=RuntimeError("connection reset")) + + result = await orch._ensure_review_pm_assigned(client, task) + + assert result is None + + +# --------------------------------------------------------------------------- +# _closure_review_pm — _maybe_spawn_pm_closure's caller-owned fallback: the +# team-resolved default must survive ANY ensure-call failure. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_closure_review_pm_adopts_resolved_value_on_success() -> None: + orch = _orch() + task = _review_task() + with patch.object( + orch, "_ensure_review_pm_assigned", new=AsyncMock(return_value="be-pm") + ): + result = await orch._closure_review_pm(cast("Any", object()), task, "main-pm") + assert result == "be-pm" + + +@pytest.mark.asyncio +async def test_closure_review_pm_keeps_default_on_route_failure() -> None: + """The exact critic-flagged regression: a transient assign-review-pm + failure must NOT overwrite the already-correct team-resolved pm_id with + a stale fallback (e.g. main-pm on a task that should be be-pm).""" + orch = _orch() + task = _review_task() + with patch.object( + orch, "_ensure_review_pm_assigned", new=AsyncMock(return_value=None) + ) as ensure: + result = await orch._closure_review_pm(cast("Any", object()), task, "be-pm") + assert result == "be-pm" + ensure.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_closure_review_pm_skips_ensure_outside_review_status() -> None: + """claimed/in_progress/paused parents got their PM from the normal + claim/delegate flow — no correction, no route call at all.""" + orch = _orch() + task = _review_task(status="in_progress") + with patch.object(orch, "_ensure_review_pm_assigned", new=AsyncMock()) as ensure: + result = await orch._closure_review_pm(cast("Any", object()), task, "be-pm") + assert result == "be-pm" + ensure.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_maybe_spawn_pm_closure_spawns_team_resolved_pm_on_route_failure() -> ( + None +): + """End-to-end: _maybe_spawn_pm_closure must still spawn the correct + (team-resolved) PM when the assign-review-pm route fails outright.""" + orch = _orch_any() + orch._is_recently_paused = MagicMock(return_value=False) + orch._fetch_all_descendants = AsyncMock( + return_value=[{"id": "leaf", "status": "completed"}] + ) + orch._all_descendants_terminal = MagicMock(return_value=True) + orch._already_promoted_for_closure = MagicMock(return_value=False) + orch._closure_pm_for_team = MagicMock(return_value="be-pm") + orch._is_agent_active = MagicMock(return_value=False) + orch._closure_handled_without_pm = AsyncMock(return_value=(False, None)) + orch._build_pm_closure_prompt = MagicMock(return_value="PROMPT") + orch._task_git_context = MagicMock(return_value=None) + orch.spawn_agent = AsyncMock() + orch._ensure_review_pm_assigned = AsyncMock(return_value=None) # route failed + task = { + "id": "parent-1", + "status": "awaiting_pm_review", + "team": "backend", + "assigned_to": str(uuid4()), # some stale value the fix must ignore + } + + await orch._maybe_spawn_pm_closure(cast("Any", object()), task) + + orch.spawn_agent.assert_awaited_once() + assert orch.spawn_agent.await_args.kwargs["agent_id"] == "be-pm" + + +# --------------------------------------------------------------------------- +# _review_pm_slug — _dispatch_pm_review_work's pre-check + own fallback +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_review_pm_slug_skips_route_when_already_correct() -> None: + """The FINDING-3 pre-check: assigned_to already names the team-resolved + owner, so no row-lock + HTTP round trip is needed this tick.""" + orch = _orch() + pm_uuid = str(uuid4()) + task = _review_task(assigned_to=pm_uuid) + with ( + patch.object(orch, "_closure_pm_for_team", return_value="be-pm"), + patch.object(orch, "_resolve_agent_slug", return_value="be-pm"), + patch.object(orch, "_ensure_review_pm_assigned", new=AsyncMock()) as ensure, + ): + result = await orch._review_pm_slug(cast("Any", object()), task) + assert result == "be-pm" + ensure.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_review_pm_slug_corrects_mismatch_via_route() -> None: + orch = _orch() + task = _review_task(assigned_to=str(uuid4())) # stale/wrong owner + with ( + patch.object(orch, "_closure_pm_for_team", return_value="be-pm"), + patch.object(orch, "_resolve_agent_slug", return_value="main-pm"), + patch.object( + orch, "_ensure_review_pm_assigned", new=AsyncMock(return_value="be-pm") + ) as ensure, + ): + result = await orch._review_pm_slug(cast("Any", object()), task) + assert result == "be-pm" + ensure.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_review_pm_slug_falls_back_to_current_on_route_failure() -> None: + """No independently-known-better default here (unlike _closure_review_pm) + — falling back to the task's own current assignee is correct.""" + orch = _orch() + stale_pm_uuid = str(uuid4()) + task = _review_task(assigned_to=stale_pm_uuid) + with ( + patch.object(orch, "_closure_pm_for_team", return_value="be-pm"), + patch.object(orch, "_resolve_agent_slug", return_value="main-pm"), + patch.object( + orch, "_ensure_review_pm_assigned", new=AsyncMock(return_value=None) + ), + ): + result = await orch._review_pm_slug(cast("Any", object()), task) + assert result == "main-pm" + + +@pytest.mark.asyncio +async def test_review_pm_slug_none_when_unassigned_and_route_fails() -> None: + orch = _orch() + task = _review_task(assigned_to=None) + with ( + patch.object(orch, "_closure_pm_for_team", return_value="be-pm"), + patch.object( + orch, "_ensure_review_pm_assigned", new=AsyncMock(return_value=None) + ), + ): + result = await orch._review_pm_slug(cast("Any", object()), task) + assert result is None + + +# --------------------------------------------------------------------------- +# _dispatch_pm_review_work — the seam replaces the old claim-then-spawn split +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_dispatch_pm_review_work_spawns_resolved_pm() -> None: + orch = _orch() + task = _review_task() + client = cast("Any", object()) + + with ( + patch.object(orch, "_fetch_tasks", new=AsyncMock(return_value=[task])), + patch.object( + orch, "_blocked_by_earlier_sibling", new=AsyncMock(return_value=False) + ), + patch.object(orch, "_review_pm_slug", new=AsyncMock(return_value="be-pm")), + patch("roboco.runtime.orchestrator.is_spawnable_agent_slug", return_value=True), + patch.object(orch, "_is_agent_active", return_value=False), + patch.object( + orch, "_pm_respawn_should_gate", new=AsyncMock(return_value=False) + ), + patch.object(orch, "_build_pm_review_prompt", return_value="prompt"), + patch.object(orch, "_task_git_context", return_value=None), + patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn, + ): + await orch._dispatch_pm_review_work(client) + + spawn.assert_awaited_once() + assert spawn.call_args.kwargs["agent_id"] == "be-pm" + assert spawn.call_args.kwargs["task_id"] == task["id"] + + +@pytest.mark.asyncio +async def test_dispatch_pm_review_work_skips_when_pm_unresolvable() -> None: + orch = _orch() + task = _review_task() + client = cast("Any", object()) + + with ( + patch.object(orch, "_fetch_tasks", new=AsyncMock(return_value=[task])), + patch.object( + orch, "_blocked_by_earlier_sibling", new=AsyncMock(return_value=False) + ), + patch.object(orch, "_review_pm_slug", new=AsyncMock(return_value=None)), + patch.object(orch, "spawn_agent", new=AsyncMock()) as spawn, + ): + await orch._dispatch_pm_review_work(client) + + spawn.assert_not_awaited() + + +if __name__ == "__main__": + pytest.main([__file__, "-q"]) diff --git a/tests/unit/services/test_task.py b/tests/unit/services/test_task.py index 0dcf1eca..d19de88a 100644 --- a/tests/unit/services/test_task.py +++ b/tests/unit/services/test_task.py @@ -1166,6 +1166,182 @@ async def test_request_changes_rejects_wrong_status() -> None: assert out is None +# --------------------------------------------------------------------------- +# pr_pass / assign_review_pm — the #740 fix (d87e2d9b) removed the illegal +# awaiting_pm_review -> claimed re-claim edge, which was also load-bearing: +# it was the only way a PM re-acquired ownership after the PR gate cleared +# it. pr_pass now hands off to the owning PM instead (mirrors pr_fail); the +# unassigned-or-stale recovery seam is assign_review_pm. +# --------------------------------------------------------------------------- + + +def _pr_pass_svc(task: MagicMock, *, owning_pm: object) -> TaskService: + """A TaskService with pr_pass's helper calls stubbed — isolates the + ownership-handoff logic from `_validate_and_set_status`'s real + enforcement-layer transition/git-requirement checks (already covered + elsewhere) and from `_record_pr_review`'s note-writing side effect.""" + svc = TaskService(MagicMock(flush=AsyncMock())) + _bind(svc, "get", AsyncMock(return_value=task)) + _bind(svc, "_validate_and_set_status", MagicMock()) + _bind(svc, "_record_pr_review", MagicMock()) + _bind(svc, "_clear_agent_current_task", AsyncMock()) + _bind(svc, "_revision_pm_for_task", AsyncMock(return_value=owning_pm)) + return svc + + +@pytest.mark.asyncio +async def test_pr_pass_assigns_owning_cell_pm() -> None: + """A cell-team assembled task hands off to the resolved cell PM instead + of clearing ownership — AWAITING_PM_REVIEW has no claim() edge back in.""" + reviewer = uuid4() + cell_pm = SimpleNamespace(id=uuid4()) + task = _build_task( + status=TaskStatus.AWAITING_PR_REVIEW, + claimed_by=reviewer, + active_claimant_id=reviewer, + ) + svc = _pr_pass_svc(task, owning_pm=cell_pm) + out = await svc.pr_pass(reviewer, task.id, "clean, ship it") + assert out is task + assert task.assigned_to == cell_pm.id + assert task.claimed_by == cell_pm.id + # The reviewer's own claim ends here — active_claimant_id stays cleared + # (mirrors pr_fail exactly; the PM's ownership is assigned_to/claimed_by). + assert task.active_claimant_id is None + + +@pytest.mark.asyncio +async def test_pr_pass_assigns_main_pm_for_root_task() -> None: + """A root (main_pm-team) assembled task routes to the Main PM — same + `_revision_pm_for_task` resolution, just a different team branch.""" + reviewer = uuid4() + main_pm = SimpleNamespace(id=uuid4()) + task = _build_task( + status=TaskStatus.AWAITING_PR_REVIEW, + claimed_by=reviewer, + active_claimant_id=reviewer, + ) + svc = _pr_pass_svc(task, owning_pm=main_pm) + out = await svc.pr_pass(reviewer, task.id, "clean, ship it") + assert out is task + assert task.assigned_to == main_pm.id + assert task.claimed_by == main_pm.id + + +@pytest.mark.asyncio +async def test_pr_pass_falls_back_to_none_when_pm_unresolvable() -> None: + """An unresolvable owning PM leaves the task unassigned rather than + crashing — matches pr_fail's own fallback.""" + reviewer = uuid4() + task = _build_task( + status=TaskStatus.AWAITING_PR_REVIEW, + claimed_by=reviewer, + active_claimant_id=reviewer, + ) + svc = _pr_pass_svc(task, owning_pm=None) + out = await svc.pr_pass(reviewer, task.id, "clean, ship it") + assert out is task + assert task.assigned_to is None + assert task.claimed_by is None + + +@pytest.mark.asyncio +async def test_assign_review_pm_assigns_unassigned_task() -> None: + """The orchestrator's pm-review dispatch seam: an unassigned + awaiting_pm_review task (pr_pass resolved no PM, or legacy pre-fix data) + gets placed with its real owner, including active_claimant_id so the + newly-assigned PM's own note()/commit() calls don't bounce.""" + task = _build_task( + status=TaskStatus.AWAITING_PM_REVIEW, + assigned_to=None, + claimed_by=None, + active_claimant_id=None, + ) + result = MagicMock() + result.scalar_one_or_none.return_value = task + session = MagicMock(flush=AsyncMock()) + session.execute = AsyncMock(return_value=result) + svc = TaskService(session) + cell_pm = SimpleNamespace(id=uuid4()) + _bind(svc, "_revision_pm_for_task", AsyncMock(return_value=cell_pm)) + clear_mock = AsyncMock() + _bind(svc, "_clear_agent_current_task", clear_mock) + out = await svc.assign_review_pm(task.id) + assert out is task + assert task.assigned_to == cell_pm.id + assert task.claimed_by == cell_pm.id + assert task.active_claimant_id == cell_pm.id + clear_mock.assert_not_awaited() # nothing to release — was never claimed + + +@pytest.mark.asyncio +async def test_assign_review_pm_corrects_stale_assignment() -> None: + """A block/escalate/unblock(restore=True) round trip can leave a review + task pointed at the wrong (escalation-target) owner with a stale active + claim — this must correct BOTH to the real team-resolved PM, releasing + the stale claimant's fleet marker.""" + stale_pm = uuid4() + task = _build_task( + status=TaskStatus.AWAITING_PM_REVIEW, + assigned_to=stale_pm, + claimed_by=stale_pm, + active_claimant_id=stale_pm, + ) + result = MagicMock() + result.scalar_one_or_none.return_value = task + session = MagicMock(flush=AsyncMock()) + session.execute = AsyncMock(return_value=result) + svc = TaskService(session) + real_pm = SimpleNamespace(id=uuid4()) + clear_mock = AsyncMock() + _bind(svc, "_revision_pm_for_task", AsyncMock(return_value=real_pm)) + _bind(svc, "_clear_agent_current_task", clear_mock) + out = await svc.assign_review_pm(task.id) + assert out is task + assert task.assigned_to == real_pm.id + assert task.claimed_by == real_pm.id + assert task.active_claimant_id == real_pm.id + clear_mock.assert_awaited_once_with(stale_pm, task.id) + + +@pytest.mark.asyncio +async def test_assign_review_pm_noop_when_already_correct() -> None: + """Already correctly owned — no redundant write/notify every dispatch tick.""" + pm = uuid4() + task = _build_task( + status=TaskStatus.AWAITING_PM_REVIEW, + assigned_to=pm, + claimed_by=pm, + active_claimant_id=pm, + ) + result = MagicMock() + result.scalar_one_or_none.return_value = task + session = MagicMock(flush=AsyncMock()) + session.execute = AsyncMock(return_value=result) + svc = TaskService(session) + _bind(svc, "_revision_pm_for_task", AsyncMock(return_value=SimpleNamespace(id=pm))) + clear_mock = AsyncMock() + _bind(svc, "_clear_agent_current_task", clear_mock) + out = await svc.assign_review_pm(task.id) + assert out is task + clear_mock.assert_not_awaited() + session.flush.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_assign_review_pm_rejects_wrong_status() -> None: + """A no-op outside awaiting_pm_review — CLAIM_RULES already covers a + real claim status; this seam is scoped to the review-only gap.""" + task = _build_task(status=TaskStatus.IN_PROGRESS) + result = MagicMock() + result.scalar_one_or_none.return_value = task + session = MagicMock(flush=AsyncMock()) + session.execute = AsyncMock(return_value=result) + svc = TaskService(session) + out = await svc.assign_review_pm(task.id) + assert out is None + + @pytest.mark.asyncio async def test_admin_set_status_pre_block_restore_syncs_active_claimant() -> None: """The pending/in_progress restore path re-owns the task to the pre-block