"""End-to-end: microphone in, speech out, over real HTTP.

Every other test in this suite injects a fake at the seam it cares about, and
every one of them passed all week while these got through to production:

  - notifications sitting unspoken for minutes (a clock stamped in the wrong
    order, two correct units)
  - the pet saying the same thing twice (server discarded prose, model repeated
    it — both sides behaving as written)
  - `[laughing]` read out loud (a tag that means something to one model and
    nothing to the next one down the pipe)
  - a device command written as prose (extractor fine, prompt fine, no marker)

They were all *interaction* bugs. So this one runs the actual pipeline against
a real socket: a threaded HTTP server that speaks the desk protocol, the real
`server_client` doing real requests (including NDJSON streaming), the real
controller loop and state machine. The only fakes are where the hardware is —
the mic stream and the speakers — because those are the two things a test
genuinely cannot have.
"""

import json
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import numpy as np
import pytest
from PySide6.QtWidgets import QApplication

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from bolt_pet import controller as controller_mod
from bolt_pet.notifications import Notification
from bolt_pet.state import PetState

_app = QApplication.instance() or QApplication(["test"])


# ── a desk server that actually listens on a port ───────────────────────────

class FakeDesk:
    """Scripted responses, real HTTP. Set `.script` per test."""

    def __init__(self):
        self.script = {}
        self.requests = []
        self._server = ThreadingHTTPServer(("127.0.0.1", 0), self._handler())
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
        self._thread.start()

    @property
    def url(self) -> str:
        host, port = self._server.server_address[:2]
        return f"http://{host}:{port}"

    def stop(self) -> None:
        self._server.shutdown()
        self._server.server_close()

    def _handler(self):
        desk = self

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, *_args):
                pass  # the test output is not an access log

            def do_GET(self):
                path = self.path.split("?")[0]
                desk.requests.append(("GET", path))
                self._json(desk.script.get(path, {"files": []}))

            def do_POST(self):
                length = int(self.headers.get("Content-Length") or 0)
                body = json.loads(self.rfile.read(length) or b"{}")
                path = self.path.split("?")[0]
                desk.requests.append(("POST", path, body))
                response = desk.script.get(path)
                if callable(response):
                    response = response(body)
                if path.endswith("converse_stream"):
                    self._ndjson(response or [])
                else:
                    self._json(response if response is not None else {"type": "reply", "text": "ok"})

            def _json(self, payload):
                data = json.dumps(payload).encode()
                self.send_response(200)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(data)))
                self.end_headers()
                self.wfile.write(data)

            def _ndjson(self, events):
                self.send_response(200)
                self.send_header("Content-Type", "application/x-ndjson")
                self.end_headers()
                for event in events:
                    self.wfile.write((json.dumps(event) + "\n").encode())
                    self.wfile.flush()

        return Handler


class FakeMic:
    """Loud frames then quiet ones, so the VAD ends the utterance on its own."""

    def __init__(self, loud=8, quiet=60):
        self.frames = ([np.full((320, 1), 4000, dtype=np.int16)] * loud
                       + [np.zeros((320, 1), dtype=np.int16)] * quiet)

    def read(self, _n):
        return (self.frames.pop(0) if self.frames
                else np.zeros((320, 1), dtype=np.int16)), None

    def __enter__(self):
        return self

    def __exit__(self, *_a):
        return False


@pytest.fixture
def pipeline(monkeypatch):
    """A controller wired to a real local server, with fake ears and mouth."""
    desk = FakeDesk()
    spoken: list[str] = []
    levels: list[float] = []

    monkeypatch.setattr(controller_mod.config, "SERVER_URL", desk.url)
    monkeypatch.setattr(controller_mod.config, "API_KEY", "test-key")
    monkeypatch.setattr(controller_mod.config, "SESSION_ID", "pet-smoke")
    monkeypatch.setattr(controller_mod.config, "SILENCE_END_SEC", 0.2)
    monkeypatch.setattr(controller_mod.config, "MIN_UTTERANCE_S", 0.0)
    monkeypatch.setattr(controller_mod.config, "RECEIVE_FILES", False)
    # Off by default so each test picks its own path; the streaming tests opt in.
    monkeypatch.setattr(controller_mod.config, "STREAMING_REPLIES", False)
    monkeypatch.setattr(controller_mod.screen_context, "context_for", lambda text: text)
    monkeypatch.setattr(controller_mod.screen_context, "is_fullscreen_active", lambda: False)
    # Deepgram and the speakers are the two things a test can't have.
    monkeypatch.setattr(controller_mod.stt, "transcribe", lambda pcm: "what's on my disk?")
    monkeypatch.setattr(controller_mod.stt_stream.StreamingTranscriber, "open",
                        classmethod(lambda cls, **kw: None))

    def fake_speak(text, on_error=None, should_stop=None, voice_id=None, on_level=None):
        spoken.append(text)
        if on_level is not None:
            on_level(0.8)   # the mouth opens while a word plays...
            levels.append(0.8)
        return True

    monkeypatch.setattr(controller_mod.tts, "speak", fake_speak)

    ctrl = controller_mod.PetController()
    ctrl._stream = FakeMic()
    try:
        yield ctrl, desk, spoken, levels
    finally:
        desk.stop()


# ── the whole path ──────────────────────────────────────────────────────────

