"""Integration tests for api.openai_realtime.websocket_router.

Uses Starlette's synchronous TestClient with WebSocket support to exercise
the full FastAPI app produced by ``create_app``. Each test gets a fresh
PipelineUnit pool (size 1, matching the single-session semantics of the
old tests) so there is no cross-test state.
"""

import asyncio
import base64
import time
from queue import Empty, Queue
from threading import Event as ThreadingEvent

import pytest
from starlette.testclient import TestClient
from starlette.websockets import WebSocketState

import speech_to_speech.api.openai_realtime.websocket_router as router_module
from speech_to_speech.api.openai_realtime.pipeline_unit import PipelineUnit
from speech_to_speech.api.openai_realtime.service import CHUNK_SIZE_BYTES, RealtimeService
from speech_to_speech.api.openai_realtime.websocket_router import create_app
from speech_to_speech.pipeline.cancel_scope import CancelScope
from speech_to_speech.pipeline.control import SESSION_END, PipelineControlMessage, is_control_message
from speech_to_speech.pipeline.events import AssistantTextEvent, SpeechStartedEvent, TokenUsageEvent
from speech_to_speech.pipeline.messages import AUDIO_RESPONSE_DONE, PIPELINE_END, AudioOutput

# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True)
def short_drain_timeout(monkeypatch):
    """Shorten the SESSION_END drain warning threshold so tests don't wait 10s.

    The constant only controls when the release task logs a warning about a
    slow-draining unit — there is no longer a release-anyway timeout, so the
    unit stays unavailable until SESSION_END actually drains.
    """
    monkeypatch.setattr(router_module, "SESSION_END_DRAIN_TIMEOUT_S", 0.1)


@pytest.fixture
def setup():
    """Return (app, service, input_queue, output_queue, text_output_queue,
    should_listen, stop_event, response_playing, cancel_scope) for a pool of one.

    There is no real handler chain in this fixture, so SESSION_END enqueued by
    the route handler on disconnect never reaches output_queue. Tests that need
    the release task to complete (verifying unit.session is cleared and the
    service unregistered) must drain SESSION_END themselves — see
    `_simulate_session_end_drain` below.
    """
    text_prompt_queue: Queue = Queue()
    should_listen = ThreadingEvent()
    should_listen.set()
    service = RealtimeService(
        text_prompt_queue=text_prompt_queue,
        should_listen=should_listen,
    )
    input_queue: Queue = Queue()
    output_queue: Queue = Queue()
    text_output_queue: Queue = Queue()
    stop_event = ThreadingEvent()
    response_playing = ThreadingEvent()
    cancel_scope = CancelScope()
    unit = PipelineUnit(
        index=0,
        service=service,
        cancel_scope=cancel_scope,
        should_listen=should_listen,
        response_playing=response_playing,
        input_queue=input_queue,
        output_queue=output_queue,
        text_output_queue=text_output_queue,
        text_prompt_queue=text_prompt_queue,
        handlers=[],
    )
    app = create_app(pool=[unit], stop_event=stop_event)
    return (
        app,
        service,
        input_queue,
        output_queue,
        text_output_queue,
        should_listen,
        stop_event,
        response_playing,
        cancel_scope,
    )


def _simulate_session_end_drain(input_queue: Queue, output_queue: Queue, timeout: float = 1.0) -> None:
    """Wait for SESSION_END to land in input_queue (from the route handler's
    release path) and forward it to output_queue — simulating the handler chain.
    The send loop will then observe SESSION_END and set `session.drained`,
    letting the release task complete (unregister + clear `unit.session`).
    """
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            item = input_queue.get(timeout=0.05)
        except Empty:
            continue
        if isinstance(item, PipelineControlMessage) and is_control_message(item, SESSION_END.kind):
            output_queue.put(item)
            return
    raise AssertionError("SESSION_END did not appear on input_queue within timeout")


def _pcm_bytes(n_samples: int) -> bytes:
    return b"\x00" * (n_samples * 2)


class _FakeWebSocket:
    application_state = WebSocketState.CONNECTED

    def __init__(self):
        self.sent: list[dict] = []

    async def send_json(self, payload: dict) -> None:
        self.sent.append(payload)


# ===================================================================
# Connection
# ===================================================================


