Harden cell-ownership invariant at reassign and dependency revival

A board/advisory role (product owner, head of marketing, auditor) has no
verb to build, document, or complete a cell task. The escalate path already
diverts such a hand-off to the pool; extend the same backstop to the two
remaining write-sites:

- reassign / reassign_active_claim: a refused board/advisory target is
  diverted to the pool for a role-matched claim instead of being planted as
  a non-workable owner. Normal handoff targets (qa, documenter, cell PM) are
  unaffected.
- dependency revival: when a blocked task's last dependency clears, resume
  in place only under a workable owner; re-home a board/advisory-or-absent
  owner on a cell task to the pool so it does not immediately re-deadlock.

All three sites share one audited pool-divert primitive so no direct status
set skips the transition audit.

Also reduce cyclomatic complexity below the project threshold for
unclaim_for_agent, the dependency-revival path, and the docs index source
expansion by extracting helpers (behavior-preserving), and add coverage for
the doc source-expansion paths.

roboco/services/optimal_brain/indexes/docs.py

@@ -73,17 +73,20 @@ class DocsIndexPlugin(BaseIndexPlugin):
         if "*" in source:
             return list(Path().glob(source))
         if source_path.is_dir():
-            md_files = [
+            return self._expand_directory(source_path)
-                f
+        return self._expand_file(source_path, source)
-                for f in source_path.rglob("*.md")
+
-                if not any(skip in f.parts for skip in SKIP_DIRECTORIES)
+    def _expand_directory(self, source_path: Path) -> list[Path]:
-            ]
+        """Recursively collect indexable doc files under a directory."""
-            txt_files = [
+        return [
-                f
+            f
-                for f in source_path.rglob("*.txt")
+            for pattern in ("*.md", "*.txt")
-                if not any(skip in f.parts for skip in SKIP_DIRECTORIES)
+            for f in source_path.rglob(pattern)
-            ]
+            if not any(skip in f.parts for skip in SKIP_DIRECTORIES)
-            return md_files + txt_files
+        ]
+
+    def _expand_file(self, source_path: Path, source: str) -> list[Path]:
+        """Resolve a single source path to an indexable doc file, or nothing."""
         if source_path.exists():
             # Only markdown/text files are docs; a recorded path to a source
             # file (e.g. a .tsx) is not indexable and is skipped quietly.

roboco/services/task.py

