# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-a2a-v1-0-agent-card-discovery-appear-to-succeed-while-cross-vendor-task-delegation-silently-fails-at-lifecycle-negotiation.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-a2a-v1-0-agent-card-discovery-appear-to-succeed-while-cross-vendor-task-delegation-silently-fails-at-lifecycle-negotiation.warrant.json
# Markdown SHA-256:   dd8b1e5e4125cda665760d97d6b2ad912086e5ce2df41ac8b296115c0b7f14eb
# Code AST SHA-256:   ca50a9d7e8d236b062d822b3ecbbc345c8c4e9d0799edf5f4ba7485687ea5a0c
# Verified Classes:   LifecycleInvariantError, TaskState, TaskGuard
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class LifecycleInvariantError(RuntimeError):
    """Raised when a task emits a terminal transition after cancel was acknowledged."""


class TaskState(str, Enum):
    SUBMITTED = "submitted"
    WORKING = "working"
    CANCELED = "canceled"
    COMPLETED = "completed"
    FAILED = "failed"


TERMINAL = {TaskState.CANCELED, TaskState.COMPLETED, TaskState.FAILED}


@dataclass
class TaskGuard:
    task_id: str
    card_capabilities: dict
    state: TaskState = TaskState.SUBMITTED
    cancel_acknowledged: bool = False
    _history: list = field(default_factory=list)

    def confirm_capability(self, capability: str) -> bool:
        # Card is a claim; only trust it for non-lifecycle-affecting fields.
        lifecycle_affecting = {"streaming", "pushNotifications", "extensions"}
        if capability in lifecycle_affecting:
            raise LifecycleInvariantError(
                f"capability {capability!r} affects lifecycle; "
                f"must be confirmed against live task manager, not card"
            )
        return bool(self.card_capabilities.get(capability))

    def apply_transition(self, new_state: TaskState) -> None:
        self._history.append((self.state, new_state))
        if self.cancel_acknowledged and new_state in TERMINAL and new_state != TaskState.CANCELED:
            raise LifecycleInvariantError(
                f"task {self.task_id}: cancel acknowledged but received terminal "
                f"state {new_state.value!r} after cancel; card advertised "
                f"capabilities={sorted(self.card_capabilities)} — route to reconciliation"
            )
        if self.state in TERMINAL and new_state in TERMINAL and new_state != self.state:
            raise LifecycleInvariantError(
                f"task {self.task_id}: terminal state {self.state.value!r} "
                f"superseded by {new_state.value!r}"
            )
        self.state = new_state

    def acknowledge_cancel(self) -> None:
        if self.state in TERMINAL:
            raise LifecycleInvariantError(
                f"task {self.task_id}: cannot cancel from terminal state {self.state.value!r}"
            )
        self.cancel_acknowledged = True
        self.state = TaskState.CANCELED


def delegate(guard: TaskGuard, intent: str, interruptible: bool) -> Optional[str]:
    if interruptible:
        # Interruptible delegation requires live confirmation, not card trust.
        if not guard.card_capabilities.get("extensions"):
            raise LifecycleInvariantError(
                f"intent {intent!r}: interruptible delegation requested but remote "
                f"advertises no lifecycle extension; downgrade to poll-only"
            )
    return guard.task_id