"""CPU contract/integration tests against upstream-shaped duplex boundaries.

These use real PCM/JPEG encoding, Pydantic parsing and FastAPI HTTP handling,
plus native adapter source-ledger and lifecycle logic. They do not claim CUDA.
"""
import base64
import io
import json
import threading
import time
from concurrent.futures import ThreadPoolExecutor

import numpy as np
import pytest
from fastapi.testclient import TestClient
from PIL import Image
from pydantic import ValidationError

from gpu_service.app import create_app
from gpu_service.contracts import AVChunk
from gpu_service.native import MODEL_LOCK, NativeDuplex, NativeContractError, prepare_text_context, text_only_duplex

TOKEN = "test-only-token-no-credentials-12345"
SESSION = "contract-session-1234"


class UpstreamDecoder:
    def __init__(self):
        self.units = 0
        self.generated_tokens = []

    def get_window_stats(self):
        return {"unit_count": min(self.units, 64), "cache_length": min(self.units, 64) * 100}

    def get_cache_length(self):
        return min(self.units, 64) * 100

    @property
    def _unit_history(self):
        return [None] * min(self.units, 64)

    def reset(self):
        self.units = 0


class UpstreamShapedDuplex:
    """Faithful signatures and first-chunk/core consumption from pinned source."""
    def __init__(self):
        self.decoder = UpstreamDecoder()
        self.audio_buffer = np.zeros(0, dtype=np.float32)
        self.calls = []
        self.current_turn_ended = True
        self.total_hidden = []
        self.total_ids = []
        self.prefill_schema_tokens = []
        self.res_ids = []
        self.stopped = False
        self.last_audio = None
        self.last_frames = None
        self.first = True
        self.generated_count = 0
        self.model = None
        self.prompts = []
        self.generate_options = []
        self.speak_token_id = 1
        self.pending_logits = np.zeros((1, 4), dtype=np.float32)

    def prepare(self, *, prefix_system_prompt):
        self.calls.append("prepare")
        assert "English" in prefix_system_prompt
        assert "chain-of-thought" in prefix_system_prompt
        self.prompts.append(prefix_system_prompt)
        self.stopped = False
        self.decoder.reset()
        self._reset_streaming_state()

    def _reset_streaming_state(self):
        self.audio_buffer = np.zeros(0, dtype=np.float32)
        self.first = True

    def set_session_stop(self):
        self.stopped = True
        self.calls.append("stop")

    def streaming_prefill(self, *, audio_waveform, frame_list, max_slice_nums, batch_vision_feed):
        self.calls.append("prefill")
        assert audio_waveform.dtype == np.float32 and audio_waveform.shape == (16000,)
        assert max_slice_nums == 2 and batch_vision_feed is True
        self.last_audio, self.last_frames = audio_waveform.copy(), frame_list
        self.pending_logits = np.zeros((1, 4), dtype=np.float32)
        self.audio_buffer = np.concatenate([self.audio_buffer, audio_waveform])
        if self.first:
            self.audio_buffer = np.concatenate([np.zeros(560, dtype=np.float32), self.audio_buffer])
            self.audio_buffer = self.audio_buffer[16480:]
            self.first = False
        else:
            self.audio_buffer = self.audio_buffer[16000:]
        return {"success": True, "cost_all": 0.0001}

    def streaming_generate(self, *, max_new_speak_tokens_per_chunk, decode_mode, listen_top_k=None, listen_prob_scale=1.0):
        self.calls.append("generate")
        assert 8 <= max_new_speak_tokens_per_chunk <= 64 and decode_mode == "sampling"
        self.generate_options.append((max_new_speak_tokens_per_chunk, listen_top_k, listen_prob_scale))
        self.generated_count += 1
        self.decoder.units += 1
        self.total_hidden.append([np.ones(10)])
        return {"is_listen": True, "text": "", "audio_waveform": np.zeros(24000), "end_of_turn": False, "current_time": self.generated_count, "cost_all": 0.0001}


