fix(gateway): rejected PM is told the exact complete() call (#170, partial)

Smoke-15 wedge: leaf 1533ce56 sat at awaiting_pm_review owned by
be-pm, but the PMs looped firing complete/unblock at the wrong
(parent) task_ids — every rejection was generic ("not assigned to
you" / "not ready for completion"), so minimax never discovered it
just needed `complete(1533ce56)`.

New _own_review_hint: on a cell_pm/main_pm complete-guard rejection
(not-owner, wrong-state, or main-pm-on-non-root), if the PM owns a
DIFFERENT task that is awaiting_pm_review, append a remediate suffix
naming it and the exact `complete(task_id='<id>', notes='...')` call.
Best-effort (never raises into the rejection path), pure guidance —
no control-flow or state-machine change.

Scope: this is the bounded, low-risk slice of #170 (fix b). The
parent-state corruption + missing recovery transition (root->paused /
cell->blocked from earlier mis-targeted verbs, fix a/c) is a lifecycle
state-machine change deferred for explicit design alignment — tracked
in #170.

roboco/services/gateway/choreographer/_impl.py

@@ -3500,6 +3500,33 @@ class Choreographer:
             context_briefing=await self._briefing_for(pm_agent_id, task_id),
         ).with_introspection(task=t, role=role)
 
+    async def _own_review_hint(self, pm_agent_id: UUID, exclude_task_id: UUID) -> str:
+        """Remediate suffix naming the PM's OWN task ready to complete.
+
+        Smoke-15 wedge (#170): a PM looped firing complete/unblock at the
+        wrong (parent) task_id while its own leaf sat at
+        ``awaiting_pm_review``, never named in any rejection — minimax
+        never found the one correct call. Surface it explicitly.
+        Best-effort: never raises into the rejection path.
+        """
+        try:
+            owned = await self.task.list_by_assignee(pm_agent_id)
+        except Exception:
+            return ""
+        ready = [
+            str(o.id)
+            for o in owned
+            if str(o.status) == "awaiting_pm_review" and o.id != exclude_task_id
+        ]
+        if not ready:
+            return ""
+        tid = ready[0]
+        return (
+            f" You OWN task {tid} which is awaiting_pm_review and ready to "
+            f"finish — call complete(task_id='{tid}', notes='...') on THAT "
+            "task, not this one."
+        )
+
     async def _cell_pm_complete_guard(
         self, pm_agent_id: UUID, task_id: UUID, t: Any, notes: str
     ) -> Envelope | None:
@@ -3515,7 +3542,10 @@ class Choreographer:
         if t.assigned_to != pm_agent_id:
             return Envelope.not_authorized(
                 message="not assigned to you",
-                remediate="claim the task or wait for it to be assigned",
+                remediate=(
+                    "claim the task or wait for it to be assigned."
+                    + await self._own_review_hint(pm_agent_id, task_id)
+                ),
                 context_briefing=await self._briefing_for(pm_agent_id, task_id),
             )
         if str(t.status) != "awaiting_pm_review":
@@ -3523,7 +3553,10 @@ class Choreographer:
                 message=(
                     f"task {task_id} is in {t.status}, expected awaiting_pm_review"
                 ),
-                remediate="this task is not ready for completion",
+                remediate=(
+                    "this task is not ready for completion."
+                    + await self._own_review_hint(pm_agent_id, task_id)
+                ),
                 context_briefing=await self._briefing_for(pm_agent_id, task_id),
             )
         if env := await self._check_complete_gates(pm_agent_id, task_id, notes):
