Core¶

@dataclass(frozen=True, slots=True)
class CircuitBreakerConfig:
    """Circuit breaker keyed per CircuitKey; one signal per logical call.

    A ``STATUS`` outcome trips the breaker only for 5xx responses.
    """

    fail_threshold: int = 5
    recovery_timeout: float = 60.0
    half_open_max_calls: int = 1
    max_keys: int = 512
    key: CircuitKey = CircuitKey.ORIGIN
    trip_kinds: frozenset[FailureKind] = DEFAULT_TRIP_KINDS

    def __post_init__(self) -> None:
        _coerce_enum(self, "key", CircuitKey)
        if self.fail_threshold < 1:
            raise ValueError(f"circuit_breaker.fail_threshold must be >= 1, got {self.fail_threshold}")
        _require_positive("circuit_breaker.recovery_timeout", self.recovery_timeout)
        if self.half_open_max_calls < 1:
            raise ValueError(f"circuit_breaker.half_open_max_calls must be >= 1, got {self.half_open_max_calls}")
        if self.max_keys < 1:
            raise ValueError(f"circuit_breaker.max_keys must be >= 1, got {self.max_keys}")

@dataclass(frozen=True, slots=True)
class ClientConfig:
    """The whole client, described once, readable by every adapter."""

    service_name: str
    base_url: str | None = None
    timeout: TimeoutConfig = field(default_factory=TimeoutConfig)
    pool: PoolConfig = field(default_factory=PoolConfig)
    retry: RetryConfig | None = field(default_factory=RetryConfig)
    circuit_breaker: CircuitBreakerConfig | None = field(default_factory=CircuitBreakerConfig)
    tls: TlsConfig = field(default_factory=TlsConfig)
    proxy: ProxyConfig | None = None
    headers: Mapping[str, str] = field(default_factory=dict)
    redirects: RedirectMode = RedirectMode.OWNED
    max_redirects: int = 5
    caller_override: CallerOverride = CallerOverride.CALLER_WINS
    deadline_header: str | None = None
    observability: ObservabilityConfig = field(default_factory=ObservabilityConfig)
    native: NativeOptions = field(default_factory=NativeOptions)
    on_unsupported: UnsupportedPolicy = UnsupportedPolicy.WARN

    def __post_init__(self) -> None:
        _coerce_enum(self, "redirects", RedirectMode)
        _coerce_enum(self, "caller_override", CallerOverride)
        _coerce_enum(self, "on_unsupported", UnsupportedPolicy)
        if not self.service_name or not self.service_name.strip():
            raise ValueError("service_name must be a non-empty string")
        if self.base_url is not None and not self.base_url.startswith(("http://", "https://")):
            raise ValueError(f"base_url must start with http:// or https://, got {self.base_url!r}")
        if self.max_redirects < 0:
            raise ValueError(f"max_redirects must be >= 0, got {self.max_redirects}")

@dataclass(frozen=True, slots=True)
class NativeOptions:
    """Raw passthrough to the native client, grouped by adapter-declared slots.

    Validated at build time: unknown slots, reserved keys, typos and collisions
    with explicitly set config fields are all errors, not silent behavior.
    """

    slots: Mapping[str, Mapping[str, object]] = field(default_factory=dict)

    @classmethod
    def of(cls, **slots: Mapping[str, object]) -> NativeOptions:
        return cls(slots={name: dict(values) for name, values in slots.items()})

    def for_slot(self, name: str) -> dict[str, object]:
        return dict(self.slots.get(name, {}))

@dataclass(frozen=True, slots=True)
class ObservabilityConfig:
    """Which telemetry channels are active; only knobs that actually work exist here.

    Two-stage URL scrubbing before anything reaches a log line or a span:
    ``sensitive_query_params`` redacts by parameter *name*, then ``url_masker``
    (if set) sees the whole redacted URL and may scrub PII by *value* - the
    email in a path segment that no name list can catch. See
    :class:`~clientwright.core.contracts.observability.MaskerProtocol` for the
    failure contract.
    """

    logging: bool = True
    metrics: bool = True
    tracing: bool = True
    success_log_level: int = _INFO_LEVEL
    sensitive_query_params: frozenset[str] = DEFAULT_SENSITIVE_QUERY_PARAMS
    url_masker: MaskerProtocol | None = None

    def __post_init__(self) -> None:
        if self.url_masker is not None and not callable(self.url_masker):
            raise ValueError(f"observability.url_masker must be callable or None, got {self.url_masker!r}")