def payload(sequence=0, start_ms=None, count=16000, epoch=0):
    start = sequence * 1000 if start_ms is None else start_ms
    # Dense continuously nonzero signal, encoded as actual little-endian float32.
    signal = (np.sin(np.arange(count) * 2 * np.pi * 313 / 16000) * 0.7).astype("<f4")
    image = Image.new("RGB", (64, 48), "green")
    stream = io.BytesIO()
    image.save(stream, "JPEG")
    return {"session_id": SESSION, "epoch": epoch, "sequence": sequence,
            "source_start_ms": start, "source_end_ms": start + count / 16,
            "sample_rate": 16000, "sample_count": count,
            "audio_b64": base64.b64encode(signal.tobytes()).decode(),
            "frames": [{"source_time_ms": start + count / 32, "jpeg_b64": base64.b64encode(stream.getvalue()).decode(), "timing_method": "requestVideoFrameCallback"}],
            "capture_time_ms": time.time() * 1000, "final_chunk": count < 16000,
            "evidence_ids": [f"e{epoch}c{sequence}"]}


def client_and_native():
    upstream = UpstreamShapedDuplex()
    adapter = NativeDuplex(upstream)
    return TestClient(create_app(adapter, TOKEN), headers={"Authorization": f"Bearer {TOKEN}"}), upstream


def start(client, epoch=0):
    response = client.post("/sessions", json={"session_id": SESSION, "epoch": epoch, "objective": "Evaluate source evidence relevant to copper demand."})
    assert response.status_code == 200, response.text


def test_lock_identifies_exact_o45_and_same_remote_code():
    assert MODEL_LOCK["model_id"] == "openbmb/MiniCPM-o-4_5"
    assert MODEL_LOCK["model_revision"] == MODEL_LOCK["inference_code_revision"]
    assert len(MODEL_LOCK["model_revision"]) == 40


def test_first_av_request_avoids_broken_optional_stats_and_retains_real_counts(monkeypatch):
    from unittest.mock import Mock

    client, upstream = client_and_native()
    broken_stats = Mock(side_effect=AttributeError("_system_prompt_template"))
    monkeypatch.setattr(upstream.decoder, "get_window_stats", broken_stats)
    with client:
        start(client)
        first = client.post(f"/sessions/{SESSION}/chunks", json=payload())
        assert first.status_code == 200, first.text
        assert upstream.last_audio.shape == (16000,) and np.any(upstream.last_audio)
        assert upstream.last_frames[0].mode == "RGB"
        assert first.json()["retained_native_units"] == 1
        assert first.json()["cache_tokens"] == 100
        # A native window drop must still change retained context; removing the
        # broken helper must not replace real counts with constants/defaults.
        upstream.decoder.units = 0
        second = client.post(f"/sessions/{SESSION}/chunks", json=payload(1))
        assert second.status_code == 200, second.text
        assert second.json()["context_start_ms"] == 1000
        assert second.json()["context_evidence_ids"] == ["e0c1"]
    broken_stats.assert_not_called()


def test_context_prefill_preserves_prompt_tokens_and_protection_with_bounded_forward_calls():
    from types import SimpleNamespace
    from unittest.mock import Mock

    # Token-shaped CPU decoder: compare the actual ordered context and protected
    # boundary, while proving long portfolios do not cause one forward per token.
    received = []
    decoder = SimpleNamespace(_window_config=SimpleNamespace(sliding_window_mode="basic"),
                              reset=Mock(side_effect=received.clear), embed_tokens=lambda ids: ids,
                              feed=Mock(side_effect=received.extend))
    protected = []
    decoder.register_system_prompt = lambda: protected.append(len(received))
    duplex = SimpleNamespace(generate_audio=False, decoder=decoder,
                             tokenizer=SimpleNamespace(encode=lambda s, **kw: list(s.encode())),
                             clear_break_event=Mock(), clear_session_stop=Mock(),
                             _reset_streaming_state=Mock(), model=SimpleNamespace(init_streaming_processor=Mock()))
    prompt = "Observed source and portfolio context. " * 60
    cancelled = threading.Event()
    prepare_text_context(duplex, prompt, cancelled)
    assert bytes(received).decode() == "<|im_start|>system\n" + prompt + "<|im_end|>"
    assert protected == [len(received)] and decoder.feed.call_count < 12
    duplex.model.init_streaming_processor.assert_called_once()
    cancelled.set()
    forwards = decoder.feed.call_count
    with pytest.raises(NativeContractError, match="cancelled"):
        prepare_text_context(duplex, prompt, cancelled)
    assert decoder.feed.call_count == forwards


