example_usage.py

"""
Live demonstration and measurement harness.

Runs the agent against real sites and reports, per scenario: step count,
wall-clock time, model calls, USD cost, pages visited, and the answer.

    export ANTHROPIC_API_KEY=sk-ant-...
    .venv/bin/python example_usage.py

Environment:
    AIL_HEADLESS=1   run without opening browser windows
    AIL_ONLY=d,e     run only the named scenarios
"""

from __future__ import annotations

import asyncio
import os
import sys
import time
from dataclasses import dataclass, field

sys.path.insert(0, os.path.dirname(__file__))

from controller import AgentConfig, AgentController  # noqa: E402
from session import SessionConfig, SessionManager  # noqa: E402

HEADLESS = os.environ.get("AIL_HEADLESS") == "1"

BOOKS = "https://books.toscrape.com"
QUOTES = "https://quotes.toscrape.com"
WIKI_TOKYO = "https://en.wikipedia.org/wiki/Tokyo"
HACKER_NEWS = "https://news.ycombinator.com"


@dataclass
class Scenario:
    key: str
    title: str
    url: str
    goal: str
    conversation: str
    note: str = ""
    expect: str = ""


SCENARIOS: list[Scenario] = [
    Scenario(
        "a", "Fresh session — cheapest in a category", BOOKS,
        "find the cheapest book in the Mystery category and return its title and price",
        "demo-001",
        note="Needs category navigation, then a min across a paginated list.",
    ),
    Scenario(
        "b", "Session reuse — same conversation, new goal", BOOKS,
        "find the most expensive book on the site and return its title and price",
        "demo-001",
        note="Same conversation_id as (a) — the browser context is reused.",
    ),
    Scenario(
        "c", "Memory hit — new conversation, same goal", BOOKS,
        "find the cheapest book in the Mystery category and return its title and price",
        "demo-002",
        note="New conversation; SiteMemory may supply a route learned in (a).",
    ),
    Scenario(
        "d", "Tag-filtered text extraction", QUOTES,
        "find the first quote tagged with 'life' and return the quote text and its author",
        "demo-003",
        note="Non-numeric goal — grounding falls back to word overlap.",
    ),
    Scenario(
        "e", "Dense article, decoy number present", WIKI_TOKYO,
        "You are already on the Tokyo Wikipedia article. Find and return the "
        "current population of Tokyo as stated on this page. Do not navigate "
        "away or search — the answer is on this page.",
        "demo-004",
        note="Should answer from the landing page with no navigation at all.",
    ),
    Scenario(
        "f", "Live ranked list, bare integers", HACKER_NEWS,
        "find the title and score of the highest-scored post on the front page",
        "demo-005",
        note="Scores carry no currency symbol — invisible to the old tracker.",
    ),
]


@dataclass
class Result:
    scenario: Scenario
    answer: str = ""
    steps: int = 0
    seconds: float = 0.0
    calls: int = 0
    cost: float = 0.0
    pages: int = 0
    status: str = "ok"
    models: dict = field(default_factory=dict)


async def run_one(scenario: Scenario, manager: SessionManager) -> Result:
    controller = AgentController(
        manager,
        config=AgentConfig(max_steps=15, max_seconds=180.0, humanize=not HEADLESS),
    )
    session_id = SessionManager.make_session_id(
        scenario.url, scenario.goal, scenario.conversation
    )

    print(f"\n{'=' * 72}")
    print(f"  ({scenario.key}) {scenario.title}")
    print(f"{'=' * 72}")
    print(f"  url          : {scenario.url}")
    print(f"  goal         : {scenario.goal[:100]}")
    print(f"  conversation : {scenario.conversation}  ->  session {session_id}")
    if scenario.note:
        print(f"  why          : {scenario.note}")

    hint = controller.memory.get_hint(scenario.url, scenario.goal)
    print(f"  memory       : {hint or '(none yet)'}")
    print("  running...", flush=True)

    result = Result(scenario=scenario)
    started = time.monotonic()
    try:
        result.answer = await asyncio.wait_for(
            controller.run(scenario.url, scenario.goal, session_id), timeout=300
        )
    except asyncio.TimeoutError:
        result.status = "wall-clock timeout"
        result.answer = "(exceeded 300s)"
    except Exception as exc:  # noqa: BLE001 — a demo must survive any failure
        result.status = f"error: {type(exc).__name__}"
        result.answer = str(exc)[:200]

    result.seconds = time.monotonic() - started
    result.steps = controller.state.step
    result.calls = controller.llm.usage.calls
    result.cost = controller.llm.usage.cost_usd
    result.pages = controller.observations.unique_urls
    result.models = dict(controller.llm.usage.by_model)

    print(f"\n  ANSWER: {result.answer}")
    print(
        f"  [{result.steps} steps · {result.seconds:.1f}s · {result.calls} calls · "
        f"${result.cost:.4f} · {result.pages} pages]"
    )
    return result


def summarise(results: list[Result]) -> None:
    print(f"\n{'=' * 88}")
    print("  SUMMARY")
    print(f"{'=' * 88}")
    header = f"  {'':2} {'scenario':<34} {'steps':>5} {'time':>7} {'calls':>6} {'cost':>9} {'pages':>6}"
    print(header)
    print(f"  {'-' * 84}")
    for r in results:
        print(
            f"  {r.scenario.key:<2} {r.scenario.title[:34]:<34} "
            f"{r.steps:>5} {r.seconds:>6.1f}s {r.calls:>6} "
            f"${r.cost:>8.4f} {r.pages:>6}"
        )
    print(f"  {'-' * 84}")
    print(
        f"  {'':2} {'TOTAL':<34} {sum(r.steps for r in results):>5} "
        f"{sum(r.seconds for r in results):>6.1f}s "
        f"{sum(r.calls for r in results):>6} "
        f"${sum(r.cost for r in results):>8.4f} "
        f"{sum(r.pages for r in results):>6}"
    )

    failures = [r for r in results if r.status != "ok"]
    if failures:
        print("\n  Non-clean finishes:")
        for r in failures:
            print(f"    ({r.scenario.key}) {r.status}")

    models: dict[str, int] = {}
    for r in results:
        for model, count in r.models.items():
            models[model] = models.get(model, 0) + count
    print(f"\n  Model calls: {models}")


async def main() -> None:
    if not os.environ.get("ANTHROPIC_API_KEY"):
        print("ANTHROPIC_API_KEY is not set — nothing to run.")
        return

    only = {k.strip() for k in os.environ.get("AIL_ONLY", "").split(",") if k.strip()}
    scenarios = [s for s in SCENARIOS if not only or s.key in only]

    print(f"Agent Interaction Layer — live run ({len(scenarios)} scenarios)")
    print(f"Browser: {'headless' if HEADLESS else 'visible'}")

    manager = SessionManager(SessionConfig(headless=HEADLESS))
    results: list[Result] = []
    try:
        for scenario in scenarios:
            results.append(await run_one(scenario, manager))
    finally:
        await manager.close_all()

    summarise(results)


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