Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 3 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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) |
Expand Down
127 changes: 127 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
@@ -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)
38 changes: 34 additions & 4 deletions tracon/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

import argparse
import json
import os
import sys
from datetime import datetime, timezone
Expand All @@ -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:
Expand All @@ -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)
Expand All @@ -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
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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")
Expand Down
Loading