def test_text_duplex_skips_unused_speech_renderer_and_restores_model_method():
    from types import SimpleNamespace

    class Model:
        config = SimpleNamespace(init_vision=True, init_audio=True, init_tts=False)
        fail_after_init = False

        def init_tts(self, **kwargs):
            raise RuntimeError("Unused speech renderer must not be initialized")

        def as_duplex(self, *, generate_audio):
            self.init_tts(enable_float16=False, n_timesteps=10)
            if self.fail_after_init:
                raise ValueError("unrelated constructor failure")
            return SimpleNamespace(generate_audio=generate_audio, model=self)

    model = Model()
    with pytest.raises(RuntimeError, match="Unused speech renderer"):
        model.as_duplex(generate_audio=False)
    duplex = text_only_duplex(model)
    assert duplex.model is model and duplex.generate_audio is False
    assert model.config.init_vision and model.config.init_audio
    assert "init_tts" not in vars(model)
    model.fail_after_init = True
    with pytest.raises(ValueError, match="unrelated constructor failure"):
        text_only_duplex(model)
    assert "init_tts" not in vars(model)
    with pytest.raises(RuntimeError, match="Unused speech renderer"):
        model.init_tts()


def test_eager_startup_error_identifies_stage_and_cause_without_secret_messages(monkeypatch):
    import importlib
    module = importlib.import_module("gpu_service.app")
    secret = "hf_" + "private-value-never-log" * 2
    monkeypatch.setenv("HF_TOKEN", secret)
    monkeypatch.setenv("NATIVE_EAGER_LOAD", "1")

    class FailingLoad(NativeDuplex):
        def load(self):
            self.load_stage = "native_duplex_setup"
            try:
                raise ValueError("private download URL and credential: " + secret)
            except ValueError:
                raise ImportError("unhelpful upstream wrapper " + secret) from None

    monkeypatch.setattr(module, "NativeDuplex", FailingLoad)
    with pytest.raises(RuntimeError) as failure:
        with TestClient(module.create_app(token=TOKEN)):
            pass
    message = str(failure.value)
    assert "native_duplex_setup" in message
    assert "ImportError" in message and "ValueError" in message
    assert "test_gpu_contract.py:" in message
    assert secret not in message and "private download URL" not in message


def test_portfolio_context_once_per_epoch_without_publication_cooldown():
    from types import SimpleNamespace
    client, native = client_and_native()
    resets = []
    native.model = SimpleNamespace(reset_session=lambda: resets.append(True))
    holding = {"id": "copper-fund", "name": "User copper research holding", "ticker": "COPPER",
               "exchange": "User supplied", "currency": "CAD", "asset_type": "fund", "weight": 100}
    context = {"session_id": SESSION, "epoch": 0, "objective": "Evaluate source evidence relevant to copper demand.",
               "holdings": [holding], "output_max_tokens_per_unit": 32, "live_update_interval_seconds": 10}
    original = native.streaming_generate
    def first_speaks(**kwargs):
        result = original(**kwargs)
        if native.generated_count == 1:
            return {**result, "text": "Evidence: Demand is discussed. Holding: copper-fund", "is_listen": False, "end_of_turn": True}
        return result
    native.streaming_generate = first_speaks
    with client:
        assert client.post("/sessions", json=context).status_code == 200
        assert client.post("/sessions", json=context).status_code == 200
        assert client.post("/sessions", json={**context, "holdings": [{**holding, "name": "Changed name"}]}).status_code == 409
        assert client.post(f"/sessions/{SESSION}/chunks", json=payload()).status_code == 200
        assert client.post(f"/sessions/{SESSION}/chunks", json=payload(1)).status_code == 200
        assert native.generate_options == [(32, None, 1.0), (32, None, 1.0)]
        assert len(native.prompts) == 1 and '"id":"copper-fund"' in native.prompts[0]
        assert len(resets) == 1
        assert "Holding: the exact user holding ID" in native.prompts[0]
        assert client.post(f"/sessions/{SESSION}/reset", json={"epoch": 1}).status_code == 200
        assert len(native.prompts) == 2 and native.prompts[1] == native.prompts[0]
        assert len(resets) >= 2
        assert client.post(f"/sessions/{SESSION}/chunks", json=payload(0, start_ms=30000, epoch=1)).status_code == 200
        assert native.generate_options[-1] == (32, None, 1.0)


