"""Run one minute of closed-loop C8 traffic on each public REST route.

Requires httpx plus benchmark_public.py in the same directory. Uses ORUK_API_KEY
or --key-file. The completed initial benchmark supplies fixture hashes; its
manifest supplies local audio paths. All requests and errors are retained.
"""
import argparse
import asyncio
import hashlib
import json
import math
import os
from pathlib import Path
import time
import uuid

import httpx
from benchmark_public import fixture, summarize


async def main(args):
    # Running this after the initial phases avoids contaminating streaming data.
    while not Path(args.initial).exists() or not json.loads(Path(args.initial).read_text()).get('completed_at'):
        await asyncio.sleep(5)
    initial = json.loads(Path(args.initial).read_text())
    key = Path(args.key_file).read_text().strip() if args.key_file else os.environ['ORUK_API_KEY']
    candidates = {row['file_sha256']: row for row in (json.loads(line) for line in Path(args.manifest).read_text().splitlines())}
    clips = [fixture(candidates[row['sha256']]['audio_filepath']) for row in initial['fixtures']]
    report = {'started_at': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), 'results': [], 'receipts': [],
              'script_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
              'method': '60 seconds of closed-loop concurrency 8 per route; include drain time in throughput denominator; same preloaded WAV clips as initial benchmark, cyclic selection, native key, no optional tasks, no retries or outlier removal; eight warmups per route retained separately.'}
    prefix = 'orukeet-sustained-' + uuid.uuid4().hex[:12] + '-'
    output = Path(args.output)
    journal = output.with_suffix('.jsonl').open('w')

    async def post(client, base, clip, warmup=False):
        ident = prefix + uuid.uuid4().hex
        row = {'id': ident, 'group': base, 'warmup': warmup, 'audio_seconds': clip['duration'], 'fixture_sha256': clip['sha256'], 'ok': False}
        start = time.perf_counter()
        try:
            response = await client.post(base + '/v1/audio/transcriptions',
                headers={'Authorization': 'Bearer ' + key, 'X-Request-ID': ident},
                data={'model': 'oruk-orukeet'}, files={'file': ('audio.wav', clip['audio'], 'audio/wav')})
            row.update(status=response.status_code, client_ms=(time.perf_counter() - start) * 1000)
            data = response.json()
            if response.status_code == 200:
                row.update({k: data[k] for k in ('asr_ms', 'server_ms')})
                row['cost_microusd'] = data['usage']['cost_microusd']
                assert row['cost_microusd'] == math.ceil(clip['duration'] / 60 * 100 - 1e-8)
                assert abs(data['duration'] - clip['duration']) < .001 and data['text'].strip()
                row['ok'] = True
            else:
                row['error_code'] = data.get('error', {}).get('code', 'http_error')
        except Exception as exc:
            row.update(error_type=type(exc).__name__, client_ms=(time.perf_counter() - start) * 1000)
        report['receipts'].append(row)
        journal.write(json.dumps(row) + '\n')
        journal.flush()
        return row

    try:
        for base in ('https://orukeet-direct.oruk.ai', 'https://speech-api.oruk.ai'):
            async with httpx.AsyncClient(timeout=90, limits=httpx.Limits(max_connections=8, max_keepalive_connections=8, keepalive_expiry=120)) as client:
                await asyncio.gather(*(post(client, base, clips[index], True) for index in range(8)))
                rows = []
                cursor = 0
                start = time.perf_counter()

                async def worker():
                    nonlocal cursor
                    while time.perf_counter() - start < 60:
                        index = cursor
                        cursor += 1
                        rows.append(await post(client, base, clips[index % len(clips)]))

                await asyncio.gather(*(worker() for _ in range(8)))
                result = {'base_url': base, 'concurrency': 8, 'offered_load_seconds': 60, **summarize(rows, time.perf_counter() - start)}
                report['results'].append(result)
                print(json.dumps(result), flush=True)
                output.write_text(json.dumps(report, indent=2) + '\n')
        report['completed_at'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
    finally:
        output.write_text(json.dumps(report, indent=2) + '\n')
        journal.close()


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    for argument in ('initial', 'manifest', 'output'):
        parser.add_argument('--' + argument, required=True)
    parser.add_argument('--key-file')
    asyncio.run(main(parser.parse_args()))