"""Tests for the headless console stream."""

import asyncio
import logging
import os
import threading
from types import SimpleNamespace
from typing import Any
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock

import numpy as np
import pytest
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient

from peachy.config import HF_AVAILABLE_VOICES, config
from peachy.console import LocalStream
from peachy.openai_realtime import OPENAI_VOICES, RetryExhaustedError
from peachy.providers import (
    ProviderDefinition,
    ProviderReadiness,
    get_provider_definition,
)
from peachy.startup_settings import (
    StartupSettings,
    load_startup_settings_into_runtime,
)
from peachy.personality_routes import mount_personality_routes


async def _wait_until(predicate: Any, timeout: float = 1.0) -> None:
    """Wait until a test predicate becomes true."""
    deadline = asyncio.get_running_loop().time() + timeout
    while asyncio.get_running_loop().time() < deadline:
        if predicate():
            return
        await asyncio.sleep(0.01)
    raise AssertionError("Timed out waiting for condition")


def test_clear_audio_queue_prefers_clear_player() -> None:
    """clear_player() is the canonical flush and is used whenever available."""
    handler = MagicMock()
    handler.output_queue = asyncio.Queue()
    handler.output_queue.put_nowait((24000, np.zeros(4, dtype=np.int16)))
    audio = SimpleNamespace(
        clear_player=MagicMock(),
        clear_output_buffer=MagicMock(),
    )
    robot = SimpleNamespace(media=SimpleNamespace(audio=audio))
    stream = LocalStream(handler, robot)

    stream.clear_audio_queue()

    audio.clear_player.assert_called_once()
    audio.clear_output_buffer.assert_not_called()
    assert handler.output_queue.empty()


def test_clear_audio_queue_falls_back_to_output_buffer() -> None:
    """Older SDKs without clear_player() still flush via clear_output_buffer()."""
    handler = MagicMock()
    handler.output_queue = asyncio.Queue()
    audio = SimpleNamespace(clear_output_buffer=MagicMock())  # no clear_player
    robot = SimpleNamespace(media=SimpleNamespace(audio=audio))
    stream = LocalStream(handler, robot)

    stream.clear_audio_queue()

    audio.clear_output_buffer.assert_called_once()
    assert handler.output_queue.empty()


def test_clear_audio_queue_drains_queue_in_place() -> None:
    """The output queue is drained in place, not replaced with a new object."""
    handler = MagicMock()
    queue: asyncio.Queue[Any] = asyncio.Queue()
    queue.put_nowait((24000, np.zeros(4, dtype=np.int16)))
    queue.put_nowait((24000, np.zeros(4, dtype=np.int16)))
    handler.output_queue = queue
    audio = SimpleNamespace(clear_player=MagicMock())
    robot = SimpleNamespace(media=SimpleNamespace(audio=audio))
    stream = LocalStream(handler, robot)

    stream.clear_audio_queue()

    assert handler.output_queue is queue  # same object, not replaced
    assert queue.empty()


def test_mic_endpoints_report_and_toggle_mute_state() -> None:
    """The mic starts live; the settings API exposes and flips the pause state."""
    app = FastAPI()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(MagicMock(), robot, settings_app=app)
    stream._init_settings_ui_if_needed()
    client = TestClient(app)

    assert client.get("/api/v1/mic").json() == {"muted": False}

    assert client.post("/api/v1/mic", json={"muted": True}).json() == {"muted": True}
    assert stream._mic_muted is True

    assert client.post("/api/v1/mic", json={"muted": False}).json() == {"muted": False}
    assert stream._mic_muted is False

    # headless streams keep the mic live
    assert LocalStream(MagicMock(), robot)._mic_muted is False


def test_settings_api_uses_versioned_routes_only() -> None:
    """Settings clients should use the versioned API paths."""
    app = FastAPI()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(MagicMock(), robot, settings_app=app)
    stream._init_settings_ui_if_needed()
    client = TestClient(app)

    assert client.get("/api/v1/mic").json() == {"muted": False}

    response = client.post("/api/v1/mic", json={"muted": True})

    assert response.status_code == 200
    assert response.json() == {"muted": True}
    assert stream._mic_muted is True
    assert client.get("/api/v1/status").json()["provider"]
    assert client.get("/api/v1/ready").status_code == 404
    assert client.get("/status").status_code == 404
    assert client.get("/ready").status_code == 404
    assert client.post("/backend_config", json={"backend": "openai"}).status_code == 404
    assert client.get("/mic").status_code == 404
    assert client.post("/mic", json={"muted": False}).status_code == 404
    assert client.get("/conversation_events").status_code == 404


def test_settings_ui_detaches_framework_catch_all_before_api_routes() -> None:
    """Framework fallback routes should not shadow the settings API."""
    app = FastAPI()

    @app.get("/{path:path}")
    def _framework_fallback(path: str) -> None:
        raise HTTPException(status_code=404)

    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(MagicMock(), robot, settings_app=app)
    stream._init_settings_ui_if_needed()
    client = TestClient(app)

    assert client.get("/").status_code == 200
    assert client.get("/static/js/api.js").status_code == 200
    assert client.get("/api/v1/mic").status_code == 200
    assert client.get("/api/v1/personalities").status_code == 200


@pytest.mark.asyncio
async def test_conversation_events_survive_handler_rebuild() -> None:
    """Activity from a rebuilt handler must reach subscribers of the original event bus."""

    class FakeHandler:
        def __init__(self) -> None:
            self.observer = None

        def set_activity_observer(self, observer: Any) -> None:
            self.observer = observer

    rebuilt = FakeHandler()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(FakeHandler(), robot, settings_app=FastAPI(), handler_factory=lambda voice: rebuilt)

    queue, unsubscribe = stream._event_bus.subscribe()
    stream._build_handler_for_current_backend()
    rebuilt.observer("assistant_audio_delta")

    assert await asyncio.wait_for(queue.get(), timeout=1.0) == "assistant_audio_delta"
    unsubscribe()


