[F043] guard escalate_up against resurrecting terminal tasks

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

@@ -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:

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":
+    if first_missing.rejection_kind == "tracing_gap":
-        return Decision.reject(
+        return Decision.tracing_gap(missing=missing, remediate=first_missing.remediate)
-            kind="not_authorized",
+    # A precondition may declare a non-tracing rejection kind (not_authorized
-            message=first_missing.remediate,
+    # for ownership, invalid_state for the terminal-state guard on escalate_up).
-            remediate=first_missing.remediate,
+    # Honor it directly so the envelope signals the right failure flavor.
-        )
+    return Decision.reject(
-    return Decision.tracing_gap(missing=missing, remediate=first_missing.remediate)
+        kind=first_missing.rejection_kind,
+        message=first_missing.remediate,
+        remediate=first_missing.remediate,
+    )
 
 
 def can_invoke_intent(

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.

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
 ):

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