class TestConnection:
    def test_connect_receives_session_created(self, setup):
        app, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                msg = ws.receive_json()
                assert msg["type"] == "session.created"
                assert msg["event_id"].startswith("event_")
                assert "session" in msg

    def test_second_connection_rejected(self, setup):
        app, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws1:
                ws1.receive_json()  # session.created
                with client.websocket_connect("/v1/realtime") as ws2:
                    msg = ws2.receive_json()
                    assert msg["type"] == "error"
                    # Rejection uses the stateless build_error_event helper —
                    # the error type identifies pool exhaustion specifically.
                    assert msg["error"]["type"] == "session_limit_reached"


# ===================================================================
# Client event dispatch
# ===================================================================


class TestClientEventDispatch:
    def test_audio_append_forwarded_to_input_queue(self, setup):
        app, _, input_queue, *_ = setup
        audio_b64 = base64.b64encode(_pcm_bytes(512)).decode("ascii")
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                ws.send_json(
                    {
                        "type": "input_audio_buffer.append",
                        "audio": audio_b64,
                    }
                )
                time.sleep(0.1)
                item = input_queue.get(timeout=1)
                assert isinstance(item, tuple) and len(item) == 2
                chunk, rt_cfg = item
                assert isinstance(chunk, bytes)
                assert len(chunk) == CHUNK_SIZE_BYTES

    def test_session_update_applied(self, setup):
        app, service, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                ws.send_json(
                    {
                        "type": "session.update",
                        "session": {
                            "type": "realtime",
                            "audio": {"output": {"voice": "coral"}},
                        },
                    }
                )
                time.sleep(0.1)
                cid = service.connection_ids[0]
                assert service._state(cid).runtime_config.session.audio.output.voice == "coral"

    def test_conversation_item_create_returns_events(self, setup):
        app, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                ws.send_json(
                    {
                        "type": "conversation.item.create",
                        "item": {
                            "type": "message",
                            "role": "user",
                            "content": [{"type": "input_text", "text": "ping"}],
                        },
                    }
                )
                msg = ws.receive_json()
                assert msg["type"] == "conversation.item.created"
                assert msg["item"]["content"][0]["text"] == "ping"

    def test_response_create_error_when_active(self, setup):
        app, service, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                conn_id = list(service._conns.keys())[0]
                service.response._ensure_response(conn_id)
                ws.send_json({"type": "response.create"})
                msg = ws.receive_json()
                assert msg["type"] == "error"
                assert "another response is in progress" in msg["error"]["message"].lower()

    def test_response_cancel_returns_events(self, setup):
        app, service, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                conn_id = list(service._conns.keys())[0]
                service.response._ensure_response(conn_id)
                ws.send_json({"type": "response.cancel"})
                msg1 = ws.receive_json()
                msg2 = ws.receive_json()
                types = {msg1["type"], msg2["type"]}
                assert "response.output_audio.done" in types
                assert "response.done" in types

    def test_response_cancel_flushes_queues(self, setup):
        app, service, _, output_queue, text_output_queue, _, _, response_playing, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                conn_id = list(service._conns.keys())[0]
                service.response._ensure_response(conn_id)
                response_playing.set()
                output_queue.put(_pcm_bytes(256))
                output_queue.put(_pcm_bytes(256))
                text_output_queue.put(AssistantTextEvent(text="stale"))
                ws.send_json({"type": "response.cancel"})
                ws.receive_json()  # response.output_audio.done
                ws.receive_json()  # response.done
                time.sleep(0.1)
                assert output_queue.empty()
                assert text_output_queue.empty()
                assert not response_playing.is_set()
                assert cancel_scope.discarding

    def test_response_cancel_spurious_does_not_set_discarding(self, setup):
        """response.cancel when no response is active must NOT enable discarding,
        otherwise it would stick True forever (no __RESPONSE_DONE__ to clear it)."""
        app, service, _, _, _, _, _, _, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                assert not service._state(list(service._conns.keys())[0]).in_response
                ws.send_json({"type": "response.cancel"})
                time.sleep(0.1)
                assert not cancel_scope.discarding

    def test_response_cancel_late_audio_is_discarded(self, setup):
        """Audio arriving after response.cancel is silently dropped (discard guard)."""
        app, service, _, output_queue, _, _, _, response_playing, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                conn_id = list(service._conns.keys())[0]
                service.response._ensure_response(conn_id)
                response_playing.set()
                ws.send_json({"type": "response.cancel"})
                ws.receive_json()  # response.output_audio.done
                ws.receive_json()  # response.done
                time.sleep(0.1)
                assert cancel_scope.discarding
                output_queue.put(_pcm_bytes(256))
                time.sleep(0.15)
                # No response.created or audio delta should appear; only
                # __RESPONSE_DONE__ will eventually clear the guard.
                output_queue.put(AUDIO_RESPONSE_DONE)
                time.sleep(0.15)
                assert not cancel_scope.discarding

    def test_unknown_event_returns_error(self, setup):
        app, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                ws.send_json({"type": "bogus.event"})
                msg = ws.receive_json()
                assert msg["type"] == "error"


