"""SDK-free pre-development multi-provider coordination core."""
from collections import Counter, deque
from dataclasses import dataclass
import time


@dataclass(frozen=True)
class ProviderFrameKey:
    run_id: str
    run_generation: int
    frame_seq: int
    pts_ns: int | None

    def __post_init__(self):
        if not isinstance(self.run_id, str) or not self.run_id:
            raise ValueError('Invalid provider run ID.')
        if type(self.run_generation) is not int or self.run_generation < 1:
            raise ValueError('Invalid provider generation.')
        if type(self.frame_seq) is not int or self.frame_seq < 0:
            raise ValueError('Invalid provider Frame sequence.')
        if self.pts_ns is not None and (type(self.pts_ns) is not int or self.pts_ns < 0):
            raise ValueError('Invalid provider PTS.')


class LatestWinsLane:
    """At most one in-flight item and one newest pending item."""

    def __init__(self, provider):
        if not isinstance(provider, str) or not provider:
            raise ValueError('Invalid provider.')
        self.provider = provider
        self.inflight = None
        self.pending = None
        self.metrics = Counter()

    def offer(self, key, payload):
        if not isinstance(key, ProviderFrameKey):
            raise ValueError('Invalid provider Frame key.')
        item = (key, payload)
        if self.inflight is None:
            self.inflight = item
            self.metrics['dispatched'] += 1
            return item
        if self.pending is not None:
            self.metrics['pending_replaced'] += 1
        self.pending = item
        self.metrics['pending_offered'] += 1
        return None

    def complete(self, key):
        if self.inflight is None or self.inflight[0] != key:
            raise ValueError('Provider completion differs from in-flight Frame.')
        self.metrics['completed'] += 1
        self.inflight = self.pending
        self.pending = None
        if self.inflight is not None:
            self.metrics['dispatched'] += 1
        return self.inflight

    def snapshot(self):
        return dict(self.metrics, inflight=int(self.inflight is not None),
                    pending=int(self.pending is not None))


@dataclass(frozen=True)
class CompositionRelease:
    key: ProviderFrameKey
    results: dict
    complete: bool
    timed_out: bool


@dataclass
class _Slot:
    first_result_ns: int
    deadline_ns: int
    results: dict


class BoundedFrameComposer:
    """Evaluate only open Frames; hold decisions until complete/timeout release."""

    def __init__(self, providers, wait_ms=50, *, clock_ns=time.monotonic_ns,
                 closed_capacity=1024):
        expected = frozenset(providers)
        if not expected or any(not isinstance(p, str) or not p for p in expected):
            raise ValueError('Invalid provider set.')
        if type(wait_ms) not in (int, float) or not 0 < wait_ms <= 1000:
            raise ValueError('Invalid composition wait.')
        if type(closed_capacity) is not int or not 16 <= closed_capacity <= 8192:
            raise ValueError('Invalid closed Frame capacity.')
        self.providers = expected
        self.wait_ns = round(wait_ms * 1_000_000)
        self.clock_ns = clock_ns
        self.closed_capacity = closed_capacity
        self.slots = {}
        self.closed_order = deque()
        self.closed = set()
        self.metrics = Counter()

    def _remember_closed(self, key):
        if key in self.closed:
            return
        self.closed.add(key)
        self.closed_order.append(key)
        while len(self.closed_order) > self.closed_capacity:
            self.closed.discard(self.closed_order.popleft())

    def _release(self, key, *, timed_out, now_ns=None):
        slot = self.slots.pop(key)
        complete = self.providers <= set(slot.results)
        self._remember_closed(key)
        now = self.clock_ns() if now_ns is None else now_ns
        self.metrics['compose_wait_ns_total'] += max(0, now - slot.first_result_ns)
        self.metrics['compose_complete' if complete else 'compose_partial'] += 1
        if timed_out and not complete:
            self.metrics['join_timeout'] += 1
        return CompositionRelease(key, dict(slot.results), complete,
                                  bool(timed_out and not complete))

    def submit(self, provider, key, evaluator, *, now_ns=None):
        """Run evaluator only while the Frame slot is still open.

        Returns (accepted, releases, value). A late result never invokes evaluator.
        """
        if provider not in self.providers or not isinstance(key, ProviderFrameKey) or not callable(evaluator):
            raise ValueError('Invalid provider composition input.')
        now = self.clock_ns() if now_ns is None else now_ns
        if type(now) is not int or now < 0:
            raise ValueError('Invalid composition clock.')
        if key in self.closed:
            self.metrics['provider_result_late_' + provider] += 1
            return False, (), None
        slot = self.slots.get(key)
        releases = []
        if slot is not None and now > slot.deadline_ns:
            releases.append(self._release(key, timed_out=True, now_ns=now))
            self.metrics['provider_result_late_' + provider] += 1
            return False, tuple(releases), None
        if slot is None:
            slot = _Slot(now, now + self.wait_ns, {})
            self.slots[key] = slot
        if provider in slot.results:
            raise ValueError('Duplicate provider result for Frame.')
        value = evaluator()
        slot.results[provider] = value
        self.metrics['provider_' + provider + '_result'] += 1
        if self.providers <= set(slot.results):
            releases.append(self._release(key, timed_out=False, now_ns=now))
        return True, tuple(releases), value

    def expire(self, *, now_ns=None):
        now = self.clock_ns() if now_ns is None else now_ns
        if type(now) is not int or now < 0:
            raise ValueError('Invalid composition clock.')
        releases = []
        for key, slot in tuple(self.slots.items()):
            if now >= slot.deadline_ns:
                releases.append(self._release(key, timed_out=True, now_ns=now))
        return tuple(releases)

    def snapshot(self):
        return dict(self.metrics, open_slots=len(self.slots),
                    closed_frames=len(self.closed),
                    join_wait_max_ms=self.wait_ns / 1_000_000)