From 2f322286730a329a531af7195c30261efcec6ce1 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 11:28:23 +0200 Subject: [PATCH] [F043] guard escalate_up against resurrecting terminal tasks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit escalate_up had composes=() and no source-status guard, so a PM could escalate a COMPLETED/CANCELLED task and apply_escalation set it back to BLOCKED — bypassing the state machine's terminal-state invariant. Defense in depth: - spec: add PRECONDITION_NON_TERMINAL to escalate_up's extra_preconditions so the lifecycle gate rejects terminal tasks (invalid_state) before the journal:decision write fires; generalize _check_intent_preconditions to honor non-tracing rejection_kind (not_authorized / invalid_state). - service: apply_escalation (the single write primitive) returns False and refuses to mutate a terminal task — covers the HTTP escalate route which bypasses the spec gate. escalate() / escalate_up_to_role() return None on refusal so the gateway emits a clean invalid_state envelope. - route: the HTTP escalate route 409s a terminal task BEFORE sending the escalation notification (so a finished task isn't yanked back, PM not pinged). --- roboco/api/routes/tasks.py | 13 +++ roboco/foundation/policy/lifecycle.py | 45 +++++++-- roboco/services/task.py | 47 +++++++-- tests/foundation/test_lifecycle_spec.py | 43 +++++++++ .../services/test_escalation_board_guard.py | 96 +++++++++++++++++++ 5 files changed, 229 insertions(+), 15 deletions(-) diff --git a/roboco/api/routes/tasks.py b/roboco/api/routes/tasks.py index 1b8511ff..cea1fe6e 100644 --- a/roboco/api/routes/tasks.py +++ b/roboco/api/routes/tasks.py @@ -2024,6 +2024,19 @@ async def escalate_task( raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Task not found" ) + # F043: a terminal task (completed / cancelled) must not be resurrected to + # BLOCKED via escalation. Refuse BEFORE sending the escalation notification + # so a finished/cancelled task isn't yanked back into the workflow (and the + # PM isn't pinged about a task that's already done). The single write + # primitive apply_escalation guards this too — defense in depth. + if task.status in (TaskStatus.COMPLETED, TaskStatus.CANCELLED): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=( + f"Task is in a terminal state ({task.status.value}) and cannot" + " be escalated — terminal tasks must not be resurrected." + ), + ) delivery = get_notification_delivery_service(db) try: diff --git a/roboco/foundation/policy/lifecycle.py b/roboco/foundation/policy/lifecycle.py index 8d7d5881..cd24d5b6 100644 --- a/roboco/foundation/policy/lifecycle.py +++ b/roboco/foundation/policy/lifecycle.py @@ -891,6 +891,21 @@ def _p_owns_task(task: Any, _agent: Any, ctx: Any) -> bool: return getattr(task, "assigned_to", None) == getattr(ctx, "actor_id", None) +def _p_non_terminal(task: Any, _agent: Any, _ctx: Any) -> bool: + """True unless the task is in a terminal state (completed / cancelled). + + Escalation is a "I'm blocked, hand this up" action — it must never resurrect + a task the lifecycle has already terminated. ``escalate_up`` has + ``composes=()`` (no composed action supplies a source-status gate), so + without this precondition the spec gate accepts a COMPLETED/CANCELLED task + and ``apply_escalation`` sets it back to BLOCKED, bypassing the state + machine's terminal-state invariant (F043). + """ + status = getattr(task, "status", None) + value = status.value if isinstance(status, Status) else str(status) + return value not in (Status.COMPLETED.value, Status.CANCELLED.value) + + PRECONDITION_PLAN = Precondition( key="plan", check=_p_has_plan_or_supplied, @@ -924,6 +939,17 @@ PRECONDITION_OWNERSHIP = Precondition( rejection_kind="not_authorized", ) +PRECONDITION_NON_TERMINAL = Precondition( + key="non_terminal", + check=_p_non_terminal, + remediate=( + "task is in a terminal state (completed / cancelled) and cannot be" + " escalated — terminal tasks must not be resurrected to blocked" + ), + missing_token="non_terminal", + rejection_kind="invalid_state", +) + _INTENT_VERBS: dict[str, IntentSpec] = { # Phase 1: developer verbs @@ -1237,7 +1263,7 @@ _INTENT_VERBS: dict[str, IntentSpec] = { allowed_roles=_PM_ROLES, description="Escalate to your role's escalation_target.", composes=(), # special - uses TaskService.escalate - extra_preconditions=(), + extra_preconditions=(PRECONDITION_NON_TERMINAL,), side_effects=(), next_hint=lambda _t: "idle until escalation target acts", ), @@ -1522,13 +1548,16 @@ def _check_intent_preconditions( first_missing = next( p for p in spec_intent.extra_preconditions if p.missing_token == missing[0] ) - if first_missing.rejection_kind == "not_authorized": - return Decision.reject( - kind="not_authorized", - message=first_missing.remediate, - remediate=first_missing.remediate, - ) - return Decision.tracing_gap(missing=missing, remediate=first_missing.remediate) + if first_missing.rejection_kind == "tracing_gap": + return Decision.tracing_gap(missing=missing, remediate=first_missing.remediate) + # A precondition may declare a non-tracing rejection kind (not_authorized + # for ownership, invalid_state for the terminal-state guard on escalate_up). + # Honor it directly so the envelope signals the right failure flavor. + return Decision.reject( + kind=first_missing.rejection_kind, + message=first_missing.remediate, + remediate=first_missing.remediate, + ) def can_invoke_intent( diff --git a/roboco/services/task.py b/roboco/services/task.py index 6d9188fd..a2aa9da2 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -210,6 +210,20 @@ def _task_type_is_code(task_type: Any) -> bool: return str(value) == TaskType.CODE.value +def _is_terminal_task(task: TaskTable) -> bool: + """True when ``task`` is in a terminal state (completed / cancelled). + + Terminal tasks must never be resurrected by a side-channel write + (escalation → BLOCKED, reassign, dependency-revival). The lifecycle spec + guards the gateway verbs, but the HTTP routes and direct service callers + bypass it, so the single write primitives consult this too (F043). Robust + to the enum-or-raw-string shapes SQLAlchemy hands back. + """ + status = getattr(task, "status", None) + value = status.value if isinstance(status, TaskStatus) else str(status) + return value in (TaskStatus.COMPLETED.value, TaskStatus.CANCELLED.value) + + _PM_OWNED_CELL_TASK_TYPES: frozenset[str] = frozenset( { TaskType.PLANNING.value, @@ -4801,7 +4815,7 @@ class TaskService(BaseService): escalator_slug: str, target_slug: str, reason: str, - ) -> None: + ) -> bool: """Apply the state mutations for a generic chain escalation. Sets the task to BLOCKED, reassigns to the escalation target, and @@ -4813,6 +4827,13 @@ class TaskService(BaseService): and the orchestrator re-spawns them. Without this, escalation loses the dev's identity permanently. + Returns True when the escalation was applied (or diverted to the pool + by the board/main-pm guard); False when refused because the task is in + a terminal state (COMPLETED / CANCELLED) — a terminal task must not be + resurrected to BLOCKED (F043). The gateway ``escalate_up`` path also + guards this in the lifecycle spec, but the HTTP escalate route bypasses + the spec gate, so the single write primitive refuses it here too. + Invariant: a board/advisory role is NEVER assigned a task it cannot own — a descendant executable task (code / documentation / design), a cell's own coordination/planning descendant, OR a Main-PM coordination root @@ -4825,6 +4846,15 @@ class TaskService(BaseService): it. Enforced here — the single write primitive — so both the gateway ``escalate`` verb and the HTTP escalate route are covered. """ + if _is_terminal_task(task): + self.log.warning( + "Refusing to escalate a terminal task (no resurrection to blocked)", + 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 ): @@ -4834,7 +4864,7 @@ class TaskService(BaseService): blocked_target_slug=target_slug, reason=reason, ) - return + return True # Impossibility backstop: a Main-PM target must never receive (back) a # main_pm + code task — a coordinator with no code verb cannot fix the # code, so escalating it to Main PM perpetuates the mismatch (the @@ -4852,7 +4882,7 @@ class TaskService(BaseService): blocked_target_slug=target_slug, reason=reason, ) - return + return True if task.assigned_to and not task.blocker_raised_by: task.blocker_raised_by = cast("Any", task.assigned_to) # Capture before mutating: the audit row must record the real prior @@ -4893,6 +4923,7 @@ class TaskService(BaseService): escalator=escalator_slug, target=target_slug, ) + return True # ========================================================================= # CEO APPROVAL WORKFLOW @@ -8039,14 +8070,14 @@ class TaskService(BaseService): # 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( + applied = await self.apply_escalation( task=task, target_agent_id=UUID(str(target.id)), escalator_slug=agent.slug, target_slug=target_slug, reason=reason, ) - return task + return task if applied else None async def _is_board_advisory_agent(self, agent_id: UUID) -> bool: """True if ``agent_id`` is a board/advisory role (PO / marketing / auditor).""" @@ -8174,14 +8205,16 @@ class TaskService(BaseService): if target is None: return None - await self.apply_escalation( + applied = await self.apply_escalation( task=task, target_agent_id=UUID(str(target.id)), escalator_slug=agent.slug, target_slug=target.slug, reason=reason, ) - return task + # apply_escalation refuses terminal tasks (F043); mirror escalate()'s + # contract so the gateway emits a clean invalid_state envelope. + return task if applied else None async def list_in_progress_for_agent(self, agent_id: UUID) -> list[TaskTable]: """Tasks the agent is still on the hook for — in_progress OR blocked. diff --git a/tests/foundation/test_lifecycle_spec.py b/tests/foundation/test_lifecycle_spec.py index 3961fd9a..e489d263 100644 --- a/tests/foundation/test_lifecycle_spec.py +++ b/tests/foundation/test_lifecycle_spec.py @@ -566,6 +566,49 @@ def test_can_invoke_intent_developer_open_pr_no_commits_tracing_gap() -> None: assert "commits>=1" in d.missing +def test_escalate_up_rejected_on_completed_task() -> None: + """F043: a PM must not resurrect a COMPLETED task via escalate_up. + + escalate_up has composes=() and historically no source-status guard, so the + spec gate accepted it on a terminal task and apply_escalation set it back to + BLOCKED — bypassing the state machine's terminal-state invariant. The spec + now rejects terminal tasks (completed / cancelled) before the journal:decision + write fires. + """ + d = spec.can_invoke_intent( + spec.Role.CELL_PM, + "escalate_up", + _stub_task(status="completed"), + context=spec.Context(notes="stuck on something"), + ) + assert d.allowed is False + assert d.rejection_kind == "invalid_state" + + +def test_escalate_up_rejected_on_cancelled_task() -> None: + """F043: cancelled is terminal — escalate_up must not resurrect it either.""" + d = spec.can_invoke_intent( + spec.Role.MAIN_PM, + "escalate_up", + _stub_task(status="cancelled"), + context=spec.Context(notes="stuck on something"), + ) + assert d.allowed is False + assert d.rejection_kind == "invalid_state" + + +def test_escalate_up_allowed_on_blocked_task() -> None: + """F043: the terminal guard must not over-restrict — BLOCKED is the natural + escalation source and must still be allowed.""" + d = spec.can_invoke_intent( + spec.Role.CELL_PM, + "escalate_up", + _stub_task(status="blocked"), + context=spec.Context(notes="stuck on something"), + ) + assert d.allowed is True + + def test_valid_next_verbs_developer_in_progress_includes_open_pr_and_i_am_done() -> ( None ): diff --git a/tests/unit/services/test_escalation_board_guard.py b/tests/unit/services/test_escalation_board_guard.py index cb53d0f6..4a058b5b 100644 --- a/tests/unit/services/test_escalation_board_guard.py +++ b/tests/unit/services/test_escalation_board_guard.py @@ -428,6 +428,102 @@ async def test_is_board_advisory_agent_classifies_roles() -> None: assert await svc._is_board_advisory_agent(uuid4()) is expected +@pytest.mark.asyncio +async def test_apply_escalation_refuses_completed_task() -> None: + # F043: a COMPLETED task is terminal — apply_escalation must not resurrect + # it to BLOCKED. The HTTP escalate route bypasses the spec gate, so the + # single write primitive must refuse terminal tasks itself. Returns False + # so callers (escalate / HTTP route) can surface a clean invalid_state / 409 + # instead of mutating a finished task. + 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.COMPLETED, + ) + flush = AsyncMock() + object.__setattr__(svc.session, "flush", flush) + _bind(svc, "_is_board_advisory_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.COMPLETED # untouched — not resurrected + assert task.assigned_to == original_assignee # no reassignment happened + flush.assert_not_called() + + +@pytest.mark.asyncio +async def test_apply_escalation_refuses_cancelled_task() -> None: + # F043: cancelled is terminal too — must not be resurrected via escalation. + svc = _service() + task = MagicMock( + id=uuid4(), + parent_task_id=uuid4(), + task_type=TaskType.CODE, + assigned_to=uuid4(), + blocker_raised_by=None, + status=TaskStatus.CANCELLED, + ) + flush = AsyncMock() + object.__setattr__(svc.session, "flush", flush) + _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False)) + + 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.CANCELLED + flush.assert_not_called() + + +@pytest.mark.asyncio +async def test_apply_escalation_blocks_non_terminal_task() -> None: + # F043: the terminal guard must not over-restrict — a normal in_progress + # task still escalates (blocked + reassigned) and returns True. + svc = _service() + target_id = uuid4() + task = MagicMock( + id=uuid4(), + parent_task_id=uuid4(), + task_type=TaskType.CODE, + assigned_to=uuid4(), + blocker_raised_by=None, + dev_notes=None, + status=TaskStatus.IN_PROGRESS, + ) + _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False)) + _bind(svc, "_emit_status_transition_audit", MagicMock()) + + applied = await svc.apply_escalation( + task=task, + target_agent_id=target_id, + escalator_slug="be-pm", + target_slug="main-pm", + reason="cell blocked", + ) + + assert applied is True + assert task.status == TaskStatus.BLOCKED + assert task.assigned_to == target_id + + @pytest.mark.asyncio async def test_apply_escalation_emits_blocked_audit_event() -> None: """A non-divert escalation sets BLOCKED and MUST record a task.blocked audit