task-service: completion hooks + escalation/cancel/audit hardening

Closes the task-service cluster (#21/#98, #99, #100, #101, #103, #216; #102
verified already-covered, #217 verified already-guarded):

- #21/#98: ceo_approve now closes the work session + triggers completion
  hooks before worktree removal (no-op when work_session_id is None), so a
  CEO-approved task lands the same close-path as PM-completed.
- #99: apply_escalation routes through the transition validator with an
  enumerated escalation exemption (_ESCALATABLE_TO_BLOCKED) instead of an
  arbitrary source->BLOCKED write; BACKLOG is refused.
- #100: branchless ceo_reject awaiting_ceo_approval->pending gets a real
  spec edge (ceo_reject_to_pool ActionSpec + _STATUS_TRANSITIONS entry) so
  future admin-override tightening can't wedge the path.
- #101: revision_count bump is documented as the single chokepoint, with
  the pre-block RESTORE path undoing it when restoring a snapshotted
  needs_revision (same cycle resuming, not a new rejection).
- #103: cancel cascade surfaces non-terminal orphans instead of swallowing
  the role violation.
- #216: _remove_task_worktree_on_terminal escalates recurring FS/permission
  failure (audit/notify after N) instead of silent-failing forever.
- #102: pinned in test_verb_runner_midverb_invalid_state.py (committed with
  the choreographer cluster) — verb-runner savepoints already surface a
  concurrent mid-verb state change as INVALID_STATE.
- #217: submit_for_qa claimed_by guard verified intact.

