Bunch of fixes we need to verify first..

.github/workflows/ci.yml

@@ -93,6 +93,16 @@ jobs:
     steps:
       - name: Checkout code
         uses: actions/checkout@v7
+        with:
+          # The release-readiness smoke test calls ``git describe --tags`` to
+          # find the most recent release tag. ``actions/checkout``'s default
+          # shallow + no-tags clone makes that return empty, which made
+          # ``test_gather_snapshot_reads_the_real_repo`` fail with
+          # ``last_tag is None`` even though master had a tagged v0.13.0.
+          # ``fetch-depth: 0`` clones full history; the default ``fetch-tags``
+          # would still skip tags on shallow clones, so we also pin it true.
+          fetch-depth: 0
+          fetch-tags: true
 
       - name: Set up Python
         uses: actions/setup-python@v6

roboco/models/llm_catalog.py

@@ -69,7 +69,7 @@ MODEL_CATALOG: tuple[CatalogEntry, ...] = (
     # --- Ollama Cloud (verbatim tags) ---
     # Pro plan active as of 2026-04-22. Drop any entry that stops working —
     # the catalog is the single source of truth the Settings dropdown renders from.
-    CatalogEntry("glm-5.1:cloud", ModelProvider.OLLAMA_CLOUD, "GLM 5.1"),
+    CatalogEntry("glm-5.2:cloud", ModelProvider.OLLAMA_CLOUD, "GLM 5.2"),
     CatalogEntry("kimi-k2.6:cloud", ModelProvider.OLLAMA_CLOUD, "Kimi K2.6"),
     CatalogEntry("kimi-k2.7-code:cloud", ModelProvider.OLLAMA_CLOUD, "Kimi K2.7 Code"),
     CatalogEntry("minimax-m3:cloud", ModelProvider.OLLAMA_CLOUD, "Minimax M3"),
@@ -107,8 +107,8 @@ def provider_type_for_model(model_name: str) -> ModelProvider | None:
 OLLAMA_ROLE_DEFAULTS: dict[str, str] = {
     # High-volume agentic coding — M3 is purpose-built for this.
     "developer": "minimax-m3:cloud",
-    # Deep code review — GLM 5.1 has the highest SWE-Bench and iterates thoroughly.
-    "qa": "glm-5.1:cloud",
+    # Deep code review — GLM 5.2 has the highest SWE-Bench and iterates thoroughly.
+    "qa": "glm-5.2:cloud",
     # Orchestration + tool coordination — Kimi K2.6's Agent Swarm is the exact fit.
     "cell_pm": "kimi-k2.6:cloud",
     "main_pm": "kimi-k2.6:cloud",
@@ -116,10 +116,10 @@ OLLAMA_ROLE_DEFAULTS: dict[str, str] = {
     "auditor": "kimi-k2.6:cloud",
     # Product reasoning — same profile as PM work.
     "product_owner": "kimi-k2.6:cloud",
-    # Writing with code-context — GLM 5.1's creative writing + SWE-Bench combo.
-    "documenter": "glm-5.1:cloud",
-    # Stylistic writing — GLM 5.1's creative-writing strength.
-    "head_marketing": "glm-5.1:cloud",
+    # Writing with code-context — GLM 5.2's creative writing + SWE-Bench combo.
+    "documenter": "glm-5.2:cloud",
+    # Stylistic writing — GLM 5.2's creative-writing strength.
+    "head_marketing": "glm-5.2:cloud",
     # CEO is human-in-the-loop; keep an entry in case someone forces
     # a route to it, but the Settings UI intentionally excludes it.
     "ceo": "kimi-k2.6:cloud",

roboco/services/prompter.py

@@ -704,18 +704,35 @@ def parse_readiness(content: str) -> tuple[str, ReadinessTag | None]:
     )
 
 
-def _cell_teams(the_work: list[dict[str, Any]]) -> list[str]:
+def _as_work_entry(entry: Any) -> dict[str, Any]:
+    """Normalize one ``the_work`` entry to a dict.
+
+    The intake agent is an LLM and sometimes emits ``the_work`` as a list of
+    bare team-name strings (``"backend"``) instead of the documented
+    ``{team, summary, items}`` objects. Treat a bare string as
+    ``{"team": <str>}`` so every consumer can keep calling ``.get("team")`` /
+    ``.get("items")`` instead of crashing with ``'str' has no 'get'``
+    (regression: ``preview-batch`` 500'd on this shape).
+    """
+    if isinstance(entry, str):
+        return {"team": entry.strip()}
+    if isinstance(entry, dict):
+        return entry
+    return {}
+
+
+def _cell_teams(the_work: list[Any]) -> list[str]:
     """Distinct cell teams (backend/frontend/ux_ui) present in the_work, in order."""
     cell_values = {t.value for t in CELL_TEAMS}
     seen: list[str] = []
     for entry in the_work:
-        team = str(entry.get("team", ""))
+        team = str(_as_work_entry(entry).get("team", ""))
         if team in cell_values and team not in seen:
             seen.append(team)
     return seen
 
 
-def derive_scale(the_work: list[dict[str, Any]]) -> str:
+def derive_scale(the_work: list[Any]) -> str:
     """'multi' when more than one cell participates, else 'single'."""
     return "multi" if len(_cell_teams(the_work)) > 1 else "single"
 
@@ -740,17 +757,22 @@ def _cell_label(team: str) -> str:
     return _TEAM_LABELS.get(team) or team.replace("_", " ").title() or "Work"
 
 
-def _render_work_entry(entry: dict[str, Any]) -> str:
-    """Render one cell's slice: a bold heading and its deliverables."""
-    head = f"**{_cell_label(_text(entry.get('team')))}**"
-    summary = _text(entry.get("summary"))
+def _render_work_entry(entry: Any) -> str:
+    """Render one cell's slice: a bold heading and its deliverables.
+
+    ``entry`` may be a bare team-name string (see ``_as_work_entry``); a bare
+    string renders as just the cell heading, with no summary/items.
+    """
+    e = _as_work_entry(entry)
+    head = f"**{_cell_label(_text(e.get('team')))}**"
+    summary = _text(e.get("summary"))
     if summary:
         head = f"{head} — {summary}"
-    items = _clean_list(entry.get("items"))
+    items = _clean_list(e.get("items"))
     return f"{head}\n{_bullets(items)}" if items else head
 
 
-def _render_the_work(the_work: list[dict[str, Any]]) -> str:
+def _render_the_work(the_work: list[Any]) -> str:
     """Render The Work section, with a board-led lead line when multi-cell."""
     blocks = [_render_work_entry(e) for e in the_work]
     if len(_cell_teams(the_work)) > 1:

roboco/services/task.py

@@ -192,6 +192,37 @@ def _board_cannot_own(task: TaskTable) -> bool:
     )
 
 
+_PM_OWNED_CELL_TASK_TYPES: frozenset[str] = frozenset(
+    {
+        TaskType.PLANNING.value,
+        TaskType.RESEARCH.value,
+        TaskType.ADMINISTRATIVE.value,
+        TaskType.DOCUMENTATION.value,
+        TaskType.DESIGN.value,
+    }
+)
+
+
+def _is_cell_pm_owned_task(task: TaskTable) -> bool:
+    """True for a descendant cell-team task that must be owned by its cell PM.
+
+    A cell team's planning / research / administrative / documentation / design
+    work is not Main-PM work — the Main PM coordinates across cells, but each
+    cell's own non-code work belongs to that cell's PM. Assigning such a child
+    to main-pm deadlocks because the Main PM's escalation chain points up, not
+    across to the cell PM who can actually decompose it. ``code`` is excluded
+    because leaf code work is delegated by the cell PM to a developer, not
+    owned by the cell PM itself.
+    """
+    if task.parent_task_id is None:
+        return False
+    team_value = getattr(task.team, "value", task.team)
+    if str(team_value) not in _CELL_TEAMS:
+        return False
+    task_type_value = getattr(task.task_type, "value", task.task_type)
+    return str(task_type_value) in _PM_OWNED_CELL_TASK_TYPES
+
+
 # Notes fields (dev_notes, qa_notes, quick_context) are append-only —
 # every revision cycle adds more. Cap total size so a task that cycles
 # dozens of times doesn't grow into megabytes. When we exceed the cap,
@@ -826,6 +857,12 @@ class TaskService(BaseService):
             title=req.title,
             team=req.team if isinstance(req.team, str) else req.team.value,
         )