def test_native_silence_is_not_overridden_and_effective_settings_are_reported():
    client, native = client_and_native()
    original = native.streaming_generate
    def decode(**kwargs):
        # Application must leave every pending logit unchanged, even after a
        # long silent stretch or the legacy interval setting.
        np.testing.assert_array_equal(native.pending_logits, np.zeros((1, 4)))
        return original(**kwargs)
    native.streaming_generate = decode
    with client:
        response = client.post('/sessions', json={'session_id': SESSION, 'epoch': 0,
            'objective': 'Evaluate supplied media.', 'live_update_interval_seconds': 1})
        assert response.status_code == 200
        runtime = response.json()['runtime']
        assert runtime['max_slice_nums'] == 2 and runtime['batch_vision_feed'] is True
        assert runtime['generate_audio'] is False and runtime['audio_understanding'] is True
        assert runtime['window_high_tokens'] == 8000 and runtime['window_low_tokens'] == 6000
        for i in range(5):
            result = client.post(f'/sessions/{SESSION}/chunks', json=payload(i)).json()
            assert result['native_silence'] and result['text'] == ''
            timing = result['timing']
            assert timing['new_source_ms'] == (995 if i == 0 else 1000)
            assert timing['processing_per_source_second'] == pytest.approx(
                (timing['preprocess_ms'] + timing['native_unit_ms']) / timing['new_source_ms'])
        assert native.calls.count('prepare') == 1 and native.calls.count('generate') == 5
        assert client.post(f'/sessions/{SESSION}/finish', json={'epoch': 0}).status_code == 200


def test_real_pcm_jpeg_decode_and_native_request(monkeypatch):
    from gpu_service.contracts import Frame
    audio_decode, frame_decode = AVChunk.decode_audio, Frame.decode
    decoded = []
    def audio_once(chunk):
        decoded.append(("audio", threading.current_thread().name))
        return audio_decode(chunk)
    def frame_once(frame):
        decoded.append(("frame", threading.current_thread().name))
        return frame_decode(frame)
    monkeypatch.setattr(AVChunk, "decode_audio", audio_once)
    monkeypatch.setattr(Frame, "decode", frame_once)
    client, native = client_and_native()
    with client:
        start(client)
        data = payload()
        second = io.BytesIO()
        Image.new("RGB", (32, 24), "blue").save(second, "JPEG")
        data["frames"].append({"source_time_ms": 750.125,
                               "jpeg_b64": base64.b64encode(second.getvalue()).decode()})
        response = client.post(f"/sessions/{SESSION}/chunks", json=data)
        assert response.status_code == 200, response.text
        result = response.json()
        assert native.last_frames[0].mode == "RGB"
        np.testing.assert_array_equal(native.last_audio, np.frombuffer(base64.b64decode(data["audio_b64"]), dtype="<f4"))
        assert [kind for kind, _ in decoded] == ["audio", "frame", "frame"]
        assert [frame.size for frame in native.last_frames] == [(64, 48), (32, 24)]
        assert result["timing"]["frame_source_times_ms"] == [500, 750.125]
        assert all(name.startswith("asyncio_") for _, name in decoded)
        assert client.post(f"/sessions/{SESSION}/chunks", json=data).json()["idempotent_replay"]
        assert len(decoded) == 3 and native.calls.count("generate") == 1
        assert result["consumed_source_end_ms"] == 995
        assert result["synthetic_padding_samples"] == 560
        assert result["context_evidence_ids"] == ["e0c0"]
        assert result["native_silence"] is True
        assert result["audio_kind"] == "audio evidence"
        assert len(native.total_hidden) == 0


