from __future__ import annotations

import asyncio
import json
import os
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Awaitable, Callable
from urllib.parse import urlsplit, urlunsplit

import aiohttp
import httpx
import websockets
from dotenv import load_dotenv
from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS

from app.livekit.adapters.xai_tts import OraclexAITTS
from app.providers.stt_config import override_config
from app.providers.stt_internal_livekit import InternalHTTPSTT, InternalSTTConfig, extract_text_from_transcript
from app.tools.local_stresstest.audio import (
    AudioSample,
    audio_metrics,
    build_variations,
    pcm_to_wav_bytes,
    read_wav_mono16,
    vad_proxy_metrics,
    write_wav,
)
from app.tools.local_stresstest.report import (
    render_markdown_report,
    write_csv,
    write_json,
    write_markdown_report,
    write_mermaid_files,
)
from app.tools.local_stresstest.scenarios import scenario_description
from app.tools.local_stresstest.text import compare_text, normalize_text
from app.tools.local_stresstest.timeline import excerpt_timeline, timeline_has_error, wait_for_timeline


DEFAULT_EXPECTED_TEXT = (
    "Por que os serviços Aia, Hexa e Banca aparecem com valor na Danf "
    "se eles são incluídos no meu plano"
)
DEFAULT_CRITICAL_TERMS = ("por", "que", "aia", "hexa", "banca", "danf", "incluídos", "plano")
DEFAULT_SHORT_TTS_TEXT = "Teste de voz do xAI para validar áudio sintetizado."
DEFAULT_LONG_TTS_TEXT = (
    "Este é um teste operacional do TTS xAI para validar latência, duração, volume e entrega "
    "de áudio antes do relatório local."
)
TERMINAL_EVENTS = {
    "bridge_failed",
    "stop_stt_unavailable",
    "stop_tts_unavailable",
    "stop_bridge_failed",
    "stop_agent_runtime_unavailable",
}


@dataclass(frozen=True, slots=True)
class StressConfig:
    env_file: Path
    report_dir: Path
    expected_text: str
    synthesis_text: str
    critical_terms: tuple[str, ...]
    stt_wer_threshold: float
    bridge_url: str
    bridge_health_url: str
    agent_health_url: str
    startup_wait_s: float
    startup_poll_s: float
    skip_local_wait: bool
    repeat: int
    concurrency: int
    e2e_turns: int
    e2e_timeout_s: float
    stress_audio: Path | None
    prefix_text: str
    prefix_words: int
    vad_proxy_threshold_dbfs: float
    vad_proxy_prefix_padding_ms: int
    vad_proxy_min_speech_ms: int


def _env_str(name: str, default: str = "") -> str:
    return str(os.getenv(name, default) or default).strip()


def _env_int(name: str, default: int) -> int:
    try:
        return int(_env_str(name, str(default)))
    except ValueError:
        return default


def _env_float(name: str, default: float) -> float:
    try:
        return float(_env_str(name, str(default)))
    except ValueError:
        return default


def _env_bool(name: str, default: bool = False) -> bool:
    value = _env_str(name)
    if not value:
        return default
    return value.lower() in {"1", "true", "yes", "on", "sim"}


def _first_words(text: str, count: int) -> str:
    words = [item for item in str(text or "").strip().split() if item]
    return " ".join(words[: max(1, count)])


def _derive_bridge_health_url(bridge_url: str) -> str:
    explicit = _env_str("STRESS_BRIDGE_HEALTH_URL")
    if explicit:
        return explicit
    parsed = urlsplit(bridge_url)
    if not parsed.netloc:
        return "http://127.0.0.1:8000/health"
    scheme = "https" if parsed.scheme == "wss" else "http"
    return urlunsplit((scheme, parsed.netloc, "/health", "", ""))


def _derive_agent_health_url() -> str:
    return _env_str(
        "STRESS_AGENT_HEALTH_URL",
        f"http://127.0.0.1:{_env_int('AGENT_SERVER_PORT', 18081)}/",
    )