@@ -70,9 +70,9 @@ _ROLE_CLAIM_STATUSES: dict[str, set[TaskStatus]] = {
 # Board / advisory roles review and advise; they never own or execute a
 # descendant code task. Handing one to them (e.g. via the main_pm→product_owner
 # escalation rung) strands the work: the board has no verb to claim, build, or
-# complete it, and the dev's finished work deadlocks (#14). A descendant code
+# complete it, and the dev's finished work deadlocks. A descendant code task
-# task that would otherwise land on one of these roles is instead released to
+# that would otherwise land on one of these roles is instead released to the
-# the pool for a role-matched cell agent to reclaim.
+# pool for a role-matched cell agent to reclaim.
 _BOARD_ADVISORY_ROLES: frozenset[AgentRole] = frozenset(
     {AgentRole.PRODUCT_OWNER, AgentRole.HEAD_MARKETING, AgentRole.AUDITOR}
 )
@@ -82,7 +82,7 @@ _BOARD_ADVISORY_ROLES: frozenset[AgentRole] = frozenset(
 # board/advisory role has no verb to build or complete. CODE → developer,
 # DOCUMENTATION → documenter, DESIGN → UX/design cell. The remaining types
 # (PLANNING / RESEARCH / ADMINISTRATIVE) route to a PM, not a cell agent, and
-# are not diverted here — the #14 guard only fires for board/advisory targets.
+# are not diverted here — the guard only fires for board/advisory targets.
 _DESCENDANT_EXECUTABLE_TASK_TYPES: frozenset[str] = frozenset(
     {TaskType.CODE.value, TaskType.DOCUMENTATION.value, TaskType.DESIGN.value}
 )
@@ -95,7 +95,7 @@ _CELL_TEAMS: frozenset[str] = frozenset({"backend", "frontend", "ux_ui"})
 
 
 def _is_descendant_executable_task(task: TaskTable) -> bool:
-    """True for a child task that does cell-executed work (#14 guard).
+    """True for a child task that does cell-executed work.
 
     A board/advisory role must never become the assignee of such a task: it has
     no verb to build, document, or complete it. ``task_type`` is a ``TaskType``
@@ -2292,51 +2292,10 @@ class TaskService(BaseService):
         task = await self.get(task_id)
         if task is None or task.assigned_to != agent_id:
             return None
-        # #176: an agent assigned a `pending` task it never claimed (any
-        # persistent claim-time rejection — e.g. a gate the agent cannot
-        # satisfy) is otherwise trapped: unclaim/i_am_idle/i_am_blocked all
-        # reject from pending-assigned, so it loops until budget-reap and
-        # the task is left orphaned (pending, assigned, no progress).
-        # Releasing the assignment is a no-status-change escape (the row is
-        # already pending; no transition, so no lifecycle validation and no
-        # WorkSession to abandon — it was never claimed). The task returns
-        # to the pool for the dispatcher to reassign.
         if task.status == TaskStatus.PENDING:
-            task.assigned_to = cast("Any", None)
+            return await self._unclaim_pending_assignment(task)
-            task.active_claimant_id = cast("Any", None)
-            await self.session.flush()
-            return task
-        # A task the agent owns but cannot advance — it is `blocked` (a blocker
-        # it cannot self-resolve, or a dependency block) — is otherwise a trap:
-        # from `blocked` the agent has no legal forward verb and the dispatcher
-        # keeps respawning it. Releasing the claim returns the task to the pool
-        # for the cell PM to re-delegate. Audited; the active WorkSession is
-        # abandoned so a re-claim does not trip the uniqueness constraint.
         if task.status == TaskStatus.BLOCKED:
-            pre_status = (
+            return await self._unclaim_from_blocked(task)
-                task.status.value
-                if isinstance(task.status, TaskStatus)
-                else str(task.status)
-            )
-            prior_owner = cast("Any", task.claimed_by or task.assigned_to)
-            if task.work_session_id:
-                await self._abandon_work_session_best_effort(
-                    task.work_session_id, reason="agent-unclaim-from-blocked"
-                )
-                task.work_session_id = cast("Any", None)
-            task.status = TaskStatus.PENDING
-            task.assigned_to = cast("Any", None)
-            task.claimed_by = cast("Any", None)
-            task.active_claimant_id = cast("Any", None)
-            await self.session.flush()
-            self._emit_status_transition_audit(
-                task,
-                from_status=pre_status,
-                to_status=TaskStatus.PENDING.value,
-                agent_role=None,
-                audit_agent_id=prior_owner,
-            )
-            return task
         if task.status not in (TaskStatus.CLAIMED, TaskStatus.IN_PROGRESS):
             return None
 
@@ -2362,7 +2321,7 @@ class TaskService(BaseService):
         # _validate_and_set_status only updates `status`; clearing the
         # claim is the unclaim's specific side effect. Also abandon the
         # active WorkSession so a re-claim doesn't trip the uniqueness
-        # constraint (audit D-41).
+        # constraint.
         if task.work_session_id:
             await self._abandon_work_session_best_effort(
                 task.work_session_id, reason="agent-unclaim"
@@ -2373,6 +2332,57 @@ class TaskService(BaseService):
         await self.session.flush()
         return task
 
+    async def _unclaim_pending_assignment(self, task: TaskTable) -> TaskTable:
+        """Release a never-claimed ``pending`` assignment (no status change).
+
+        An agent assigned a ``pending`` task it never claimed (a persistent
+        claim-time rejection it cannot satisfy) is otherwise trapped: unclaim /
+        i_am_idle / i_am_blocked all reject from pending-assigned, so it loops
+        until budget-reap and the task is orphaned. The row is already pending,
+        so clearing the assignment is a no-status-change escape — no transition,
+        no lifecycle validation, no WorkSession to abandon (it was never
+        claimed). The task returns to the pool for the dispatcher to reassign.
+        """
+        task.assigned_to = cast("Any", None)
+        task.active_claimant_id = cast("Any", None)
+        await self.session.flush()
+        return task
+
+    async def _unclaim_from_blocked(self, task: TaskTable) -> TaskTable:
+        """Release a ``blocked`` claim back to the pool (audited).
+
+        A task the agent owns but cannot advance (a blocker it cannot
+        self-resolve, or a dependency block) is otherwise a trap: from
+        ``blocked`` the agent has no legal forward verb and the dispatcher keeps
+        respawning it. Releasing the claim returns the task to the pool for the
+        cell PM to re-delegate. The active WorkSession is abandoned so a re-claim
+        does not trip the uniqueness constraint.
+        """
+        pre_status = (
+            task.status.value
+            if isinstance(task.status, TaskStatus)
+            else str(task.status)
+        )
+        prior_owner = cast("Any", task.claimed_by or task.assigned_to)
+        if task.work_session_id:
+            await self._abandon_work_session_best_effort(
+                task.work_session_id, reason="agent-unclaim-from-blocked"
+            )
+            task.work_session_id = cast("Any", None)
+        task.status = TaskStatus.PENDING
+        task.assigned_to = cast("Any", None)
+        task.claimed_by = cast("Any", None)
+        task.active_claimant_id = cast("Any", None)
+        await self.session.flush()
+        self._emit_status_transition_audit(
+            task,
+            from_status=pre_status,
+            to_status=TaskStatus.PENDING.value,
+            agent_role=None,
+            audit_agent_id=prior_owner,
+        )
+        return task
+
     async def resume_for_agent(self, task_id: UUID, agent_id: UUID) -> TaskTable | None:
         """Voluntary resume: transition paused task → in_progress for the assignee.
 
@@ -3508,7 +3518,7 @@ class TaskService(BaseService):
         and the orchestrator re-spawns them. Without this, escalation
         loses the dev's identity permanently.
 
-        #14 invariant: a descendant executable task (code / documentation /
+        Invariant: a descendant executable task (code / documentation /
         design) is NEVER assigned to a board/advisory role (they cannot own
         cell-executed work). Such an escalation is diverted to a pool release so
         a role-matched cell agent reclaims it. Enforced here — the single write
@@ -4072,7 +4082,7 @@ class TaskService(BaseService):
             ]
             # If no more dependencies, unblock (system action - no role validation)
             if not task.dependency_ids and task.status == TaskStatus.BLOCKED:
-                self._validate_and_set_status(task, TaskStatus.IN_PROGRESS, None)
+                await self._revive_unblocked_dependent(task)
                 self.log.info(
                     "Task auto-unblocked",
                     task_id=str(task.id),
@@ -4081,6 +4091,30 @@ class TaskService(BaseService):
 
         await self.session.flush()
 
+    async def _revive_unblocked_dependent(self, task: TaskTable) -> None:
+        """Resume — or re-home — a task whose last dependency just cleared.
+
+        Resume in place when a workable owner still holds it. Re-home to the
+        pool when the owner is board/advisory or absent on a cell task: such an
+        owner has no verb to work it, so resuming would re-deadlock the task the
+        instant its dependency lands.
+        """
+        owner = cast("Any", task.claimed_by or task.assigned_to)
+        needs_rehome = owner is None or await self._is_board_advisory_agent(owner)
+        if needs_rehome and (
+            _is_descendant_executable_task(task) or _is_cell_team_task(task)
+        ):
+            await self._divert_owned_task_to_pool(
+                task,
+                note=(
+                    "\n\n[REVIVAL REDIRECTED] dependency cleared but the owner"
+                    " could not work this cell task (board/advisory or"
+                    " unassigned). Released to the pool for a role-matched claim."
+                ),
+            )
+        else:
+            self._validate_and_set_status(task, TaskStatus.IN_PROGRESS, None)
+
     # =========================================================================
     # PROGRESS AND CHECKPOINTS
     # =========================================================================
@@ -5453,6 +5487,31 @@ class TaskService(BaseService):
         task = await self.get(task_id)
         if task is None:
             return None
+        # Invariant backstop: never plant a board/advisory role as the owner of a
+        # cell task. `apply_escalation` guards the escalate path; this guards the
+        # direct reassign setter (gateway handoffs + HTTP route). A refused
+        # hand-off is diverted to the pool for a role-matched claim. Normal
+        # handoff targets (qa/documenter/cell_pm) are not board roles, so the
+        # guard never fires for them.
+        if (
+            new_assignee is not None
+            and (_is_descendant_executable_task(task) or _is_cell_team_task(task))
+            and await self._is_board_advisory_agent(new_assignee)
+        ):
+            await self._divert_owned_task_to_pool(
+                task,
+                note=(
+                    "\n\n[REASSIGN REDIRECTED] attempted to assign this cell task"
+                    " to a board/advisory role that cannot own cell-executed work."
+                    " Released to the pool for a role-matched claim instead."
+                ),
+            )
+            self.log.info(
+                "Cell task reassign to a board/advisory role diverted to pool",
+                task_id=str(task_id),
+                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
         await self.session.flush()
@@ -5480,6 +5539,25 @@ class TaskService(BaseService):
             return None
         if task.status not in (TaskStatus.CLAIMED, TaskStatus.IN_PROGRESS):
             return None
+        # Invariant backstop (mirrors `reassign`): an active claim must not be
+        # handed to a board/advisory role on a cell task — divert to the pool.
+        if (
+            _is_descendant_executable_task(task) or _is_cell_team_task(task)
+        ) and await self._is_board_advisory_agent(new_assignee):
+            await self._divert_owned_task_to_pool(
+                task,
+                note=(
+                    "\n\n[REASSIGN REDIRECTED] attempted to hand this active cell"
+                    " task to a board/advisory role that cannot own cell-executed"
+                    " work. Released to the pool for a role-matched claim instead."
+                ),
+            )
+            self.log.info(
+                "Active cell task reassign to a board/advisory role diverted",
+                task_id=str(task_id),
+                refused_assignee=str(new_assignee),
+            )
+            return task
         now = datetime.now(UTC)
         task.assigned_to = cast("Any", new_assignee)
         task.claimed_by = cast("Any", new_assignee)
@@ -5736,7 +5814,7 @@ class TaskService(BaseService):
         if target is None:
             return None
 
-        # The board/advisory guard (#14) lives in apply_escalation so the HTTP
+        # The board/advisory guard lives in apply_escalation so the HTTP
         # escalate route is covered too; nothing extra to do here.
         await self.apply_escalation(
             task=task,
@@ -5755,21 +5833,16 @@ class TaskService(BaseService):
         role = result.scalar_one_or_none()
         return role in _BOARD_ADVISORY_ROLES
 
-    async def _release_code_task_to_pool(
+    async def _divert_owned_task_to_pool(self, task: TaskTable, *, note: str) -> None:
-        self,
+        """Clear ownership and return ``task`` to PENDING for a role-matched claim.
-        *,
-        task: TaskTable,
-        escalator_slug: str,
-        blocked_target_slug: str,
-        reason: str,
-    ) -> None:
-        """Release a descendant executable task to PENDING for a role-matched claim.
 
-        Used instead of escalating a code / documentation / design task onto a
+        Shared backstop for the cell-ownership invariant: a board/advisory
-        board/advisory role (#14). Clears the assignee so the orchestrator's
+        role must never own — or be revived as the owner of — a cell task. The
-        role-matched dispatch picks it up cleanly, sets PENDING (a valid
+        escalation, reassign, and dependency-revival write-sites all funnel a
-        re-dispatch source), and appends an audit note explaining why the board
+        refused hand-off here. Sets PENDING directly (bypassing the strict
-        hand-off was refused.
+        transition validator), so emits the ``task.pending`` audit explicitly —
+        no status change may skip the audit log. ``note`` is appended to
+        ``dev_notes`` explaining why the hand-off was refused.
         """
         pre_status = (
             task.status.value
@@ -5781,19 +5854,8 @@ class TaskService(BaseService):
         task.claimed_by = cast("Any", None)
         task.active_claimant_id = cast("Any", None)
         task.status = TaskStatus.PENDING
-        existing_notes = task.dev_notes or ""
+        task.dev_notes = (task.dev_notes or "") + note
-        note = (
-            f"\n\n[ESCALATION REDIRECTED] {escalator_slug} escalated this"
-            f" executable task toward {blocked_target_slug} (a board/advisory role"
-            f" that cannot own cell-executed work). Released to the pool for a"
-            f" role-matched claim instead."
-            f"\nReason: {reason}"
-        )
-        task.dev_notes = existing_notes + note
         await self.session.flush()
-        # This path sets PENDING directly (bypassing the strict transition
-        # validator), so emit the task.pending audit explicitly — no status
-        # change may skip the audit log.
         self._emit_status_transition_audit(
             task,
             from_status=pre_status,
@@ -5801,6 +5863,33 @@ class TaskService(BaseService):
             agent_role=None,
             audit_agent_id=prior_owner,
         )
+
+    async def _release_code_task_to_pool(
+        self,
+        *,
+        task: TaskTable,
+        escalator_slug: str,
+        blocked_target_slug: str,
+        reason: str,
+    ) -> None:
+        """Release a descendant executable task to PENDING for a role-matched claim.
+
+        Used instead of escalating a code / documentation / design task onto a
+        board/advisory role. Clears the assignee so the orchestrator's
+        role-matched dispatch picks it up cleanly, sets PENDING (a valid
+        re-dispatch source), and appends an audit note explaining why the board
+        hand-off was refused.
+        """
+        await self._divert_owned_task_to_pool(
+            task,
+            note=(
+                f"\n\n[ESCALATION REDIRECTED] {escalator_slug} escalated this"
+                f" executable task toward {blocked_target_slug} (a board/advisory"
+                f" role that cannot own cell-executed work). Released to the pool"
+                f" for a role-matched claim instead."
+                f"\nReason: {reason}"
+            ),
+        )
         self.log.info(
             "Descendant executable task released to pool instead of board escalation",
             task_id=str(task.id),

tests/unit/services/optimal_brain/test_docs_index_expand_source.py

@@ -0,0 +1,65 @@
+"""Source-expansion behavior for the documentation index plugin.
+
+Locks the glob / directory / single-file resolution paths so the
+complexity-driven extraction into ``_expand_directory`` / ``_expand_file``
+stays behavior-preserving.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from roboco.services.optimal_brain.indexes.docs import DocsIndexPlugin
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _plugin() -> DocsIndexPlugin:
+    return DocsIndexPlugin()
+
+
+def test_expand_single_markdown_file(tmp_path: Path) -> None:
+    doc = tmp_path / "guide.md"
+    doc.write_text("# hi", encoding="utf-8")
+    assert _plugin()._expand_source(str(doc)) == [doc]
+
+
+def test_expand_single_text_file(tmp_path: Path) -> None:
+    doc = tmp_path / "notes.txt"
+    doc.write_text("hi", encoding="utf-8")
+    assert _plugin()._expand_source(str(doc)) == [doc]
+
+
+def test_expand_non_doc_file_is_skipped(tmp_path: Path) -> None:
+    src = tmp_path / "component.tsx"
+    src.write_text("export {}", encoding="utf-8")
+    assert _plugin()._expand_source(str(src)) == []
+
+
+def test_expand_missing_path_is_empty(tmp_path: Path) -> None:
+    assert _plugin()._expand_source(str(tmp_path / "nope.md")) == []
+
+
+def test_expand_directory_collects_md_and_txt(tmp_path: Path) -> None:
+    (tmp_path / "a.md").write_text("a", encoding="utf-8")
+    (tmp_path / "b.txt").write_text("b", encoding="utf-8")
+    (tmp_path / "c.tsx").write_text("c", encoding="utf-8")
+    nested = tmp_path / "sub"
+    nested.mkdir()
+    (nested / "d.md").write_text("d", encoding="utf-8")
+
+    found = {p.name for p in _plugin()._expand_source(str(tmp_path))}
+
+    assert found == {"a.md", "b.txt", "d.md"}
+
+
+def test_expand_directory_skips_ignored_dirs(tmp_path: Path) -> None:
+    (tmp_path / "keep.md").write_text("k", encoding="utf-8")
+    ignored = tmp_path / "node_modules"
+    ignored.mkdir()
+    (ignored / "dep.md").write_text("x", encoding="utf-8")
+
+    found = {p.name for p in _plugin()._expand_source(str(tmp_path))}
+
+    assert found == {"keep.md"}

tests/unit/services/test_escalation_board_guard.py

@@ -1,4 +1,4 @@
-"""#14: a descendant executable task is never assigned to a board/advisory role.
+"""A descendant executable task is never assigned to a board/advisory role.
 
 The main_pm -> product_owner escalation rung used to hand an in_progress child
 code task to the Product Owner and mark it BLOCKED. The board has no verb to
@@ -66,13 +66,13 @@ def test_non_cell_team_task_is_not_flagged() -> None:
 
 
 def test_descendant_documentation_task_is_flagged() -> None:
-    # #14 broaden: documentation is cell-executed (documenter), not board work.
+    # Documentation is cell-executed (documenter), not board work.
     task = MagicMock(parent_task_id=uuid4(), task_type=TaskType.DOCUMENTATION)
     assert _is_descendant_executable_task(task) is True
 
 
 def test_descendant_design_task_is_flagged() -> None:
-    # #14 broaden: design is cell-executed (UX/design cell), not board work.
+    # Design is cell-executed (UX/design cell), not board work.
     task = MagicMock(parent_task_id=uuid4(), task_type=TaskType.DESIGN)
     assert _is_descendant_executable_task(task) is True
 
@@ -153,8 +153,8 @@ async def test_apply_escalation_diverts_descendant_code_to_board() -> None:
 
 @pytest.mark.asyncio
 async def test_apply_escalation_diverts_descendant_documentation_to_board() -> None:
-    # #14 broaden: a descendant DOCUMENTATION task escalated to a board role is
+    # A descendant DOCUMENTATION task escalated to a board role is diverted too
-    # diverted too — the board has no verb to write/complete docs either.
+    # — the board has no verb to write/complete docs either.
     svc = _service()
     target_id = uuid4()
     task = MagicMock(
@@ -184,7 +184,7 @@ async def test_apply_escalation_diverts_descendant_documentation_to_board() -> N
 
 @pytest.mark.asyncio
 async def test_apply_escalation_diverts_descendant_design_to_board() -> None:
-    # #14 broaden: a descendant DESIGN task escalated to a board role is diverted.
+    # A descendant DESIGN task escalated to a board role is diverted.
     svc = _service()
     target_id = uuid4()
     task = MagicMock(
@@ -422,3 +422,253 @@ async def test_unblock_with_restore_emits_audit_event() -> None:
     assert kwargs["event_type"] == "task.in_progress"
     assert kwargs["details"]["from_status"] == "blocked"
     assert kwargs["details"]["to_status"] == "in_progress"
+
+
+# ---------------------------------------------------------------------------
+# reassign / reassign_active_claim board-role divert — same invariant at the
+# direct reassign setters, not just the escalate path
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_reassign_diverts_cell_task_to_board_role() -> None:
+    svc = _service()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=uuid4(),
+        task_type=TaskType.CODE,
+        assigned_to=uuid4(),
+        claimed_by=uuid4(),
+        active_claimant_id=uuid4(),
+        dev_notes="prior",
+        status=TaskStatus.IN_PROGRESS,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=True))
+
+    out = await svc.reassign(task.id, uuid4())
+
+    assert out is task
+    assert task.status == TaskStatus.PENDING
+    assert task.assigned_to is None
+    assert task.claimed_by is None
+    assert task.active_claimant_id is None
+    assert "REASSIGN REDIRECTED" in task.dev_notes
+
+
+@pytest.mark.asyncio
+async def test_reassign_assigns_non_board_target_normally() -> None:
+    svc = _service()
+    new_assignee = uuid4()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=uuid4(),
+        task_type=TaskType.CODE,
+        assigned_to=uuid4(),
+        claimed_by=uuid4(),
+        status=TaskStatus.IN_PROGRESS,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False))
+
+    out = await svc.reassign(task.id, new_assignee)
+
+    assert out is task
+    assert task.assigned_to == new_assignee
+    assert task.claimed_by == new_assignee
+    assert task.status == TaskStatus.IN_PROGRESS  # status untouched by a handoff
+
+
+@pytest.mark.asyncio
+async def test_reassign_none_clears_without_consulting_board_check() -> None:
+    svc = _service()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=uuid4(),
+        task_type=TaskType.CODE,
+        assigned_to=uuid4(),
+        claimed_by=uuid4(),
+        status=TaskStatus.IN_PROGRESS,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    board_check = AsyncMock(return_value=True)
+    _bind(svc, "_is_board_advisory_agent", board_check)
+
+    out = await svc.reassign(task.id, None)
+
+    # new_assignee=None short-circuits the guard (clearing assignment is the
+    # documented "escalated to CEO, acts via UI" path).
+    board_check.assert_not_called()
+    assert out is task
+    assert task.assigned_to is None
+    assert task.claimed_by is None
+
+
+@pytest.mark.asyncio
+async def test_reassign_active_claim_diverts_cell_task_to_board_role() -> None:
+    svc = _service()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=uuid4(),
+        task_type=TaskType.CODE,
+        assigned_to=uuid4(),
+        claimed_by=uuid4(),
+        active_claimant_id=uuid4(),
+        dev_notes="prior",
+        status=TaskStatus.IN_PROGRESS,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=True))
+
+    out = await svc.reassign_active_claim(task.id, uuid4())
+
+    assert out is task
+    assert task.status == TaskStatus.PENDING
+    assert task.assigned_to is None
+    assert "REASSIGN REDIRECTED" in task.dev_notes
+
+
+@pytest.mark.asyncio
+async def test_reassign_active_claim_assigns_non_board_target_normally() -> None:
+    svc = _service()
+    new_assignee = uuid4()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=uuid4(),
+        task_type=TaskType.CODE,
+        assigned_to=uuid4(),
+        claimed_by=uuid4(),
+        active_claimant_id=uuid4(),
+        status=TaskStatus.IN_PROGRESS,
+    )
+    _bind(svc, "get", AsyncMock(return_value=task))
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False))
+
+    out = await svc.reassign_active_claim(task.id, new_assignee)
+
+    assert out is task
+    assert task.assigned_to == new_assignee
+    assert task.claimed_by == new_assignee
+    assert task.active_claimant_id == new_assignee
+
+
+# ---------------------------------------------------------------------------
+# _unblock_dependents revival re-home — a dependency clearing must not revive a
+# cell task under a board/advisory or absent owner
+# ---------------------------------------------------------------------------
+
+
+def _blocked_dependent(task: MagicMock) -> AsyncMock:
+    """A session.execute that yields ``task`` as the only dependency-blocked row."""
+    result = MagicMock()
+    result.scalars.return_value.all.return_value = [task]
+    return AsyncMock(return_value=result)
+
+
+@pytest.mark.asyncio
+async def test_unblock_dependents_rehomes_board_owned_cell_task() -> None:
+    svc = _service()
+    completed_id = uuid4()
+    board_owner = uuid4()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=uuid4(),
+        task_type=TaskType.CODE,
+        dependency_ids=[completed_id],
+        status=TaskStatus.BLOCKED,
+        assigned_to=board_owner,
+        claimed_by=board_owner,
+        active_claimant_id=board_owner,
+        dev_notes="prior",
+    )
+    svc.session.execute = _blocked_dependent(task)
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=True))
+
+    await svc._unblock_dependents(completed_id)
+
+    assert task.dependency_ids == []
+    assert task.status == TaskStatus.PENDING
+    assert task.assigned_to is None
+    assert task.claimed_by is None
+    assert "REVIVAL REDIRECTED" in task.dev_notes
+
+
+@pytest.mark.asyncio
+async def test_unblock_dependents_rehomes_ownerless_cell_task() -> None:
+    svc = _service()
+    completed_id = uuid4()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=uuid4(),
+        task_type=TaskType.CODE,
+        dependency_ids=[completed_id],
+        status=TaskStatus.BLOCKED,
+        assigned_to=None,
+        claimed_by=None,
+        active_claimant_id=None,
+        dev_notes="",
+    )
+    svc.session.execute = _blocked_dependent(task)
+    board_check = AsyncMock(return_value=False)
+    _bind(svc, "_is_board_advisory_agent", board_check)
+
+    await svc._unblock_dependents(completed_id)
+
+    # owner is None → needs_rehome short-circuits True without the board check.
+    board_check.assert_not_called()
+    assert task.status == TaskStatus.PENDING
+    assert "REVIVAL REDIRECTED" in task.dev_notes
+
+
+@pytest.mark.asyncio
+async def test_unblock_dependents_resumes_dev_owned_cell_task() -> None:
+    svc = _service()
+    completed_id = uuid4()
+    dev_owner = uuid4()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=uuid4(),
+        task_type=TaskType.CODE,
+        dependency_ids=[completed_id],
+        status=TaskStatus.BLOCKED,
+        assigned_to=dev_owner,
+        claimed_by=dev_owner,
+    )
+    svc.session.execute = _blocked_dependent(task)
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False))
+    validate_mock = MagicMock()
+    _bind(svc, "_validate_and_set_status", validate_mock)
+
+    await svc._unblock_dependents(completed_id)
+
+    # Workable owner → resume in place, owner preserved (not cleared).
+    validate_mock.assert_called_once()
+    assert validate_mock.call_args.args[1] == TaskStatus.IN_PROGRESS
+    assert task.assigned_to == dev_owner
+
+
+@pytest.mark.asyncio
+async def test_unblock_dependents_resumes_board_owned_root_task() -> None:
+    # A ROOT task legitimately owned by a board role (e.g. a product root the PO
+    # reviews) must resume in place — the cell guard targets descendants only.
+    svc = _service()
+    completed_id = uuid4()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=None,
+        task_type=TaskType.CODE,
+        team=Team.BOARD,
+        dependency_ids=[completed_id],
+        status=TaskStatus.BLOCKED,
+        assigned_to=uuid4(),
+        claimed_by=uuid4(),
+    )
+    svc.session.execute = _blocked_dependent(task)
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=True))
+    validate_mock = MagicMock()
+    _bind(svc, "_validate_and_set_status", validate_mock)
+
+    await svc._unblock_dependents(completed_id)
+
+    validate_mock.assert_called_once()
+    assert validate_mock.call_args.args[1] == TaskStatus.IN_PROGRESS