@@ -3629,7 +3662,10 @@ class Choreographer:
         if t.assigned_to != main_pm_agent_id:
             return Envelope.not_authorized(
                 message="not assigned to you",
-                remediate="wait for assignment or claim",
+                remediate=(
+                    "wait for assignment or claim."
+                    + await self._own_review_hint(main_pm_agent_id, root_task_id)
+                ),
                 context_briefing=await self._briefing_for(
                     main_pm_agent_id, root_task_id
                 ),
@@ -3639,7 +3675,10 @@ class Choreographer:
                 message=(
                     f"task {root_task_id} is in {t.status}, expected awaiting_pm_review"
                 ),
-                remediate="this task is not ready for main-PM completion",
+                remediate=(
+                    "this task is not ready for main-PM completion."
+                    + await self._own_review_hint(main_pm_agent_id, root_task_id)
+                ),
                 context_briefing=await self._briefing_for(
                     main_pm_agent_id, root_task_id
                 ),
@@ -3650,8 +3689,9 @@ class Choreographer:
                     "main_pm complete only operates on root tasks (no parent_task_id)"
                 ),
                 remediate=(
-                    "cell PM should complete this task;"
+                    "cell PM should complete this task; main PM only"
-                    " main PM only completes root tasks"
+                    " completes root tasks."
+                    + await self._own_review_hint(main_pm_agent_id, root_task_id)
                 ),
                 context_briefing=await self._briefing_for(
                     main_pm_agent_id, root_task_id

tests/unit/gateway/test_pm_own_review_hint.py

@@ -0,0 +1,157 @@
+"""#170: a rejected PM must be told the exact complete() call to make.
+
+Smoke-15 wedge: leaf 1533ce56 sat at awaiting_pm_review owned by
+be-pm, but be-pm/main-pm looped firing complete/unblock at the wrong
+(parent) task_ids — no rejection ever named the one actionable task,
+so minimax never issued `complete(1533ce56)`. The complete guards now
+append `_own_review_hint`: if the PM owns a DIFFERENT task that is
+awaiting_pm_review, the remediate names it + the exact call.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+
+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(),
+        "messaging": AsyncMock(),
+    }
+    base.update(overrides)
+    repo = base["evidence_repo"]
+    for m 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, m).return_value = []
+    _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 _owned(task_id: Any, status: str) -> SimpleNamespace:
+    return SimpleNamespace(id=task_id, status=status)
+
+
+@pytest.mark.asyncio
+async def test_hint_names_the_owned_awaiting_review_task() -> None:
+    pm = uuid4()
+    ready_id = uuid4()
+    rejected_id = uuid4()
+    deps = _make_deps()
+    deps.task.list_by_assignee = AsyncMock(
+        return_value=[
+            _owned(rejected_id, "in_progress"),
+            _owned(ready_id, "awaiting_pm_review"),
+        ]
+    )
+    c = Choreographer(deps)
+
+    hint = await c._own_review_hint(pm, rejected_id)
+    assert str(ready_id) in hint
+    assert f"complete(task_id='{ready_id}'" in hint
+
+
+@pytest.mark.asyncio
+async def test_hint_excludes_the_rejected_task_itself() -> None:
+    pm = uuid4()
+    same = uuid4()
+    deps = _make_deps()
+    deps.task.list_by_assignee = AsyncMock(
+        return_value=[_owned(same, "awaiting_pm_review")]
+    )
+    c = Choreographer(deps)
+    assert await c._own_review_hint(pm, same) == ""
+
+
+@pytest.mark.asyncio
+async def test_hint_empty_when_nothing_ready() -> None:
+    pm = uuid4()
+    deps = _make_deps()
+    deps.task.list_by_assignee = AsyncMock(
+        return_value=[_owned(uuid4(), "in_progress")]
+    )
+    c = Choreographer(deps)
+    assert await c._own_review_hint(pm, uuid4()) == ""
+
+
+@pytest.mark.asyncio
+async def test_hint_best_effort_swallows_errors() -> None:
+    pm = uuid4()
+    deps = _make_deps()
+    deps.task.list_by_assignee = AsyncMock(side_effect=RuntimeError("db down"))
+    c = Choreographer(deps)
+    assert await c._own_review_hint(pm, uuid4()) == ""
+
+
+@pytest.mark.asyncio
+async def test_cell_pm_complete_guard_not_owner_surfaces_hint() -> None:
+    """The smoke-15 case: main-pm calls complete on a leaf it doesn't
+    own while its real task is awaiting_pm_review elsewhere."""
+    caller = uuid4()
+    other_owner = uuid4()
+    rejected_id = uuid4()
+    ready_id = uuid4()
+    deps = _make_deps()
+    deps.task.list_by_assignee = AsyncMock(
+        return_value=[_owned(ready_id, "awaiting_pm_review")]
+    )
+    c = Choreographer(deps)
+    t = MagicMock(
+        id=rejected_id,
+        assigned_to=other_owner,
+        status="awaiting_pm_review",
+        team="backend",
+        task_type="code",
+    )
+
+    env = await c._cell_pm_complete_guard(caller, rejected_id, t, "notes")
+    assert env is not None
+    body = env.as_dict()
+    assert body["error"] == "not_authorized"
+    assert f"complete(task_id='{ready_id}'" in body["remediate"]
+
+
+@pytest.mark.asyncio
+async def test_cell_pm_complete_guard_wrong_state_surfaces_hint() -> None:
+    pm = uuid4()
+    rejected_id = uuid4()
+    ready_id = uuid4()
+    deps = _make_deps()
+    deps.task.list_by_assignee = AsyncMock(
+        return_value=[_owned(ready_id, "awaiting_pm_review")]
+    )
+    c = Choreographer(deps)
+    t = MagicMock(
+        id=rejected_id,
+        assigned_to=pm,
+        status="in_progress",
+        team="backend",
+        task_type="planning",
+    )
+
+    env = await c._cell_pm_complete_guard(pm, rejected_id, t, "notes")
+    assert env is not None
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+    assert f"complete(task_id='{ready_id}'" in body["remediate"]