def load_config() -> StressConfig:
    env_file = Path(_env_str("STRESS_ENV_FILE", ".env.dev"))
    if env_file.exists():
        load_dotenv(env_file, override=False)

    report_dir = Path(_env_str("STRESS_REPORT_DIR", ".run/local-stresstest"))
    os.environ.setdefault("STT_DUMP_DIR", str(report_dir / "stt_dumps"))
    os.environ.setdefault("FLOW_LOG_VAD_DECISIONS", "1")
    os.environ.setdefault("FLOW_LOG_VAD_ACTIVITY", "1")
    os.environ.setdefault("FLOW_LOG_VAD_ACTIVITY_MIN_PROB", "0.03")
    audio_value = _env_str("STRESS_AUDIO")
    critical_terms = tuple(
        item.strip()
        for item in _env_str("STRESS_CRITICAL_TERMS", ",".join(DEFAULT_CRITICAL_TERMS)).split(",")
        if item.strip()
    )
    bridge_url = _env_str("STRESS_BRIDGE_URL", "ws://127.0.0.1:8000/ws/agent")
    expected_text = _env_str("STRESS_EXPECTED_TEXT", DEFAULT_EXPECTED_TEXT)
    synthesis_text = _env_str("STRESS_SYNTH_TEXT", expected_text)
    prefix_words = max(1, _env_int("STRESS_PREFIX_WORDS", 2))
    vad_prefix_padding_default_ms = round(_env_float("VAD_PREFIX_PADDING_DURATION", 1.5) * 1000)
    return StressConfig(
        env_file=env_file,
        report_dir=report_dir,
        expected_text=expected_text,
        synthesis_text=synthesis_text,
        critical_terms=critical_terms,
        stt_wer_threshold=_env_float("STRESS_WER_THRESHOLD", 0.20),
        bridge_url=bridge_url,
        bridge_health_url=_derive_bridge_health_url(bridge_url),
        agent_health_url=_derive_agent_health_url(),
        startup_wait_s=max(0.0, _env_float("STRESS_STARTUP_WAIT_S", 180.0)),
        startup_poll_s=max(0.5, _env_float("STRESS_STARTUP_POLL_S", 3.0)),
        skip_local_wait=_env_bool("STRESS_SKIP_LOCAL_WAIT", False),
        repeat=max(1, _env_int("STRESS_REPEAT", 1)),
        concurrency=max(1, _env_int("STRESS_CONCURRENCY", 2)),
        e2e_turns=max(1, _env_int("STRESS_E2E_TURNS", 3)),
        e2e_timeout_s=max(5.0, _env_float("STRESS_E2E_TIMEOUT_S", 45.0)),
        stress_audio=Path(audio_value) if audio_value else None,
        prefix_text=_env_str("STRESS_PREFIX_TEXT", _first_words(expected_text, prefix_words)),
        prefix_words=prefix_words,
        vad_proxy_threshold_dbfs=_env_float("STRESS_VAD_PROXY_DBFS", -45.0),
        vad_proxy_prefix_padding_ms=max(
            0,
            _env_int("STRESS_VAD_PREFIX_PADDING_MS", vad_prefix_padding_default_ms),
        ),
        vad_proxy_min_speech_ms=max(0, _env_int("STRESS_VAD_MIN_SPEECH_MS", 100)),
    )


def _require_env(*names: str) -> str:
    for name in names:
        value = _env_str(name)
        if value:
            return value
    raise RuntimeError(f"Missing required environment variable: {' or '.join(names)}")


ServiceProbe = Callable[[str], Awaitable[tuple[bool, str]]]
SleepFn = Callable[[float], Awaitable[None]]


async def _http_ready_probe(url: str) -> tuple[bool, str]:
    timeout = httpx.Timeout(2.5, connect=1.0)
    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.get(url)
    ok = 200 <= response.status_code < 300
    return ok, f"HTTP {response.status_code}"