def test_stop_during_cpu_preparation_discards_decoded_input_before_native_work(monkeypatch):
    from concurrent.futures import ThreadPoolExecutor
    entered, release = threading.Event(), threading.Event()
    decode = NativeDuplex.decode
    def blocked(chunk):
        entered.set()
        assert release.wait(3)
        return decode(chunk)
    monkeypatch.setattr(NativeDuplex, "decode", staticmethod(blocked))
    client, native = client_and_native()
    with client, ThreadPoolExecutor(max_workers=1) as pool:
        start(client)
        request = pool.submit(client.post, f"/sessions/{SESSION}/chunks", json=payload())
        try:
            assert entered.wait(2)
            before = time.perf_counter()
            assert client.post(f"/sessions/{SESSION}/stop", json={"epoch": 0}).status_code == 200
            assert time.perf_counter() - before < 1
        finally:
            release.set()
        assert request.result(timeout=3).status_code == 409
        assert "prefill" not in native.calls and "generate" not in native.calls


def test_ten_minute_dense_audio_actual_encoder_http_and_final_five_ms():
    client, native = client_and_native()
    with client:
        start(client)
        for sequence in range(600):
            response = client.post(f"/sessions/{SESSION}/chunks", json=payload(sequence))
            assert response.status_code == 200, (sequence, response.text)
        result = response.json()
        assert result["consumed_source_end_ms"] == 599995
        assert len(native.total_hidden) == 0
        finish = client.post(f"/sessions/{SESSION}/finish", json={"epoch": 0})
        assert finish.status_code == 200, finish.text
        report = finish.json()
        assert report["model_consumed_sample_count"] == 600 * 16000
        assert report["complete"] and report["model_consumed_end_ms"] == 600000
        assert report["outputs"][0]["consumed_source_start_ms"] == 599995
        assert report["outputs"][0]["consumed_source_end_ms"] == 600000
        assert report["outputs"][0]["synthetic_tail"]
        assert report["outputs"][0]["context_end_ms"] == 600000
        assert native.calls.count("prepare") == 1
        assert native.calls.count("generate") == 601


@pytest.mark.parametrize("count", [1, 79, 80, 81, 15999])
def test_final_partial_preserves_every_sample(count):
    client, native = client_and_native()
    with client:
        start(client)
        assert client.post(f"/sessions/{SESSION}/chunks", json=payload(0)).status_code == 200
        response = client.post(f"/sessions/{SESSION}/chunks", json=payload(1, count=count))
        assert response.status_code == 200, response.text
        completion = client.post(f"/sessions/{SESSION}/finish", json={"epoch": 0}).json()
        assert completion["complete"]
        assert completion["model_consumed_sample_count"] == 16000 + count
        assert completion["model_consumed_end_ms"] == 1000 + count / 16


@pytest.mark.parametrize("mutation", [
    {"sample_rate": 48000}, {"audio_b64": "not-base64"}, {"sample_count": 5},
    {"audio_b64": base64.b64encode(np.full(16000, np.nan, dtype="<f4").tobytes()).decode()},
    {"source_end_ms": 2000}, {"source_end_ms": float("inf")},
])
def test_bad_encoding_intervals_rejected_before_model(mutation):
    client, native = client_and_native()
    with client:
        start(client)
        data = payload()
        data.update(mutation)
        if mutation.get("source_end_ms") == float("inf"):
            with pytest.raises(ValidationError):
                AVChunk(**data)
        else:
            response = client.post(f"/sessions/{SESSION}/chunks", json=data)
            assert response.status_code == 422
        assert "prefill" not in native.calls


def test_idempotent_retry_no_second_inference_and_conflict_rejected():
    client, native = client_and_native()
    with client:
        start(client)
        data = payload()
        first = client.post(f"/sessions/{SESSION}/chunks", json=data)
        repeat = client.post(f"/sessions/{SESSION}/chunks", json=data)
        assert first.status_code == repeat.status_code == 200
        assert repeat.json()["idempotent_replay"]
        assert native.calls.count("generate") == 1
        data["audio_b64"] = base64.b64encode(np.zeros(16000, dtype="<f4").tobytes()).decode()
        assert client.post(f"/sessions/{SESSION}/chunks", json=data).status_code == 409


