scripts/agent_e2e_smoke.py

"""Throwaway smoke: drive the real agent harness against the live llama.cpp server.

Loads a fixture session and runs one SPEAK turn and one ACT turn through
``run_agent_turn_streaming``, printing streamed tokens, tool activity, and the
final TurnResult. Proves the full agent-forward pipeline end-to-end.

    $env:PYTHONPATH="src"; & ./.venv/Scripts/python.exe scripts/agent_e2e_smoke.py
"""

from __future__ import annotations

import asyncio
from pathlib import Path

from loosecanvas.agent_harness import (
    FinalEvent,
    TokenEvent,
    ToolActivityEvent,
    run_agent_turn_streaming,
)
from loosecanvas.turn_logic import load_fixture_session

_SMALL = Path(__file__).resolve().parents[1] / "fixtures" / "small_graph.json"


async def _turn(sid: str, msg: str) -> None:
    print(f"\n{'=' * 70}\nUSER: {msg}\n{'-' * 70}")
    async for ev in run_agent_turn_streaming(sid, {}, msg):
        if isinstance(ev, TokenEvent):
            print(ev.text, end="", flush=True)
        elif isinstance(ev, ToolActivityEvent):
            print(f"\n  [activity: {ev.message}]", flush=True)
        elif isinstance(ev, FinalEvent):
            tr = ev.turn_result
            print(f"\n{'-' * 70}")
            if tr is None:
                print("FINAL: pure-speech turn (no canvas patch)")
            else:
                ops = len(tr.renderer_patch.operations)
                print(f"FINAL: status={tr.status} ops={ops} warnings={tr.warnings}")


async def main() -> None:
    sid, _ = await load_fixture_session(fixture_path=_SMALL)
    print(f"session={sid}")
    await _turn(sid, "Hi — in one sentence, what can you help me do here?")
    await _turn(sid, "What's on the canvas right now?")
    await _turn(sid, "Reveal the node about learning rate, please.")


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