"""Controlled B1 integration proof: simulated provider capacity, no production quota."""

from __future__ import annotations

import asyncio
import inspect
import json
import time
from pathlib import Path
from types import SimpleNamespace

import httpx
import openai
import pytest
from src.core.provider_capacity import (
    MemoryCapacityStore,
    ProviderCapacityUnavailableError,
    capacity_bucket,
    override_capacity_store,
)
from src.core.provider_capacity.types import CapacityObservation
from src.domain.projects.discovery_b1 import capacity_summary, score_b1
from src.domain.projects.discovery_schemas import DiscoveryTurnResult
from src.integrations import project_intake_client as client_module

ROOT = Path(__file__).resolve().parents[1]
WITNESS = json.loads(
    (ROOT / "docs/audits/fixtures/13a_run_02/trials/13A-004.json").read_text(encoding="utf-8")
)


def _settings(**overrides):
    values = dict(
        openai_api_key="test-key",
        project_intake_model="gpt-4o-mini",
        redis_url=None,
        provider_capacity_enabled=True,
        provider_capacity_safety_fraction=0.9,
        provider_capacity_wait_seconds=1.0,
        provider_capacity_unknown_max_in_flight=2,
        provider_capacity_stale_observation_seconds=120,
    )
    values.update(overrides)
    return SimpleNamespace(**values)


def _success_response():
    parsed = DiscoveryTurnResult.model_validate(WITNESS["discovery"]["raw_turn"])
    message = SimpleNamespace(refusal=None, parsed=parsed)
    return SimpleNamespace(
        choices=[SimpleNamespace(message=message)],
        usage=SimpleNamespace(
            prompt_tokens=100,
            completion_tokens=20,
            total_tokens=120,
            model_dump=lambda **_kw: {
                "prompt_tokens": 100,
                "completion_tokens": 20,
                "total_tokens": 120,
            },
        ),
        model_dump=lambda **_kw: {"ok": True, "usage": {"total_tokens": 120}},
    )


def _install_fake_parse(monkeypatch, parse):
    captured = {}

    class FakeClient:
        def __init__(self, **kwargs):
            captured["timeout"] = kwargs.get("timeout")
            captured["max_retries"] = kwargs.get("max_retries")
            self.chat = SimpleNamespace(completions=SimpleNamespace(parse=parse))

        async def __aenter__(self):
            return self

        async def __aexit__(self, *_exc):
            return False

    monkeypatch.setattr(openai, "AsyncOpenAI", FakeClient)
    return captured


def _tokens_bucket():
    return capacity_bucket(
        provider="openai", secret="test-key", model="gpt-4o-mini", dimension="tokens"
    )


@pytest.fixture
def fixed_estimate(monkeypatch):
    monkeypatch.setattr(
        "src.integrations.openai_capacity.estimate_openai_request_tokens",
        lambda *a, **k: 8_000,
    )


@pytest.mark.asyncio
async def test_capacity_queue_engages_and_resumes_without_429(monkeypatch, fixed_estimate):
    store = MemoryCapacityStore()
    override_capacity_store(store)
    now = int(time.time() * 1000)
    await store.reconcile(
        _tokens_bucket(),
        CapacityObservation(
            "tokens", limit=20_000, remaining=12_000, reset_in_seconds=60, observed_at_ms=now
        ),
        now,
    )
    hold = asyncio.Event()
    started = asyncio.Event()
    calls: list[int] = []

    async def parse(**_kwargs):
        calls.append(1)
        started.set()
        await hold.wait()
        return _success_response()

    captured = _install_fake_parse(monkeypatch, parse)
    trace_a: dict = {}
    trace_b: dict = {}

    async def run(trace):
        return await client_module._project_structured_call(
            {"messages": []},
            _settings(),
            "prompt",
            DiscoveryTurnResult,
            8,
            trace=trace,
        )

    first = asyncio.create_task(run(trace_a))
    await started.wait()
    second = asyncio.create_task(run(trace_b))
    await asyncio.sleep(0.08)
    assert second.done() is False
    assert (trace_b.get("provider_capacity") or {}).get("capacity_queued") is True
    hold.set()
    await asyncio.gather(first, second)
    assert captured["timeout"] == 8
    assert captured["max_retries"] == 0
    assert len(calls) == 2
    assert trace_a["provider_capacity"]["capacity_granted"] is True
    assert trace_b["provider_capacity"]["capacity_granted"] is True
    assert trace_b["provider_capacity"]["capacity_wait_ms"] > 0
    assert trace_a.get("provider_429_final") is False
    assert trace_b.get("provider_429_final") is False
    summary = capacity_summary(
        [
            {
                "discovery": {
                    "status": "success",
                    "provider_capacity": trace_a["provider_capacity"],
                    "provider_429_final": False,
                }
            },
            {
                "discovery": {
                    "status": "success",
                    "provider_capacity": trace_b["provider_capacity"],
                    "provider_429_final": False,
                }
            },
        ]
    )
    assert summary["capacity_queued"] >= 1
    assert summary["provider_429_final"] == 0
    assert summary["capacity_deadline_exceeded"] == 0