async def wait_for_local_services(
    config: StressConfig,
    *,
    probe: ServiceProbe | None = None,
    sleep: SleepFn = asyncio.sleep,
) -> dict[str, dict[str, Any]]:
    services = {
        "bridge": config.bridge_health_url,
        "agent_runtime": config.agent_health_url,
    }
    states: dict[str, dict[str, Any]] = {
        name: {
            "service": name,
            "url": url,
            "ok": False,
            "status": "pending",
            "attempts": 0,
            "detail": "",
        }
        for name, url in services.items()
    }
    if config.skip_local_wait:
        for state in states.values():
            state.update({"ok": True, "status": "skipped", "detail": "STRESS_SKIP_LOCAL_WAIT habilitado"})
        return states

    active_probe = probe or _http_ready_probe
    deadline = time.monotonic() + config.startup_wait_s
    print(
        "[local-stresstest] Aguardando servicos locais "
        f"(timeout={config.startup_wait_s:.0f}s, poll={config.startup_poll_s:.1f}s)"
    )
    while True:
        for name, url in services.items():
            state = states[name]
            if state["ok"]:
                continue
            state["attempts"] += 1
            try:
                ok, detail = await active_probe(url)
            except Exception as exc:
                ok = False
                detail = f"{type(exc).__name__}: {exc}"
            state.update(
                {
                    "ok": ok,
                    "status": "ready" if ok else "waiting",
                    "detail": detail,
                }
            )

        if all(state["ok"] for state in states.values()):
            print("[local-stresstest] Servicos locais prontos: bridge, agent_runtime")
            return states

        if time.monotonic() >= deadline:
            failed = "; ".join(
                f"{name} em {state['url']} -> {state['detail'] or state['status']}"
                for name, state in states.items()
                if not state["ok"]
            )
            raise RuntimeError(
                f"Timeout aguardando servicos locais por {config.startup_wait_s:.0f}s: {failed}"
            )

        remaining = max(0.1, deadline - time.monotonic())
        await sleep(min(config.startup_poll_s, remaining))


def _stt_config() -> InternalSTTConfig:
    return InternalSTTConfig(
        url=_require_env("STT_URL"),
        api_key=_env_str("STT_KEY") or _env_str("STT_API_KEY", "unknow"),
        language=_env_str("STT_LANG", "portuguese"),
        timeout_s=_env_float("STRESS_STT_TIMEOUT_S", _env_float("STT_TIMEOUT_S", 30.0)),
        min_prob_single_word=_env_float("STT_MIN_PROB_SINGLE_WORD", 0.10),
        initial_prompt=_env_str("STT_INITIAL_PROMPT"),
        config_override=_env_str("STT_CONFIG_OVERRIDE"),
        output_mode=_env_str("STT_OUTPUT_MODE", "threshold_text"),
    )


def _stt_preprocessor_profile(cfg: InternalSTTConfig) -> str:
    raw = (os.getenv("STT_FORCE_OVERRIDE_CONFIG", "") or cfg.config_override or "").strip()
    payload: dict[str, Any]
    if raw:
        try:
            parsed = json.loads(raw)
            payload = parsed if isinstance(parsed, dict) else {}
        except Exception:
            return "custom_override_unparseable"
    else:
        payload = override_config

    preprocessors = payload.get("pre_processors")
    if not isinstance(preprocessors, list):
        return "none"
    names = [
        str(item.get("strategy") or "").strip()
        for item in preprocessors
        if isinstance(item, dict) and str(item.get("strategy") or "").strip()
    ]
    return ",".join(names) if names else "none"


def _prefix_ok(*, expected_prefix: str, transcript: str) -> bool:
    normalized_expected = normalize_text(expected_prefix)
    normalized_actual = normalize_text(transcript)
    return bool(normalized_expected) and normalized_actual.startswith(normalized_expected)


async def _collect_tts_bytes(tts: OraclexAITTS, text: str) -> bytes:
    frame = await tts.synthesize(text, conn_options=DEFAULT_API_CONNECT_OPTIONS).collect()
    return bytes(getattr(frame, "data", b""))


async def _synthesize_xai_to_sample(text: str) -> AudioSample:
    async with aiohttp.ClientSession() as session:
        tts = OraclexAITTS(
            api_key=_require_env("XAI_API_KEY"),
            voice=_env_str("XAI_TTS_VOICE"),
            language=_env_str("XAI_TTS_LANGUAGE", "pt-BR"),
            http_session=session,
        )
        try:
            pcm = await _collect_tts_bytes(tts, text)
            sample = AudioSample(name="xai_tts", pcm=pcm, sample_rate=tts.sample_rate, channels=tts.num_channels)
            tmp_path = Path(".run/local-stresstest/.tmp_xai_collect.wav")
            write_wav(tmp_path, sample)
            return read_wav_mono16(tmp_path)
        finally:
            await tts.aclose()