def test_backend_config_requests_in_process_restart_with_handler_factory(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A rebuild-capable LocalStream should reconnect in process after a connection change."""
    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "deployed")
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", None)
    monkeypatch.delenv("HF_REALTIME_CONNECTION_MODE", raising=False)
    monkeypatch.delenv("HF_REALTIME_WS_URL", raising=False)

    app = FastAPI()
    handler = MagicMock()
    handler.shutdown = AsyncMock()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(
        handler,
        robot,
        settings_app=app,
        instance_path=str(tmp_path),
        handler_factory=lambda _voice: handler,
    )
    stream._init_settings_ui_if_needed()

    response = TestClient(app).post(
        "/api/v1/backend_config",
        json={
            "provider": "huggingface",
            "hf_mode": "local",
            "hf_host": "localhost",
            "hf_port": 8765,
        },
    )

    assert response.status_code == 200
    data = response.json()
    assert data["ok"] is True
    assert data["message"] == "Connection saved. Reconnecting backend."
    assert data["provider"] == "huggingface"
    assert data["requires_restart"] is False
    assert data["can_proceed"] is True
    assert data["backend_connection_state"] == "connecting"
    assert stream._restart_requested.is_set()


def test_backend_config_persists_local_hf_selection_and_status(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Settings API should persist a direct Hugging Face websocket target."""
    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "deployed")
    monkeypatch.setattr(config, "HF_REALTIME_SESSION_URL", None)
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", None)
    monkeypatch.delenv("HF_REALTIME_CONNECTION_MODE", raising=False)
    monkeypatch.delenv("HF_REALTIME_SESSION_URL", raising=False)
    monkeypatch.delenv("HF_REALTIME_WS_URL", raising=False)

    app = FastAPI()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(MagicMock(), robot, settings_app=app, instance_path=str(tmp_path))
    stream._init_settings_ui_if_needed()

    client = TestClient(app)
    response = client.post(
        "/api/v1/backend_config",
        json={
            "provider": "huggingface",
            "hf_mode": "local",
            "hf_host": "localhost",
            "hf_port": 8765,
        },
    )

    assert response.status_code == 200
    data = response.json()
    assert data["ok"] is True
    assert data["provider"] == "huggingface"
    assert data["has_hf_ws_url"] is True
    assert data["has_hf_connection"] is True
    assert data["hf_connection_mode"] == "local"
    assert data["hf_direct_host"] == "localhost"
    assert data["hf_direct_port"] == 8765

    env_text = (tmp_path / ".env").read_text(encoding="utf-8")
    assert "HF_REALTIME_CONNECTION_MODE=local" in env_text
    assert "HF_REALTIME_WS_URL=ws://localhost:8765/v1/realtime" in env_text


def test_backend_config_persists_deployed_mode_without_clearing_local_hf_ws_url(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Saving deployed mode should make env selection explicit and remove stale allocator URLs."""
    env_path = tmp_path / ".env"
    env_path.write_text(
        "HF_REALTIME_SESSION_URL=https://lb.example.test/session\n"
        "HF_REALTIME_WS_URL=ws://localhost:8765/v1/realtime\n",
        encoding="utf-8",
    )

    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "deployed")
    monkeypatch.setattr(config, "HF_REALTIME_SESSION_URL", "https://lb.example.test/session")
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", "ws://localhost:8765/v1/realtime")
    monkeypatch.delenv("HF_REALTIME_CONNECTION_MODE", raising=False)
    monkeypatch.setenv("HF_REALTIME_SESSION_URL", "https://lb.example.test/session")
    monkeypatch.setenv("HF_REALTIME_WS_URL", "ws://localhost:8765/v1/realtime")

    app = FastAPI()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(MagicMock(), robot, settings_app=app, instance_path=str(tmp_path))
    stream._init_settings_ui_if_needed()

    client = TestClient(app)
    response = client.post(
        "/api/v1/backend_config",
        json={
            "provider": "huggingface",
            "hf_mode": "deployed",
        },
    )

    assert response.status_code == 200
    data = response.json()
    assert data["ok"] is True
    assert data["has_hf_session_url"] is True
    assert data["has_hf_ws_url"] is True
    assert data["hf_connection_mode"] == "deployed"

    env_text = env_path.read_text(encoding="utf-8")
    assert "HF_REALTIME_CONNECTION_MODE=deployed" in env_text
    assert "HF_REALTIME_SESSION_URL=" not in env_text
    assert "HF_REALTIME_WS_URL=ws://localhost:8765/v1/realtime" in env_text


def test_deployed_hf_cleanup_failure_is_transactional(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A failed stale-session cleanup must not commit any provider change."""
    original_env_text = (
        "PEACHY_REALTIME_PROVIDER=openai\n"
        "HF_REALTIME_CONNECTION_MODE=local\n"
        "HF_REALTIME_SESSION_URL=https://lb.example.test/session\n"
        "HF_REALTIME_WS_URL=ws://localhost:8765/v1/realtime\n"
    )
    env_path = tmp_path / ".env"
    env_path.write_text(original_env_text, encoding="utf-8")
    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "local")
    monkeypatch.setattr(config, "HF_REALTIME_SESSION_URL", "https://lb.example.test/session")
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", "ws://localhost:8765/v1/realtime")
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "openai")
    monkeypatch.setenv("HF_REALTIME_CONNECTION_MODE", "local")
    monkeypatch.setenv("HF_REALTIME_SESSION_URL", "https://lb.example.test/session")
    monkeypatch.setenv("HF_REALTIME_WS_URL", "ws://localhost:8765/v1/realtime")

    app = FastAPI()
    handler = MagicMock()
    handler.shutdown = AsyncMock()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(
        handler,
        robot,
        settings_app=app,
        instance_path=str(tmp_path),
        handler_factory=lambda _voice: handler,
    )
    stream._init_settings_ui_if_needed()

    real_write_text = Path.write_text

    def fail_cleanup_write(path: Path, text: str, *args: Any, **kwargs: Any) -> int:
        if "HF_REALTIME_SESSION_URL=" not in text:
            raise OSError("simulated cleanup-only disk failure")
        return real_write_text(path, text, *args, **kwargs)

    monkeypatch.setattr(Path, "write_text", fail_cleanup_write)

    response = TestClient(app).post(
        "/api/v1/backend_config",
        json={"provider": "huggingface", "hf_mode": "deployed"},
    )

    assert response.status_code == 500
    assert response.json() == {"ok": False, "error": "persistence_failed"}
    assert os.environ["PEACHY_REALTIME_PROVIDER"] == "openai"
    assert os.environ["HF_REALTIME_CONNECTION_MODE"] == "local"
    assert os.environ["HF_REALTIME_SESSION_URL"] == "https://lb.example.test/session"
    assert os.environ["HF_REALTIME_WS_URL"] == "ws://localhost:8765/v1/realtime"
    assert env_path.read_text(encoding="utf-8") == original_env_text
    assert list(tmp_path.glob(".env.*.tmp")) == []
    assert not stream._restart_requested.is_set()


