Spaces:
Sleeping
Sleeping
File size: 1,727 Bytes
2415446 a1bab2d 2415446 a1bab2d 2415446 0a54372 2415446 0cc4495 2415446 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | """Protocol-neutral execution failure semantics."""
from __future__ import annotations
from dataclasses import FrozenInstanceError, dataclass
from enum import StrEnum
class FailureKind(StrEnum):
"""Stable failure categories shared across execution and wire adapters."""
INVALID_REQUEST = "invalid_request"
CONTEXT_WINDOW_EXCEEDED = "context_window_exceeded"
AUTHENTICATION = "authentication"
PERMISSION = "permission"
RATE_LIMIT = "rate_limit"
OVERLOADED = "overloaded"
TIMEOUT = "timeout"
UPSTREAM = "upstream"
UNAVAILABLE = "unavailable"
@dataclass(slots=True, eq=False)
class ExecutionFailure(Exception):
"""A finalized provider-execution failure independent of any wire protocol."""
kind: FailureKind
status_code: int
message: str
retryable: bool
def __post_init__(self) -> None:
Exception.__init__(self, self.message)
def __setattr__(self, name: str, value: object) -> None:
# Exception machinery must be able to update __traceback__, __cause__,
# and __context__ while semantic failure fields remain immutable.
if name in self.__slots__ and hasattr(self, name):
raise FrozenInstanceError(f"cannot assign to field {name!r}")
object.__setattr__(self, name, value)
def find_execution_failure(exc: BaseException) -> ExecutionFailure | None:
"""Return the first canonical failure in an exception or nested group."""
pending = [exc]
while pending:
current = pending.pop()
if isinstance(current, ExecutionFailure):
return current
if isinstance(current, BaseExceptionGroup):
pending.extend(reversed(current.exceptions))
return None
|