async def prepare_baseline(config: StressConfig) -> dict[str, Any]:
    assets_dir = config.report_dir / "assets"
    assets_dir.mkdir(parents=True, exist_ok=True)
    if config.stress_audio is not None:
        sample = read_wav_mono16(config.stress_audio)
        path = write_wav(assets_dir / "baseline_from_stress_audio.wav", sample)
        return {"sample": sample, "path": path, "synthetic": False}

    sample = await _synthesize_xai_to_sample(config.synthesis_text)
    path = write_wav(assets_dir / "synthetic_baseline.wav", sample)
    return {"sample": sample, "path": path, "synthetic": True}


async def run_stt_scenario(
    *,
    config: StressConfig,
    scenario: AudioSample,
    output_path: Path,
) -> dict[str, Any]:
    started = time.perf_counter()
    metrics = audio_metrics(scenario.pcm, sample_rate=scenario.sample_rate, channels=scenario.channels)
    vad_metrics = vad_proxy_metrics(
        scenario,
        threshold_dbfs=config.vad_proxy_threshold_dbfs,
        prefix_padding_ms=config.vad_proxy_prefix_padding_ms,
        min_speech_ms=config.vad_proxy_min_speech_ms,
    )
    write_wav(output_path, scenario)
    stt_cfg = _stt_config()
    row: dict[str, Any] = {
        "scenario": scenario.name,
        "description": scenario_description(scenario.name),
        "audio_path": str(output_path),
        "duration_ms": metrics.duration_ms,
        "rms_dbfs": metrics.rms_dbfs,
        "peak_dbfs": metrics.peak_dbfs,
        "clipping_ratio": metrics.clipping_ratio,
        "vad_proxy_threshold_dbfs": vad_metrics.threshold_dbfs,
        "vad_proxy_prefix_padding_ms": vad_metrics.prefix_padding_ms,
        "vad_proxy_first_voice_ms": vad_metrics.first_voice_ms,
        "vad_proxy_padded_start_ms": vad_metrics.padded_start_ms,
        "vad_proxy_unrecovered_prefix_ms": vad_metrics.unrecovered_prefix_ms,
        "vad_proxy_speech_ms": vad_metrics.speech_ms,
        "vad_proxy_initial_200ms_dbfs": vad_metrics.initial_200ms_dbfs,
        "vad_proxy_initial_500ms_dbfs": vad_metrics.initial_500ms_dbfs,
        "vad_proxy_low_start_risk": vad_metrics.low_start_risk,
        "expected_prefix": config.prefix_text,
        "stt_preprocessors": _stt_preprocessor_profile(stt_cfg),
    }
    client = httpx.AsyncClient()
    try:
        stt = InternalHTTPSTT(stt_cfg, client=client)
        event = await stt._post_internal_http(
            scenario.pcm,
            pcm_to_wav_bytes(scenario),
            req_id=f"stress-{scenario.name}",
            started_ns=time.time_ns(),
            language=_env_str("STT_LANG", "portuguese"),
            audio_duration_ms=metrics.duration_ms,
            level_dbfs=metrics.rms_dbfs,
        )
        transcript = event.alternatives[0].text if event.alternatives else ""
        transcript = extract_text_from_transcript(transcript)
        prefix_ok = _prefix_ok(expected_prefix=config.prefix_text, transcript=transcript)
        comparison = compare_text(
            expected=config.expected_text,
            actual=transcript,
            critical_terms=config.critical_terms,
        )
        passed = bool(transcript) and comparison.terms_ok and comparison.wer <= config.stt_wer_threshold
        row.update(
            {
                "passed": passed,
                "transcript": transcript,
                "actual_prefix": _first_words(transcript, config.prefix_words),
                "prefix_ok": prefix_ok,
                "wer": round(comparison.wer, 4),
                "substitutions": comparison.substitutions,
                "deletions": comparison.deletions,
                "insertions": comparison.insertions,
                "missing_terms": ",".join(comparison.missing_terms),
                "latency_ms": round((time.perf_counter() - started) * 1000),
                "error": "",
            }
        )
    except Exception as exc:
        row.update(
            {
                "passed": False,
                "transcript": "",
                "actual_prefix": "",
                "prefix_ok": False,
                "wer": "",
                "missing_terms": "",
                "latency_ms": round((time.perf_counter() - started) * 1000),
                "error": str(exc),
            }
        )
    finally:
        await client.aclose()
    return row