def test_backend_config_switches_to_saved_local_hf_connection_without_payload_target(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Switching back to a saved local Hugging Face backend should reuse the persisted target."""
    env_path = tmp_path / ".env"
    env_path.write_text(
        "HF_REALTIME_CONNECTION_MODE=local\nHF_REALTIME_WS_URL=ws://192.168.1.42:8766/v1/realtime\n",
        encoding="utf-8",
    )

    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "local")
    monkeypatch.setattr(config, "HF_REALTIME_SESSION_URL", None)
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", "ws://192.168.1.42:8766/v1/realtime")
    monkeypatch.setenv("HF_REALTIME_CONNECTION_MODE", "local")
    monkeypatch.setenv("HF_REALTIME_WS_URL", "ws://192.168.1.42:8766/v1/realtime")

    app = FastAPI()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(MagicMock(), robot, settings_app=app, instance_path=str(tmp_path))
    stream._init_settings_ui_if_needed()

    client = TestClient(app)
    response = client.post(
        "/api/v1/backend_config",
        json={"provider": "huggingface"},
    )

    assert response.status_code == 200
    data = response.json()
    assert data["ok"] is True
    assert data["provider"] == "huggingface"
    assert data["hf_connection_mode"] == "local"
    assert data["hf_direct_host"] == "192.168.1.42"
    assert data["hf_direct_port"] == 8766

    env_text = env_path.read_text(encoding="utf-8")
    assert "HF_REALTIME_CONNECTION_MODE=local" in env_text
    assert "HF_REALTIME_WS_URL=ws://192.168.1.42:8766/v1/realtime" in env_text


def test_backend_config_rejects_invalid_hf_port_zero(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Settings API should reject invalid local Hugging Face ports from direct callers."""
    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "deployed")
    monkeypatch.setattr(config, "HF_REALTIME_SESSION_URL", None)
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", None)

    app = FastAPI()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(MagicMock(), robot, settings_app=app, instance_path=str(tmp_path))
    stream._init_settings_ui_if_needed()

    client = TestClient(app)
    response = client.post(
        "/api/v1/backend_config",
        json={
            "provider": "huggingface",
            "hf_mode": "local",
            "hf_host": "localhost",
            "hf_port": 0,
        },
    )

    assert response.status_code == 400
    assert response.json()["error"] == "invalid_hf_port"


def test_status_reports_direct_hf_ws_url_as_ready(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Settings API should treat a direct Hugging Face websocket as a valid configuration."""
    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "local")
    monkeypatch.setattr(config, "HF_REALTIME_SESSION_URL", None)
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", "ws://127.0.0.1:8765/v1/realtime")
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "huggingface")

    app = FastAPI()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(MagicMock(), robot, settings_app=app, instance_path=str(tmp_path))
    stream._init_settings_ui_if_needed()

    client = TestClient(app)
    response = client.get("/api/v1/status")

    assert response.status_code == 200
    data = response.json()
    assert data["provider"] == "huggingface"
    assert data["has_hf_session_url"] is False
    assert data["has_hf_ws_url"] is True
    assert data["has_hf_connection"] is True
    assert data["hf_connection_mode"] == "local"
    assert data["provider_state"] == "ready"
    assert data["can_proceed"] is True


def test_status_reports_backend_connection_failure(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Settings API should expose backend connection failures without hiding controls."""
    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "local")
    monkeypatch.setattr(config, "HF_REALTIME_SESSION_URL", None)
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", "ws://127.0.0.1:8765/v1/realtime")
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "huggingface")

    app = FastAPI()
    handler = MagicMock()
    handler.connection = None
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(handler, robot, settings_app=app, instance_path=str(tmp_path))
    stream._set_backend_connection_state("disconnected", RuntimeError("connect failed"))
    stream._init_settings_ui_if_needed()

    client = TestClient(app)
    response = client.get("/api/v1/status")

    assert response.status_code == 200
    data = response.json()
    assert data["provider"] == "huggingface"
    assert data["backend_connected"] is False
    assert data["backend_connection_state"] == "disconnected"
    assert data["backend_error"] == "RuntimeError: connect failed"
    assert data["can_proceed"] is True
    assert data["provider_state"] == "ready"
    assert data["can_proceed"] is True


def test_status_redacts_configured_openai_key_from_backend_error(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    secret = "sk-test-status-must-be-redacted"
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "openai")
    monkeypatch.setenv("OPENAI_API_KEY", secret)

    app = FastAPI()
    handler = MagicMock()
    handler.connection = None
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(handler, robot, settings_app=app)
    stream._set_backend_connection_state(
        "disconnected",
        RuntimeError(f"authentication failed for {secret}"),
    )
    stream._init_settings_ui_if_needed()

    response = TestClient(app).get("/api/v1/status")

    assert response.status_code == 200
    assert secret not in response.text
    assert response.json()["backend_error"] == "RuntimeError: authentication failed for [redacted]"