TDD: test_task.py, test_worktree_cleanup_on_complete.py,
test_escalation_board_guard.py (#99), test_task_service_* integration,
test_lifecycle_spec.py.

roboco/foundation/policy/lifecycle.py

@@ -350,6 +350,19 @@ _STATUS_TRANSITIONS: tuple[StatusTransition, ...] = (
         "ceo_reject",
         frozenset({Role.CEO}),
     ),
+    # A branchless coordination root (product integration root / MegaTask
+    # umbrella) has no developer to revise it, so CEO rejection routes it to
+    # PENDING for the Main PM to re-plan — not NEEDS_REVISION (dev-claim-only,
+    # would deadlock the root). The runtime applies this via the audited
+    # privileged override (admin_set_status); the edge is in the spec so future
+    # admin-override tightening can't wedge the path, and so the transition is
+    # acknowledged as CEO-legal rather than an unsanctioned out-of-band write.
+    StatusTransition(
+        Status.AWAITING_CEO_APPROVAL,
+        Status.PENDING,
+        "ceo_reject_to_pool",
+        frozenset({Role.CEO}),
+    ),
     # Direct PM submission for non-dev tasks
     StatusTransition(
         Status.IN_PROGRESS,
@@ -639,6 +652,16 @@ _ATOMIC_ACTIONS: dict[str, ActionSpec] = {
         self_review_block=False,
         needs_team_match=False,
     ),
+    "ceo_reject_to_pool": ActionSpec(
+        name="ceo_reject_to_pool",
+        allowed_roles=frozenset({Role.CEO}),
+        source_statuses=frozenset({Status.AWAITING_CEO_APPROVAL}),
+        target_status=Status.PENDING,
+        allowed_task_types=None,
+        preconditions=(),
+        self_review_block=False,
+        needs_team_match=False,
+    ),
     "cancel": ActionSpec(
         name="cancel",
         allowed_roles=frozenset(_PM_ROLES | {Role.CEO}),

roboco/services/task.py

@@ -51,6 +51,7 @@ from roboco.models.base import (
     BlockerResolverType,
     Complexity,
     JournalEntryType,
+    NotificationPriority,
     TaskNature,
     TaskStatus,
     TaskType,
@@ -102,6 +103,31 @@ _ROLE_CLAIM_STATUSES: dict[str, set[TaskStatus]] = {
     },
 }
 
+# Statuses a task may be escalated FROM into BLOCKED. The strict transition
+# validator only allows IN_PROGRESS→BLOCKED, but a chain escalation legitimately
+# blocks a task from any active work state (and re-escalates an already-blocked
+# task to a new target). BACKLOG is excluded — a never-activated task has no
+# business being escalated to BLOCKED (no spec edge, nonsensical). Terminal
+# statuses are refused separately by ``_is_terminal_task``. This is the explicit
+# escalation exemption to the validator: an enumerated source set, not an
+# arbitrary source→BLOCKED write.
+_ESCALATABLE_TO_BLOCKED: frozenset[TaskStatus] = frozenset(
+    {
+        TaskStatus.PENDING,
+        TaskStatus.CLAIMED,
+        TaskStatus.IN_PROGRESS,
+        TaskStatus.VERIFYING,
+        TaskStatus.AWAITING_QA,
+        TaskStatus.NEEDS_REVISION,
+        TaskStatus.AWAITING_DOCUMENTATION,
+        TaskStatus.AWAITING_PR_REVIEW,
+        TaskStatus.AWAITING_PM_REVIEW,
+        TaskStatus.AWAITING_CEO_APPROVAL,
+        TaskStatus.PAUSED,
+        TaskStatus.BLOCKED,
+    }
+)
+
 
 # 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
@@ -541,6 +567,14 @@ class TaskService(BaseService):
 
     service_name: ClassVar[str] = "task"
     _background_tasks: ClassVar[set[asyncio.Task[None]]] = set()
+    # #216: consecutive FS/permission failures during terminal worktree
+    # cleanup. A systemic FS issue (stuck mount, perms) would otherwise leak
+    # per-task worktrees indefinitely with only a warning log; after this many
+    # consecutive OSError failures a CEO alert fires once (suppressed until a
+    # success resets the streak). Class-level so it spans all cleanup calls.
+    _worktree_cleanup_fail_streak: ClassVar[int] = 0
+    _worktree_cleanup_escalated: ClassVar[bool] = False
+    _WORKTREE_CLEANUP_ESCALATE_AFTER: ClassVar[int] = 3
 
     # =========================================================================
     # STATUS TRANSITION HELPER
@@ -699,7 +733,9 @@ class TaskService(BaseService):
         # Rework counter: a bounce INTO needs_revision (not a re-entry) is one
         # rework cycle. Incremented at this single chokepoint — every transition
         # path funnels its audit through here exactly once — so the rework rate
-        # is an O(1) column read. Synchronous (part of this unit of work).
+        # is an O(1) column read. Synchronous (part of this unit of work). The
+        # pre-block RESTORE path undoes this bump when it restores to a
+        # snapshotted needs_revision (same cycle resuming, not a new rejection).
         if (
             to_status == TaskStatus.NEEDS_REVISION.value
             and from_status != TaskStatus.NEEDS_REVISION.value
@@ -5009,6 +5045,23 @@ class TaskService(BaseService):
                 target=target_slug,
             )
             return False
+        current = (
+            task.status
+            if isinstance(task.status, TaskStatus)
+            else TaskStatus(str(task.status))
+        )
+        if current not in _ESCALATABLE_TO_BLOCKED:
+            # No arbitrary source→BLOCKED write: a status outside the escalation
+            # exemption (e.g. BACKLOG, never activated) has no spec edge into
+            # BLOCKED. Refuse rather than bypass the validator silently.
+            self.log.warning(
+                "Refusing escalation from non-escalatable status",
+                task_id=str(task.id),
+                status=str(task.status),
+                escalator=escalator_slug,
+                target=target_slug,
+            )
+            return False
         if _board_cannot_own(task) and await self._is_board_advisory_agent(
             target_agent_id
         ):
@@ -5220,13 +5273,14 @@ class TaskService(BaseService):
         task.completed_at = datetime.now(UTC)
         # Validate transition with CEO role requirement
         self._validate_and_set_status(task, TaskStatus.COMPLETED, "ceo")
-        await self.session.flush()
+        # Mirror ``complete()``: close the work session (the PR is merged) and
+        # run the full completion-hook fan-out (commits + dev_notes indexing),
+        # not a lone learnings task. Without this the work session stayed ACTIVE
+        # forever after a CEO approval and code/decision RAG indexing was skipped.
+        await self._close_work_session_for_task(task, reason="ceo approved")
         await self._remove_task_worktree_on_terminal(task)
-
-        # Extract learnings (fire-and-forget)
-        bg_task = asyncio.create_task(self._extract_completion_learnings(task, None))
-        self._background_tasks.add(bg_task)
-        bg_task.add_done_callback(self._background_tasks.discard)
+        await self.session.flush()
+        await self._trigger_completion_hooks(task, None)
 
         # Unblock any tasks waiting on this one
         await self._unblock_dependents(task_id)
@@ -5356,6 +5410,7 @@ class TaskService(BaseService):
         released = False
         for child in await self.get_subtasks(cast("UUID", umbrella.id)):
             if child.batch_id is not None and child.status == TaskStatus.BACKLOG:
+                pre_status = child.status.value
                 child.status = TaskStatus.PENDING
                 child.team = cast("Any", Team.MAIN_PM)
                 # a board-routed root-subtask is created in BACKLOG with
@@ -5371,6 +5426,20 @@ class TaskService(BaseService):
                         task_id=str(child.id),
                     )
                     child.task_type = cast("Any", TaskType.PLANNING)
+                # No status change may bypass the audit log — record the
+                # backlog→pending activation so the child's lifecycle
+                # reconstruction (the metric source of truth) keeps its start.
+                # audit_agent_id is None: approve_and_start receives no approver
+                # id, and the audit_log.agent_id FK is nullable — attributing a
+                # hardcoded ceo uuid would FK-violate when the ceo agent row is
+                # absent. agent_role="ceo" still labels the actor.
+                self._emit_status_transition_audit(
+                    child,
+                    from_status=pre_status,
+                    to_status=TaskStatus.PENDING.value,
+                    agent_role="ceo",
+                    audit_agent_id=None,
+                )
                 released = True
         if released:
             await self.session.flush()
@@ -5641,12 +5710,63 @@ class TaskService(BaseService):
             if not project_slug:
                 return
             await self._remove_task_worktree_best_effort(task, project_slug)
+        except OSError as e:
+            # FS/permission failure (stuck mount, perms) — systemic, not
+            # task-specific. Track the streak and escalate a CEO alert once
+            # after N consecutive failures so worktrees don't leak silently.
+            TaskService._worktree_cleanup_fail_streak += 1
+            self.log.warning(
+                "Terminal worktree cleanup skipped (FS/permission)",
+                task_id=str(task.id),
+                streak=TaskService._worktree_cleanup_fail_streak,
+                error=str(e),
+            )
+            if (
+                TaskService._worktree_cleanup_fail_streak
+                >= TaskService._WORKTREE_CLEANUP_ESCALATE_AFTER
+                and not TaskService._worktree_cleanup_escalated
+            ):
+                TaskService._worktree_cleanup_escalated = True
+                self.log.error(
+                    "Recurring worktree-cleanup FS failure escalating to CEO",
+                    streak=TaskService._worktree_cleanup_fail_streak,
+                    task_id=str(task.id),
+                )
+                await self._escalate_worktree_cleanup_failure(task, e)
         except Exception as e:
+            # Non-FS failure (git, DB) — task-specific, best-effort, no streak.
+            TaskService._worktree_cleanup_fail_streak = 0
             self.log.warning(
                 "Terminal worktree cleanup skipped",
                 task_id=str(task.id),
                 error=str(e),
             )
+        else:
+            TaskService._worktree_cleanup_fail_streak = 0
+            TaskService._worktree_cleanup_escalated = False
+
+    async def _escalate_worktree_cleanup_failure(
+        self, task: TaskTable, exc: BaseException
+    ) -> None:
+        """Best-effort CEO alert on recurring worktree-cleanup FS failure."""
+        try:
+            from roboco.services.notification import NotificationService
+
+            await NotificationService().send_ack_notification(
+                from_agent="system",
+                to_agent="ceo",
+                body=(
+                    f"Recurring worktree-cleanup FS failure "
+                    f"(last error: {str(exc)[:160]}). "
+                    f"Task {task.id} terminal worktree could not be removed — "
+                    "per-task worktrees may be leaking on disk. Investigate the "
+                    "workspace mount / permissions."
+                ),
+                priority=NotificationPriority.HIGH,
+                task_id=str(task.id),
+            )
+        except Exception as e:
+            self.log.warning("Worktree-cleanup escalation notify failed", error=str(e))
 
     async def _close_work_session_for_task(self, task: TaskTable, reason: str) -> None:
         """Close the task's work session on successful completion.
@@ -5690,9 +5810,13 @@ class TaskService(BaseService):
         # Cancel all descendants first (children, grandchildren, etc.)
         # Skip tasks already in terminal states (completed or cancelled).
         # Route every descendant through _validate_and_set_status so role
-        # restrictions (e.g., only CEO can cancel awaiting_ceo_approval) still
-        # apply to cascaded cancels — skip descendants that fail validation
-        # rather than bypassing the rules.
+        # restrictions still apply to cascaded cancels. A non-terminal
+        # descendant the caller's role can't cancel REFUSES the whole
+        # cancel — never silently skip it and leave an orphaned subtree
+        # under a cancelled parent. (The spec gates every cancel edge to
+        # {cell_pm, main_pm, ceo} uniformly, so a PM/CEO cancel won't hit
+        # this; the narrow catch + refusal keeps a future per-edge role
+        # gate from silently orphaning.) Non-validation errors propagate.
         descendants = await self.get_all_descendants(task_id)
         cancelled_count = 0
         for descendant in descendants:
@@ -5702,15 +5826,23 @@ class TaskService(BaseService):
                 self._validate_and_set_status(
                     descendant, TaskStatus.CANCELLED, agent_role
                 )
-            except Exception as e:
+            except TaskLifecycleError as e:
                 self.log.warning(
-                    "Skipping cascade-cancel of descendant; role not permitted",
+                    "Refusing cascade-cancel: caller role cannot cancel descendant",
                     descendant_id=str(descendant.id),
                     descendant_status=descendant.status.value,
                     agent_role=agent_role,
                     error=str(e),
                 )
-                continue
+                raise TaskLifecycleError(
+                    current_status=descendant.status.value,
+                    target_status=TaskStatus.CANCELLED.value,
+                    message=(
+                        f"Cannot cancel descendant {descendant.id} in state "
+                        f"{descendant.status.value} with role {agent_role!r}; "
+                        "refusing parent cancel to avoid an orphaned subtree."
+                    ),
+                ) from e
             cancelled_count += 1
             await self._abandon_work_session_for_task(
                 descendant, reason="parent task cancelled"
@@ -8269,7 +8401,9 @@ class TaskService(BaseService):
         await self.session.flush()
         # This restore path sets the status directly (bypassing the strict
         # transition validator), so emit the audit explicitly — no status
-        # change may skip the audit log.
+        # change may skip the audit log. A restore to a snapshotted
+        # needs_revision is the same rework cycle resuming, not a fresh
+        # rejection, so undo the bump the chokepoint applied above.
         self._emit_status_transition_audit(
             task,
             from_status=pre_status,
@@ -8277,6 +8411,11 @@ class TaskService(BaseService):
             agent_role=None,
             audit_agent_id=restored_owner,
         )
+        if (
+            restored_status == TaskStatus.NEEDS_REVISION
+            and pre_status != TaskStatus.NEEDS_REVISION.value
+        ):
+            task.revision_count = max((task.revision_count or 1) - 1, 0)
         return task
 
     async def cell_pm_complete(

tests/integration/test_task_service_lifecycle_misc.py

@@ -24,6 +24,7 @@ from roboco.db.tables import (
     SessionTaskTable,
     WorkSessionTable,
 )
+from roboco.exceptions import TaskLifecycleError
 from roboco.models import AgentRole, AgentStatus, Team
 from roboco.models.base import (
     ChannelType,
@@ -846,6 +847,52 @@ async def test_cancel_descendants_cascades_for_authorized_pm(
     assert refreshed_child.status == TaskStatus.CANCELLED
 
 
+@pytest.mark.asyncio
+async def test_cancel_refuses_when_descendant_role_forbidden(
+    task_setup: dict, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """#103: a non-terminal descendant the caller's role can't cancel refuses
+    the whole cancel — never silently skips the descendant and leaves an
+    orphaned subtree under a cancelled parent.
+
+    The current spec gates every cancel edge to {cell_pm, main_pm, ceo}
+    uniformly, so a PM cancel won't naturally hit a role-forbidden
+    descendant. Simulate the future-regression shape (a per-edge role gate
+    that re-excludes a state) by stubbing ``_validate_and_set_status`` to
+    raise ``TaskLifecycleError`` for the descendant only — the parent's
+    cancel stays valid. The broad ``except Exception`` swallow used to
+    skip the descendant and cancel the parent anyway (orphan); it must
+    now refuse.
+    """
+    svc = task_setup["svc"]
+    parent = await svc.create(_req(task_setup))
+    child = await svc.create(_req(task_setup, parent_task_id=parent.id))
+    await db_session.flush()
+
+    real_validate = svc._validate_and_set_status
+
+    def stub_validate(task: Any, new_status: Any, agent_role: Any) -> Any:
+        if task.id == child.id:
+            raise TaskLifecycleError(
+                current_status=task.status.value,
+                target_status=new_status.value,
+                message="simulated per-edge role gate excludes this descendant",
+            )
+        return real_validate(task, new_status, agent_role)
+
+    monkeypatch.setattr(svc, "_validate_and_set_status", stub_validate)
+
+    with pytest.raises(TaskLifecycleError, match="orphaned subtree"):
+        await svc.cancel(parent.id, agent_role="cell_pm")
+
+    refreshed_parent = await svc.get(parent.id)
+    assert refreshed_parent is not None
+    assert refreshed_parent.status != TaskStatus.CANCELLED
+    refreshed_child = await svc.get(child.id)
+    assert refreshed_child is not None
+    assert refreshed_child.status != TaskStatus.CANCELLED
+
+
 # ---------------------------------------------------------------------------
 # Helper: simple async context manager
 # ---------------------------------------------------------------------------

tests/unit/services/test_escalation_board_guard.py

@@ -490,6 +490,43 @@ async def test_apply_escalation_refuses_cancelled_task() -> None:
     flush.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_apply_escalation_refuses_backlog_task() -> None:
+    """#99: escalation sets BLOCKED directly, bypassing the transition
+    validator. A BACKLOG task was never activated — escalating it to BLOCKED is
+    a nonsense transition with no spec edge. The primitive guards itself
+    (HTTP route bypasses the spec gate) and returns False, leaving the task
+    untouched. Active work statuses still escalate (next test)."""
+    svc = _service()
+    original_assignee = uuid4()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=uuid4(),
+        task_type=TaskType.CODE,
+        assigned_to=original_assignee,
+        blocker_raised_by=None,
+        status=TaskStatus.BACKLOG,
+    )
+    flush = AsyncMock()
+    object.__setattr__(svc.session, "flush", flush)
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False))
+    _bind(svc, "_is_main_pm_agent", AsyncMock(return_value=False))
+    _bind(svc, "_emit_status_transition_audit", MagicMock())
+
+    applied = await svc.apply_escalation(
+        task=task,
+        target_agent_id=uuid4(),
+        escalator_slug="be-pm",
+        target_slug="main-pm",
+        reason="please review",
+    )
+
+    assert applied is False
+    assert task.status == TaskStatus.BACKLOG  # untouched
+    assert task.assigned_to == original_assignee
+    flush.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_apply_escalation_blocks_non_terminal_task() -> None:
     # the terminal guard must not over-restrict: a normal in_progress task

tests/unit/services/test_task.py

@@ -595,6 +595,69 @@ async def test_admin_set_status_non_blocked_is_bare_status_set() -> None:
     assert task.assigned_to == owner
 
 
+@pytest.mark.asyncio
+async def test_pre_block_restore_skips_revision_count_bump() -> None:
+    """#101 Gap B: restoring a blocked task to its snapshotted needs_revision
+    state is a RESTORE, not a rework bounce — ``revision_count`` must not
+    increment. The rework counter counts rejections INTO needs_revision, and
+    this task was already rejected before it was blocked; unblocking it back to
+    needs_revision is the same rework cycle resuming, not a new one."""
+    REWORK_BOUNCES_BEFORE_BLOCK = 2
+    dev = uuid4()
+    task = _build_task(
+        status=TaskStatus.BLOCKED,
+        assigned_to=dev,
+        claimed_by=dev,
+        branch_name="feature/backend/abc--def--ghi",
+        pre_block_state="needs_revision",
+        pre_block_assignee=dev,
+        revision_count=REWORK_BOUNCES_BEFORE_BLOCK,
+    )
+    svc = TaskService(MagicMock(flush=AsyncMock()))
+    _bind(svc, "get", AsyncMock(return_value=task))
+    out = await svc.unblock_with_restore(task.id, uuid4(), restore=True)
+    assert out is task
+    assert task.status == TaskStatus.NEEDS_REVISION
+    # A restore must NOT bump the rework counter — only a fresh rejection does.
+    assert task.revision_count == REWORK_BOUNCES_BEFORE_BLOCK
+
+
+@pytest.mark.asyncio
+async def test_activate_batch_root_subtasks_emits_audit_for_activated_child() -> None:
+    """#101 Gap A: ``_activate_batch_root_subtasks`` sets a held root-subtask
+    BACKLOG→PENDING directly. No status change may bypass the audit log — the
+    transition journey (the metric source of truth) must record the activation,
+    or the child's lifecycle reconstruction silently drops its start point."""
+    batch = uuid4()
+    child = _build_task(
+        status=TaskStatus.BACKLOG,
+        batch_id=batch,
+        team=Team.BOARD,
+        task_type=TaskType.CODE,
+    )
+    umbrella = _build_task(
+        status=TaskStatus.PENDING,
+        batch_id=batch,
+        parent_task_id=None,
+        team=Team.BOARD,
+        task_type=TaskType.PLANNING,
+    )
+    added: list[object] = []
+    session = MagicMock()
+    session.flush = AsyncMock()
+    session.add.side_effect = added.append
+    svc = TaskService(session)
+    _bind(svc, "get_subtasks", AsyncMock(return_value=[child]))
+
+    await svc._activate_batch_root_subtasks(umbrella)
+
+    assert child.status == TaskStatus.PENDING
+    rows = [r for r in added if isinstance(r, AuditLogTable)]
+    assert any(
+        r.event_type == "task.pending" and r.target_id == child.id for r in rows
+    ), "batch root-subtask activation must emit a task.pending audit row"
+
+
 @pytest.mark.asyncio
 async def test_create_generates_ac_ids_and_carries_parent_ac_refs() -> None:
     # Every task gets one stable id per acceptance criterion (1:1), and a

tests/unit/services/test_worktree_cleanup_on_complete.py

@@ -12,7 +12,7 @@ failure never blocks completion.
 
 from __future__ import annotations
 
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 from uuid import uuid4
 
 import pytest
@@ -139,7 +139,8 @@ async def test_ceo_approve_removes_assignee_worktree_best_effort() -> None:
     svc, _ = _svc(AsyncMock(return_value=_slug_row("roboco-api")))
     _bind(svc, "get", AsyncMock(return_value=task))
     _bind(svc, "_validate_and_set_status", MagicMock())
-    _bind(svc, "_extract_completion_learnings", AsyncMock())
+    _bind(svc, "_close_work_session_for_task", AsyncMock())
+    _bind(svc, "_trigger_completion_hooks", AsyncMock())
     _bind(svc, "_unblock_dependents", AsyncMock())
     _bind(svc, "_emit_task_event", AsyncMock())
     remove = AsyncMock()
@@ -157,7 +158,8 @@ async def test_ceo_approve_skips_worktree_cleanup_for_branchless_task() -> None:
     svc, _ = _svc(AsyncMock())
     _bind(svc, "get", AsyncMock(return_value=task))
     _bind(svc, "_validate_and_set_status", MagicMock())
-    _bind(svc, "_extract_completion_learnings", AsyncMock())
+    _bind(svc, "_close_work_session_for_task", AsyncMock())
+    _bind(svc, "_trigger_completion_hooks", AsyncMock())
     _bind(svc, "_unblock_dependents", AsyncMock())
     _bind(svc, "_emit_task_event", AsyncMock())
     remove = AsyncMock()
@@ -167,3 +169,134 @@ async def test_ceo_approve_skips_worktree_cleanup_for_branchless_task() -> None:
 
     assert result is task
     remove.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_ceo_approve_closes_work_session_and_triggers_completion_hooks() -> None:
+    """#21/#98: ceo_approve must close the work session and run the full
+    completion-hook fan-out (commits + dev_notes indexing), mirroring
+    ``complete()`` — not hand-roll a lone learnings task and leave the work
+    session ACTIVE forever. The PR-merged gate already ran above."""
+    task = _build_task(status=TaskStatus.AWAITING_CEO_APPROVAL, work_session_id=None)
+    svc, _ = _svc(AsyncMock(return_value=_slug_row("roboco-api")))
+    _bind(svc, "get", AsyncMock(return_value=task))
+    _bind(svc, "_validate_and_set_status", MagicMock())
+    close_ws = AsyncMock()
+    hooks = AsyncMock()
+    _bind(svc, "_close_work_session_for_task", close_ws)
+    _bind(svc, "_trigger_completion_hooks", hooks)
+    _bind(svc, "_unblock_dependents", AsyncMock())
+    _bind(svc, "_emit_task_event", AsyncMock())
+    _bind(svc, "_remove_task_worktree_best_effort", AsyncMock())
+
+    result = await svc.ceo_approve(task.id)
+
+    assert result is task
+    close_ws.assert_awaited_once()
+    # The hooks fire with the CEO actor (None) — same as complete() for a
+    # non-agent completion path.
+    hooks.assert_awaited_once_with(task, None)
+
+
+# ---------------------------------------------------------------------------
+# #216: recurring FS/permission cleanup failure escalates to the CEO
+# ---------------------------------------------------------------------------
+
+
+def _reset_cleanup_streak() -> None:
+    TaskService._worktree_cleanup_fail_streak = 0
+    TaskService._worktree_cleanup_escalated = False
+
+
+@pytest.mark.asyncio
+async def test_recurring_worktree_cleanup_failure_escalates_to_ceo() -> None:
+    """#216: a recurring FS/permission failure (stuck mount, perms) used to
+    leak worktrees indefinitely with only a warning log. After N consecutive
+    OSError failures a CEO alert fires once; further failures are suppressed
+    until a success resets the streak."""
+    _reset_cleanup_streak()
+    n = TaskService._WORKTREE_CLEANUP_ESCALATE_AFTER
+    task = _build_task(status=TaskStatus.COMPLETED)
+    svc, _ = _svc(AsyncMock(return_value=_slug_row("roboco-api")))
+    _bind(
+        svc,
+        "_remove_task_worktree_best_effort",
+        AsyncMock(side_effect=PermissionError("denied")),
+    )
+
+    with patch(
+        "roboco.services.notification.NotificationService.send_ack_notification",
+        new=AsyncMock(),
+    ) as notifier:
+        for _ in range(n - 1):
+            await svc._remove_task_worktree_on_terminal(task)
+        notifier.assert_not_awaited()
+        await svc._remove_task_worktree_on_terminal(task)  # Nth failure
+        notifier.assert_awaited_once()
+        call = notifier.await_args
+        assert call is not None
+        assert call.kwargs["to_agent"] == "ceo"
+        assert str(task.id) in str(call.kwargs.get("task_id") or call.args)
+        # Suppressed on the (N+1)th — one escalation per failure streak.
+        await svc._remove_task_worktree_on_terminal(task)
+        notifier.assert_awaited_once()
+
+    _reset_cleanup_streak()
+
+
+@pytest.mark.asyncio
+async def test_worktree_cleanup_success_resets_failure_streak() -> None:
+    """A successful cleanup resets the streak + escalation flag, so a later
+    failure streak must re-accumulate to the full threshold before escalating."""
+    _reset_cleanup_streak()
+    n = TaskService._WORKTREE_CLEANUP_ESCALATE_AFTER
+    task = _build_task(status=TaskStatus.COMPLETED)
+    svc, _ = _svc(AsyncMock(return_value=_slug_row("roboco-api")))
+    remove = AsyncMock(side_effect=PermissionError("denied"))
+    _bind(svc, "_remove_task_worktree_best_effort", remove)
+
+    with patch(
+        "roboco.services.notification.NotificationService.send_ack_notification",
+        new=AsyncMock(),
+    ) as notifier:
+        for _ in range(n - 1):
+            await svc._remove_task_worktree_on_terminal(task)
+        notifier.assert_not_awaited()
+        # A success resets the streak.
+        remove.side_effect = None
+        await svc._remove_task_worktree_on_terminal(task)
+        # Re-introduce failures: needs a full N again, not 1.
+        remove.side_effect = PermissionError("denied")
+        for _ in range(n - 1):
+            await svc._remove_task_worktree_on_terminal(task)
+        notifier.assert_not_awaited()
+        await svc._remove_task_worktree_on_terminal(task)  # Nth of new streak
+        notifier.assert_awaited_once()
+
+    _reset_cleanup_streak()
+
+
+@pytest.mark.asyncio
+async def test_non_fs_worktree_cleanup_failure_does_not_escalate() -> None:
+    """A non-FS failure (git RuntimeError, DB error) is task-specific, not the
+    systemic operator-actionable FS case — it logs but never escalates, and it
+    resets the streak so an interleaved FS streak can't ride on it."""
+    _reset_cleanup_streak()
+    n = TaskService._WORKTREE_CLEANUP_ESCALATE_AFTER
+    task = _build_task(status=TaskStatus.COMPLETED)
+    svc, _ = _svc(AsyncMock(return_value=_slug_row("roboco-api")))
+    _bind(
+        svc,
+        "_remove_task_worktree_best_effort",
+        AsyncMock(side_effect=RuntimeError("git worktree remove failed")),
+    )
+
+    with patch(
+        "roboco.services.notification.NotificationService.send_ack_notification",
+        new=AsyncMock(),
+    ) as notifier:
+        for _ in range(n + 2):
+            await svc._remove_task_worktree_on_terminal(task)
+        notifier.assert_not_awaited()
+
+    _reset_cleanup_streak()