watch_model_until_failure.py

#!/usr/bin/env python3
"""Poll an LLM model until the CLI command fails."""

from __future__ import annotations

import argparse
import subprocess
import sys
import time
from datetime import datetime, timezone


def timestamp() -> str:
    return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Run `uv run llm` repeatedly until it exits with a non-zero status."
    )
    parser.add_argument(
        "-m",
        "--model",
        default="claude-fable-5",
        help="Model name to poll. Defaults to claude-fable-5.",
    )
    parser.add_argument(
        "-i",
        "--interval",
        type=float,
        default=60.0,
        help="Seconds to wait between successful calls. Defaults to 60.",
    )
    parser.add_argument(
        "prompt",
        nargs="?",
        default="hi",
        help="Prompt to send. Defaults to 'hi'.",
    )
    args = parser.parse_args()

    attempt = 1
    while True:
        started_at = timestamp()
        command = ["uv", "run", "llm", "-m", args.model, args.prompt]
        print(
            f"[{started_at}] attempt {attempt}: running {' '.join(command)}",
            flush=True,
        )

        completed = subprocess.run(command, text=True, capture_output=True)
        finished_at = timestamp()

        if completed.returncode != 0:
            print(
                f"[{finished_at}] FAILED after attempt {attempt} "
                f"with exit code {completed.returncode}",
                flush=True,
            )
            if completed.stdout:
                print("\nstdout:\n" + completed.stdout.rstrip(), flush=True)
            if completed.stderr:
                print("\nstderr:\n" + completed.stderr.rstrip(), flush=True)
            return completed.returncode

        first_line = completed.stdout.strip().splitlines()[0:1]
        if first_line:
            print(f"[{finished_at}] success: {first_line[0]}", flush=True)
        else:
            print(f"[{finished_at}] success: no output", flush=True)

        attempt += 1
        time.sleep(args.interval)


if __name__ == "__main__":
    sys.exit(main())