+        # NOTE: cell-team PM-owned invariants are enforced on the reassign path
+        # (`reassign` / `reassign_active_claim` → `_resolve_cell_pm_redirect`)
+        # — not at create. Direct task.assigned_to writes in the escalation
+        # chain (e.g. the orchestrator's `_dispatch_revision_coordination_roots`)
+        # bypass this; they are TODO-listed at the call sites.
+
         await self._attach_baseline_constraints(task)
         return task
 
@@ -6956,16 +6993,124 @@ class TaskService(BaseService):
                 refused_assignee=str(new_assignee),
             )
             return task
-        task.assigned_to = cast("Any", new_assignee) if new_assignee else None
-        task.claimed_by = cast("Any", new_assignee) if new_assignee else None
+        # Invariant backstop: cell-team planning/research/administrative children
+        # must be owned by their cell PM. If the caller tried to hand such a task
+        # to main-pm (or another mismatched owner), redirect to the cell PM.
+        redirect = await self._resolve_cell_pm_redirect(task, new_assignee)
+        effective_assignee = redirect.effective_assignee
+        if redirect.redirected:
+            self.log.info(
+                "Reassign redirected to cell PM",
+                task_id=str(task_id),
+                requested_assignee=str(new_assignee),
+                effective_assignee=str(effective_assignee),
+                reason=redirect.reason,
+            )
+            if redirect.dev_notes_line is not None:
+                task.dev_notes = (task.dev_notes or "") + redirect.dev_notes_line
+
+        task.assigned_to = (
+            cast("Any", effective_assignee) if effective_assignee else None
+        )
+        task.claimed_by = (
+            cast("Any", effective_assignee) if effective_assignee else None
+        )
         await self.session.flush()
         self.log.info(
             "Task reassigned",
             task_id=str(task_id),
-            new_assignee=str(new_assignee) if new_assignee else None,
+            new_assignee=str(effective_assignee) if effective_assignee else None,
         )
         return task
 
