scripts/modal_smoke.py

from __future__ import annotations

import os
import sys
from pathlib import Path
from time import perf_counter


ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src"
ROOT_ENV = ROOT / ".env"
LOCAL_ENV = ROOT / "data" / "local.env"

if str(SRC) not in sys.path:
    sys.path.insert(0, str(SRC))


def main() -> int:
    _load_local_env(ROOT_ENV, override=True)
    _load_local_env(LOCAL_ENV, override=True)
    os.environ["TIME_MACHINE_ADAPTER_PROFILE"] = "modal"

    audio_path = Path(os.getenv("TIME_MACHINE_MODAL_SMOKE_AUDIO", ROOT / "data" / "audio" / "stt_probe.wav"))
    if not audio_path.exists():
        print(f"Missing smoke audio: {audio_path}")
        return 2

    from time_machine.application.container import create_container
    from time_machine.domain.events import AudioEvent, ConversationTextEvent, TranscriptEvent

    container = create_container(adapter_profile="modal")
    launch_started = perf_counter()
    launch_events = list(
        container.encounter_service.start_encounter(
            mode=os.getenv("TIME_MACHINE_MODAL_SMOKE_MODE", "strange"),
            coordinate_prompt=os.getenv(
                "TIME_MACHINE_MODAL_SMOKE_PROMPT",
                "a noisy workshop with a radio channel open",
            ),
            session_id="modal-smoke",
        )
    )
    launch_elapsed = perf_counter() - launch_started
    encounter_id = launch_events[0].metadata["encounter_id"]

    turn_started = perf_counter()
    turn_events = list(
        container.speech_orchestrator.handle_audio_turn(
            encounter_id=encounter_id,
            audio=str(audio_path),
        )
    )
    turn_elapsed = perf_counter() - turn_started

    transcript = _single_event(turn_events, TranscriptEvent).transcript
    response = _single_event(turn_events, ConversationTextEvent)
    audio = _single_event(turn_events, AudioEvent).audio

    print("Modal smoke passed.")
    print(f"Transcript: {transcript.text}")
    print(f"Reply: {response.text[:240]}")
    print(f"Audio: {audio.path} ({audio.duration_seconds}s)")
    print(f"Audio note: {audio.description}")
    print(f"Launch wall time: {launch_elapsed:.2f}s")
    print(f"Audio turn wall time: {turn_elapsed:.2f}s")
    print("Turn stage timings:")
    for timing in _event_timings(turn_events):
        print(f"  - {timing['stage']}: {timing['elapsed_ms'] / 1000:.2f}s")
    return 0


def _load_local_env(path: Path, *, override: bool = False) -> None:
    if not path.exists():
        return
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip().strip('"').strip("'")
        if key and (override or key not in os.environ):
            os.environ[key] = value


def _single_event(events: list[object], event_type: type) -> object:
    matches = [event for event in events if isinstance(event, event_type)]
    if not matches:
        raise RuntimeError(f"Expected event {event_type.__name__}, got {[type(e).__name__ for e in events]}")
    return matches[-1]


def _event_timings(events: list[object]) -> list[dict[str, object]]:
    timings: list[dict[str, object]] = []
    seen: set[tuple[str, float]] = set()
    for event in events:
        metadata = getattr(event, "metadata", {})
        for timing in metadata.get("timings", []):
            stage = timing.get("stage")
            elapsed_ms = timing.get("elapsed_ms")
            key = (stage, elapsed_ms)
            if key in seen:
                continue
            seen.add(key)
            timings.append(timing)
    return timings


if __name__ == "__main__":
    raise SystemExit(main())