# ===================================================================
# Send loop (pipeline -> client)
# ===================================================================


class TestSendLoop:
    def test_audio_output_ignores_session_end_control_message(self, setup):
        app, _, _, output_queue, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                output_queue.put(SESSION_END)
                output_queue.put(_pcm_bytes(256))

                msg1 = ws.receive_json()
                assert msg1["type"] == "response.created"
                msg2 = ws.receive_json()
                assert msg2["type"] == "response.output_audio.delta"

    def test_audio_output_sends_response_created_and_delta(self, setup):
        app, _, _, output_queue, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                output_queue.put(_pcm_bytes(256))
                msg1 = ws.receive_json()
                assert msg1["type"] == "response.created"
                assert msg1["response"]["status"] == "in_progress"
                msg2 = ws.receive_json()
                assert msg2["type"] == "response.output_audio.delta"
                assert "delta" in msg2

    def test_audio_output_batches_immediately_available_chunks(self, setup):
        app, _, _, output_queue, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                output_queue.put(_pcm_bytes(256))
                output_queue.put(_pcm_bytes(256))
                output_queue.put(PIPELINE_END)

                msg1 = ws.receive_json()
                assert msg1["type"] == "response.created"

                msg2 = ws.receive_json()
                assert msg2["type"] == "response.output_audio.delta"
                decoded = base64.b64decode(msg2["delta"])
                assert len(decoded) == len(_pcm_bytes(512))

                msg3 = ws.receive_json()
                msg4 = ws.receive_json()
                types = {msg3["type"], msg4["type"]}
                assert "response.output_audio.done" in types
                assert "response.done" in types

    def test_end_marker_sends_finish_events(self, setup):
        app, _, _, output_queue, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                output_queue.put(_pcm_bytes(256))
                ws.receive_json()  # response.created
                ws.receive_json()  # audio delta
                output_queue.put(PIPELINE_END)
                msg1 = ws.receive_json()
                msg2 = ws.receive_json()
                types = {msg1["type"], msg2["type"]}
                assert "response.output_audio.done" in types
                assert "response.done" in types

    def test_text_output_sends_pipeline_events(self, setup):
        app, _, _, _, text_output_queue, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                text_output_queue.put(SpeechStartedEvent())
                msg = ws.receive_json()
                assert msg["type"] == "input_audio_buffer.speech_started"
                assert msg["audio_start_ms"] == 0

    def test_barge_in_discard_clears_after_response_done(self, setup):
        """After barge-in sets discarding=True, __RESPONSE_DONE__ must clear it back to False."""
        app, service, _, output_queue, text_output_queue, _, _, response_playing, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                conn_id = list(service._conns.keys())[0]
                service.response._ensure_response(conn_id)
                response_playing.set()
                # Trigger barge-in
                text_output_queue.put(SpeechStartedEvent())
                ws.receive_json()  # input_audio_buffer.speech_started
                ws.receive_json()  # response.output_audio.done
                ws.receive_json()  # response.done
                time.sleep(0.1)
                assert cancel_scope.discarding
                output_queue.put(AUDIO_RESPONSE_DONE)
                time.sleep(0.15)
                assert not cancel_scope.discarding

    def test_speech_started_cancels_pending_implicit_response(self, setup):
        app, service, _, output_queue, text_output_queue, _, _, response_playing, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                conn_id = list(service._conns.keys())[0]
                stale_generation = cancel_scope.generation
                service._state(conn_id).response_pending = True

                text_output_queue.put(SpeechStartedEvent())
                msg = ws.receive_json()

                assert msg["type"] == "input_audio_buffer.speech_started"
                time.sleep(0.15)
                assert cancel_scope.discarding
                assert cancel_scope.generation == stale_generation + 1
                assert service._state(conn_id).response_pending is False
                assert service._state(conn_id).in_response is False
                assert not response_playing.is_set()

                output_queue.put(AudioOutput(audio=AUDIO_RESPONSE_DONE, cancel_generation=stale_generation))
                time.sleep(0.15)
                assert not cancel_scope.discarding

    def test_speech_started_does_not_cancel_pending_when_internal_non_interrupt(self, setup):
        app, service, _, _, text_output_queue, _, _, _, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                conn_id = list(service._conns.keys())[0]
                service._state(conn_id).response_pending = True

                text_output_queue.put(SpeechStartedEvent(interrupt_response=False))
                msg = ws.receive_json()

                assert msg["type"] == "input_audio_buffer.speech_started"
                time.sleep(0.15)
                assert not cancel_scope.discarding
                assert service._state(conn_id).response_pending is True

    def test_stale_tagged_audio_is_dropped_after_interruption(self, setup):
        app, _, _, output_queue, _, _, _, _, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                stale_generation = cancel_scope.generation
                cancel_scope.cancel()
                current_generation = cancel_scope.generation
                output_queue.put(AudioOutput(audio=_pcm_bytes(64), cancel_generation=stale_generation))
                output_queue.put(AudioOutput(audio=_pcm_bytes(512), cancel_generation=current_generation))

                assert ws.receive_json()["type"] == "response.created"
                delta = ws.receive_json()

                assert delta["type"] == "response.output_audio.delta"
                assert len(base64.b64decode(delta["delta"])) == len(_pcm_bytes(512))

    def test_current_generation_text_survives_stuck_discarding(self, setup):
        """Regression: a fresh response's transcript must survive a stuck discard guard.

        A superseded speculative turn can leave ``cancel_scope.discarding`` stuck True
        (its TTS dropped the stale ``EndOfResponse`` without emitting AUDIO_RESPONSE_DONE,
        so ``response_done()`` never cleared the flag). The next response's audio is tagged
        with the current generation and streams fine, but the assistant text used to be
        blanket-dropped while discarding — leaving audio + ``response.done`` with no
        ``response.output_audio_transcript.done``. The text is now discarded by the same
        generation-aware rule as audio, so a current-generation transcript is kept.
        """
        app, _, _, output_queue, text_output_queue, _, _, _, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                cancel_scope.cancel()  # discarding=True, generation bumped; sentinel never arrived
                current_generation = cancel_scope.generation
                assert cancel_scope.discarding

                text_output_queue.put(AssistantTextEvent(text="hello there", cancel_generation=current_generation))
                output_queue.put(AudioOutput(audio=_pcm_bytes(256), cancel_generation=current_generation))
                output_queue.put(AudioOutput(audio=AUDIO_RESPONSE_DONE, cancel_generation=current_generation))

                types: list[str] = []
                transcript = None
                for _ in range(8):
                    msg = ws.receive_json()
                    types.append(msg["type"])
                    if msg["type"] == "response.output_audio_transcript.done":
                        transcript = msg["transcript"]
                    if msg["type"] == "response.done":
                        break
                assert "response.output_audio_transcript.done" in types
                assert transcript == "hello there"

    def test_stale_tagged_response_done_does_not_finish_current_response(self, setup):
        app, service, _, output_queue, _, _, _, _, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                conn_id = list(service._conns.keys())[0]
                stale_generation = cancel_scope.generation
                service.response._ensure_response(conn_id)
                service.finish_response(conn_id, status="cancelled")
                cancel_scope.cancel()
                current_response_id, _ = service.response._ensure_response(conn_id)

                output_queue.put(AudioOutput(audio=AUDIO_RESPONSE_DONE, cancel_generation=stale_generation))
                time.sleep(0.15)

                state = service._state(conn_id)
                assert state.in_response
                assert state.current_response_id == current_response_id

    def test_response_done_drains_pending_token_usage_before_finish(self, setup):
        app, service, _, output_queue, text_output_queue, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                conn_id = list(service._conns.keys())[0]

                text_output_queue.put(
                    AssistantTextEvent(
                        text="",
                        tools=[{"type": "function_call", "call_id": "c1", "name": "f1", "arguments": "{}"}],
                    )
                )
                text_output_queue.put(TokenUsageEvent(input_tokens=10, output_tokens=5))
                output_queue.put(AUDIO_RESPONSE_DONE)

                assert ws.receive_json()["type"] == "response.function_call_arguments.done"
                msg1 = ws.receive_json()
                msg2 = ws.receive_json()
                assert {msg1["type"], msg2["type"]} == {"response.output_audio.done", "response.done"}

                assert service.total_usage.input_tokens == 10
                assert service.total_usage.output_tokens == 5
                assert service._state(conn_id).response_usage.input_tokens == 0
                assert service._state(conn_id).response_usage.output_tokens == 0

    def test_response_completion_drain_sends_pending_tool_before_done(self, setup):
        _, service, input_queue, output_queue, text_output_queue, should_listen, _, response_playing, cancel_scope = (
            setup
        )
        unit = PipelineUnit(
            index=0,
            service=service,
            cancel_scope=cancel_scope,
            should_listen=should_listen,
            response_playing=response_playing,
            input_queue=input_queue,
            output_queue=output_queue,
            text_output_queue=text_output_queue,
            text_prompt_queue=Queue(),
            handlers=[],
        )
        conn_id = service.register()
        response_id, _ = service.response._ensure_response(conn_id)
        text_output_queue.put(
            AssistantTextEvent(
                text="",
                tools=[
                    {
                        "type": "function_call",
                        "call_id": "c1",
                        "name": "play_emotion",
                        "arguments": '{"emotion":"loving"}',
                    }
                ],
            )
        )
        text_output_queue.put(TokenUsageEvent(input_tokens=10, output_tokens=5))
        ws = _FakeWebSocket()

        asyncio.run(router_module._drain_pending_response_events(ws, unit, conn_id))
        done_events = service.finish_response(conn_id)

        assert [payload["type"] for payload in ws.sent] == ["response.function_call_arguments.done"]
        assert [event.type for event in done_events] == ["response.output_audio.done", "response.done"]
        assert ws.sent[0]["response_id"] == response_id
        assert done_events[1].response.id == response_id
        assert done_events[1].response.usage.input_tokens == 10
        assert done_events[1].response.usage.output_tokens == 5
        assert text_output_queue.empty()

    def test_response_completion_drain_preserves_usage_across_non_response_boundary(self, setup):
        _, service, input_queue, output_queue, text_output_queue, should_listen, _, response_playing, cancel_scope = (
            setup
        )
        unit = PipelineUnit(
            index=0,
            service=service,
            cancel_scope=cancel_scope,
            should_listen=should_listen,
            response_playing=response_playing,
            input_queue=input_queue,
            output_queue=output_queue,
            text_output_queue=text_output_queue,
            text_prompt_queue=Queue(),
            handlers=[],
        )
        conn_id = service.register()
        response_id, _ = service.response._ensure_response(conn_id)
        text_output_queue.put(
            AssistantTextEvent(
                text="",
                tools=[{"type": "function_call", "call_id": "c1", "name": "play_emotion", "arguments": "{}"}],
            )
        )
        text_output_queue.put(SpeechStartedEvent())
        text_output_queue.put(TokenUsageEvent(input_tokens=10, output_tokens=5))
        text_output_queue.put(AssistantTextEvent(text="queued after boundary"))
        ws = _FakeWebSocket()

        asyncio.run(router_module._drain_pending_response_events(ws, unit, conn_id))
        done_events = service.finish_response(conn_id)

        assert [payload["type"] for payload in ws.sent] == ["response.function_call_arguments.done"]
        assert ws.sent[0]["response_id"] == response_id
        assert done_events[1].response.usage.input_tokens == 10
        assert done_events[1].response.usage.output_tokens == 5

        boundary = text_output_queue.get_nowait()
        queued_assistant = text_output_queue.get_nowait()
        assert isinstance(boundary, SpeechStartedEvent)
        assert isinstance(queued_assistant, AssistantTextEvent)
        assert queued_assistant.text == "queued after boundary"
        assert text_output_queue.empty()

    def test_speech_started_does_not_cancel_when_interrupt_disabled(self, setup):
        """With interrupt_response=False, speech during playback should NOT cancel or flush."""
        from openai.types.realtime.realtime_audio_input_turn_detection import ServerVad

        app, service, _, output_queue, text_output_queue, _, _, response_playing, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()  # session.created
                conn_id = list(service._conns.keys())[0]
                service._state(conn_id).runtime_config.session.audio.input.turn_detection = ServerVad(
                    type="server_vad",
                    interrupt_response=False,
                )
                _, response_item_id = service.response._ensure_response(conn_id)
                response_playing.set()
                text_output_queue.put(SpeechStartedEvent())
                msg = ws.receive_json()
                assert msg["type"] == "input_audio_buffer.speech_started"
                time.sleep(0.15)
                assert response_playing.is_set(), "response_playing should remain set"
                assert not cancel_scope.discarding, "cancel_scope should not be discarding"
                assert service._state(conn_id).in_response, "response should still be active"
                assert service._state(conn_id).current_item_id == response_item_id


