diff --git a/CLAUDE.md b/CLAUDE.md index 4146d63..4d4aeca 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -9,6 +9,7 @@ web page. Python 3.12+, managed with uv. ```sh uv run pytest # all tests; fast, offline, no Claude Code needed uv run tracon --once --no-judge # snapshot of the newest session, no TUI +uv run tracon --once --json --no-judge # snapshot as JSON (for status bars / scripts) uv run tracon --once --no-judge --at "HH:MM" # replay a moment (needs a real session) uv run tracon web --no-judge # browser view only uv run tracon --demo # made-up run; no session needed diff --git a/README.md b/README.md index bfa543c..4ea93de 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,7 @@ tracon --list # recent sessions that spawned subagents tracon -s 4f2a # pick a session by id prefix tracon --at "2026-10-11 07:20" # replay a session as it stood at that time tracon --once # print one snapshot and exit +tracon --once --json # snapshot as JSON (for scripts and status bars) ``` A good layout: your orchestrating session on one side, `tracon` in a split next to it @@ -313,6 +314,8 @@ messages from your own user on macOS and Linux. | `--project PATH` | current dir | which project's sessions to watch (its parent folders too, without this) | | `--session ID` | next split, else newest | session id or prefix | | `--at TIME` | live | replay as of a local time (`07:20` or `2026-10-11 07:20`) | +| `--once` | | print one snapshot and exit (no TUI) | +| `--json` | | output snapshot as JSON (for scripts and status bars) | | `--idle MIN` | 5 | silence before IDLE | | `--stall MIN` | 20 | running without change before STALLED | | `--notify` | off | desktop notifications when an agent needs you (`osascript` on macOS, `notify-send` on Linux) | diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..96bd383 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,127 @@ +import json +from datetime import datetime, timezone + +import pytest + +from tracon.cli import build_parser, main, snapshot_json +from tracon.demo import Demo +from tracon.lane import assess_all +from tracon.session import Session +from tracon.signals import Thresholds + + +def test_build_parser_json_flag(): + p = build_parser() + args = p.parse_args(["--once", "--json"]) + assert args.once is True + assert args.json is True + + +def test_snapshot_json_empty(tmp_path): + demo = Demo(root=str(tmp_path)) + s = Session(demo.ref) + data = snapshot_json(s, {}) + assert data["session"] == s.ref.session_id + assert data["agents"] == [] + + +def test_snapshot_json_structure(tmp_path): + demo = Demo(root=str(tmp_path)) + s = Session(demo.ref) + s.refresh() + wts = [a.worktree for a in s.agents.values()] + statuses, _ = assess_all(s, None, demo.prober.probe(wts), Thresholds(), demo.judge) + + at_dt = datetime(2026, 10, 12, 1, 0, tzinfo=timezone.utc) + data = snapshot_json(s, statuses, at=at_dt) + assert set(data.keys()) == {"session", "lane", "as_of", "agents"} + assert data["session"] == s.ref.session_id + assert data["lane"] == "checkout-v2" + assert data["as_of"] == at_dt.astimezone().isoformat() + + agents = data["agents"] + assert len(agents) == 7 + + # Check that sort order matches sort_key: most urgent first (BLOCKED, IDLE, etc.) + states = [item["state"] for item in agents] + assert states[0] == "BLOCKED" + assert states[1] == "IDLE" + assert states[-1] == "DONE" + + for item in agents: + assert set(item.keys()) == {"id", "description", "state", "severity", "flags", "quiet_s", "judge"} + assert isinstance(item["id"], str) + assert isinstance(item["description"], str) + assert isinstance(item["state"], str) + assert isinstance(item["severity"], str) + assert isinstance(item["flags"], list) + for f in item["flags"]: + assert set(f.keys()) == {"code", "severity", "detail"} + assert isinstance(f["code"], str) + assert isinstance(f["severity"], str) + assert isinstance(f["detail"], str) + assert isinstance(item["quiet_s"], int) + if item["judge"] is not None: + assert set(item["judge"].keys()) == {"verdict", "dismissed"} + assert isinstance(item["judge"]["verdict"], str) + assert isinstance(item["judge"]["dismissed"], bool) + + by_desc = {item["description"].split()[0]: item for item in agents} + assert by_desc["#15"]["state"] == "BLOCKED" + assert by_desc["#15"]["severity"] == "bad" + assert by_desc["#15"]["judge"] == {"verdict": "on_track", "dismissed": False} + + flag_codes_16 = {f["code"] for f in by_desc["#16"]["flags"]} + assert "LOOP" in flag_codes_16 + assert "STUCK" in flag_codes_16 + assert by_desc["#16"]["judge"] == {"verdict": "stuck", "dismissed": False} + + flag_codes_13 = {f["code"] for f in by_desc["#13"]["flags"]} + assert "DRIFT" in flag_codes_13 + assert by_desc["#13"]["judge"] == {"verdict": "drifting", "dismissed": False} + assert by_desc["#17"]["judge"] is None # not judged in demo + + +def test_cli_once_json(capsys, monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + rc = main(["--demo", "--once", "--json"]) + assert rc == 0 + captured = capsys.readouterr() + assert captured.err == "" + + data = json.loads(captured.out) + assert set(data.keys()) == {"session", "lane", "as_of", "agents"} + agents = data["agents"] + assert len(agents) == 7 + blocked = [a for a in agents if a["state"] == "BLOCKED"] + assert len(blocked) == 1 + assert blocked[0]["description"].startswith("#15") + + +def test_cli_json_without_once_errors(capsys, monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + with pytest.raises(SystemExit) as exc: + main(["--demo", "--json"]) + assert exc.value.code == 2 + captured = capsys.readouterr() + assert "--json needs --once" in captured.err + + +def test_cli_once_json_no_judge(capsys, monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + rc = main(["--demo", "--once", "--json", "--no-judge"]) + assert rc == 0 + captured = capsys.readouterr() + data = json.loads(captured.out) + assert all(a["judge"] is None for a in data["agents"]) + + +def test_cli_once_text_not_json(capsys, monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + rc = main(["--demo", "--once"]) + assert rc == 0 + captured = capsys.readouterr() + assert "lane: checkout-v2" in captured.out + assert "BLOCKED" in captured.out + with pytest.raises(json.JSONDecodeError): + json.loads(captured.out) diff --git a/tracon/cli.py b/tracon/cli.py index 21ce8a9..1279d25 100644 --- a/tracon/cli.py +++ b/tracon/cli.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse +import json import os import sys from datetime import datetime, timezone @@ -14,7 +15,7 @@ from .lane import assess_all from .probe import Prober from .session import Chooser, Session, find_sessions -from .signals import Thresholds, fmt_duration +from .signals import Status, Thresholds, fmt_duration def parse_at(s: str) -> datetime: @@ -39,6 +40,7 @@ def build_parser() -> argparse.ArgumentParser: p.add_argument("--at", help="replay as of this local time, e.g. '2026-10-11 01:30' or '01:30'") p.add_argument("--list", action="store_true", help="list recent sessions that have subagents") p.add_argument("--once", action="store_true", help="print one snapshot and exit (no TUI)") + p.add_argument("--json", action="store_true", help="output snapshot as JSON (with --once)") # Off by default: they're sent through osascript, so macOS shows them as coming from Script Editor. p.add_argument("--notify", action="store_true", help="desktop notifications when an agent needs you (off by default)") p.add_argument("--no-notify", action="store_true", help=argparse.SUPPRESS) @@ -57,7 +59,10 @@ def build_parser() -> argparse.ArgumentParser: def main(argv: list[str] | None = None) -> int: - args = build_parser().parse_args(argv) + parser = build_parser() + args = parser.parse_args(argv) + if args.json and not args.once: + parser.error("--json needs --once") demo = chooser = None if args.demo: from .demo import Demo @@ -87,7 +92,7 @@ def main(argv: list[str] | None = None) -> int: elif not args.no_judge: judge = Judge(model=args.judge_model, backend=args.judge_backend, budget_usd=args.judge_budget, min_interval_s=args.judge_every * 60) if args.once: - return print_snapshot(Session(refs[0]), at, th, judge, demo.prober if demo else None) + return print_snapshot(Session(refs[0]), at, th, judge, demo.prober if demo else None, as_json=args.json) if args.command == "web": import webbrowser @@ -120,13 +125,38 @@ def _write(seq: str) -> None: sys.stdout.flush() -def print_snapshot(session: Session, at: datetime | None, th: Thresholds, judge: Judge | None = None, prober=None) -> int: +def snapshot_json(session: Session, statuses: dict[str, Status], at: datetime | None = None) -> dict: + rows = sorted(statuses.values(), key=lambda s: s.sort_key()) + now = at or datetime.now(timezone.utc) + return { + "session": session.ref.session_id, + "lane": session.lane(), + "as_of": now.astimezone().isoformat(), + "agents": [ + { + "id": s.agent.id, + "description": s.agent.description, + "state": s.state, + "severity": s.severity, + "flags": [{"code": f.code, "severity": f.severity, "detail": f.detail} for f in s.flags], + "quiet_s": round(s.age_s), + "judge": {"verdict": s.judgment.verdict, "dismissed": s.dismissed} if s.judgment else None, + } + for s in rows + ], + } + + +def print_snapshot(session: Session, at: datetime | None, th: Thresholds, judge: Judge | None = None, prober=None, as_json: bool = False) -> int: session.refresh() probes = {} if at is None: wts = [a.worktree for a in session.agents.values() if a.worktree] probes = (prober or Prober()).probe(wts) statuses, _ = assess_all(session, at, probes, th, judge) + if as_json: + print(json.dumps(snapshot_json(session, statuses, at), indent=2)) + return 0 rows = sorted(statuses.values(), key=lambda s: s.sort_key()) when = (at or datetime.now(timezone.utc)).astimezone().strftime("%Y-%m-%d %H:%M") print(f"lane: {session.lane() or '?'} session {session.ref.session_id[:8]} as of {when}\n")