# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-backpressure-saturation-in-a2a-v1-0-streaming-task-updates-over-http-native-transport.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-backpressure-saturation-in-a2a-v1-0-streaming-task-updates-over-http-native-transport.warrant.json
# Markdown SHA-256:   b14c7459c1fecaa5457bc1a430aa0d307008f2dad5b1f5acc021cddab274b010
# Code AST SHA-256:   54997a5c67b6535177689c11f94d772b8bbcf88d3b095b8f73359b1a9890c5b9
# Verified Classes:   StreamSaturationError, StreamHealth
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field
from typing import Optional


class StreamSaturationError(RuntimeError):
    """Raised when an A2A v1.0 stream shows heartbeat liveness but sequence lag
    with a pinned server-side outbound queue AND a blocked task executor.
    This is the abort-and-resync boundary; drain-and-continue is NOT valid here."""


@dataclass
class StreamHealth:
    heartbeat_interval_s: float = 15.0
    max_lag_heartbeats: int = 2
    max_outbound_queue: int = 128

    last_heartbeat: Optional[float] = None
    last_sequence: int = -1
    server_queue_depth: int = 0
    executor_advancing: bool = True


def check_saturation(health: StreamHealth, now: Optional[float] = None) -> None:
    """Evaluate the A2A v1.0 backpressure decision boundary.

    Raises StreamSaturationError iff:
      - heartbeats are present (transport alive), AND
      - sequence has not advanced for > max_lag_heartbeats intervals, AND
      - server outbound queue is pinned at max, AND
      - the task executor is NOT advancing (deadlock-shaped saturation).

    If the executor is still advancing, this returns normally: the caller
    should drain-and-continue rather than abort.
    """
    now = now if now is not None else time.monotonic()

    if health.last_heartbeat is None:
        # No heartbeat yet: cannot disambiguate; caller must wait.
        return

    heartbeat_age = now - health.last_heartbeat
    transport_alive = heartbeat_age <= health.heartbeat_interval_s * 1.5
    if not transport_alive:
        # Dead transport is a different failure class; not our boundary.
        return

    lag_heartbeats = heartbeat_age / health.heartbeat_interval_s
    sequence_stalled = lag_heartbeats > health.max_lag_heartbeats
    queue_pinned = health.server_queue_depth >= health.max_outbound_queue

    if sequence_stalled and queue_pinned and not health.executor_advancing:
        raise StreamSaturationError(
            "A2A v1.0 stream saturated: heartbeat alive "
            f"(age={heartbeat_age:.1f}s), sequence stalled at {health.last_sequence} "
            f"for {lag_heartbeats:.1f} heartbeat intervals, "
            f"outbound queue pinned at {health.server_queue_depth}/{health.max_outbound_queue}, "
            "executor not advancing. Abort-and-resync via tasks/get; do not drain."
        )
    # Otherwise: drain-and-continue is the correct action.