From 3d6330843c5772d3e72558f60cebdac1457f0f79 Mon Sep 17 00:00:00 2001 From: Renn F Date: Mon, 29 Jun 2026 23:49:42 +0200 Subject: [PATCH] task-service: completion hooks + escalation/cancel/audit hardening MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 23 +++ roboco/services/task.py | 167 ++++++++++++++++-- .../test_task_service_lifecycle_misc.py | 47 +++++ .../services/test_escalation_board_guard.py | 37 ++++ tests/unit/services/test_task.py | 63 +++++++ .../test_worktree_cleanup_on_complete.py | 139 ++++++++++++++- 6 files changed, 459 insertions(+), 17 deletions(-) diff --git a/roboco/foundation/policy/lifecycle.py b/roboco/foundation/policy/lifecycle.py index 0cb78ee3..562c8fd3 100644 --- a/roboco/foundation/policy/lifecycle.py +++ b/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}), diff --git a/roboco/services/task.py b/roboco/services/task.py index 4bc9df89..f5d943c6 100644 --- a/roboco/services/task.py +++ b/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( diff --git a/tests/integration/test_task_service_lifecycle_misc.py b/tests/integration/test_task_service_lifecycle_misc.py index 615ba934..5474aad3 100644 --- a/tests/integration/test_task_service_lifecycle_misc.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/tests/unit/services/test_escalation_board_guard.py b/tests/unit/services/test_escalation_board_guard.py index c0e03b87..bb774faf 100644 --- a/tests/unit/services/test_escalation_board_guard.py +++ b/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 diff --git a/tests/unit/services/test_task.py b/tests/unit/services/test_task.py index db431896..33c6d6a2 100644 --- a/tests/unit/services/test_task.py +++ b/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 diff --git a/tests/unit/services/test_worktree_cleanup_on_complete.py b/tests/unit/services/test_worktree_cleanup_on_complete.py index 86589e8d..0b38b5ff 100644 --- a/tests/unit/services/test_worktree_cleanup_on_complete.py +++ b/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()