From 11b3e97e4de43015d41877c772ff08151c86c8d6 Mon Sep 17 00:00:00 2001 From: Thanvish Date: Mon, 12 Oct 2026 00:50:39 +0530 Subject: [PATCH 1/2] cli: add --json option to --once snapshot for status bars and scripts Adds `--json` to `tracon --once` (or standalone `tracon --json`) to emit a structured JSON array of agent objects with id, description, state, flags, quiet seconds, and judge verdict, sorted by urgency. This enables integrations with status bars (tmux, sketchybar, Waybar) and external scripts. Co-Authored-By: Claude Sonnet 5 --- CLAUDE.md | 1 + README.md | 3 ++ tests/test_cli.py | 104 ++++++++++++++++++++++++++++++++++++++++++++++ tracon/cli.py | 35 ++++++++++++++-- 4 files changed, 139 insertions(+), 4 deletions(-) create mode 100644 tests/test_cli.py diff --git a/CLAUDE.md b/CLAUDE.md index 189c90d..0d30ca6 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 0a8b31d..10cb893 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 @@ -301,6 +302,8 @@ messages from your own user on macOS and Linux. | `--project PATH` | current dir | which project's sessions to watch | | `--session ID` | 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..68c5122 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,104 @@ +import json + +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(): + assert snapshot_json({}) == [] + assert snapshot_json([]) == [] + + +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) + + data = snapshot_json(statuses) + assert len(data) == 7 + + # Check that sort order matches sort_key: most urgent first (BLOCKED, IDLE, etc.) + states = [item["state"] for item in data] + assert states[0] == "BLOCKED" + assert states[1] == "IDLE" + assert states[-1] == "DONE" + + for item in data: + assert set(item.keys()) == {"id", "description", "state", "flags", "quiet_s", "judge"} + assert isinstance(item["id"], str) + assert isinstance(item["description"], str) + assert isinstance(item["state"], str) + assert isinstance(item["flags"], list) + assert all(isinstance(f, str) for f in item["flags"]) + assert isinstance(item["quiet_s"], int) + assert item["judge"] is None or isinstance(item["judge"], str) + + by_desc = {item["description"].split()[0]: item for item in data} + assert by_desc["#15"]["state"] == "BLOCKED" + assert by_desc["#15"]["judge"] == "on_track" + assert "LOOP" in by_desc["#16"]["flags"] + assert "STUCK" in by_desc["#16"]["flags"] + assert by_desc["#16"]["judge"] == "stuck" + assert "DRIFT" in by_desc["#13"]["flags"] + assert by_desc["#13"]["judge"] == "drifting" + 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 isinstance(data, list) + assert len(data) == 7 + blocked = [a for a in data if a["state"] == "BLOCKED"] + assert len(blocked) == 1 + assert blocked[0]["description"].startswith("#15") + + +def test_cli_json_implies_once(capsys, monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + rc = main(["--demo", "--json"]) + assert rc == 0 + captured = capsys.readouterr() + data = json.loads(captured.out) + assert len(data) == 7 + + +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) + + +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 + # Not valid JSON + try: + json.loads(captured.out) + assert False, "Should not be JSON" + except json.JSONDecodeError: + pass diff --git a/tracon/cli.py b/tracon/cli.py index e18dc2d..a4dd9af 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 @@ -13,7 +14,7 @@ from .lane import assess_all from .probe import Prober from .session import Session, find_sessions -from .signals import Thresholds, fmt_duration +from .signals import Status, Thresholds, fmt_duration def parse_at(s: str) -> datetime: @@ -38,6 +39,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) @@ -83,8 +85,8 @@ def main(argv: list[str] | None = None) -> int: judge = demo.judge 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) + if args.once or args.json: + 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 @@ -111,13 +113,38 @@ def main(argv: list[str] | None = None) -> int: return 0 -def print_snapshot(session: Session, at: datetime | None, th: Thresholds, judge: Judge | None = None, prober=None) -> int: +def snapshot_json(statuses: dict[str, Status] | list[Status]) -> list[dict]: + rows = sorted(statuses.values() if isinstance(statuses, dict) else statuses, key=lambda s: s.sort_key()) + return [ + { + "id": s.agent.id, + "description": s.agent.description, + "state": s.state, + "flags": [f.code for f in s.flags], + "quiet_s": round(s.age_s), + "judge": s.judgment.verdict 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(statuses), 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") From ea2427a77eb537b45d4984d69e1ee798af854257 Mon Sep 17 00:00:00 2001 From: Thanvish Date: Mon, 12 Oct 2026 01:44:46 +0530 Subject: [PATCH 2/2] cli: refine --once --json output shape and validation - Require `--once` when using `--json` via parser.error. - Nest flag objects with `code`, `severity`, and `detail` matching web view. - Add `severity` to each agent object for direct status bar styling. - Nest `judge` with `verdict` and `dismissed` flag. - Wrap output in top-level object containing `session`, `lane`, `as_of`, and `agents` list. - Keep `print_snapshot` definition on a single line. - Use `pytest.raises` for JSONDecodeError in tests. Co-Authored-By: Claude Sonnet 5 --- tests/test_cli.py | 85 ++++++++++++++++++++++++++++++----------------- tracon/cli.py | 55 +++++++++++++++--------------- 2 files changed, 83 insertions(+), 57 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index 68c5122..96bd383 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,4 +1,7 @@ import json +from datetime import datetime, timezone + +import pytest from tracon.cli import build_parser, main, snapshot_json from tracon.demo import Demo @@ -14,9 +17,12 @@ def test_build_parser_json_flag(): assert args.json is True -def test_snapshot_json_empty(): - assert snapshot_json({}) == [] - assert snapshot_json([]) == [] +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): @@ -26,33 +32,53 @@ def test_snapshot_json_structure(tmp_path): wts = [a.worktree for a in s.agents.values()] statuses, _ = assess_all(s, None, demo.prober.probe(wts), Thresholds(), demo.judge) - data = snapshot_json(statuses) - assert len(data) == 7 + 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 data] + states = [item["state"] for item in agents] assert states[0] == "BLOCKED" assert states[1] == "IDLE" assert states[-1] == "DONE" - for item in data: - assert set(item.keys()) == {"id", "description", "state", "flags", "quiet_s", "judge"} + 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) - assert all(isinstance(f, str) for f in item["flags"]) + 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) - assert item["judge"] is None or isinstance(item["judge"], str) + 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 data} + by_desc = {item["description"].split()[0]: item for item in agents} assert by_desc["#15"]["state"] == "BLOCKED" - assert by_desc["#15"]["judge"] == "on_track" - assert "LOOP" in by_desc["#16"]["flags"] - assert "STUCK" in by_desc["#16"]["flags"] - assert by_desc["#16"]["judge"] == "stuck" - assert "DRIFT" in by_desc["#13"]["flags"] - assert by_desc["#13"]["judge"] == "drifting" + 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 @@ -64,20 +90,21 @@ def test_cli_once_json(capsys, monkeypatch, tmp_path): assert captured.err == "" data = json.loads(captured.out) - assert isinstance(data, list) - assert len(data) == 7 - blocked = [a for a in data if a["state"] == "BLOCKED"] + 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_implies_once(capsys, monkeypatch, tmp_path): +def test_cli_json_without_once_errors(capsys, monkeypatch, tmp_path): monkeypatch.chdir(tmp_path) - rc = main(["--demo", "--json"]) - assert rc == 0 + with pytest.raises(SystemExit) as exc: + main(["--demo", "--json"]) + assert exc.value.code == 2 captured = capsys.readouterr() - data = json.loads(captured.out) - assert len(data) == 7 + assert "--json needs --once" in captured.err def test_cli_once_json_no_judge(capsys, monkeypatch, tmp_path): @@ -86,7 +113,7 @@ def test_cli_once_json_no_judge(capsys, monkeypatch, tmp_path): assert rc == 0 captured = capsys.readouterr() data = json.loads(captured.out) - assert all(a["judge"] is None for a in data) + assert all(a["judge"] is None for a in data["agents"]) def test_cli_once_text_not_json(capsys, monkeypatch, tmp_path): @@ -96,9 +123,5 @@ def test_cli_once_text_not_json(capsys, monkeypatch, tmp_path): captured = capsys.readouterr() assert "lane: checkout-v2" in captured.out assert "BLOCKED" in captured.out - # Not valid JSON - try: + with pytest.raises(json.JSONDecodeError): json.loads(captured.out) - assert False, "Should not be JSON" - except json.JSONDecodeError: - pass diff --git a/tracon/cli.py b/tracon/cli.py index a4dd9af..50f1350 100644 --- a/tracon/cli.py +++ b/tracon/cli.py @@ -58,7 +58,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 = None if args.demo: from .demo import Demo @@ -85,7 +88,7 @@ def main(argv: list[str] | None = None) -> int: judge = demo.judge 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 or args.json: + if args.once: 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 @@ -113,29 +116,29 @@ def main(argv: list[str] | None = None) -> int: return 0 -def snapshot_json(statuses: dict[str, Status] | list[Status]) -> list[dict]: - rows = sorted(statuses.values() if isinstance(statuses, dict) else statuses, key=lambda s: s.sort_key()) - return [ - { - "id": s.agent.id, - "description": s.agent.description, - "state": s.state, - "flags": [f.code for f in s.flags], - "quiet_s": round(s.age_s), - "judge": s.judgment.verdict 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: +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: @@ -143,7 +146,7 @@ def print_snapshot( probes = (prober or Prober()).probe(wts) statuses, _ = assess_all(session, at, probes, th, judge) if as_json: - print(json.dumps(snapshot_json(statuses), indent=2)) + 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")