async def run_stt_suite(config: StressConfig, baseline: AudioSample) -> list[dict[str, Any]]:
    audio_dir = config.report_dir / "stt_audio"
    variations = []
    for repeat_idx in range(1, config.repeat + 1):
        for variation in build_variations(baseline):
            if config.repeat > 1:
                variations.append(
                    AudioSample(
                        name=f"{variation.name}_r{repeat_idx}",
                        pcm=variation.pcm,
                        sample_rate=variation.sample_rate,
                        channels=variation.channels,
                    )
                )
            else:
                variations.append(variation)
    rows: list[dict[str, Any]] = []
    semaphore = asyncio.Semaphore(config.concurrency)

    async def _run(item: AudioSample, idx: int) -> dict[str, Any]:
        async with semaphore:
            return await run_stt_scenario(
                config=config,
                scenario=item,
                output_path=audio_dir / f"{idx:02d}_{item.name}.wav",
            )

    tasks = [
        _run(variation, idx)
        for idx, variation in enumerate(variations, start=1)
    ]
    for result in await asyncio.gather(*tasks):
        rows.append(result)
    return rows


async def run_tts_suite(config: StressConfig) -> list[dict[str, Any]]:
    scenarios = [
        ("short", DEFAULT_SHORT_TTS_TEXT),
        ("long", DEFAULT_LONG_TTS_TEXT),
        ("reuse_1", DEFAULT_SHORT_TTS_TEXT),
        ("reuse_2", DEFAULT_SHORT_TTS_TEXT),
    ]
    rows: list[dict[str, Any]] = []
    out_dir = config.report_dir / "tts_audio"
    async with aiohttp.ClientSession() as session:
        tts = OraclexAITTS(
            api_key=_require_env("XAI_API_KEY"),
            voice=_env_str("XAI_TTS_VOICE"),
            language=_env_str("XAI_TTS_LANGUAGE", "pt-BR"),
            http_session=session,
        )
        try:
            for name, text in scenarios:
                started = time.perf_counter()
                row: dict[str, Any] = {"scenario": name, "text_len": len(text)}
                row["description"] = scenario_description(name)
                try:
                    pcm = await _collect_tts_bytes(tts, text)
                    sample = AudioSample(name=name, pcm=pcm, sample_rate=tts.sample_rate, channels=tts.num_channels)
                    path = write_wav(out_dir / f"{name}.wav", sample)
                    normalized = read_wav_mono16(path)
                    metrics = audio_metrics(normalized.pcm)
                    expected_min_ms = max(250, len(text) * 18)
                    expected_max_ms = max(2500, len(text) * 180)
                    passed = (
                        bool(pcm)
                        and metrics.audible
                        and metrics.clipping_ratio <= _env_float("STRESS_TTS_MAX_CLIPPING", 0.02)
                        and expected_min_ms <= metrics.duration_ms <= expected_max_ms
                    )
                    row.update(
                        {
                            "passed": passed,
                            "audio_path": str(path),
                            "duration_ms": metrics.duration_ms,
                            "latency_ms": round((time.perf_counter() - started) * 1000),
                            "rms_dbfs": metrics.rms_dbfs,
                            "peak_dbfs": metrics.peak_dbfs,
                            "clipping_ratio": metrics.clipping_ratio,
                            "error": "",
                        }
                    )
                except Exception as exc:
                    row.update(
                        {
                            "passed": False,
                            "audio_path": "",
                            "duration_ms": 0,
                            "latency_ms": round((time.perf_counter() - started) * 1000),
                            "rms_dbfs": "",
                            "clipping_ratio": "",
                            "error": str(exc),
                        }
                    )
                rows.append(row)
        finally:
            await tts.aclose()
    return rows