# ===================================================================
# Cleanup
# ===================================================================


class TestCleanup:
    def test_new_connection_resets_discard_after_invalidating_generation(self, setup):
        """connect-time _clean_unit cancels+resets: stale work is invalidated, discarding cleared."""
        app, _, *_rest, cancel_scope = setup
        cancel_scope.cancel()
        assert cancel_scope.discarding
        assert cancel_scope.generation == 1
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                assert not cancel_scope.discarding
                assert cancel_scope.generation == 2

    def test_disconnect_bumps_cancel_scope_generation(self, setup):
        """_clean_unit() on disconnect calls cancel() so in-flight generations go stale."""
        app, _, _, _, _, _, _, _, cancel_scope = setup
        assert cancel_scope.generation == 0
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                assert cancel_scope.generation == 1
            # disconnect triggers _clean_unit again + drain (short timeout in tests)
            time.sleep(0.3)
        assert cancel_scope.generation == 2

    def test_disconnect_unregisters(self, setup):
        app, service, input_queue, output_queue, *_ = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                assert len(service._conns) == 1
            # Simulate the handler chain consuming SESSION_END so the release
            # task can complete and unregister the session.
            _simulate_session_end_drain(input_queue, output_queue)
            time.sleep(0.3)
            assert len(service._conns) == 0

    def test_last_disconnect_cancels_and_clears_response_state(self, setup):
        app, service, input_queue, output_queue, text_output_queue, _, _, response_playing, cancel_scope = setup
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws:
                ws.receive_json()
                conn_id = list(service._conns.keys())[0]
                service.response._ensure_response(conn_id)
                response_playing.set()
                output_queue.put(_pcm_bytes(256))
                text_output_queue.put(AssistantTextEvent(text="stale"))
            _simulate_session_end_drain(input_queue, output_queue)
            time.sleep(0.3)

        assert not cancel_scope.discarding
        assert cancel_scope.generation == 2
        assert not response_playing.is_set()
        assert text_output_queue.empty()


