"""Opt-in single-provider trial: bounded completed results, no pre-infer slots."""
import math
import os

from app.multi_provider_results import MultiProviderCoordinator, _OpenFrame


def linear_trial_enabled():
    value = os.environ.get('CONTEXTRA_PROVIDER_LINEAR_TRIAL', 'false')
    if value not in ('true', 'false'):
        raise ValueError('Invalid linear trial option.')
    return value == 'true'


class LinearProviderResults(MultiProviderCoordinator):
    linear_trial = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if len(self.plan.providers) != 1 or self.freshness_enabled:
            raise ValueError('Linear trial requires one provider and freshness OFF.')
        self.capacity = 32
        self.queue_max_age = float(os.environ.get('CONTEXTRA_COMPLETED_QUEUE_MAX_AGE_SECONDS', '1'))
        if not math.isfinite(self.queue_max_age) or not 0 < self.queue_max_age <= 5:
            raise ValueError('Completed queue age must be finite and within (0, 5] seconds.')
        self._last_received = -1

    def admit(self, frame, admitted_at=None):
        now = self.clock() if admitted_at is None else admitted_at
        with self._cv:
            if self._closed or not self.is_current() or frame.frame_seq <= self._last_admitted:
                self._metrics['admission_stale'] += 1
                return False
            self._last_admitted = frame.frame_seq
            row = next(iter(self._providers.values()))
            if row['state'] != 'ACTIVE':
                return False
            if row['first_input_at'] is None:
                row['first_input_at'] = now
            row['last_input_at'] = now
            self._metrics['admitted'] += 1
            return True

    def note_request(self, key):
        with self._cv:
            row = self._providers[key]
            row.setdefault('first_request_at', self.clock())
            row['requests'] = row.get('requests', 0) + 1

    def submit(self, provider, frame, copied, response_at=None, *, timing=None):
        selected = timing[0] if timing else None
        if selected is None or not math.isfinite(selected):
            raise ValueError('Linear result lacks monotonic admission time.')
        identity = tuple(getattr(frame, k) for k in
                         ('server_id', 'source_id', 'stream_id', 'run_id', 'run_generation'))
        if identity != self.ingress.identity:
            raise ValueError('Foreign result identity.')
        if ((frame.frame_width, frame.frame_height) != (copied.width, copied.height)
                or frame.pts_ns != copied.pts or len(copied.detections) > 300):
            raise ValueError('Result association differs.')
        if provider != self.plan.providers[0].key:
            raise ValueError('Foreign result provider.')
        now = self.clock() if response_at is None else response_at
        with self._cv:
            row = self._providers[provider]
            row['last_response_at'] = now
            row.setdefault('first_response_at', now)
            row['responses'] += 1
            self._timing_locked(provider, 'selected_to_response', now-selected)
            if now-selected > self.freshness_s:
                self._metrics['over_freshness_'+provider] += 1
            reason = ('closed_frame' if self._closed or not self.is_current() else
                      'provider_disabled' if row['state'] != 'ACTIVE' else
                      'backward' if frame.frame_seq <= self._last_received else None)
            if reason:
                row['late'] += 1
                self._metrics['rejected_'+provider+'_'+reason] += 1
                return False
            self._last_received = frame.frame_seq
            while len(self._frames) >= self.capacity:
                _, old = self._frames.popitem(last=False)
                self._evict_locked(old)
                self._metrics['completed_queue_drop'] += 1
            # Only completed, owned numeric results enter this queue.
            self._frames[frame.frame_seq] = _OpenFrame(
                frame, now, now+self.queue_max_age, {provider}, {provider: copied},
                selected_at=selected)
            self._metrics['accepted_result'] += 1
            from app.pipeline_diagnostics import diagnostics
            diagnostics.for_provider(provider).mark('accepted')
            self._cv.notify_all()
        if self.on_provider_frame is not None:
            import time
            self.on_provider_frame(provider, time.time())
        return True

    def _ready_locked(self, now):
        while self._frames:
            seq, state = next(iter(self._frames.items()))
            if now-state.admitted_at <= self.queue_max_age:
                return seq
            self._frames.pop(seq)
            self._evict_locked(state)
            self._metrics['completed_queue_expired'] += 1
        return None

    def _check_stalls_locked(self, now):
        stalled = set(super()._check_stalls_locked(now))
        for key, row in self._providers.items():
            if row['state'] != 'ACTIVE' or row['first_input_at'] is None:
                continue
            responded = row['last_response_at'] is not None
            if (responded and row.get('requests', 0) <= row['responses']
                    and self._last_admitted <= self._last_received):
                continue
            reference = (max(row['last_response_at'], row.get('last_progress_at', 0))
                         if responded else row['first_input_at'])
            # Backpressure can stop selected input while an SDK request is stuck.
            if now-reference >= (self.stall_s if responded else self.startup_s):
                stalled.add(key)
        return sorted(stalled)

    def _finalize(self, state):
        provider = self.plan.providers[0].key
        from app.pipeline_diagnostics import diagnostics
        metric = diagnostics.for_provider(provider)
        with metric.lock:
            if metric.collecting():
                metric._duration('completed_queue_wait', round((self.clock()-state.admitted_at)*1e9))
        # Person gaps are already consumed by the common finalizer; Fire also
        # needs selected-but-unprocessed inputs, not just result queue drops.
        if provider == 'fire' and self.ingress.result_gap(state.frame):
            state.gaps.add('fire')
        super()._finalize(state)

    def status(self):
        value = super().status()
        value.update(execution_path='provider-linear-trial', queue_max_age_s=self.queue_max_age,
                     inflight_age_s=None, pending_age_s=None)
        with self._cv:
            value['completed_queue_age_s'] = max(
                (self.clock()-s.admitted_at for s in self._frames.values()), default=0)
        return value