391 lines
15 KiB
Python
Executable File
391 lines
15 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# #region Scripts.PrettyCoT [C:2] [TYPE Script] [SEMANTICS logging,pretty,agent,cli]
|
|
# @BRIEF Agent-first pretty printer and trace visualizer for Molecular CoT JSON logs.
|
|
# @RELATION IMPLEMENTS -> [molecular-cot-logging:CLI Reader]
|
|
# @NOTE
|
|
# python scripts/pretty_cot.py backend/logs/app.log --last 80
|
|
# tail -f logs/app.log | python scripts/pretty_cot.py
|
|
# python scripts/pretty_cot.py backend/logs/app.log --follow
|
|
# python scripts/pretty_cot.py backend/logs/app.log* --stats [--json]
|
|
# python scripts/pretty_cot.py backend/logs/app.log* --digest [--top 20] [--json]
|
|
# python scripts/pretty_cot.py backend/logs/app.log* --trajectory ScenarioExecution.BrowserProvider
|
|
# python scripts/pretty_cot.py backend/logs/app.log --trace <id> --story
|
|
# @INVARIANT Output is always a readable narrative grouped by trace_id.
|
|
# @RELATION DEPENDS_ON -> [Core.LogStats]
|
|
"""
|
|
pretty_cot.py — Agent-centric pretty printer for Molecular CoT logs.
|
|
|
|
Usage:
|
|
python scripts/pretty_cot.py backend/logs/app.log --last 100
|
|
python scripts/pretty_cot.py --trace <trace_id> < log.txt
|
|
python scripts/pretty_cot.py backend/logs/app.log --follow
|
|
tail -f backend/logs/app.log | python scripts/pretty_cot.py
|
|
|
|
Makes raw JSON CoT lines human- and agent-readable with icons, grouping,
|
|
and optional filtering. Designed so an agent can quickly understand a trace.
|
|
|
|
This is the reference tool for "agent view" of logs (see molecular-cot-logging skill).
|
|
"""
|
|
# #endregion Scripts.PrettyCoT
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from collections import defaultdict
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, Optional
|
|
|
|
# ── Central suppression (mirrors backend/src/core/cot_logger.py) ──
|
|
# This is the canonical list. If you update this, update the backend module too.
|
|
_ROUTINE_PHRASES = (
|
|
"Reusing cached Superset auth tokens",
|
|
"Resolve authenticated user principal",
|
|
"User principal resolved",
|
|
"Resolving current user preference",
|
|
"Loading current user's dashboard preference",
|
|
"Validated ENCRYPTION_KEY",
|
|
"Superset client ready",
|
|
"Superset client initialized",
|
|
"SupersetClientRegistry.get_client",
|
|
"Initialized ResourceService",
|
|
"ResourceService initialized",
|
|
"Created shared HTTP client",
|
|
"Ensured directory",
|
|
)
|
|
|
|
|
|
def _is_routine(intent: str) -> bool:
|
|
"""Check if intent matches known infrastructure noise (defense-in-depth)."""
|
|
return any(p in intent for p in _ROUTINE_PHRASES)
|
|
|
|
|
|
# ── Analytics engine (SSOT: backend/src/core/log_stats.py — never duplicate) ──
|
|
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def _load_engine():
|
|
"""Load log_stats.py by file path — stdlib-only module, zero package side effects
|
|
(works from any python3 without the backend venv; zombie-mode friendly)."""
|
|
import importlib.util
|
|
|
|
engine_path = _REPO_ROOT / "backend" / "src" / "core" / "log_stats.py"
|
|
try:
|
|
spec = importlib.util.spec_from_file_location("cot_log_stats", engine_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
except Exception as exc: # pragma: no cover - environment dependent
|
|
print(f"Error: cannot load {engine_path} ({exc})", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
ICONS = {
|
|
"REASON": "→",
|
|
"REFLECT": "✓",
|
|
"EXPLORE": "⚠",
|
|
}
|
|
|
|
LEVEL_COLOR = {
|
|
"INFO": "",
|
|
"WARNING": "⚡ ",
|
|
"ERROR": "🔥 ",
|
|
"DEBUG": "… ",
|
|
}
|
|
|
|
|
|
def parse_line(line: str) -> Optional[dict[str, Any]]:
|
|
line = line.strip()
|
|
if not line:
|
|
return None
|
|
try:
|
|
# Handle lines that may be prefixed by docker timestamps etc.
|
|
if line.startswith("{"):
|
|
return json.loads(line)
|
|
# Try to find embedded JSON
|
|
start = line.find("{")
|
|
if start != -1:
|
|
candidate = line[start:]
|
|
end = candidate.rfind("}") + 1
|
|
if end > 1:
|
|
return json.loads(candidate[:end])
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def format_record(rec: dict[str, Any], compact: bool = False) -> str:
|
|
ts = rec.get("ts", "")
|
|
if isinstance(ts, str) and "T" in ts:
|
|
try:
|
|
dt = datetime.fromisoformat(ts.replace("Z", "+00:00").split(".")[0])
|
|
ts = dt.strftime("%H:%M:%S")
|
|
except Exception:
|
|
ts = ts[-12:]
|
|
|
|
level = rec.get("level", "INFO")
|
|
marker = rec.get("marker", "REASON")
|
|
src = rec.get("src", "?")
|
|
intent = rec.get("intent", "")
|
|
payload = rec.get("payload")
|
|
error = rec.get("error")
|
|
elapsed = rec.get("elapsed_ms")
|
|
|
|
# Suppress routine infra noise (defense-in-depth; call-site cleanup is primary)
|
|
if _is_routine(intent):
|
|
return ""
|
|
|
|
icon = ICONS.get(marker, "·")
|
|
lvl = LEVEL_COLOR.get(level, "") + level[:4]
|
|
|
|
trace = rec.get("trace_id", "")
|
|
trace_short = trace[:8] if trace and trace != "no-trace" else ""
|
|
|
|
base = f"{ts} {icon} {lvl:5} {src:30.30} {intent}"
|
|
|
|
if elapsed is not None:
|
|
base += f" ⏱{elapsed}ms"
|
|
|
|
if error:
|
|
base += f" | error={error}"
|
|
|
|
error_code = rec.get("error_code")
|
|
claim = rec.get("claim")
|
|
loc = rec.get("loc")
|
|
contract = rec.get("contract_id")
|
|
|
|
if error_code:
|
|
base += f" | #{error_code}"
|
|
if claim:
|
|
base += f" | claim={claim}"
|
|
|
|
if payload and not compact:
|
|
try:
|
|
p = json.dumps(payload, ensure_ascii=False, default=str)[:180]
|
|
base += f" | {p}"
|
|
except Exception:
|
|
pass
|
|
|
|
if loc and not compact:
|
|
base += f" | loc={loc}"
|
|
if contract and contract != src and not compact:
|
|
base += f" | cid={contract}"
|
|
|
|
if trace_short and not compact:
|
|
base = f"[{trace_short}] {base}"
|
|
|
|
return base
|
|
|
|
|
|
def pretty_print(
|
|
records: Iterable[dict[str, Any]],
|
|
group_by_trace: bool = True,
|
|
compact: bool = False,
|
|
max_per_trace: int = 60,
|
|
) -> None:
|
|
if not group_by_trace:
|
|
for r in records:
|
|
line = format_record(r, compact=compact)
|
|
if line:
|
|
print(line)
|
|
return
|
|
|
|
# Group and limit per trace to keep signal high for agent
|
|
by_trace: dict[str, list[dict]] = defaultdict(list)
|
|
for r in records:
|
|
tid = r.get("trace_id") or "no-trace"
|
|
by_trace[tid].append(r)
|
|
|
|
for tid, items in by_trace.items():
|
|
filtered = [r for r in items if format_record(r, compact=compact)]
|
|
if not filtered:
|
|
continue
|
|
if tid != "no-trace":
|
|
print(f"\n=== TRACE {tid} ({len(filtered)} events) ===")
|
|
shown = 0
|
|
for r in filtered:
|
|
if shown >= max_per_trace:
|
|
print(f" ... ({len(filtered) - shown} more events truncated for agent focus)")
|
|
break
|
|
line = format_record(r, compact=compact)
|
|
if line:
|
|
print(" " + line)
|
|
shown += 1
|
|
|
|
|
|
def follow_file(file_path: str) -> None:
|
|
"""Tail -f equivalent for a single log file, processing new lines in real time."""
|
|
with open(file_path, encoding="utf-8", errors="replace") as f:
|
|
# Seek to end
|
|
f.seek(0, os.SEEK_END)
|
|
print(f"Following {file_path}... (Ctrl-C to stop)", file=sys.stderr)
|
|
buf = ""
|
|
try:
|
|
while True:
|
|
chunk = f.read()
|
|
if chunk:
|
|
buf += chunk
|
|
lines = buf.split("\n")
|
|
buf = lines[-1] # keep incomplete last line
|
|
for line in lines[:-1]:
|
|
rec = parse_line(line)
|
|
if rec:
|
|
line_out = format_record(rec)
|
|
if line_out:
|
|
print(line_out)
|
|
else:
|
|
time.sleep(0.25)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Pretty-print Molecular CoT logs for agents")
|
|
parser.add_argument("files", nargs="*", help="Log files (JSON lines). Use - for stdin.")
|
|
parser.add_argument("--last", type=int, default=None,
|
|
help="Only process last N lines (narrative default: 200; analytics: all)")
|
|
parser.add_argument("--trace", help="Filter to a specific trace_id")
|
|
parser.add_argument("--compact", action="store_true", help="Less verbose output")
|
|
parser.add_argument("--no-group", action="store_true", help="Do not group by trace_id")
|
|
parser.add_argument("--follow", action="store_true", help="Follow (like tail -f)")
|
|
parser.add_argument("--stats", action="store_true",
|
|
help="Log-economy + bond-structure stats (engine: backend/src/core/log_stats.py)")
|
|
parser.add_argument("--evidence", action="store_true",
|
|
help="Uncapped redacted EXPLORE/REFLECT evidence metadata (no payload or free text)")
|
|
parser.add_argument("--digest", action="store_true",
|
|
help="Prioritized EXPLORE fix-map digest")
|
|
parser.add_argument("--top", type=int, default=20, help="Digest group limit (default 20)")
|
|
parser.add_argument("--trajectory", metavar="CONTRACT_ID",
|
|
help="Belief trajectory (ts, marker, claim) for one contract")
|
|
parser.add_argument("--story", action="store_true",
|
|
help="With --trace: chronological story ending in a VERDICT echo of EXPLOREs")
|
|
parser.add_argument("--json", action="store_true",
|
|
help="Machine-readable JSON output (with --stats/--digest/--trajectory)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.evidence and (args.last is not None or args.trace or args.stats or args.digest
|
|
or args.trajectory or args.story or args.follow):
|
|
parser.error("--evidence cannot be combined with filters, caps, or other output modes")
|
|
|
|
# ── Follow mode ────────────────────────────────────────────────────────
|
|
if args.follow and args.files:
|
|
for f in args.files:
|
|
follow_file(f)
|
|
return
|
|
|
|
analytics_mode = bool(args.stats or args.digest or args.trajectory or args.story or args.evidence)
|
|
effective_last = args.last if args.last else (None if analytics_mode else 200)
|
|
|
|
# ── Batch mode ─────────────────────────────────────────────────────────
|
|
sources: list[Any] = []
|
|
source_read_errors = 0
|
|
if not args.files:
|
|
sources.append(sys.stdin)
|
|
else:
|
|
for f in args.files:
|
|
if f == "-":
|
|
sources.append(sys.stdin)
|
|
else:
|
|
try:
|
|
sources.append(Path(f).open(encoding="utf-8", errors="replace"))
|
|
except OSError:
|
|
if args.evidence:
|
|
source_read_errors += 1
|
|
else:
|
|
raise
|
|
|
|
all_records: list[dict] = []
|
|
input_lines = 0
|
|
parsed_records = 0
|
|
malformed_json_candidates = 0
|
|
for src in sources:
|
|
if src is None:
|
|
continue
|
|
try:
|
|
lines = src if hasattr(src, "readline") else src.read_text(encoding="utf-8", errors="replace").splitlines()
|
|
if effective_last and isinstance(lines, list):
|
|
lines = lines[-effective_last:]
|
|
for line in lines:
|
|
input_lines += 1
|
|
text_line = line if isinstance(line, str) else str(line)
|
|
rec = parse_line(text_line)
|
|
if rec:
|
|
parsed_records += 1
|
|
if not args.evidence and args.trace and args.trace not in (rec.get("trace_id") or ""):
|
|
continue
|
|
if not args.evidence or isinstance(rec, dict):
|
|
all_records.append(rec)
|
|
elif "{" in text_line and text_line.strip():
|
|
malformed_json_candidates += 1
|
|
except Exception as exc:
|
|
source_read_errors += 1
|
|
if not args.evidence:
|
|
print(f"Error reading: {exc}", file=sys.stderr)
|
|
|
|
# Close files
|
|
for s in sources:
|
|
if hasattr(s, "close") and s is not sys.stdin:
|
|
try:
|
|
s.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Analytics modes (single engine implementation — log_stats.py) ─────
|
|
if args.stats or args.digest or args.trajectory or args.evidence:
|
|
engine = _load_engine()
|
|
if engine is None:
|
|
sys.exit(2)
|
|
if args.evidence:
|
|
result = engine.compute_evidence(
|
|
all_records,
|
|
input_lines=input_lines,
|
|
parsed_records=parsed_records,
|
|
malformed_json_candidates=malformed_json_candidates,
|
|
)
|
|
result["complete"] = source_read_errors == 0
|
|
result["counts"]["read_errors"] = source_read_errors
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
if source_read_errors:
|
|
sys.exit(2)
|
|
elif args.stats:
|
|
stats = engine.compute_stats(all_records)
|
|
print(json.dumps(stats, ensure_ascii=False, indent=2) if args.json
|
|
else engine.render_stats(stats))
|
|
elif args.digest:
|
|
groups = engine.compute_digest(all_records, top=args.top)
|
|
print(json.dumps(groups, ensure_ascii=False, indent=2) if args.json
|
|
else engine.render_digest(groups))
|
|
else:
|
|
rows = engine.compute_trajectory(all_records, args.trajectory)
|
|
print(json.dumps(rows, ensure_ascii=False, indent=2) if args.json
|
|
else engine.render_trajectory(args.trajectory, rows))
|
|
return
|
|
|
|
if args.story:
|
|
# Chronological causal story; ends with a VERDICT echo (recency for
|
|
# constrained attention — the falsification must be the last thing read).
|
|
all_records.sort(key=lambda r: (r.get("ts") or ""))
|
|
for r in all_records:
|
|
line = format_record(r, compact=args.compact)
|
|
if line:
|
|
print(line)
|
|
verdicts = [r for r in all_records if r.get("marker") == "EXPLORE"]
|
|
if verdicts:
|
|
print("\n== VERDICT (falsified beliefs) ==")
|
|
for r in verdicts:
|
|
line = format_record(r, compact=False)
|
|
if line:
|
|
print(" " + line)
|
|
else:
|
|
print("\n== VERDICT: no EXPLORE in window — no falsified beliefs observed ==")
|
|
return
|
|
|
|
pretty_print(all_records, group_by_trace=not args.no_group, compact=args.compact)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|