def _start_payload(config: StressConfig) -> dict[str, Any]:
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
    return {
        "type": "start",
        "data": {
            "agent": "conta",
            "ani": "5511999990000",
            "gsm": "5511999990000",
            "routerCallKeyDay": datetime.now().strftime("%Y%m%d"),
            "routerCallKey": f"RCK-stress-{stamp}",
            "callIdGed": f"GED-stress-{stamp}",
            "session_id": f"stress-{stamp}",
            "agentData": {"idFatura": "FAT-STRESS-001"},
        },
        "audioFormat": {"encoding": "linear16", "sampleRateHz": 16000, "channels": 1},
        "callConfig": {
            "agentBackend": "remote_ws_fake",
            "stt": {"provider": "internal_http"},
            "tts": {"provider": "xai"},
        },
    }


async def _send_pcm_turns(ws: Any, sample: AudioSample, *, turns: int) -> None:
    frame_bytes = round(sample.sample_rate * 20 / 1000) * 2
    silence = b"\x00" * frame_bytes
    for _ in range(turns):
        for offset in range(0, len(sample.pcm), frame_bytes):
            chunk = sample.pcm[offset : offset + frame_bytes]
            if len(chunk) < frame_bytes:
                chunk = chunk + b"\x00" * (frame_bytes - len(chunk))
            await ws.send(chunk)
            await asyncio.sleep(0.02)
        for _ in range(60):
            await ws.send(silence)
            await asyncio.sleep(0.02)


async def run_e2e_suite(config: StressConfig, baseline: AudioSample) -> list[dict[str, Any]]:
    scenario = AudioSample(name="clean", pcm=baseline.pcm)
    started = time.perf_counter()
    row: dict[str, Any] = {
        "scenario": scenario.name,
        "description": scenario_description(scenario.name),
        "ready_received": False,
        "stop_received": False,
        "non_silent_frames": 0,
        "timeline_transcript": "",
        "audio_path": "",
        "error": "",
    }
    received_audio = bytearray()
    text_messages: list[dict[str, Any]] = []
    room = ""

    try:
        async with websockets.connect(config.bridge_url, max_size=2**24) as ws:
            await ws.send(json.dumps(_start_payload(config), ensure_ascii=False))
            first = await asyncio.wait_for(ws.recv(), timeout=10.0)
            if isinstance(first, bytes):
                raise RuntimeError("Expected ready/stop text message, got binary")
            payload = json.loads(first)
            text_messages.append(payload)
            if payload.get("type") == "stop":
                row["error"] = json.dumps(payload, ensure_ascii=False)
                row["stop_received"] = True
                return [dict(row, passed=False)]
            if payload.get("type") != "ready":
                raise RuntimeError(f"Expected ready, got {payload}")
            row["ready_received"] = True
            room = str(payload.get("room") or "")

            sender = asyncio.create_task(_send_pcm_turns(ws, scenario, turns=config.e2e_turns))
            deadline = time.monotonic() + config.e2e_timeout_s
            while time.monotonic() < deadline:
                try:
                    message = await asyncio.wait_for(ws.recv(), timeout=0.5)
                except asyncio.TimeoutError:
                    if sender.done():
                        continue
                    continue
                if isinstance(message, bytes):
                    received_audio.extend(message)
                    if audio_metrics(message).audible:
                        row["non_silent_frames"] += 1
                    continue

                payload = json.loads(message)
                text_messages.append(payload)
                if payload.get("type") == "stop":
                    row["stop_received"] = True
                    break

            sender.cancel()
            await asyncio.gather(sender, return_exceptions=True)

        if received_audio:
            audio_path = config.report_dir / "e2e_audio" / f"{scenario.name}_bridge_response.wav"
            write_wav(audio_path, AudioSample(name="bridge_response", pcm=bytes(received_audio)))
            row["audio_path"] = str(audio_path)

        timeline_dir = Path(_env_str("CALL_TIMELINE_DIR", "./timeline"))
        records = wait_for_timeline(timeline_dir / f"{room}.jsonl", timeout_s=5.0) if room else []
        transcripts = [
            str(record.get("text") or record.get("transcript") or "")
            for record in records
            if str(record.get("event") or "") in {"user_transcript_final", "stt_final"}
        ]
        row["timeline_transcript"] = " | ".join(item for item in transcripts if item)
        row["timeline_path"] = str(timeline_dir / f"{room}.jsonl") if room else ""
        row["timeline_error"] = timeline_has_error(records)
        row["text_messages"] = json.dumps(text_messages, ensure_ascii=False)
        terminal_error = any(
            str(message.get("data", {}).get("status") or "") in TERMINAL_EVENTS
            for message in text_messages
            if isinstance(message, dict)
        )
        row["passed"] = bool(
            row["ready_received"]
            and row["stop_received"]
            and row["non_silent_frames"] > 0
            and row["timeline_transcript"]
            and not row["timeline_error"]
            and not terminal_error
        )
        if not row["passed"] and not row["error"]:
            row["error"] = "missing expected ready/stop/audio/timeline signal"

        excerpt_path = config.report_dir / "timeline_excerpt.jsonl"
        excerpt_path.parent.mkdir(parents=True, exist_ok=True)
        with excerpt_path.open("w", encoding="utf-8") as handle:
            for record in excerpt_timeline(records):
                handle.write(json.dumps(record, ensure_ascii=False))
                handle.write("\n")
    except Exception as exc:
        row["passed"] = False
        row["error"] = str(exc)

    row["latency_ms"] = round((time.perf_counter() - started) * 1000)
    return [row]


