"""Real HTTP encoder/client/native service boundary, with a fake CUDA decoder only."""
import asyncio

import httpx
import pytest
from pydantic import SecretStr

from backend.perception import PerceptionClient
from backend.schemas import AVChunk, StartRequest
from backend.session import Session
from backend.settings import Settings
from gpu_service.app import create_app
from gpu_service.native import NativeDuplex
from tests.test_api import media_chunk, start_body


class Decoder:
    def __init__(self):
        self.units = 0
    def get_window_stats(self):
        return {"unit_count": min(self.units, 30), "cache_length": min(self.units * 200, 8000)}
    def get_cache_length(self):
        return min(self.units * 200, 8000)
    @property
    def _unit_history(self):
        return [None] * min(self.units, 30)
    def reset(self):
        self.units = 0


class DuplexFixture:
    """Mirrors the locked native first-chunk/sample-remainder contract, not media encoding."""
    def __init__(self):
        import numpy as np
        self.decoder = Decoder()
        self.audio_buffer = np.zeros(0, dtype=np.float32)
        self.current_turn_ended = True
        self.first = True
        self.calls = 0
        self.speak_token_id = 1
        self.pending_logits = np.zeros((1, 4), dtype=np.float32)
    def prepare(self, **kwargs):
        self._reset_streaming_state()
    def _reset_streaming_state(self):
        import numpy as np
        self.audio_buffer = np.zeros(0, dtype=np.float32)
        self.first = True
        self.decoder.reset()
    def set_session_stop(self):
        pass
    def streaming_prefill(self, audio_waveform, frame_list, **kwargs):
        import numpy as np
        assert audio_waveform.dtype == np.float32
        assert audio_waveform.ndim == 1
        assert all(frame.mode == "RGB" for frame in frame_list)
        prefix = np.zeros(560 if self.first else 0, dtype=np.float32)
        self.audio_buffer = np.concatenate([self.audio_buffer, prefix, audio_waveform])
        self.audio_buffer = self.audio_buffer[16480 if self.first else 16000:]
        self.first = False
        self.pending_logits = np.zeros((1, 4), dtype=np.float32)
        self.decoder.units += 1
        return {"success": True}
    def streaming_generate(self, **kwargs):
        self.calls += 1
        return {"text": "Evidence: The fixture reports an uncertain supply outlook.\nHolding: xom\nImplication: If supply disruptions affect the issuer operations, delivery reliability could weaken.\nUncertainty: The issuer exposure is unverified.\nVerify: Check affected facility locations in issuer disclosures." if self.calls % 20 == 1 else "",
                "is_listen": self.calls % 20 != 1, "end_of_turn": True, "current_time": self.calls}


@pytest.mark.asyncio
async def test_full_600_seconds_through_app_client_and_native_encoder_contract():
    native = NativeDuplex(DuplexFixture())
    gpu_app = create_app(native=native, token="fixture-service-token-long-enough")
    cfg = Settings(app_mode="real", inference_service_token=SecretStr("fixture-service-token-long-enough"),
                   mock_latency_seconds=0)
    # Real perception HTTP adapter + actual GPU-service parser/decode/worker.
    # Only the expensive CUDA decoder is replaced by a deterministic fixture.
    async with gpu_app.router.lifespan_context(gpu_app):
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=gpu_app), base_url="http://gpu",
                                    headers={"X-Inference-Token": "fixture-service-token-long-enough"}) as http:
            session = Session(StartRequest.model_validate(start_body(600000)), cfg)
            session.perception = PerceptionClient(cfg, session.id, client=http)
            await session.start()
            for sequence in range(600):
                await session.accept(AVChunk.model_validate(media_chunk(sequence)))
                # Simulates finite-source pacing instead of discarding queued speech.
                await session.queue.join()
            await session.complete()
            await session.completion_task
            assert session.state == "completed"
            assert session.coverage["processed_chunks"] == 600
            assert session.coverage["processed_audio_ms"] == 600000
            assert session.coverage["model_consumed_end_ms"] == 600000
            assert session.coverage["complete"], session.coverage
            assert native.sample_count_consumed == 9_600_000
            assert native.duplex.calls == 601  # Native retained5ms is drained at EOF.
            assert session.brief is not None
            assert session.brief["implications"][0]["holding_id"] == "xom"
            assert session.coverage["final_summary_requests"] == 0
            assert session.coverage["attempts"] == 600
            assert len(session.evidence) <= 720