"""End-to-end test of the pre-warmed agent pool (voice/agent_pool.py).

Proves the three things that were broken:
  1. PRE-WARMED    — a spare connects at boot; dispatch on it is ~instant.
  2. SUPERVISED    — live task_progress events while it works; the final
                     report is delivered on the inbox rail by itself.
  3. KILLABLE      — kill() ends the session and clears the chip at once.

Run:  .venv/bin/python test_agent_pool.py
"""
from __future__ import annotations

import asyncio
import time

from voice import agent_pool
from voice.agent_pool import pool

EVENTS: list[tuple[str, dict]] = []


def _spy(event_type: str, **data):
    if event_type.startswith("task_"):
        EVENTS.append((event_type, data))
        print(f"    EVT {event_type} {data}")


async def main() -> int:
    # spy on the event rail (what the HUD/registry see)
    import voice.events as events
    events.emit = _spy
    agent_pool.emit = _spy

    inbox: asyncio.Queue[str] = asyncio.Queue()
    pool.report_sink = inbox.put_nowait

    print("1) WARMING a spare…")
    t0 = time.monotonic()
    pool.start()
    while not pool._spares and time.monotonic() - t0 < 90:
        await asyncio.sleep(0.5)
    assert pool._spares, "no warm spare after 90s"
    warm_s = time.monotonic() - t0
    print(f"   spare connected in {warm_s:.1f}s (paid ONCE, at boot)\n")

    print("2) DISPATCH on the warm spare…")
    t0 = time.monotonic()
    out = await pool.dispatch(
        "Run `echo jarvis-pool-ok` with Bash and report exactly what it "
        "printed. Nothing else.", "pool smoke test")
    disp_s = time.monotonic() - t0
    print(f"   dispatch returned in {disp_s:.2f}s -> {out}\n")
    assert disp_s < 3.0, f"dispatch took {disp_s:.1f}s — not instant"
    assert "pre-warmed" in out

    print("   (a replacement spare warms in the background — N+1 always ready)")

    print("\n3) SUPERVISION — waiting for the report to arrive by itself…")
    report = await asyncio.wait_for(inbox.get(), timeout=180)
    total = time.monotonic() - t0
    print(f"\n   REPORT ({total:.0f}s): {report[:300]}\n")

    kinds = [e for e, _ in EVENTS]
    assert "task_started" in kinds, "no task_started event"
    assert "task_progress" in kinds, "NO LIVE PROGRESS — the whole point"
    assert "task_done" in kinds, "chip never cleared"
    assert "jarvis-pool-ok" in report, f"worker didn't do the job: {report[:200]}"
    print(f"   events seen: {kinds}")
    print("   ✓ progress streamed live, report delivered, chip cleared")

    print("\n4) KILL — dispatch a long one, then kill it…")
    await pool.dispatch("Sleep for 10 minutes using `sleep 600` via Bash, "
                        "then say done.", "long sleeper")
    await asyncio.sleep(2)
    assert pool._running, "worker not running"
    killed = await pool.kill("latest")
    print(f"   {killed}")
    assert not pool._running, "worker still running after kill"
    assert EVENTS[-1][0] == "task_done", "chip not cleared on kill"
    print("   ✓ killed instantly, chip cleared in the same breath")

    await pool.stop()
    print("\nALL GOOD — pre-warmed, supervised, killable.")
    return 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))