def test_out_of_order_source_gap_seek_and_stale_epoch():
    client, native = client_and_native()
    with client:
        start(client)
        assert client.post(f"/sessions/{SESSION}/chunks", json=payload(1)).status_code == 409
        assert client.post(f"/sessions/{SESSION}/chunks", json=payload(0)).status_code == 200
        assert client.post(f"/sessions/{SESSION}/reset", json={"epoch": 1}).status_code == 200
        assert client.post(f"/sessions/{SESSION}/chunks", json=payload(1)).status_code == 409
        new = client.post(f"/sessions/{SESSION}/chunks", json=payload(0, start_ms=50000, epoch=1))
        assert new.status_code == 200, new.text
        assert new.json()["context_start_ms"] == 50000
        assert new.json()["context_evidence_ids"] == ["e1c0"]
        gap = client.post(f"/sessions/{SESSION}/chunks", json=payload(1, start_ms=53000, epoch=1))
        assert gap.status_code == 502


def test_stop_no_additional_generation_finish_or_prepare():
    client, native = client_and_native()
    with client:
        start(client)
        assert client.post(f"/sessions/{SESSION}/chunks", json=payload()).status_code == 200
        calls = native.calls.count("generate")
        response = client.post(f"/sessions/{SESSION}/stop", json={"epoch": 0})
        assert response.status_code == 200
        assert response.json()["gpu_uptime_billing_paused"] is False
        assert client.post(f"/sessions/{SESSION}/finish", json={"epoch": 0}).status_code == 409
        assert client.post(f"/sessions/{SESSION}/chunks", json=payload(1)).status_code == 409
        assert native.calls.count("generate") == calls
        assert native.calls.count("prepare") == 1


def test_stop_interrupts_control_immediately_and_discards_late_result():
    client, native = client_and_native()
    entered, release = threading.Event(), threading.Event()
    original_generate = native.streaming_generate
    def slow(**kwargs):
        entered.set()
        release.wait(timeout=5)
        return original_generate(**kwargs)
    native.streaming_generate = slow
    with client, ThreadPoolExecutor(max_workers=1) as pool:
        start(client)
        future = pool.submit(client.post, f"/sessions/{SESSION}/chunks", json=payload())
        assert entered.wait(2)
        before = time.monotonic()
        response = client.post(f"/sessions/{SESSION}/stop", json={"epoch": 0})
        assert response.status_code == 200
        assert time.monotonic() - before < 1
        release.set()
        assert future.result(timeout=5).status_code == 409


def test_authentication_and_validation_never_echo_payload():
    client, native = client_and_native()
    with client:
        assert client.get("/health", headers={"Authorization": "Bearer wrong"}).status_code == 401
        assert client.post("/sessions", json={"session_id": "PRIVATE_SENTINEL", "epoch": -1, "objective": "PRIVATE_SENTINEL"}).status_code == 422
        response = client.post("/sessions", json={"session_id": "PRIVATE_SENTINEL", "epoch": -1, "objective": "PRIVATE_SENTINEL"})
        assert "PRIVATE_SENTINEL" not in response.text


def test_malformed_native_result_fails_closed_and_requires_new_epoch():
    client, native = client_and_native()
    native.streaming_generate = lambda **_: {"text": "unchecked", "is_listen": "false"}
    with client:
        start(client)
        response = client.post(f"/sessions/{SESSION}/chunks", json=payload())
        assert response.status_code == 502
        assert "unchecked" not in response.text
        assert client.post(f"/sessions/{SESSION}/chunks", json=payload(1)).status_code == 409


def test_separate_inference_secret_supports_private_hf_ingress():
    client, native = client_and_native()
    with client:
        response = client.get("/health", headers={"Authorization": "Bearer hf_ingress_placeholder", "X-Inference-Token": TOKEN})
        assert response.status_code == 200
        rejected = client.get("/health", headers={"Authorization": "Bearer " + TOKEN, "X-Inference-Token": "wrong"})
        assert rejected.status_code == 401