def _artifact_paths(config: StressConfig) -> dict[str, str]:
    artifacts = {
        "summary_json": str(config.report_dir / "summary.json"),
        "stt_csv": str(config.report_dir / "stt_results.csv"),
        "tts_csv": str(config.report_dir / "tts_results.csv"),
        "e2e_csv": str(config.report_dir / "e2e_results.csv"),
        "timeline_excerpt": str(config.report_dir / "timeline_excerpt.jsonl"),
        "report_md": str(config.report_dir / "report.md"),
    }
    artifacts.update(write_mermaid_files(config.report_dir))
    return artifacts


async def run() -> int:
    config = load_config()
    config.report_dir.mkdir(parents=True, exist_ok=True)
    started = time.perf_counter()
    started_at = datetime.now(timezone.utc).isoformat()

    startup_checks = await wait_for_local_services(config)
    baseline_info = await prepare_baseline(config)
    baseline = baseline_info["sample"]
    stt_results = await run_stt_suite(config, baseline)
    tts_results = await run_tts_suite(config)
    e2e_results = await run_e2e_suite(config, baseline)

    artifacts = _artifact_paths(config)
    write_csv(artifacts["stt_csv"], stt_results)
    write_csv(artifacts["tts_csv"], tts_results)
    write_csv(artifacts["e2e_csv"], e2e_results)

    summary = {
        "started_at": started_at,
        "duration_ms": round((time.perf_counter() - started) * 1000),
        "expected_text": config.expected_text,
        "synthesis_text": config.synthesis_text,
        "critical_terms": list(config.critical_terms),
        "wer_threshold": config.stt_wer_threshold,
        "stt_scenarios_per_repeat": len(build_variations(baseline)),
        "stt_total_calls": len(stt_results),
        "stt_prefix_text": config.prefix_text,
        "vad_proxy": {
            "threshold_dbfs": config.vad_proxy_threshold_dbfs,
            "prefix_padding_ms": config.vad_proxy_prefix_padding_ms,
            "min_speech_ms": config.vad_proxy_min_speech_ms,
            "note": "Proxy por energia RMS; nao substitui o Silero VAD do LiveKit.",
        },
        "diagnostics": {
            "stt_dump_dir": _env_str("STT_DUMP_DIR"),
            "flow_log_vad_decisions": _env_str("FLOW_LOG_VAD_DECISIONS"),
            "flow_log_vad_activity": _env_str("FLOW_LOG_VAD_ACTIVITY"),
            "flow_log_vad_activity_min_prob": _env_str("FLOW_LOG_VAD_ACTIVITY_MIN_PROB"),
        },
        "startup_checks": startup_checks,
        "baseline": {
            "path": str(baseline_info["path"]),
            "synthetic": bool(baseline_info["synthetic"]),
            "metrics": audio_metrics(baseline.pcm),
        },
        "stt_results": stt_results,
        "tts_results": tts_results,
        "e2e_results": e2e_results,
        "artifacts": artifacts,
    }
    summary["passed"] = all(row.get("passed") is True for row in [*stt_results, *tts_results, *e2e_results])
    write_json(artifacts["summary_json"], summary)
    write_markdown_report(artifacts["report_md"], summary)

    print(render_markdown_report(summary))
    return 0 if summary["passed"] else 1
