fix(gateway): cell PM completes its own cell task; drop main-PM handoff

submit_up bubbled the cell task to Main PM (_handoff_to_main_pm), but
main_pm_complete rejects any task with a parent_task_id ("only operates
on root tasks"), so the cell->root PR had no one to merge it and the
cell task wedged at awaiting_pm_review. _maybe_advance_parent_to_pm_review
already intends the CELL PM to complete it.

Cell PM now owns cell completion:
- submit_up no longer hands off to Main PM; the cell task stays assigned
  to the cell PM, which is respawned to complete() it. Removed the
  now-unused _handoff_to_main_pm.
- cell_pm_complete resolves the merge target from the parent task's real
  branch_name (shared merge_chain.resolve_parent_branch, also used by the
  PR side-effects) so the cell->root PR merges into feature/main_pm/...,
  not the team-mis-derived feature/<cellteam>/... (same root cause as the
  prior PR-base fix).
- submit_up description + next_hint updated; lifecycle artifacts regen.

Main PM still only completes the ROOT (root->master + escalate-to-CEO).
First run to reach cell-PM bubble-up exposed this.

tests/unit/gateway/test_choreographer_impl_branches.py

@@ -704,25 +704,6 @@ async def test_submit_up_no_branch_rejected() -> None:
     assert "no branch" in body["message"]
 
 
-# ---------------------------------------------------------------------------
-# _handoff_to_main_pm: main_pm_agent returns None (line 1567)
-# ---------------------------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_handoff_to_main_pm_no_main_pm_returns_silently() -> None:
-    pm_id = uuid4()
-    task_id = uuid4()
-    task_svc = AsyncMock()
-    task_svc.main_pm_agent.return_value = None  # No main PM in DB
-    deps = _make_deps(task=task_svc)
-    c = Choreographer(deps)
-    # Should silently return without error.
-    await c._handoff_to_main_pm(pm_id, task_id)
-    # reassign + a2a never called.
-    task_svc.reassign.assert_not_called()
-
-
 # ---------------------------------------------------------------------------
 # _pm_next_hint: each branch (lines 1612-1616)
 # ---------------------------------------------------------------------------

tests/unit/gateway/test_choreographer_pm.py

@@ -251,18 +251,25 @@ async def test_unblock_restore_false_returns_legacy_message() -> None:
 async def test_cell_pm_complete_merges_then_completes() -> None:
     pm_id = uuid4()
     task_id = uuid4()
+    parent_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=uuid4(),
+        parent_task_id=parent_id,
         team="backend",
     )
     after = MagicMock(**{**t.__dict__, "status": "completed"})
+    # #181/#182: the merge target is the PARENT task's real branch_name —
+    # here under a DIFFERENT team prefix, which the old parent_branch_for
+    # would have mis-derived as feature/backend/abc.
+    parent = MagicMock(
+        id=parent_id, branch_name="feature/main_pm/abc", parent_task_id=None
+    )
     task_svc = AsyncMock()
-    task_svc.get.return_value = t
+    task_svc.get.side_effect = lambda tid: parent if tid == parent_id else t
     task_svc.all_subtasks_terminal.return_value = True
     task_svc.cell_pm_complete.return_value = after
     git_svc = AsyncMock()
@@ -278,7 +285,7 @@ async def test_cell_pm_complete_merges_then_completes() -> None:
     assert env.error is None
     assert env.status == "completed"
     git_svc.pr_merge.assert_awaited_once_with(
-        8, target="feature/backend/abc", actor_agent_id=pm_id
+        8, target="feature/main_pm/abc", actor_agent_id=pm_id
     )
 
 

tests/unit/gateway/test_choreographer_pm_extras.py

@@ -882,10 +882,13 @@ async def test_delegate_invalid_team_enum_rejected() -> None:
 
 
 @pytest.mark.asyncio
-async def test_submit_up_opens_pr_and_reassigns_to_main_pm() -> None:
+async def test_submit_up_opens_pr_and_keeps_cell_pm_assignment() -> None:
+    """#182: submit_up opens the cell→root PR and moves the cell task to
+    awaiting_pm_review, but does NOT hand it off to Main PM. The cell PM
+    owns cell completion (it is respawned to `complete` the task), so the
+    assignment stays put — no reassign."""
     pm_id = uuid4()
     task_id = uuid4()