@dataclass(frozen=True, slots=True)
class PoolConfig:
    """Connection pool shape; unset knobs defer to native defaults."""

    max_connections: Maybe[int | None] = 100
    max_keepalive: Maybe[int | None] = 20
    keepalive_expiry: Maybe[float | None] = 30.0
    max_connections_per_host: Maybe[int | None] = UNSET
    http2: Maybe[bool] = UNSET

    def __post_init__(self) -> None:
        for name in ("max_connections", "max_keepalive", "keepalive_expiry", "max_connections_per_host"):
            _require_positive(f"pool.{name}", getattr(self, name))

@dataclass(frozen=True, slots=True)
class ProxyConfig:
    """Explicit proxy or environment-driven proxies (mutually exclusive)."""

    url: str | None = None
    from_env: bool = False

    def __post_init__(self) -> None:
        if self.url is not None and self.from_env:
            raise ValueError("proxy.url and proxy.from_env are mutually exclusive")

@dataclass(frozen=True, slots=True)
class RetryConfig:
    """Owned retry policy; the engine runs the loop, adapters only send."""

    max_attempts: int = 3
    initial_backoff: float = 0.1
    max_backoff: float = 10.0
    multiplier: float = 2.0
    jitter: float = 0.2
    retryable_kinds: frozenset[FailureKind] = DEFAULT_RETRYABLE_KINDS
    retryable_status: frozenset[int] = DEFAULT_RETRYABLE_STATUS
    methods: frozenset[str] = IDEMPOTENT_METHODS
    respect_retry_after: bool = True
    retry_after_max: float = 60.0
    budget_ratio: float | None = 0.1
    require_replayable_body: bool = True
    mode: RetryMode = RetryMode.OWNED

    def __post_init__(self) -> None:
        _coerce_enum(self, "mode", RetryMode)
        if self.max_attempts < 1:
            raise ValueError(f"retry.max_attempts must be >= 1, got {self.max_attempts}")
        _require_positive("retry.initial_backoff", self.initial_backoff)
        _require_positive("retry.max_backoff", self.max_backoff)
        if self.multiplier < 1.0:
            raise ValueError(f"retry.multiplier must be >= 1, got {self.multiplier}")
        if not 0.0 <= self.jitter <= 1.0:
            raise ValueError(f"retry.jitter must be within [0, 1], got {self.jitter}")
        if self.budget_ratio is not None and not 0.0 < self.budget_ratio <= 1.0:
            raise ValueError(f"retry.budget_ratio must be within (0, 1], got {self.budget_ratio}")

@dataclass(frozen=True, slots=True)
class TimeoutConfig:
    """Timeout budget of a logical call.

    ``total`` is guaranteed everywhere: the engine enforces it with a monotonic
    deadline shared by all attempts, backoff sleeps and redirect hops.
    Phase knobs left ``UNSET`` defer to the adapter's native defaults.
    """

    total: float | None = 30.0
    attempt: Maybe[float | None] = UNSET
    connect: Maybe[float | None] = 5.0
    read: Maybe[float | None] = UNSET
    write: Maybe[float | None] = UNSET
    pool_acquire: Maybe[float | None] = UNSET

    def __post_init__(self) -> None:
        for name in ("total", "attempt", "connect", "read", "write", "pool_acquire"):
            _require_positive(f"timeout.{name}", getattr(self, name))

@dataclass(frozen=True, slots=True)
class Attempt:
    """One physical attempt inside a logical call."""

    index: int
    started: float
    duration: float
    outcome: Outcome
    hop: int = 0
    conn: ConnMetrics | None = None

class CircuitKey(StrEnum):
    """Granularity of the circuit-breaker key."""

    ORIGIN = "origin"
    ORIGIN_ROUTE = "origin_route"
    ORIGIN_METHOD = "origin_method"

@dataclass(frozen=True, slots=True)
class ConnMetrics:
    """Optional per-connection timings; ``None`` fields mean the adapter cannot see them."""

    dns: float | None = None
    connect: float | None = None
    tls: float | None = None
    pool_wait: float | None = None
    reused: bool | None = None
    http_version: str | None = None