+    @dataclass(frozen=True)
+    class _CellPmRedirect:
+        """Outcome of an `_resolve_cell_pm_redirect` call.
+
+        ``effective_assignee`` is the UUID the caller should persist (possibly
+        ``None`` when the task was queued for a missing cell PM).
+        ``redirected`` is ``True`` iff ``effective_assignee`` differs from the
+        caller's requested assignee — used by callers to decide whether to log
+        a redirect. ``reason`` is a short tag that names the redirect reason
+        ("to_cell_pm", "queued_no_cell_pm"). ``dev_notes_line`` is the audit
+        text to append to ``task.dev_notes`` (only set when a redirect
+        happened) so a redirect is visible in the task body, not just in logs.
+        """
+
+        effective_assignee: UUID | None
+        redirected: bool
+        reason: str
+        dev_notes_line: str | None
+
+    async def _resolve_cell_pm_redirect(
+        self, task: TaskTable, requested_assignee: UUID | None
+    ) -> "TaskService._CellPmRedirect":
+        """Decide whether ``task`` must be redirected to its cell PM.
+
+        Returns a dataclass describing the outcome. Behavior:
+
+        - Not a cell-team PM-owned child → keep ``requested_assignee`` unchanged.
+        - No cell PM exists for the team → **queue the task** (``None``
+          assignee, ``ERROR`` log) rather than silently assigning to
+          ``requested_assignee``. The system is in a bad state; this is
+          observable in logs and the task sits in ``PENDING`` until the agent
+          table is repaired.
+        - Cell PM exists and matches → keep ``requested_assignee``.
+        - Cell PM exists and differs → redirect to the cell PM and write the
+          standard ``[ASSIGNMENT REDIRECTED]`` line to ``dev_notes`` so the
+          audit is visible in the task body.
+        """
+        noop = TaskService._CellPmRedirect(
+            effective_assignee=requested_assignee,
+            redirected=False,
+            reason="noop",
+            dev_notes_line=None,
+        )
+        if not _is_cell_pm_owned_task(task):
+            return noop
+        assert task.team is not None  # guarded by _is_cell_pm_owned_task
+        team_enum = Team(str(getattr(task.team, "value", task.team)))
+        cell_pm = await self.cell_pm_for_team(team_enum)
+        if cell_pm is None:
+            self.log.error(
+                "Cell-team PM-owned task has no cell PM agent row; "
+                "queueing without an assignee. Repair agents table.",
+                task_id=str(getattr(task, "id", None)),
+                team=team_enum.value,
+            )
+            type_value = (
+                task.task_type.value
+                if isinstance(task.task_type, TaskType)
+                else task.task_type
+            )
+            return TaskService._CellPmRedirect(
+                effective_assignee=None,
+                redirected=(requested_assignee is not None),
+                reason="queued_no_cell_pm",
+                dev_notes_line=(
+                    f"\n\n[ASSIGNMENT REDIRECTED] cell-team {team_enum.value} "
+                    f"{type_value} task queued with no assignee because no "
+                    f"cell PM agent row exists; was {requested_assignee}."
+                ),
+            )
+        if requested_assignee == cell_pm.id:
+            return noop
+        type_value = (
+            task.task_type.value
+            if isinstance(task.task_type, TaskType)
+            else task.task_type
+        )
+        return TaskService._CellPmRedirect(
+            effective_assignee=cast("UUID", cell_pm.id),
+            redirected=True,
+            reason="to_cell_pm",
+            dev_notes_line=(
+                f"\n\n[ASSIGNMENT REDIRECTED] cell-team {team_enum.value} "
+                f"{type_value} task must be owned by its cell PM; reassigned "
+                f"from {requested_assignee} to {cell_pm.slug}."
+            ),
+        )
+
     async def reassign_active_claim(
         self, task_id: UUID, new_assignee: UUID
     ) -> TaskTable | None:
@@ -7002,17 +7147,32 @@ class TaskService(BaseService):
                 refused_assignee=str(new_assignee),
             )
             return task
+        # Invariant backstop: an active claim on a cell-team planning/research/
+        # administrative task must land on the cell PM, not main-pm.
+        redirect = await self._resolve_cell_pm_redirect(task, new_assignee)
+        effective_assignee = redirect.effective_assignee
+        if redirect.redirected:
+            self.log.info(
+                "Active claim redirected",
+                task_id=str(task_id),
+                requested_assignee=str(new_assignee),
+                effective_assignee=str(effective_assignee),
+                reason=redirect.reason,
+            )
+            if redirect.dev_notes_line is not None:
+                task.dev_notes = (task.dev_notes or "") + redirect.dev_notes_line
+
         now = datetime.now(UTC)
-        task.assigned_to = cast("Any", new_assignee)
-        task.claimed_by = cast("Any", new_assignee)
+        task.assigned_to = cast("Any", effective_assignee)
+        task.claimed_by = cast("Any", effective_assignee)
         task.claimed_at = now
         task.last_heartbeat_at = now
-        task.active_claimant_id = cast("Any", new_assignee)
+        task.active_claimant_id = cast("Any", effective_assignee)
         await self.session.flush()
         self.log.info(
             "Active task reassigned to a fresh claimant",
             task_id=str(task_id),
-            new_assignee=str(new_assignee),
+            new_assignee=str(effective_assignee),
         )
         return task
 

tests/unit/services/test_prompter.py