-    main_pm_id = uuid4()
     t = MagicMock(
         id=task_id,
         status="in_progress",
@@ -907,7 +910,6 @@ async def test_submit_up_opens_pr_and_reassigns_to_main_pm() -> None:
     task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
     task_svc.all_subtasks_terminal.return_value = True
     task_svc.submit_pm_review.return_value = after
-    task_svc.main_pm_agent.return_value = MagicMock(id=main_pm_id)
     git_svc = AsyncMock()
     git_svc.create_pr.return_value = {"pr_number": 12, "pr_url": "x"}
     journal_svc = AsyncMock()
@@ -922,7 +924,7 @@ async def test_submit_up_opens_pr_and_reassigns_to_main_pm() -> None:
     assert env.error is None
     assert env.status == "awaiting_pm_review"
     git_svc.create_pr.assert_awaited_once()
-    task_svc.reassign.assert_awaited_once()
+    task_svc.reassign.assert_not_called()
 
 
 @pytest.mark.asyncio

tests/unit/gateway/test_choreographer_reassignment.py

@@ -438,12 +438,15 @@ async def test_cell_pm_complete_reassigns_parent_when_all_subtasks_done() -> Non
         id=parent_id,
         team="backend",
         parent_task_id=None,
+        branch_name="feature/backend/abc",
     )
 
     task_svc = AsyncMock()
-    # First .get is for the leaf (status check), then for the parent walk-up.
-    task_svc.get.side_effect = [leaf, parent]
-    task_svc.all_subtasks_terminal.side_effect = [True, True]
+    # get(leaf) for the status check, get(parent) for the merge-target
+    # resolution (#181/#182) and the parent walk-up. Keyed by id so it is
+    # robust to call count/order.
+    task_svc.get.side_effect = lambda tid: parent if tid == parent_id else leaf
+    task_svc.all_subtasks_terminal.return_value = True
     task_svc.cell_pm_complete.return_value = after
     task_svc.cell_pm_for_team.return_value = MagicMock(id=new_pm_id)
 

tests/unit/gateway/test_merge_chain.py

@@ -2,8 +2,15 @@
 
 from __future__ import annotations
 
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
 import pytest
-from roboco.services.gateway.merge_chain import branch_depth, parent_branch_for
+from roboco.services.gateway.merge_chain import (
+    branch_depth,
+    parent_branch_for,
+    resolve_parent_branch,
+)
 
 _DEPTH_ROOT = 1
 _DEPTH_ONE_SUBTASK = 2
@@ -44,6 +51,47 @@ class TestParentBranchFor:
             parent_branch_for("not-a-branch")
 
 
+class TestResolveParentBranch:
+    """#181/#182: base/target comes from the parent TASK's branch_name, which
+    is correct across a team boundary; parent_branch_for is the fallback."""
+
+    @pytest.mark.asyncio
+    async def test_uses_parent_task_branch_across_team(self) -> None:
+        root_id = uuid4()
+        task = MagicMock(
+            parent_task_id=root_id,
+            branch_name="feature/backend/ROOT0001--CELL0001",
+        )
+        task_service = AsyncMock()
+        # Root lives under a DIFFERENT team prefix than the cell.
+        task_service.get = AsyncMock(
+            return_value=MagicMock(branch_name="feature/main_pm/ROOT0001")
+        )
+        result = await resolve_parent_branch(task, task_service)
+        assert result == "feature/main_pm/ROOT0001"
+        task_service.get.assert_awaited_once_with(root_id)
+
+    @pytest.mark.asyncio
+    async def test_falls_back_when_no_parent(self) -> None:
+        task = MagicMock(parent_task_id=None, branch_name="feature/backend/ROOT0001")
+        task_service = AsyncMock()
+        # No parent → root→master.
+        assert await resolve_parent_branch(task, task_service) == "master"
+        task_service.get.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_falls_back_when_parent_has_no_branch(self) -> None:
+        task = MagicMock(
+            parent_task_id=uuid4(),
+            branch_name="feature/backend/ROOT0001--CELL0001",
+        )
+        task_service = AsyncMock()
+        task_service.get = AsyncMock(return_value=MagicMock(branch_name=None))
+        # Parent exists but has no branch yet → string derivation.
+        result = await resolve_parent_branch(task, task_service)
+        assert result == "feature/backend/ROOT0001"
+
+
 def test_branch_depth_master_is_zero() -> None:
     """Line 28: master returns 0."""
     assert branch_depth("master") == 0