class FailureKind(StrEnum):
    """Shared alphabet of call outcomes.

    An alphabet, not a shared partition: each adapter declares which kinds it
    can actually emit (``AdapterCapabilities.emits``) and which finer kinds
    collapse into coarser ones (``AdapterCapabilities.collapses``).
    """

    CONNECT_TIMEOUT = "connect_timeout"
    READ_TIMEOUT = "read_timeout"
    WRITE_TIMEOUT = "write_timeout"
    POOL_TIMEOUT = "pool_timeout"
    TOTAL_TIMEOUT = "total_timeout"
    CONNECT_ERROR = "connect_error"
    DNS_ERROR = "dns_error"
    TLS_ERROR = "tls_error"
    PROTOCOL_ERROR = "protocol_error"
    DISCONNECTED = "disconnected"
    BODY_ERROR = "body_error"
    STATUS = "status"
    CANCELLED = "cancelled"
    CIRCUIT_OPEN = "circuit_open"
    UNKNOWN = "unknown"

@dataclass(frozen=True, slots=True)
class Outcome:
    """Result of a single attempt. ``kind is None`` means success."""

    kind: FailureKind | None
    status_code: int | None = None
    retry_after: float | None = None
    exception: BaseException | None = None

    @property
    def ok(self) -> bool:
        return self.kind is None

@dataclass(frozen=True, slots=True)
class RequestInfo:
    """Low-cardinality identity of a logical call."""

    method: str
    origin: str
    url: str
    route: str | None = None
    idempotent: bool = True

    def circuit_key(self, mode: CircuitKey) -> str:
        if mode is CircuitKey.ORIGIN_ROUTE:
            return f"{self.origin} {self.route or 'unknown'}"
        if mode is CircuitKey.ORIGIN_METHOD:
            return f"{self.origin} {self.method}"
        return self.origin

@dataclass(frozen=True, slots=True)
class ResolvedTimeouts:
    """Per-attempt timeout plan handed to the adapter before each send.

    ``attempt`` is the hard ceiling of the whole attempt; async engines enforce
    it with task cancellation, sync engines can only clamp the phases below.
    """

    connect: float | None = None
    read: float | None = None
    write: float | None = None
    pool_acquire: float | None = None
    attempt: float | None = None

class CallError(ClientwrightError):
    """Base for errors raised while executing a call; adapters dual-inherit these."""

class CircuitOpenError(CallError):
    """Local refusal: the circuit for this key is open."""

    def __init__(self, key: str, retry_after: float | None = None) -> None:
        suffix = f" (retry in ~{retry_after:.1f}s)" if retry_after is not None else ""
        super().__init__(f"Circuit open for {key!r}{suffix}")
        self.key = key
        self.retry_after = retry_after

class ClientwrightError(Exception):
    """Base class for everything clientwright raises on its own authority."""

class DeadlineExceededError(CallError):
    """The total deadline of the logical call was exhausted."""

    def __init__(self, total: float) -> None:
        super().__init__(f"Deadline of {total:.3f}s exhausted")
        self.total = total

class NativeConfigError(ClientwrightError):
    """Base for invalid native passthrough configuration."""

class NotReplayableError(CallError):
    """The request body cannot be replayed, so the required repeat is impossible."""

class UnknownAdapterError(ClientwrightError):
    """Requested adapter name is not registered."""

    def __init__(self, name: str, known: tuple[str, ...]) -> None:
        super().__init__(f"Unknown adapter {name!r}; registered adapters: {', '.join(known) or '<none>'}")
        self.name = name
        self.known = known

class UnsupportedCapabilityError(ClientwrightError):
    """Config asks for something the chosen adapter cannot express (STRICT mode)."""

@dataclass(frozen=True, slots=True)
class AdapterCapabilities:
    """Everything an adapter admits about itself, in one frozen record."""

    adapter: str
    seam: str
    granularity: SeamGranularity
    boundary: DurationBoundary
    support: Mapping[Capability, Support]
    emits: frozenset[FailureKind]
    collapses: Mapping[FailureKind, FailureKind] = field(default_factory=dict)
    notes: Mapping[str, str] = field(default_factory=dict)

    def support_of(self, capability: Capability) -> Support:
        return self.support.get(capability, Support.ABSENT)

