fix(lifecycle): pr_pass hands ownership to the owning PM — closes the passed-PR completion wedge

Removing the awaiting_pm_review re-claim edge (d87e2d9b, the #740
review-loop fix) exposed that the edge was load-bearing: pr_pass
cleared assigned_to/claimed_by to None and the re-claim was the only
way a PM ever re-acquired the task. Since then every assembled task
passing the PR gate wedged: the closure PM's complete rejected with
'not assigned to you', its fallback claim rejected (edge gone), and
its only exit was escalate_up — BLOCKING the task onto main-pm, who
is not the assignee either and burned spawns doing nothing. Live:
PR #741's task (7 ownership rejections, escalated 13:59Z), plus two
more tasks with 25 and 2 rejections in the same shape.

pr_pass now resolves the owning PM via _revision_pm_for_task and
assigns it, exactly as pr_fail always did — one chokepoint covering
cell (submit_up) and root (submit_root) tasks. mark_pr_created's
ready_for_pm branch (leaf docs-path, PR-arrives-second) had the same
clear-to-None wedge and now resolves via _resolve_pm_for_review like
its docs-first sibling. No claim edge is reintroduced; the #740
parity test is untouched.

Recovery for already-wedged tasks needs no DB surgery: new idempotent
TaskService.assign_review_pm + POST /tasks/{id}/assign-review-pm
(ASSIGN-gated, explicit commit) corrects an unassigned OR mis-assigned
awaiting_pm_review task to its owning PM; _dispatch_pm_review_work
ensures assignment before every spawn (cheap pre-check skips the
round-trip when the fetched assigned_to already matches), replacing
the dead _claim_task_for_agent call whose lifecycle claim the removed
edge now always rejects; _maybe_spawn_pm_closure routes through
_closure_review_pm, which keeps the team-resolved PM whenever the
assign route fails so a transient error can never spawn a stale
assignee.

Gate: 15506 passed, 459 skipped; xenon/ruff/mypy/vulture/bandit/
pip-audit/deptry/import-linter/foundation-check green.

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