"""Provider request filtering, latest-usage timing, and retry backoff.

Request rows expose attempt boundaries and pagination state that the
conversation history cannot show.
"""

import json
import unittest
import urllib.error
from datetime import datetime
from urllib.parse import urlencode

from harness import Albedo, Provider, Reply, text

FIRST = {
    "input_tokens": 100,
    "output_tokens": 20,
    "input_tokens_details": {"cached_tokens": 40},
    "output_tokens_details": {"reasoning_tokens": 7},
}
THIRD = {
    "input_tokens": 400,
    "output_tokens": 50,
    "input_tokens_details": {"cached_tokens": 160},
    "output_tokens_details": {"reasoning_tokens": 5},
}


def reply(request):
    inputs = request["input"]
    user = next(
        (
            str(item.get("content", ""))
            for item in reversed(inputs)
            if item.get("role") == "user"
        ),
        "",
    )
    if user == "hit the limit":
        # A spent quota fails at once; a brief limit would wait out the backoff.
        return Reply("error", "limit exceeded: insufficient_quota", status=429)
    if user == "third":
        return text("third reply", usage=THIRD)
    return text("first reply", usage=FIRST)


def instant(timestamp):
    return datetime.fromisoformat(timestamp.replace("Z", "+00:00"))


class ProviderRequestsTest(unittest.TestCase):
    def rows(self, app, session, query=""):
        with app.api(f"/sessions/{session}/context?view=requests{query}") as response:
            return json.load(response)

    def test_since_ms_filters_finished_attempts_and_binds_pagination(self):
        provider = Provider(reply)
        try:
            with Albedo(provider, protocol="responses") as app:
                session = app.session()
                for prompt in ("first", "third", "hit the limit"):
                    app.prompt(session, prompt).close()
                    app.idle(session)
                rows = self.rows(app, session)["items"]
                self.assertEqual(len(rows), 3)
                since_ms = round(instant(rows[1]["ended_at"]).timestamp() * 1000)
                self.assertLess(
                    instant(rows[0]["ended_at"]), instant(rows[1]["ended_at"])
                )
                query = f"&since_ms={since_ms}&limit=1"
                page = self.rows(app, session, query)
                self.assertEqual(page["items"], [rows[1]])
                self.assertIsNotNone(page["next"])
                continuation = "&" + urlencode({"next": page["next"]})
                following = self.rows(app, session, query + continuation)
                self.assertEqual(following["items"], [rows[2]])
                self.assertIsNone(following["next"])
                self.assertEqual(rows[2]["outcome"], "failed")
                after = self.rows(app, session, query + f"&after={rows[1]['sequence']}")
                self.assertEqual(after["items"], [rows[2]])
                ended_ms = round(instant(rows[2]["ended_at"]).timestamp() * 1000)
                self.assertEqual(
                    self.rows(app, session, f"&since_ms={ended_ms + 1}")["items"], []
                )
                for invalid in (
                    f"&since_ms={since_ms + 1}&limit=1" + continuation,
                    "&limit=1" + continuation,
                    "&since_ms=-1",
                    "&since_ms=1.5",
                    "&since_ms=9007199254740992",
                ):
                    with self.subTest(query=invalid):
                        with self.assertRaises(urllib.error.HTTPError) as rejected:
                            self.rows(app, session, invalid)
                        self.assertEqual(rejected.exception.code, 400)
                        rejected.exception.close()
                for view in ("summary", "section"):
                    with self.subTest(view=view):
                        with self.assertRaises(urllib.error.HTTPError) as rejected:
                            app.api(
                                f"/sessions/{session}/context?view={view}&since_ms=0"
                            ).close()
                        self.assertEqual(rejected.exception.code, 400)
                        rejected.exception.close()
        finally:
            provider.close()

    def test_the_latest_usage_reports_its_calls_span_and_rate(self):
        provider = Provider(reply)
        try:
            with Albedo(provider, protocol="responses") as app:
                session = app.session()
                app.prompt(session, "third").close()
                app.idle(session)
                row = self.rows(app, session)["items"][-1]
                with app.api(f"/sessions/{session}?tail=0") as response:
                    usage = json.load(response)["usage"]
                started, ended = instant(row["started_at"]), instant(row["ended_at"])
                span = round((ended - started).total_seconds() * 1000)
                self.assertEqual(usage["completion_tokens"], 50)
                self.assertEqual(usage["elapsed_ms"], span)
                if span:
                    self.assertAlmostEqual(usage["tokens_per_second"], 50 * 1000 / span)
                else:
                    self.assertIsNone(usage["tokens_per_second"])
        finally:
            provider.close()

    def test_a_failing_gateway_is_retried_with_doubling_waits(self):
        failures = []

        def script(request):
            if len(failures) < 4:
                failures.append(1)
                return Reply("error", "gateway down", status=503)
            return text("recovered")

        provider = Provider(script)
        try:
            with Albedo(provider, protocol="responses") as app:
                session = app.session()
                app.prompt(session, "hello").close()
                app.idle(session, timeout=60)
                rows = self.rows(app, session)["items"]
                self.assertEqual(
                    [row["outcome"] for row in rows], ["failed"] * 4 + ["completed"]
                )
                self.assertEqual({row["http_status"] for row in rows[:4]}, {503})
                self.assertTrue(rows[4]["transcript_positions"])
                # Each wait doubles from 250ms, measured from the failed
                # attempt's end to the next one's start.
                gaps = [
                    (
                        datetime.fromisoformat(
                            after["started_at"].replace("Z", "+00:00")
                        )
                        - datetime.fromisoformat(
                            before["ended_at"].replace("Z", "+00:00")
                        )
                    ).total_seconds()
                    * 1000
                    for before, after in zip(rows, rows[1:])
                ]
                for gap, wait in zip(gaps, (250, 500, 1000, 2000)):
                    self.assertGreaterEqual(gap, wait)
        finally:
            provider.close()