@dataclass(frozen=True, slots=True)
class ConfigApplicationReport:
    """What actually happened when a config met an adapter."""

    adapter: str
    applied_natively: frozenset[Capability] = frozenset()
    emulated: frozenset[Capability] = frozenset()
    dropped: Mapping[Capability, str] = field(default_factory=dict)
    dead_retryable_kinds: frozenset[FailureKind] = frozenset()
    collapsed_kinds: Mapping[FailureKind, FailureKind] = field(default_factory=dict)
    native_overrides: Mapping[str, tuple[str, ...]] = field(default_factory=dict)

    @property
    def has_issues(self) -> bool:
        return bool(self.dropped or self.dead_retryable_kinds)

    def issues(self) -> list[str]:
        problems = [f"{capability.value}: {reason}" for capability, reason in self.dropped.items()]
        if self.dead_retryable_kinds:
            dead = ", ".join(sorted(kind.value for kind in self.dead_retryable_kinds))
            problems.append(f"retryable_kinds never emitted by {self.adapter}: {dead}")
        return problems

    def enforce(self, policy: UnsupportedPolicy) -> None:
        """Apply the on_unsupported policy: raise, warn or stay silent."""
        if not self.has_issues or policy is UnsupportedPolicy.IGNORE:
            return
        problems = self.issues()
        if policy is UnsupportedPolicy.STRICT:
            raise UnsupportedCapabilityError(
                f"Adapter {self.adapter!r} cannot express the requested config: " + "; ".join(problems)
            )
        for problem in problems:
            logger.warning("Adapter %s: %s", self.adapter, problem)

@dataclass(frozen=True, slots=True)
class CallPlan:
    """Everything the engine needs, compiled once at build time."""

    config: ClientConfig
    capabilities: AdapterCapabilities
    planner: TimeoutPlanner
    retry_policy: DefaultRetryPolicy | None
    report: ConfigApplicationReport
    use_origin_limiter: bool
    # False under RetryMode.DELEGATED: attempts live below the seam, inside the
    # native retry machinery, and a single fake attempt record would be a lie.
    emit_attempt_metrics: bool = True

@dataclass(slots=True)
class ClientHandle[ClientT]:
    """The real native client plus everything clientwright knows about it."""

    client: ClientT
    adapter: str
    capabilities: AdapterCapabilities
    report: ConfigApplicationReport
    runtime: ClientRuntime
    plan: CallPlan
    aclose: Callable[[], Awaitable[None]] | None = None
    close: Callable[[], None] | None = None

class ClientRuntime:
    """State that must outlive REQUEST-scoped clients: circuits, budgets, limiters.

    Give this APP scope in DI. A fresh runtime per client makes the circuit
    breaker and retry budget decorative.
    """

    def __init__(
        self,
        *,
        clock: Callable[[], float] | None = None,
        rng: Random | None = None,
        circuits: CircuitRegistry | None = None,
        retry_budgets: RetryBudgetRegistry | None = None,
        async_limiter: AsyncOriginLimiter | None = None,
        sync_limiter: SyncOriginLimiter | None = None,
    ) -> None:
        self.clock: Callable[[], float] = clock or time.monotonic
        self.rng = rng or Random()  # noqa: S311 - jitter, not cryptography
        self.circuits = circuits
        self.retry_budgets = retry_budgets
        self.async_limiter = async_limiter
        self.sync_limiter = sync_limiter

    @classmethod
    def for_config(
        cls,
        config: ClientConfig,
        *,
        clock: Callable[[], float] | None = None,
        rng: Random | None = None,
        circuit_listener: StateListener | None = None,
    ) -> ClientRuntime:
        resolved_clock = clock or time.monotonic
        circuits = (
            CircuitRegistry(config.circuit_breaker, resolved_clock, circuit_listener)
            if config.circuit_breaker is not None
            else None
        )
        retry_budgets = (
            RetryBudgetRegistry(config.retry.budget_ratio)
            if config.retry is not None and config.retry.budget_ratio is not None
            else None
        )
        per_host = resolve(config.pool.max_connections_per_host, None)
        async_limiter = AsyncOriginLimiter(per_host) if per_host is not None else None
        sync_limiter = SyncOriginLimiter(per_host) if per_host is not None else None
        return cls(
            clock=resolved_clock,
            rng=rng,
            circuits=circuits,
            retry_budgets=retry_budgets,
            async_limiter=async_limiter,
            sync_limiter=sync_limiter,
        )