@pytest.mark.asyncio
async def test_capacity_deadline_is_not_a_provider_timeout_or_semantic_failure(
    monkeypatch, fixed_estimate
):
    store = MemoryCapacityStore()
    override_capacity_store(store)
    now = int(time.time() * 1000)
    await store.reconcile(
        _tokens_bucket(),
        CapacityObservation(
            "tokens", limit=10_000, remaining=50, reset_in_seconds=60, observed_at_ms=now
        ),
        now,
    )

    async def parse(**_kwargs):
        raise AssertionError("provider must not run when capacity is unavailable")

    captured = _install_fake_parse(monkeypatch, parse)
    trace: dict = {}
    with pytest.raises(ProviderCapacityUnavailableError):
        await client_module._project_structured_call(
            {"messages": []},
            _settings(provider_capacity_wait_seconds=0.05),
            "prompt",
            DiscoveryTurnResult,
            8,
            trace=trace,
        )
    assert "timeout" not in captured
    assert trace["provider_capacity"]["capacity_deadline_exceeded"] is True
    scored = score_b1(
        [
            {
                "id": "c",
                "prompt": "x",
                "oracle_kind": "admitted",
                "discovery": {
                    "status": "error",
                    "error_type": "ProviderCapacityUnavailableError",
                    "provider_capacity": trace["provider_capacity"],
                },
            }
        ]
    )
    assert scored["discovery_timeout_rate"]["numerator"] == 0
    assert scored["provider_capacity"]["capacity_deadline_exceeded"] == 1
    assert scored["discovery_error_types"] == {"ProviderCapacityUnavailableError": 1}
    assert scored["timeout_policy_changed"] is False


@pytest.mark.asyncio
async def test_provider_429_remains_distinguishable_from_capacity_queue(
    monkeypatch, fixed_estimate
):
    store = MemoryCapacityStore()
    override_capacity_store(store)
    now = int(time.time() * 1000)
    await store.reconcile(
        _tokens_bucket(),
        CapacityObservation(
            "tokens",
            limit=200_000,
            remaining=180_000,
            reset_in_seconds=60,
            observed_at_ms=now,
        ),
        now,
    )
    request = httpx.Request("POST", "https://api.openai.com/v1/chat/completions")

    async def parse(**_kwargs):
        raise openai.RateLimitError(
            "rate limited tokens",
            response=httpx.Response(
                429,
                request=request,
                headers={
                    "retry-after": "0.01",
                    "x-ratelimit-limit-tokens": "200000",
                    "x-ratelimit-remaining-tokens": "100",
                },
            ),
            body=None,
        )

    captured = _install_fake_parse(monkeypatch, parse)
    trace: dict = {}
    with pytest.raises(openai.RateLimitError):
        await client_module._project_structured_call(
            {"messages": []},
            _settings(),
            "prompt",
            DiscoveryTurnResult,
            8,
            trace=trace,
        )
    assert captured["timeout"] == 8
    assert captured["max_retries"] == 0
    assert trace["provider_429_initial"] is True
    assert trace["provider_429_final"] is True
    assert trace["provider_capacity"]["capacity_granted"] is True
    scored = score_b1(
        [
            {
                "id": "d",
                "prompt": "x",
                "oracle_kind": "admitted",
                "discovery": {
                    "status": "error",
                    "error_type": "RateLimitError",
                    "provider_429_initial": True,
                    "provider_429_final": True,
                    "provider_capacity": trace["provider_capacity"],
                },
            }
        ]
    )
    assert scored["discovery_timeout_rate"]["numerator"] == 0
    assert scored["provider_capacity"]["provider_429_final"] == 1
    assert scored["provider_capacity"]["capacity_deadline_exceeded"] == 0


def test_structured_call_keeps_eight_second_timeout_and_bounded_retry():
    source = inspect.getsource(client_module._project_structured_call)
    assert "timeout=timeout" in source
    assert "max_retries=0" in source
    assert "max_retries=1" not in source
    from src.workers.tasks import project_discovery as worker

    worker_source = inspect.getsource(worker.run_project_discovery)
    assert "asyncio.wait_for" not in worker_source
    assert "model_facing_discovery_payload(payload)" in worker_source
    classify_source = inspect.getsource(worker._classify_discovery_failure)
    assert "provider_capacity_unavailable" in classify_source