# ===================================================================
# Pool semantics (new in pool refactor)
# ===================================================================


def _make_unit(index: int) -> PipelineUnit:
    text_prompt_queue: Queue = Queue()
    should_listen = ThreadingEvent()
    should_listen.set()
    return PipelineUnit(
        index=index,
        service=RealtimeService(text_prompt_queue=text_prompt_queue, should_listen=should_listen),
        cancel_scope=CancelScope(),
        should_listen=should_listen,
        response_playing=ThreadingEvent(),
        input_queue=Queue(),
        output_queue=Queue(),
        text_output_queue=Queue(),
        text_prompt_queue=text_prompt_queue,
        handlers=[],
    )


class TestPool:
    def test_pool_endpoint_reports_idle_state(self):
        pool = [_make_unit(0), _make_unit(1)]
        app = create_app(pool=pool, stop_event=ThreadingEvent())
        with TestClient(app) as client:
            r = client.get("/v1/pool")
            assert r.status_code == 200
            data = r.json()
            assert data["size"] == 2
            assert data["in_use"] == 0
            assert [u["session_id"] for u in data["units"]] == [None, None]

    def test_two_clients_claim_two_slots_third_rejected(self):
        pool = [_make_unit(0), _make_unit(1)]
        app = create_app(pool=pool, stop_event=ThreadingEvent())
        with TestClient(app) as client:
            with client.websocket_connect("/v1/realtime") as ws1:
                ws1.receive_json()  # session.created
                with client.websocket_connect("/v1/realtime") as ws2:
                    ws2.receive_json()  # session.created (different unit)
                    with client.websocket_connect("/v1/realtime") as ws3:
                        msg = ws3.receive_json()
                        assert msg["type"] == "error"
                        assert msg["error"]["type"] == "session_limit_reached"
                    # Pool now reports 2 in_use
                    r = client.get("/v1/pool")
                    assert r.json()["in_use"] == 2

    def test_usage_aggregates_errors_by_type_across_units(self):
        pool = [_make_unit(0), _make_unit(1)]
        pool[0].service.total_usage.record_error("foo")
        pool[0].service.total_usage.record_error("foo")
        pool[1].service.total_usage.record_error("bar")
        app = create_app(pool=pool, stop_event=ThreadingEvent())
        with TestClient(app) as client:
            data = client.get("/v1/usage").json()
            assert data["errors_by_type"] == {"foo": 2, "bar": 1}
            assert data["total_errors"] == 3
