"""Read-only terminal stream for the factory SQLite trace.""" from __future__ import annotations import argparse import json import sqlite3 import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Any, TextIO from .control import _resolve_db_from_config def open_readonly(db_path: str | Path) -> sqlite3.Connection: """Open an existing WAL database without creating or mutating anything.""" uri = Path(db_path).resolve().as_uri() + "?mode=ro" conn = sqlite3.connect(uri, uri=True, isolation_level=None, timeout=5.0) conn.row_factory = sqlite3.Row conn.execute("PRAGMA busy_timeout=5000") conn.execute("PRAGMA query_only=ON") return conn def select_session(conn: sqlite3.Connection, adw_id: str | None) -> tuple[sqlite3.Row, str]: if adw_id: row = conn.execute("SELECT * FROM sessions WHERE adw_id=?", (adw_id,)).fetchone() if row is None: raise ValueError(f"unknown adw_id {adw_id}") return row, "requested explicitly" row = conn.execute( "SELECT * FROM sessions WHERE status='running' AND ended_at IS NULL " "ORDER BY started_at DESC, rowid DESC LIMIT 1" ).fetchone() if row is not None: return row, "most recent running session" row = conn.execute("SELECT * FROM sessions ORDER BY started_at DESC, rowid DESC LIMIT 1").fetchone() if row is None: raise ValueError("trace database contains no sessions") return row, "no session is running; showing the most recent session history" def _json(value: str | None, fallback: Any) -> Any: if value is None: return fallback try: return json.loads(value) except (TypeError, json.JSONDecodeError): return fallback def _elapsed(started_at: str | None, ended_at: str | None) -> str: if not started_at: return "unknown" try: start = datetime.fromisoformat(started_at.replace("Z", "+00:00")) end = (datetime.fromisoformat(ended_at.replace("Z", "+00:00")) if ended_at else datetime.now(timezone.utc)) return f"{max(0.0, (end - start).total_seconds()):.1f}s" except (TypeError, ValueError): return "unknown" def _artifact(path_value: Any) -> dict[str, Any] | None: if not isinstance(path_value, str) or not path_value: return None result: dict[str, Any] = {"path": path_value} try: result["status"] = "present" if Path(path_value).exists() else "absent" except OSError as exc: result["status"] = "absent" result["error"] = str(exc) return result def event_object(conn: sqlite3.Connection, row: sqlite3.Row) -> dict[str, Any]: payload = _json(row["payload_json"], {}) obj: dict[str, Any] = { "kind": "event", "rowid": row["rowid"], "event_id": row["event_id"], "adw_id": row["adw_id"], "phase_id": row["phase_id"], "type": row["type"], "name": row["name"], "started_at": row["started_at"], "ended_at": row["ended_at"], "tokens": row["tokens"] if row["tokens"] is not None else "unknown", "payload": payload, } if row["phase_id"]: phase = conn.execute( "SELECT name,status,error,started_at,ended_at FROM phases WHERE phase_id=?", (row["phase_id"],), ).fetchone() if phase: obj["phase"] = {"name": phase["name"], "status": phase["status"], "error": phase["error"], "elapsed": _elapsed(phase["started_at"], phase["ended_at"]), "open": phase["ended_at"] is None} artifacts = [] if isinstance(payload, dict): for key in ("output_artifact", "raw_output_path"): item = _artifact(payload.get(key)) if item: artifacts.append(item) for path in payload.get("artifacts", []) if isinstance(payload.get("artifacts"), list) else []: item = _artifact(path) if item: artifacts.append(item) if artifacts: obj["artifacts"] = artifacts return obj def human_line(obj: dict[str, Any]) -> str: payload = obj["payload"] if isinstance(obj["payload"], dict) else {} phase = obj.get("phase", {}) prefix = obj.get("started_at") or "unknown-time" if phase: prefix += f" [{phase['name']} {phase['elapsed']}{' OPEN' if phase['open'] else ''}]" typ, name = obj["type"], obj["name"] or "" if typ == "phase_start": detail = f"phase START {name}: {payload.get('description') or payload.get('owner') or ''}".rstrip() elif typ == "phase_end": detail = f"phase {str(payload.get('status', 'unknown')).upper()} {name} elapsed={phase.get('elapsed', 'unknown')}" elif typ == "agent_start": detail = f"agent {name} START model={payload.get('model', 'unknown')} purpose={payload.get('purpose', 'unknown')}" elif typ == "agent_end": detail = f"agent {name} END tokens={obj['tokens']} cost={payload.get('cost', 'unknown')}" elif typ == "tool_call": detail = f"agent {payload.get('agent', 'unknown')} tool {name}" if "command" in payload: detail += f" command={payload['command']} exit={payload.get('returncode', 'unknown')} passed={payload.get('passed', 'unknown')}" elif "args" in payload: detail += f" args={json.dumps(payload['args'], ensure_ascii=False)}" if payload.get("ok") is False and payload.get("result_snippet"): detail += f" error={payload['result_snippet']}" elif typ == "tool_call_start": detail = f"agent {payload.get('agent', 'unknown')} tool {name} RUNNING" if "command" in payload: detail += f" command={payload['command']}" elif payload.get("args"): detail += f" args={json.dumps(payload['args'], ensure_ascii=False)}" elif typ == "agent_attempt_start": attempt = payload.get("attempt", {}) detail = f"agent {payload.get('agent', 'unknown')} attempt START command={attempt.get('command', 'unknown')}" elif typ == "agent_attempt_end": attempt = payload.get("attempt", {}) outcome = "TIMEOUT" if attempt.get("timedOut") else f"exit={attempt.get('returncode', 'unknown')}" detail = (f"agent {payload.get('agent', 'unknown')} attempt END {outcome}" f" tokens={attempt.get('tokens') if attempt.get('tokens') is not None else 'unknown'}") provider_failure = attempt.get("providerFailure") if isinstance(attempt, dict) else None usage = attempt.get("usage") if isinstance(attempt, dict) else None if attempt.get("signal"): detail += f" signal={attempt['signal']}" if attempt.get("error"): detail += f" error={attempt['error']}" if isinstance(provider_failure, dict): detail += f" provider_failure={provider_failure.get('kind', 'unknown')}" if provider_failure.get("resume_at"): detail += f" resume_at={provider_failure['resume_at']}" if isinstance(usage, dict) and usage.get("usage_estimated"): detail += (f" context_estimate={usage.get('total_tokens', 'unknown')}/" f"{usage.get('context_window', 'unknown')}" f" billing={usage.get('billing_status', 'unknown')}") if attempt.get("stderrPath"): detail += f" stderr={attempt['stderrPath']}" elif typ in ("gate_pass", "gate_fail"): verdict = "PASS" if typ == "gate_pass" else "FAIL" checks = payload.get("checks", []) checked = "; ".join(f"{c.get('item')}: {'ok' if c.get('ok') else 'FAIL'}{(' — ' + str(c.get('note'))) if c.get('note') else ''}" for c in checks if isinstance(c, dict)) actual = payload.get("violations") or payload.get("reason") detail = f"gate {name} {verdict}" if checked: detail += f" checked=[{checked}]" if actual: detail += f" error={actual}" elif typ == "error": detail = f"ERROR {name}: {payload.get('error') or payload.get('reason') or payload}" elif typ == "handoff": detail = f"agent {name} handoff: {payload.get('summary', 'unknown')}" else: detail = f"{typ} {name}: {json.dumps(payload, ensure_ascii=False)}" for artifact in obj.get("artifacts", []): detail += f" artifact={artifact['path']} ({artifact['status']})" return f"{prefix} {detail}" def watch(conn: sqlite3.Connection, adw_id: str | None = None, *, follow: bool = True, since_start: bool = False, json_output: bool = False, poll_ms: int = 500, output: TextIO = sys.stdout, _ready: Any = None) -> int: session, reason = select_session(conn, adw_id) chosen = session["adw_id"] finished_fallback = adw_id is None and reason.startswith("no session") if finished_fallback: follow, since_start = False, True selection = {"kind": "session", "adw_id": chosen, "reason": reason, "status": session["status"], "started_at": session["started_at"], "ended_at": session["ended_at"]} print(json.dumps(selection, ensure_ascii=False) if json_output else f"factory watch: session {chosen} — {reason} (status={session['status']})", file=output, flush=True) cursor = 0 if not since_start: cursor = conn.execute("SELECT COALESCE(MAX(rowid),0) FROM events WHERE adw_id=?", (chosen,)).fetchone()[0] if _ready is not None: _ready.set() try: while True: rows = conn.execute( "SELECT rowid,event_id,adw_id,phase_id,parent_id,type,name,payload_json,tokens,started_at,ended_at " "FROM events WHERE adw_id=? AND rowid>? ORDER BY rowid LIMIT 500", (chosen, cursor) ).fetchall() for row in rows: obj = event_object(conn, row) print(json.dumps(obj, ensure_ascii=False) if json_output else human_line(obj), file=output, flush=True) cursor = row["rowid"] current = conn.execute("SELECT ended_at FROM sessions WHERE adw_id=?", (chosen,)).fetchone() if not follow or (current and current["ended_at"] is not None and not rows): return 0 time.sleep(poll_ms / 1000) except KeyboardInterrupt: return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="factory watch", description="Stream a factory run's trace") parser.add_argument("adw_id", nargs="?", help="Session id (defaults to latest running)") parser.add_argument("--config", help="Path to sssf.config.yaml") parser.add_argument("--preset") parser.add_argument("--follow", action=argparse.BooleanOptionalAction, default=True) parser.add_argument("--since-start", action="store_true", help="Print history before following") parser.add_argument("--json", action="store_true", help="Emit one JSON object per line") parser.add_argument("--poll-ms", type=int, default=500, metavar="N") return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) if args.poll_ms < 1: print("factory watch: --poll-ms must be positive", file=sys.stderr) return 2 config = args.config or str(Path(__file__).resolve().parents[1] / "sssf.config.yaml") try: conn = open_readonly(_resolve_db_from_config(config, args.preset)) try: return watch(conn, args.adw_id, follow=args.follow, since_start=args.since_start, json_output=args.json, poll_ms=args.poll_ms) finally: conn.close() except (OSError, sqlite3.Error, ValueError) as exc: print(f"factory watch: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())