@@ -31,6 +31,7 @@ from roboco.seeds.initial_data import AGENT_UUIDS
 from roboco.services.base import ServiceError, ValidationError
 from roboco.services.prompter import (
     PrompterService,
+    _cell_teams,
     compose_description,
     derive_scale,
     get_prompter_service,
@@ -93,6 +94,49 @@ def test_derive_scale_single_vs_multi() -> None:
     assert derive_scale([]) == "single"
 
 
+# -----------------------------------------------------------------------------
+# the_work shape tolerance — the intake agent is an LLM and sometimes emits
+# the_work as a list of bare team-name strings ("backend") instead of the
+# documented {team, summary, items} objects. Every consumer must tolerate that
+# without raising (regression: preview-batch used to 500 with
+# "'str' object has no attribute 'get'").
+# -----------------------------------------------------------------------------
+
+def test_cell_teams_tolerates_bare_string_entries() -> None:
+    # The LLM emitted the_work as a list of team names, not objects.
+    assert _cell_teams(["backend", "frontend", "backend"]) == ["backend", "frontend"]
+    # A bare string that isn't a cell is skipped, just like a non-cell dict.
+    assert _cell_teams(["backend", "main_pm"]) == ["backend"]
+    assert _cell_teams(["nonsense"]) == []
+
+
+def test_lead_cell_team_tolerates_bare_string_entries() -> None:
+    draft = {"the_work": ["frontend", "backend"]}
+    assert PrompterService._lead_cell_team(draft, default=Team.BACKEND) is Team.FRONTEND
+    # First valid cell wins; an invalid bare string is skipped.
+    draft = {"the_work": ["nonsense", "ux_ui"]}
+    assert PrompterService._lead_cell_team(draft, default=Team.BACKEND) is Team.UX_UI
+
+
+def test_derive_scale_tolerates_bare_string_entries() -> None:
+    assert derive_scale(["backend"]) == "single"
+    assert derive_scale(["backend", "frontend"]) == "multi"
+
+
+def test_compose_description_renders_bare_string_work_entries() -> None:
+    draft = {
+        "objective": "Fix the intake batch preview.",
+        "the_work": ["backend", "frontend"],
+        "acceptance_criteria": ["Preview no longer 500s"],
+    }
+    md = compose_description(draft)
+    # Each bare string renders as a cell heading; multi-cell gets the board-led line.
+    assert "## The Work" in md
+    assert "**Backend**" in md
+    assert "**Frontend**" in md
+    assert "Board-led" in md
+
+
 def test_compose_description_single_cell_markdown() -> None:
     draft = {
         "objective": "Let humans track token usage.",
@@ -630,3 +674,26 @@ def test_preview_batch_rejects_empty() -> None:
     service = get_prompter_service()
     with pytest.raises(ValidationError):
         service.preview_batch([])
+
+
+def test_preview_batch_tolerates_bare_string_the_work() -> None:
+    """Regression: the LLM sometimes emits the_work as bare team-name strings.
+    preview_batch must not 500 on that shape (it did: 'str' has no 'get')."""
+    service = get_prompter_service()
+    drafts: list[dict[str, Any]] = [
+        {
+            "title": "A",
+            "project_id": str(uuid4()),
+            "the_work": ["backend"],
+            "intends_to_touch": ["a.py"],
+        },
+        {
+            "title": "B",
+            "project_id": str(uuid4()),
+            "the_work": ["backend", "frontend"],
+            "intends_to_touch": ["b.py"],
+        },
+    ]
+    result = service.preview_batch(drafts)
+    assert isinstance(result["waves"], list)
+    assert isinstance(result["warnings"], list)

tests/unit/services/test_task_assignment_invariants.py

@@ -0,0 +1,308 @@
+"""Cell-team PM-owned children must be assigned to their cell PM.
+
+A descendant task whose ``team`` is a cell team and whose ``task_type`` is a
+non-code work type (``planning`` / ``research`` / ``administrative`` /
+``documentation`` / ``design``) belongs to that cell's PM. Assigning it to
+``main-pm`` (or any other mismatched owner) deadlocks the escalation chain,
+because the Main PM escalates up to the board, which cannot own cell
+coordination work. The service-layer invariant redirects such assignments to
+the correct cell PM at reassign / reassign-active-claim time and writes an
+``[ASSIGNMENT REDIRECTED]`` line to ``dev_notes`` so the audit is visible in
+the task body, not only in logs. The create path does NOT redirect (per the
+decision that reassign is the single backstop) and is regression-tested as
+such. Direct ORM writes from the escalation chain are out of scope.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+from uuid import UUID, uuid4
+
+import pytest
+from roboco.models.base import TaskStatus, TaskType, Team
+from roboco.services.task import TaskService, _is_cell_pm_owned_task
+
+
+def _bind(svc: TaskService, name: str, value: object) -> None:
+    object.__setattr__(svc, name, value)
+
+
+def _service() -> TaskService:
+    session = MagicMock()
+    session.flush = AsyncMock()
+    return TaskService(session)
+
+
+def _task(
+    *,
+    team: Team,
+    task_type: TaskType,
+    assigned_to: UUID | None = None,
+    status: TaskStatus = TaskStatus.CLAIMED,
+    parent_task_id: object = "USE_DEFAULT",
+) -> MagicMock:
+    """Build a task-shaped mock. ``parent_task_id`` defaults to a fresh UUID
+    (a descendant); pass an explicit ``None`` for a root task. The sentinel
+    avoids the truthiness collision where ``None`` and "use a default" both
+    look the same to ``or`` / ternary ``if-else``.
+    """
+    resolved_parent = uuid4() if parent_task_id == "USE_DEFAULT" else parent_task_id
+    return MagicMock(
+        id=uuid4(),
+        parent_task_id=resolved_parent,
+        team=team,
+        task_type=task_type,
+        assigned_to=assigned_to,
+        claimed_by=assigned_to,
+        active_claimant_id=assigned_to,
+        dev_notes="",
+        status=status,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Pure predicate — every non-code cell-team child type is PM-owned
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "task_type",
+    [
+        TaskType.PLANNING,
+        TaskType.RESEARCH,
+        TaskType.ADMINISTRATIVE,
+        TaskType.DOCUMENTATION,
+        TaskType.DESIGN,
+    ],
+)
+def test_cell_non_code_child_is_pm_owned(task_type: TaskType) -> None:
+    """Every non-code cell-team child task is PM-owned by its cell PM."""
+    task = _task(team=Team.BACKEND, task_type=task_type)
+    assert _is_cell_pm_owned_task(task) is True
+
+
+def test_cell_code_child_is_not_pm_owned() -> None:
+    """Leaf code work is delegated by the cell PM to a developer, not owned
+    by the cell PM — it is NOT subject to the redirect."""
+    task = _task(team=Team.BACKEND, task_type=TaskType.CODE)
+    assert _is_cell_pm_owned_task(task) is False
+
+
+def test_root_cell_planning_is_not_pm_owned() -> None:
+    """Roots are Main-PM coordination work; only descendants are PM-owned."""
+    task = _task(
+        team=Team.BACKEND,
+        task_type=TaskType.PLANNING,
+        parent_task_id=None,
+    )
+    assert _is_cell_pm_owned_task(task) is False
+
+
+def test_main_pm_team_planning_is_not_pm_owned() -> None:
+    """The invariant is cell-team-specific; main-pm team tasks skip it."""
+    task = _task(team=Team.MAIN_PM, task_type=TaskType.PLANNING)
+    assert _is_cell_pm_owned_task(task) is False
+
+
+# ---------------------------------------------------------------------------
+# Service-layer redirect at reassign — happy path
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_reassign_redirects_main_pm_to_cell_pm_for_planning_child() -> None:
+    svc = _service()
+    be_pm_id = uuid4()
+    main_pm_id = uuid4()
+    task = _task(
+        team=Team.BACKEND,
+        task_type=TaskType.PLANNING,
+        assigned_to=main_pm_id,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False))
+    _bind(
+        svc,
+        "cell_pm_for_team",
+        AsyncMock(return_value=MagicMock(id=be_pm_id, slug="be-pm")),
+    )
+
+    result = await svc.reassign(task.id, main_pm_id)
+
+    assert result is task
+    assert task.assigned_to == be_pm_id
+    assert task.claimed_by == be_pm_id
+    # Reassign now writes the audit line so the redirect is visible in the
+    # task body, not just in logs.
+    assert "[ASSIGNMENT REDIRECTED]" in task.dev_notes
+    assert "be-pm" in task.dev_notes
+
+
+@pytest.mark.asyncio
+async def test_reassign_keeps_cell_pm_for_planning_child() -> None:
+    """A cell PM request on a cell-team PM-owned task is a no-op redirect."""
+    svc = _service()
+    be_pm_id = uuid4()
+    task = _task(
+        team=Team.BACKEND,
+        task_type=TaskType.PLANNING,
+        assigned_to=None,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False))
+    _bind(
+        svc,
+        "cell_pm_for_team",
+        AsyncMock(return_value=MagicMock(id=be_pm_id, slug="be-pm")),
+    )
+
+    result = await svc.reassign(task.id, be_pm_id)
+
+    assert result is task
+    assert task.assigned_to == be_pm_id
+    # No redirect → no audit line.
+    assert "[ASSIGNMENT REDIRECTED]" not in task.dev_notes
+
+
+@pytest.mark.asyncio
+async def test_reassign_active_claim_redirects_main_pm_to_cell_pm() -> None:
+    svc = _service()
+    be_pm_id = uuid4()
+    main_pm_id = uuid4()
+    task = _task(
+        team=Team.BACKEND,
+        task_type=TaskType.RESEARCH,
+        assigned_to=main_pm_id,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False))
+    _bind(
+        svc,
+        "cell_pm_for_team",
+        AsyncMock(return_value=MagicMock(id=be_pm_id, slug="be-pm")),
+    )
+
+    result = await svc.reassign_active_claim(task.id, main_pm_id)
+
+    assert result is task
+    assert task.assigned_to == be_pm_id
+    assert task.claimed_by == be_pm_id
+    assert task.active_claimant_id == be_pm_id
+    assert "[ASSIGNMENT REDIRECTED]" in task.dev_notes
+
+
+# ---------------------------------------------------------------------------
+# Service-layer redirect — failure modes
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_reassign_queues_task_when_no_cell_pm_exists() -> None:
+    """When no cell PM agent row exists for the team, queue the task (no
+    assignee) and log an error rather than silently routing to the caller's
+    requested assignee. The task stays PENDING until the agents table is
+    repaired; this is observable in logs and dev_notes.
+    """
+    svc = _service()
+    main_pm_id = uuid4()
+    task = _task(
+        team=Team.BACKEND,
+        task_type=TaskType.PLANNING,
+        assigned_to=main_pm_id,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False))
+    _bind(svc, "cell_pm_for_team", AsyncMock(return_value=None))
+
+    result = await svc.reassign(task.id, main_pm_id)
+
+    assert result is task
+    # Queued: assignee cleared so the orchestrator does not respawn into a
+    # misrouted hand-off.
+    assert task.assigned_to is None
+    assert task.claimed_by is None
+    assert "[ASSIGNMENT REDIRECTED]" in task.dev_notes
+    assert "queued" in task.dev_notes
+
+
+@pytest.mark.asyncio
+async def test_reassign_active_claim_queues_when_no_cell_pm_exists() -> None:
+    """Same missing-cell-PM fallback for the active-claim path."""
+    svc = _service()
+    main_pm_id = uuid4()
+    task = _task(
+        team=Team.BACKEND,
+        task_type=TaskType.DESIGN,
+        assigned_to=main_pm_id,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False))
+    _bind(svc, "cell_pm_for_team", AsyncMock(return_value=None))
+
+    result = await svc.reassign_active_claim(task.id, main_pm_id)
+
+    assert result is task
+    assert task.assigned_to is None
+    assert task.claimed_by is None
+    assert task.active_claimant_id is None
+    assert "[ASSIGNMENT REDIRECTED]" in task.dev_notes
+
+
+@pytest.mark.asyncio
+async def test_reassign_board_divert_fires_before_cell_pm_redirect() -> None:
+    """The board-divert guard must run BEFORE the cell-PM redirect. If a
+    board agent is asked to own a cell task, the divert-to-pool path is the
+    one that fires — not a cell-PM redirect that never matches because the
+    board agent is not the cell PM. This test exercises the ordering so a
+    future refactor that reorders the guards fails loudly.
+    """
+    svc = _service()
+    board_agent_id = uuid4()
+    task = _task(
+        team=Team.BACKEND,
+        task_type=TaskType.PLANNING,
+        assigned_to=board_agent_id,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    # Board guard says YES, this is a board agent.
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=True))
+    # Cell PM lookup must NOT be reached — if it is, the test fails because
+    # the AsyncMock has no return_value configured.
+    _bind(
+        svc,
+        "cell_pm_for_team",
+        AsyncMock(
+            side_effect=AssertionError(
+                "cell_pm_for_team must not run when board-divert fires first"
+            )
+        ),
+    )
+    diverted = AsyncMock()
+    _bind(svc, "_divert_owned_task_to_pool", diverted)
+
+    result = await svc.reassign(task.id, board_agent_id)
+
+    assert result is task
+    diverted.assert_awaited_once()
+    # Board-divert path returned before the cell-PM redirect mutated fields.
+    assert task.assigned_to == board_agent_id
+
+
+# ---------------------------------------------------------------------------
+# Regression — the create path does NOT redirect
+# ---------------------------------------------------------------------------
+
+
+def test_create_does_not_redirect_misassigned_cell_planning_child() -> None:
+    """Regression for the create-path backstop decision. The decision (see
+    audit, Gap 1) was: the create path no longer redirects to the cell PM,
+    because the reassign path is the single backstop and a second redirect
+    on create is duplicate coverage. If this test ever fails, someone has
+    re-added the create-path redirect and should be challenged on why.
+    """
+    svc = _service()
+    # If a `_redirect_cell_team_pm_task` method ever reappears, this assert
+    # should make the failure obvious in the diff.
+    assert not hasattr(svc, "_redirect_cell_team_pm_task"), (
+        "create-path redirect was re-added; remove it (see audit Gap 1)."
+    )