def test_real_application_http_client_to_native_service_contract():
    import asyncio
    from types import SimpleNamespace
    import httpx
    from pydantic import SecretStr
    from backend.perception import PerceptionClient
    from backend.schemas import AVChunk as AppChunk

    async def exercise():
        upstream = UpstreamShapedDuplex()
        app = create_app(NativeDuplex(upstream), TOKEN)
        settings = SimpleNamespace(app_mode="real", inference_service_token=SecretStr(TOKEN), hf_token=None,
                                   minicpm_service_url="http://127.0.0.1:8090", inference_timeout_seconds=5)
        async with app.router.lifespan_context(app):
            async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url=settings.minicpm_service_url,
                                         headers={"X-Inference-Token": TOKEN}) as http:
                client = PerceptionClient(settings, SESSION, client=http)
                await client.start("Report English audiovisual evidence")
                for sequence in range(3):
                    encoded = payload(sequence)
                    encoded.pop("session_id")
                    encoded.pop("evidence_ids")
                    chunk = AppChunk(**encoded)
                    response = await client.process(chunk)
                    assert response["sequence"] == sequence
                    assert response["context_evidence_ids"][-1] == f"{SESSION}:e0:c{sequence}"
                report = await client.finish(0)
                assert report["complete"] and report["model_consumed_sample_count"] == 48000
                assert await client.stop()
                assert upstream.calls.count("prepare") == 1
    asyncio.run(exercise())


def test_native_timeout_does_not_retry_an_uncertain_request(monkeypatch):
    monkeypatch.setenv("NATIVE_CALL_TIMEOUT_SECONDS", "0.05")
    client, native = client_and_native()
    release = threading.Event()
    original_generate = native.streaming_generate
    def timeout(**kwargs):
        release.wait(2)
        return original_generate(**kwargs)
    native.streaming_generate = timeout
    with client:
        start(client)
        try:
            first = client.post(f"/sessions/{SESSION}/chunks", json=payload())
            assert first.status_code == 504
            assert client.post(f"/sessions/{SESSION}/chunks", json=payload()).status_code == 409
            assert client.post(f"/sessions/{SESSION}/finish", json={"epoch": 0}).status_code == 409
        finally:
            release.set()


def test_stop_during_model_loading_never_prefills_prompt_after_stop():
    upstream = UpstreamShapedDuplex()
    adapter = NativeDuplex(upstream)
    entered, release = threading.Event(), threading.Event()
    def slow_load():
        entered.set()
        release.wait(3)
    adapter.load = slow_load
    client = TestClient(create_app(adapter, TOKEN), headers={"X-Inference-Token": TOKEN})
    with client, ThreadPoolExecutor(max_workers=1) as pool:
        future = pool.submit(client.post, "/sessions", json={"session_id": SESSION, "epoch": 0, "objective": "Read evidence"})
        assert entered.wait(2)
        stopped = client.post(f"/sessions/{SESSION}/stop", json={"epoch": 0})
        assert stopped.status_code == 200
        release.set()
        assert future.result(timeout=5).status_code == 409
        assert "prepare" not in upstream.calls
        assert "prefill" not in upstream.calls
        assert "generate" not in upstream.calls


def test_bounded_native_debug_history_preserves_selected_repetition_window():
    upstream = UpstreamShapedDuplex()
    adapter = NativeDuplex(upstream)
    upstream.decoder.generated_tokens = list(range(12000))
    upstream.decoder.generated_special_tokens = list(range(12000))
    upstream.decoder.context = "x" * 20000
    upstream.total_hidden = [object()] * 100
    upstream.res_ids = list(range(3000))
    recent = upstream.decoder.generated_tokens[-512:]
    adapter._bound_debug_state()
    assert upstream.decoder.generated_tokens[-512:] == recent
    assert len(upstream.decoder.generated_tokens) == 4096
    assert len(upstream.decoder.generated_special_tokens) == 4096
    assert len(upstream.decoder.context) == 8192
    assert len(upstream.res_ids) == 512 and upstream.total_hidden == []