def test_a_plain_turn_goes_mic_to_speaker(pipeline):
    ctrl, desk, spoken, _levels = pipeline
    desk.script["/desk/converse"] = {"type": "reply", "text": "About sixty percent full."}

    ctrl._handle_conversation_turn()

    assert spoken == ["About sixty percent full."]
    assert ctrl._state.state == PetState.IDLE
    posted = [r for r in desk.requests if r[0] == "POST"]
    assert posted[0][1] == "/desk/converse"
    assert "what's on my disk?" in posted[0][2]["text"]
    assert ctrl.history.entries()[-1].text == "About sixty percent full."


def test_a_streamed_turn_speaks_each_sentence_as_it_lands(pipeline, monkeypatch):
    """The NDJSON is parsed by the real client over a real socket — the layer
    that a mocked `converse` can never exercise."""
    ctrl, desk, spoken, _levels = pipeline
    monkeypatch.setattr(controller_mod.config, "STREAMING_REPLIES", True)
    desk.script["/desk/converse_stream"] = [
        {"type": "say", "text": "The disk is fine."},
        {"type": "say", "text": "About sixty percent used."},
        {"type": "reply", "text": "The disk is fine. About sixty percent used.",
         "already_spoken": True},
    ]

    ctrl._handle_conversation_turn()

    assert spoken == ["The disk is fine.", "About sixty percent used."]
    # The final reply must not be spoken a third time.
    assert len(spoken) == 2


def test_a_tool_turn_says_give_me_a_sec_then_the_answer(pipeline, monkeypatch):
    """Holding line, relayed command, and the real answer — in that order."""
    ctrl, desk, spoken, _levels = pipeline
    ran = []
    monkeypatch.setattr(controller_mod.server_client, "run_local_command",
                        lambda cmd: ran.append(cmd) or "[exit 0]\n60% used")
    desk.script["/desk/converse"] = {
        "type": "command", "command": "df -h /", "token": "tok",
        "say": "Let me check that for you.",
    }
    desk.script["/desk/tool_result"] = {"type": "reply", "text": "Sixty percent used."}

    ctrl._handle_conversation_turn()

    assert spoken == ["Let me check that for you.", "Sixty percent used."]
    assert ran == ["df -h /"]
    relayed = next(r for r in desk.requests if r[0] == "POST" and r[1] == "/desk/tool_result")
    assert relayed[2]["output"].endswith("60% used")


def test_a_device_command_never_reaches_the_shell(pipeline, monkeypatch):
    ctrl, desk, spoken, _levels = pipeline
    ran = []
    monkeypatch.setattr(controller_mod.server_client, "run_local_command", lambda cmd: ran.append(cmd))
    actions = []
    ctrl.action.connect(actions.append)
    desk.script["/desk/converse"] = {
        "type": "command", "command": "petctl emote wave", "token": "tok"}
    desk.script["/desk/tool_result"] = {"type": "reply", "text": "There you go."}

    ctrl._handle_conversation_turn()

    assert ran == []
    assert actions == [{"action": "emote", "emote": "wave"}]
    assert spoken == ["There you go."]


def test_the_mouth_moves_with_the_audio(pipeline):
    """Lip-sync is only real if the level actually reaches the window."""
    ctrl, desk, _spoken, levels = pipeline
    mouth = []
    ctrl.mouth.connect(mouth.append)
    desk.script["/desk/converse"] = {"type": "reply", "text": "Talking now."}

    ctrl._handle_conversation_turn()

    assert levels, "TTS was never given a level callback"
    assert 0.8 in mouth              # opened while speaking...
    assert mouth[-1] == 0.0          # ...and closed at the end


def test_a_notification_is_forwarded_and_spoken(pipeline):
    """The path that was silently sitting for five to ten minutes."""
    ctrl, desk, spoken, _levels = pipeline
    ctrl._notification_gate = controller_mod.notifications.NotificationGate("", 0)
    desk.script["/desk/converse"] = {"type": "reply", "text": "Harry says he's around."}

    ctrl._queue_notification(Notification(app="Signal", summary="Harry", body="you there?"))
    ctrl._maybe_heartbeat()

    assert spoken == ["Harry says he's around."]
    forwarded = next(r for r in desk.requests if r[0] == "POST" and r[1] == "/desk/converse")
    assert "Harry" in forwarded[2]["text"]


def test_streaming_falls_back_when_the_server_is_older(pipeline, monkeypatch):
    """A server without /desk/converse_stream (or one that answers with nothing)
    must not cost a turn — the client drops to the plain endpoint. This is how
    the pet keeps working against a container that hasn't been updated yet."""
    ctrl, desk, spoken, _levels = pipeline
    monkeypatch.setattr(controller_mod.config, "STREAMING_REPLIES", True)
    desk.script["/desk/converse_stream"] = []          # nothing streamed back
    desk.script["/desk/converse"] = {"type": "reply", "text": "Still here."}

    ctrl._handle_conversation_turn()

    assert spoken == ["Still here."]
    paths = [r[1] for r in desk.requests if r[0] == "POST"]
    assert paths == ["/desk/converse_stream", "/desk/converse"]


def test_a_dead_server_leaves_the_pet_usable(pipeline):
    """It should flash an error and go back to listening, not wedge."""
    ctrl, desk, spoken, _levels = pipeline
    desk.stop()   # the server disappears mid-session
    states = []
    ctrl.state_changed.connect(states.append)

    ctrl._handle_conversation_turn()

    assert states[-2:] == ["error", "idle"]
    assert spoken == []