@pytest.mark.asyncio
async def test_startup_warning_redacts_configured_openai_key(
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    secret = "sk-test-log-must-be-redacted"
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "openai")
    monkeypatch.setenv("OPENAI_API_KEY", secret)

    handler = MagicMock()
    handler.connection = None
    handler.shutdown = AsyncMock()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(handler, robot)
    stream._backend_retry_delay = 0

    async def fail_and_stop() -> None:
        stream._stop_event.set()
        raise RuntimeError(f"authentication failed for {secret}")

    handler.start_up = AsyncMock(side_effect=fail_and_stop)
    caplog.set_level(logging.WARNING, logger="peachy.console")

    await stream._run_handler_startup_loop()

    assert secret not in caplog.text
    assert "authentication failed for [redacted]" in caplog.text
    assert secret not in (stream._backend_error or "")


def test_voice_apply_redacts_configured_openai_key_from_http_and_logs(
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    secret = "sk-test-voice-apply-must-be-redacted"
    monkeypatch.setenv("OPENAI_API_KEY", secret)

    app = FastAPI()
    handler = MagicMock()
    handler.change_voice = AsyncMock(side_effect=RuntimeError(f"voice update rejected for {secret}"))
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(handler, robot, settings_app=app)
    stream._init_settings_ui_if_needed()

    loop = asyncio.new_event_loop()
    started = threading.Event()

    def _run_loop() -> None:
        asyncio.set_event_loop(loop)
        started.set()
        loop.run_forever()

    thread = threading.Thread(target=_run_loop, daemon=True)
    thread.start()
    started.wait(timeout=1.0)
    stream._asyncio_loop = loop
    caplog.set_level(logging.ERROR, logger="peachy.console")

    try:
        response = TestClient(app).post("/api/v1/voices/apply?voice=marin")
    finally:
        loop.call_soon_threadsafe(loop.stop)
        thread.join(timeout=1.0)
        loop.close()

    assert response.status_code == 200
    assert response.json() == {
        "ok": True,
        "status": "Failed to change voice: voice update rejected for [redacted]",
    }
    assert secret not in response.text
    assert secret not in caplog.text
    assert "voice update rejected for [redacted]" in caplog.text


def test_personality_validation_redacts_openai_key_from_http_and_logs(
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    secret = "sk-test-personality-validation-redaction"
    monkeypatch.setenv("OPENAI_API_KEY", secret)
    monkeypatch.setattr(config, "REACHY_MINI_CUSTOM_PROFILE", None)
    monkeypatch.setattr("peachy.config.set_custom_profile", lambda _profile: None)

    def reject_instructions(*_args: Any, **_kwargs: Any) -> str:
        raise RuntimeError("personality rejected " + secret)

    monkeypatch.setattr("peachy.prompts.get_session_instructions", reject_instructions)
    app = FastAPI()
    handler = MagicMock()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(handler, robot, settings_app=app)
    stream._init_settings_ui_if_needed()
    loop = asyncio.new_event_loop()
    started = threading.Event()

    def _run_loop() -> None:
        asyncio.set_event_loop(loop)
        started.set()
        loop.run_forever()

    thread = threading.Thread(target=_run_loop, daemon=True)
    thread.start()
    started.wait(timeout=1.0)
    stream._asyncio_loop = loop
    caplog.set_level(logging.ERROR, logger="peachy.console")
    try:
        response = TestClient(app).post(
            "/api/v1/personalities/apply",
            json={"name": "sorry_bro"},
        )
    finally:
        loop.call_soon_threadsafe(loop.stop)
        thread.join(timeout=1.0)
        loop.close()

    assert response.status_code == 200
    assert secret not in response.text
    assert secret not in caplog.text
    assert response.json()["status"] == "Failed to apply personality: personality rejected [redacted]"
    assert "personality rejected [redacted]" in caplog.text


@pytest.mark.asyncio
async def test_handler_shutdown_log_redacts_configured_openai_key(
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    secret = "sk-test-shutdown-must-be-redacted"
    monkeypatch.setenv("OPENAI_API_KEY", secret)

    handler = MagicMock()
    handler.shutdown = AsyncMock(side_effect=RuntimeError(f"shutdown rejected for {secret}"))
    stream = LocalStream(handler, MagicMock())
    caplog.set_level(logging.DEBUG, logger="peachy.console")

    await stream._shutdown_active_handler()

    assert secret not in caplog.text
    assert "shutdown rejected for [redacted]" in caplog.text


def test_backend_startup_failure_is_recorded_without_raising(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Backend startup failures should become status state instead of killing LocalStream."""
    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "local")
    monkeypatch.setattr(config, "HF_REALTIME_SESSION_URL", None)
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", "ws://127.0.0.1:8765/v1/realtime")
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "huggingface")

    app = FastAPI()
    handler = MagicMock()
    handler.connection = None
    handler.shutdown = AsyncMock()
    media = SimpleNamespace(
        audio=None,
        backend=None,
        start_recording=MagicMock(),
        start_playing=MagicMock(),
    )
    robot = SimpleNamespace(media=media)
    stream = LocalStream(handler, robot, settings_app=app, instance_path=str(tmp_path))
    stream._backend_retry_delay = 0
    stream.record_loop = AsyncMock(return_value=None)  # type: ignore[method-assign]
    stream.play_loop = AsyncMock(return_value=None)  # type: ignore[method-assign]
    monkeypatch.setattr("peachy.console.apply_audio_startup_config", MagicMock())

    async def fail_and_stop() -> None:
        stream._stop_event.set()
        raise RuntimeError("local server unavailable")

    handler.start_up = AsyncMock(side_effect=fail_and_stop)

    try:
        stream.launch()
    finally:
        asyncio.set_event_loop(asyncio.new_event_loop())

    handler.start_up.assert_awaited_once()
    client = TestClient(app)
    response = client.get("/api/v1/status")

    assert response.status_code == 200
    data = response.json()
    assert data["backend_connected"] is False
    assert data["backend_connection_state"] == "disconnected"
    assert data["backend_error"] == "RuntimeError: local server unavailable"


@pytest.mark.asyncio
async def test_startup_loop_rebuilds_handler_on_restart_request(monkeypatch: pytest.MonkeyPatch) -> None:
    """LocalStream should shut down and rebuild the handler when a restart is requested."""
    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "local")
    monkeypatch.setattr(config, "HF_REALTIME_SESSION_URL", None)
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", "ws://127.0.0.1:8765/v1/realtime")
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "huggingface")

    class FakeHandler:
        def __init__(self) -> None:
            self.connection = None
            self.output_queue = asyncio.Queue()
            self.started = asyncio.Event()
            self.stopped = asyncio.Event()
            self.shutdown_calls = 0

        async def start_up(self) -> None:
            self.connection = object()
            self.started.set()
            await self.stopped.wait()
            self.connection = None

        async def shutdown(self) -> None:
            self.shutdown_calls += 1
            self.stopped.set()

        async def receive(self, _frame: Any) -> None:
            return None

        async def emit(self) -> None:
            return None

    handlers: list[FakeHandler] = []

    def handler_factory(_voice: str | None) -> FakeHandler:
        handler = FakeHandler()
        handlers.append(handler)
        return handler

    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    initial_handler = handler_factory(None)
    stream = LocalStream(initial_handler, robot, handler_factory=handler_factory)
    stream._backend_retry_delay = 0.01

    startup_task = asyncio.create_task(stream._run_handler_startup_loop())
    try:
        await _wait_until(lambda: initial_handler.started.is_set())

        await stream.request_backend_restart("backend_config_changed")

        await _wait_until(lambda: len(handlers) == 2 and handlers[1].started.is_set())

        assert initial_handler.shutdown_calls >= 1
        assert stream.handler is handlers[1]
        assert stream._backend_connected() is True
    finally:
        stream._stop_event.set()
        await stream._shutdown_active_handler()
        startup_task.cancel()
        try:
            await startup_task
        except asyncio.CancelledError:
            pass


def test_personality_routes_return_hf_voices() -> None:
    """Headless personality UI should expose the Hugging Face voices."""
    app = FastAPI()
    handler = MagicMock()
    mount_personality_routes(app, handler, lambda: None)

    client = TestClient(app)
    response = client.get("/voices")

    assert response.status_code == 200
    assert response.json() == HF_AVAILABLE_VOICES


def test_personality_routes_mount_versioned_paths() -> None:
    """Personality and voice endpoints should use the configured API prefix."""
    app = FastAPI()
    handler = MagicMock()

    mount_personality_routes(
        app,
        handler,
        lambda: None,
        api_prefix="/api/v1",
    )

    client = TestClient(app)
    response = client.get("/api/v1/voices")
    delete_response = client.delete("/api/v1/personalities", params={"name": "mad_scientist_assistant"})

    assert response.status_code == 200
    assert delete_response.status_code == 404
    assert delete_response.json()["error"] == "not_deletable"
    assert client.get("/personalities").status_code == 404
    assert client.get("/voices").status_code == 404
    assert client.delete("/personalities", params={"name": "mad_scientist_assistant"}).status_code == 404
    assert client.post("/voices/apply?voice=cedar").status_code == 404


def test_personality_routes_load_builtin_default_tools() -> None:
    """Headless personality UI should expose built-in default tools on initial load."""
    app = FastAPI()
    handler = MagicMock()
    mount_personality_routes(app, handler, lambda: None)

    client = TestClient(app)
    response = client.get("/personalities/load", params={"name": "(built-in default)"})

    assert response.status_code == 200
    data = response.json()
    assert data["tools_text"]
    assert "dance" in data["enabled_tools"]
    assert "camera" in data["enabled_tools"]


def test_personality_routes_apply_voice_accepts_query_param() -> None:
    """Headless personality UI should apply a voice change from a POST query param."""
    app = FastAPI()
    handler = MagicMock()
    handler.change_voice = AsyncMock(return_value="Voice changed to cedar.")

    loop = asyncio.new_event_loop()
    started = threading.Event()

    def _run_loop() -> None:
        asyncio.set_event_loop(loop)
        started.set()
        loop.run_forever()

    thread = threading.Thread(target=_run_loop, daemon=True)
    thread.start()
    started.wait(timeout=1.0)

    try:
        mount_personality_routes(app, handler, lambda: loop, api_prefix="/api/v1")

        client = TestClient(app)
        response = client.post("/api/v1/voices/apply?voice=cedar")

        assert response.status_code == 200
        assert response.json() == {"ok": True, "status": "Voice changed to cedar."}
        handler.change_voice.assert_awaited_once_with("cedar")
    finally:
        loop.call_soon_threadsafe(loop.stop)
        thread.join(timeout=1.0)
        loop.close()


def test_personality_routes_persist_startup_with_voice_override() -> None:
    """Saving a startup personality should persist the active manual voice override."""
    app = FastAPI()
    handler = MagicMock()
    handler.apply_personality = AsyncMock(return_value="Applied personality and restarted realtime session.")
    handler.get_current_voice = MagicMock(return_value="shimmer")
    persist_personality = MagicMock()

    loop = asyncio.new_event_loop()
    started = threading.Event()

    def _run_loop() -> None:
        asyncio.set_event_loop(loop)
        started.set()
        loop.run_forever()

    thread = threading.Thread(target=_run_loop, daemon=True)
    thread.start()
    started.wait(timeout=1.0)

    try:
        mount_personality_routes(app, handler, lambda: loop, persist_personality=persist_personality)

        client = TestClient(app)
        response = client.post("/personalities/apply", json={"name": "sorry_bro", "persist": True})

        assert response.status_code == 200
        assert response.json()["ok"] is True
        handler.apply_personality.assert_awaited_once_with("sorry_bro")
        persist_personality.assert_called_once_with("sorry_bro", "shimmer")
    finally:
        loop.call_soon_threadsafe(loop.stop)
        thread.join(timeout=1.0)
        loop.close()


def test_personality_routes_apply_same_profile_does_not_restart(monkeypatch: pytest.MonkeyPatch) -> None:
    """Re-applying the active personality should be a no-op for the realtime handler."""
    monkeypatch.setattr(config, "REACHY_MINI_CUSTOM_PROFILE", "sorry_bro")
    app = FastAPI()
    handler = MagicMock()
    handler.apply_personality = AsyncMock(return_value="should not be called")
    handler.get_current_voice = MagicMock(return_value="shimmer")
    mount_personality_routes(app, handler, lambda: None)

    response = TestClient(app).post("/personalities/apply", json={"name": "sorry_bro"})

    assert response.status_code == 200
    assert response.json()["status"] == "Personality unchanged."
    handler.apply_personality.assert_not_awaited()
    handler.get_current_voice.assert_not_called()


def test_personality_routes_startup_choice_survives_runtime_profile_change(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Runtime profile switching should not redefine the saved startup personality."""
    monkeypatch.setattr(config, "REACHY_MINI_CUSTOM_PROFILE", "captain_circuit")
    app = FastAPI()
    handler = MagicMock()
    mount_personality_routes(app, handler, lambda: None)
    client = TestClient(app)

    initial_response = client.get("/personalities")
    assert initial_response.status_code == 200
    assert initial_response.json()["current"] == "captain_circuit"
    assert initial_response.json()["startup"] == "captain_circuit"

    monkeypatch.setattr(config, "REACHY_MINI_CUSTOM_PROFILE", "chess_coach")

    switched_response = client.get("/personalities")
    assert switched_response.status_code == 200
    assert switched_response.json()["current"] == "chess_coach"
    assert switched_response.json()["startup"] == "captain_circuit"


def test_headless_personality_routes_can_use_stream_callbacks() -> None:
    """Headless personality routes can delegate apply/restart ownership to LocalStream."""
    app = FastAPI()
    handler = MagicMock()
    handler.apply_personality = AsyncMock(return_value="handler should not be called")
    apply_personality = AsyncMock(return_value="Applied personality and restarting backend.")
    get_current_voice = MagicMock(return_value="cedar")

    loop = asyncio.new_event_loop()
    started = threading.Event()

    def _run_loop() -> None:
        asyncio.set_event_loop(loop)
        started.set()
        loop.run_forever()

    thread = threading.Thread(target=_run_loop, daemon=True)
    thread.start()
    started.wait(timeout=1.0)

    try:
        mount_personality_routes(
            app,
            handler,
            lambda: loop,
            apply_personality=apply_personality,
            get_current_voice=get_current_voice,
        )

        response = TestClient(app).post("/personalities/apply", json={"name": "sorry_bro"})

        assert response.status_code == 200
        assert response.json()["status"] == "Applied personality and restarting backend."
        apply_personality.assert_awaited_once_with("sorry_bro")
        handler.apply_personality.assert_not_awaited()
    finally:
        loop.call_soon_threadsafe(loop.stop)
        thread.join(timeout=1.0)
        loop.close()


@pytest.mark.asyncio
async def test_apply_personality_propagates_restart_cancellation(monkeypatch: pytest.MonkeyPatch) -> None:
    """Cancellation during backend restart should not be converted into a status string."""
    monkeypatch.setattr("peachy.config.set_custom_profile", lambda _profile: None)
    monkeypatch.setattr(
        "peachy.prompts.get_session_instructions", lambda _instance_path=None: "instructions"
    )
    monkeypatch.setattr("peachy.prompts.get_session_voice", lambda default: default)

    stream = LocalStream(MagicMock(), MagicMock())

    async def cancel_restart(_reason: str) -> None:
        raise asyncio.CancelledError

    monkeypatch.setattr(stream, "request_backend_restart", cancel_restart)

    with pytest.raises(asyncio.CancelledError):
        await stream.apply_personality("sorry_bro")


@pytest.mark.asyncio
async def test_local_stream_change_voice_delegates_without_backend_restart() -> None:
    """LocalStream voice changes should update the active handler without rebuilding it."""
    handler = MagicMock()
    handler.change_voice = AsyncMock(return_value="Voice changed to Serena.")
    handler.get_current_voice = MagicMock(return_value="Serena")
    stream = LocalStream(handler, MagicMock())

    status = await stream.change_voice("Serena")

    assert status == "Voice changed to Serena."
    handler.change_voice.assert_awaited_once_with("Serena")
    assert stream._voice_override == "Serena"
    assert not stream._restart_requested.is_set()


@pytest.mark.asyncio
async def test_post_change_voice_sync_log_redacts_configured_openai_key(
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    secret = "sk-test-voice-sync-must-be-redacted"
    monkeypatch.setenv("OPENAI_API_KEY", secret)
    handler = MagicMock()
    handler.change_voice = AsyncMock(return_value="Voice changed to marin.")
    handler.get_current_voice = MagicMock(
        side_effect=RuntimeError(f"voice sync rejected for {secret}")
    )
    stream = LocalStream(handler, MagicMock())
    caplog.set_level(logging.DEBUG, logger="peachy.console")

    status = await stream.change_voice("marin")

    assert status == "Voice changed to marin."
    assert secret not in caplog.text
    assert "voice sync rejected for [redacted]" in caplog.text


@pytest.mark.parametrize(
    ("voices", "current_voice"),
    [
        (list(OPENAI_VOICES), "marin"),
        (HF_AVAILABLE_VOICES, "Aiden"),
    ],
    ids=("openai", "huggingface"),
)
@pytest.mark.asyncio
async def test_local_stream_voice_reads_delegate_to_active_provider(
    voices: list[str],
    current_voice: str,
) -> None:
    handler = MagicMock()
    handler.get_available_voices = AsyncMock(return_value=voices)
    handler.get_current_voice = MagicMock(return_value=current_voice)
    stream = LocalStream(handler, MagicMock(), startup_voice="stale-local-override")

    assert await stream.get_available_voices() == voices
    assert stream.get_current_voice() == current_voice
    handler.get_available_voices.assert_awaited_once_with()
    handler.get_current_voice.assert_called_once_with()


def test_local_stream_persist_personality_stores_voice_override(tmp_path) -> None:
    """Persisting startup settings should write both profile and voice override."""
    stream = LocalStream(MagicMock(), MagicMock(), instance_path=str(tmp_path))

    stream._persist_personality("sorry_bro", "shimmer")

    settings_path = tmp_path / "startup_settings.json"
    assert settings_path.exists()
    assert settings_path.read_text(encoding="utf-8") == '{\n  "profile": "sorry_bro",\n  "voice": "shimmer"\n}\n'
    assert stream._read_persisted_personality() == "sorry_bro"


def test_local_stream_persist_personality_clears_legacy_startup_env_overrides(tmp_path, monkeypatch) -> None:
    """Saving startup settings should remove legacy `.env` profile and voice overrides."""
    env_path = tmp_path / ".env"
    env_path.write_text(
        "HF_TOKEN=test-token\n"
        "REACHY_MINI_CUSTOM_PROFILE=mad_scientist_assistant\n"
        "REACHY_MINI_VOICE_OVERRIDE=shimmer\n",
        encoding="utf-8",
    )
    stream = LocalStream(MagicMock(), MagicMock(), instance_path=str(tmp_path))

    stream._persist_personality(None, "Aiden")

    env_text = env_path.read_text(encoding="utf-8")
    assert "HF_TOKEN=test-token" in env_text
    assert "REACHY_MINI_CUSTOM_PROFILE=" not in env_text
    assert "REACHY_MINI_VOICE_OVERRIDE=" not in env_text

    applied_profiles: list[str | None] = []
    monkeypatch.delenv("REACHY_MINI_CUSTOM_PROFILE", raising=False)
    monkeypatch.setattr(
        "peachy.config.set_custom_profile",
        lambda profile: applied_profiles.append(profile),
    )

    settings = load_startup_settings_into_runtime(tmp_path)

    assert settings == StartupSettings(voice="Aiden")
    assert applied_profiles == [None]


def test_local_stream_launch_starts_media_while_provider_waits_for_config(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Startup should keep media and settings live while provider configuration is missing."""
    monkeypatch.setattr(config, "HF_REALTIME_CONNECTION_MODE", "local")
    monkeypatch.setattr(config, "HF_REALTIME_SESSION_URL", None)
    monkeypatch.setattr(config, "HF_REALTIME_WS_URL", None)
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "huggingface")

    media = SimpleNamespace(
        start_recording=MagicMock(),
        start_playing=MagicMock(),
    )
    robot = SimpleNamespace(media=media)
    handler = MagicMock()
    handler.shutdown = AsyncMock()
    stream = LocalStream(handler, robot, settings_app=FastAPI(), instance_path=str(tmp_path))

    init_settings_ui = MagicMock()
    monkeypatch.setattr(stream, "_init_settings_ui_if_needed", init_settings_ui)
    monkeypatch.setattr(stream, "_run_handler_startup_loop", AsyncMock(return_value=None))
    monkeypatch.setattr(stream, "record_loop", AsyncMock(return_value=None))
    monkeypatch.setattr(stream, "play_loop", AsyncMock(return_value=None))
    monkeypatch.setattr("peachy.console.apply_audio_startup_config", MagicMock())

    try:
        stream.launch()
    finally:
        asyncio.set_event_loop(asyncio.new_event_loop())

    init_settings_ui.assert_called_once()
    media.start_recording.assert_called_once()
    media.start_playing.assert_called_once()
    handler.start_up.assert_not_called()


def test_settings_status_and_save_select_openai_without_echoing_key(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    secret = "sk-test-settings-secret"
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "huggingface")
    app = FastAPI()
    handler = MagicMock()
    handler.shutdown = AsyncMock()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(
        handler,
        robot,
        settings_app=app,
        instance_path=str(tmp_path),
        handler_factory=lambda _voice: handler,
        provider_resolver=get_provider_definition,
    )
    stream._init_settings_ui_if_needed()
    client = TestClient(app)

    response = client.post(
        "/api/v1/backend_config",
        json={
            "provider": "openai",
            "openai_api_key": secret,
            "openai_model": "gpt-realtime",
        },
    )

    assert response.status_code == 200
    body = response.json()
    assert body["provider"] == "openai"
    assert body["provider_state"] == "ready"
    assert body["has_openai_api_key"] is True
    assert secret not in response.text
    assert stream._restart_requested.is_set()
    env_text = (tmp_path / ".env").read_text(encoding="utf-8")
    assert "PEACHY_REALTIME_PROVIDER=openai" in env_text
    assert "OPENAI_REALTIME_MODEL=gpt-realtime" in env_text
    assert secret in env_text


def test_backend_config_persistence_failure_is_transactional(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    original_env_text = (
        "PEACHY_REALTIME_PROVIDER=huggingface\n"
        "OPENAI_API_KEY=sk-test-original-key\n"
        "OPENAI_REALTIME_MODEL=gpt-realtime-original\n"
    )
    env_path = tmp_path / ".env"
    env_path.write_text(original_env_text, encoding="utf-8")
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "huggingface")
    monkeypatch.setenv("OPENAI_API_KEY", "sk-test-original-key")
    monkeypatch.setenv("OPENAI_REALTIME_MODEL", "gpt-realtime-original")

    app = FastAPI()
    handler = MagicMock()
    handler.shutdown = AsyncMock()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(
        handler,
        robot,
        settings_app=app,
        instance_path=str(tmp_path),
        handler_factory=lambda _voice: handler,
        provider_resolver=get_provider_definition,
    )
    stream._init_settings_ui_if_needed()

    def fail_write_text(*_args: Any, **_kwargs: Any) -> int:
        raise OSError("simulated full disk")

    monkeypatch.setattr(Path, "write_text", fail_write_text)

    response = TestClient(app).post(
        "/api/v1/backend_config",
        json={
            "provider": "openai",
            "openai_api_key": "sk-test-rejected-key",
            "openai_model": "gpt-realtime-rejected",
        },
    )

    assert response.status_code == 500
    assert response.json() == {"ok": False, "error": "persistence_failed"}
    assert os.environ["PEACHY_REALTIME_PROVIDER"] == "huggingface"
    assert os.environ["OPENAI_API_KEY"] == "sk-test-original-key"
    assert os.environ["OPENAI_REALTIME_MODEL"] == "gpt-realtime-original"
    assert env_path.read_text(encoding="utf-8") == original_env_text
    assert list(tmp_path.glob(".env.*.tmp")) == []
    assert not stream._restart_requested.is_set()


def test_settings_reject_unknown_provider_without_mutating_selection(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("PEACHY_REALTIME_PROVIDER", "openai")
    app = FastAPI()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(
        MagicMock(),
        robot,
        settings_app=app,
        instance_path=str(tmp_path),
        provider_resolver=get_provider_definition,
    )
    stream._init_settings_ui_if_needed()

    response = TestClient(app).post("/api/v1/backend_config", json={"provider": "mystery"})

    assert response.status_code == 400
    assert response.json()["error"] == "invalid_provider"
    assert os.environ["PEACHY_REALTIME_PROVIDER"] == "openai"


def test_packaged_settings_assets_have_provider_specific_controls() -> None:
    static_root = Path(__file__).parents[1] / "src" / "peachy" / "static" / "js"
    settings_js = (static_root / "views" / "settings.js").read_text(encoding="utf-8")
    api_js = (static_root / "api.js").read_text(encoding="utf-8")
    assert 'name: "provider"' in settings_js
    assert 'name: "openai_api_key"' in settings_js
    assert 'name: "openai_model"' in settings_js
    assert "huggingface" in settings_js
    assert "backend_config" in api_js
    assert "persistence_failed" in api_js
    assert "reconnectBackend" in api_js
    assert 'role: "status"' in settings_js
    assert '"Reconnect"' in settings_js


@pytest.mark.asyncio
async def test_openai_exhaustion_waits_for_explicit_restart(monkeypatch: pytest.MonkeyPatch) -> None:
    secret = "sk-test-do-not-log"
    monkeypatch.setenv("OPENAI_API_KEY", secret)
    handler = MagicMock()
    handler.start_up = AsyncMock(side_effect=RetryExhaustedError("failed 3 attempts " + secret))
    handler.shutdown = AsyncMock()
    handler.output_queue = asyncio.Queue()
    handler.last_activity_time = 0.0
    definition = ProviderDefinition(
        name="openai",
        readiness=lambda: ProviderReadiness("ready"),
        redacted_status=lambda: {"has_openai_api_key": True},
        build_handler=lambda _deps, _path, _voice: handler,
        auto_reconnect=False,
    )
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(handler, robot, provider_resolver=lambda: definition)

    task = asyncio.create_task(stream._run_handler_startup_loop())
    await _wait_until(lambda: stream._backend_connection_state == "disconnected")
    await asyncio.sleep(0.05)
    assert handler.start_up.await_count == 1
    assert secret not in (stream._backend_error or "")
    stream._stop_event.set()
    stream._restart_requested.set()
    await asyncio.wait_for(task, timeout=1.0)


def test_manual_reconnect_endpoint_requests_restart() -> None:
    app = FastAPI()
    handler = MagicMock()
    handler.shutdown = AsyncMock()
    robot = SimpleNamespace(media=SimpleNamespace(audio=None, backend=None))
    stream = LocalStream(handler, robot, settings_app=app, provider_resolver=get_provider_definition)
    stream._init_settings_ui_if_needed()

    response = TestClient(app).post("/api/v1/reconnect")

    assert response.status_code == 200
    assert response.json()["ok"] is True
    assert stream._restart_requested.is_set()


@pytest.mark.asyncio
async def test_status_and_logs_never_contain_openai_key(
    caplog: pytest.LogCaptureFixture,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    secret = "sk-test-redaction-canary"
    monkeypatch.setenv("OPENAI_API_KEY", secret)
    handler = MagicMock()
    handler.start_up = AsyncMock(side_effect=RetryExhaustedError("rejected " + secret))
    handler.shutdown = AsyncMock()
    handler.output_queue = asyncio.Queue()
    handler.last_activity_time = 0.0
    definition = ProviderDefinition(
        name="openai",
        readiness=lambda: ProviderReadiness("ready"),
        redacted_status=lambda: {"has_openai_api_key": True},
        build_handler=lambda _deps, _path, _voice: handler,
        auto_reconnect=False,
    )
    stream = LocalStream(
        handler,
        SimpleNamespace(media=SimpleNamespace(audio=None, backend=None)),
        settings_app=FastAPI(),
        provider_resolver=lambda: definition,
    )
    stream._init_settings_ui_if_needed()
    task = asyncio.create_task(stream._run_handler_startup_loop())
    await _wait_until(lambda: stream._backend_connection_state == "disconnected")
    response = TestClient(stream._settings_app).get("/api/v1/status")
    stream._stop_event.set()
    stream._restart_requested.set()
    await asyncio.wait_for(task, timeout=1.0)

    assert secret not in response.text
    assert secret not in caplog.text
    assert secret not in (stream._backend_error or "")


def test_provider_status_redacts_string_values_before_http_response(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    secret = "sk-test-provider-status-boundary"
    monkeypatch.setenv("OPENAI_API_KEY", secret)
    definition = ProviderDefinition(
        name="openai",
        readiness=lambda: ProviderReadiness("ready"),
        redacted_status=lambda: {"openai_model": "model-" + secret},
        build_handler=MagicMock(),
        auto_reconnect=False,
    )
    stream = LocalStream(
        MagicMock(),
        SimpleNamespace(media=SimpleNamespace(audio=None, backend=None)),
        settings_app=FastAPI(),
        provider_resolver=lambda: definition,
    )
    stream._init_settings_ui_if_needed()

    response = TestClient(stream._settings_app).get("/api/v1/status")

    assert response.status_code == 200
    assert secret not in response.text
    assert response.json()["openai_model"] == "model-[redacted]"