e16fee4db1
Hermes now routes scratch space through HERMES_HOME/cache/scratch (exported as TMPDIR), so every production path that still spelled out /tmp bypassed that and kept teaching the agent the habit. Fallbacks in tool_result_storage, code_execution_tool, process_registry, the ACP child HOME, mini_swe_runner's local cwd, and the CI/profiling scripts now use tempfile.gettempdir(); shell installers fall back to $TMPDIR (then HERMES_HOME) when mktemp is missing, and repro/eval shells use `mktemp -d -t`. User-facing help text and sample payloads (hermes send, approvals test, hooks test, voice-mode WSL hints, meet_bot debug line) no longer suggest /tmp. Container-side paths (mini_swe_runner docker cwd, sandbox base env, remote sync tarballs) keep the literal because they name the sandbox filesystem, not the host.
627 lines
23 KiB
Python
Executable File
627 lines
23 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Drive the Hermes TUI under HERMES_DEV_PERF and summarize the pipeline.
|
|
|
|
Usage:
|
|
scripts/profile-tui.py [--session SID] [--hold KEY] [--seconds N] [--rate HZ]
|
|
|
|
Defaults: picks the session with the most messages, holds PageUp for 8s at
|
|
~30 Hz (matching xterm key-repeat), summarizes ~/.hermes/perf.log on exit.
|
|
|
|
The --tui build must exist (run `npm run build` in ui-tui first). This script
|
|
launches `node dist/entry.js` directly with HERMES_TUI_RESUME set so it
|
|
bypasses the hermes_cli wrapper — we want repeatable timing, not the CLI's
|
|
session-picker flow.
|
|
|
|
Environment overrides:
|
|
HERMES_PERF_LOG (default ~/.hermes/perf.log)
|
|
HERMES_PERF_NODE (default node from $PATH)
|
|
HERMES_TUI_DIR (default: <repo>/ui-tui relative to this script)
|
|
|
|
Exit code is 0 if the harness ran and parsed results, 2 if the TUI crashed
|
|
or produced no perf data (suggests HERMES_DEV_PERF wiring is broken).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import pty
|
|
import select
|
|
import signal
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(_PROJECT_ROOT))
|
|
try:
|
|
from hermes_constants import get_hermes_home
|
|
except ImportError:
|
|
def get_hermes_home() -> Path: # type: ignore[misc]
|
|
val = (os.environ.get("HERMES_HOME") or "").strip()
|
|
return Path(val) if val else Path.home() / ".hermes"
|
|
|
|
DEFAULT_TUI_DIR = Path(
|
|
os.environ.get("HERMES_TUI_DIR")
|
|
or str(Path(__file__).resolve().parent.parent / "ui-tui")
|
|
)
|
|
DEFAULT_LOG = Path(os.environ.get("HERMES_PERF_LOG", str(get_hermes_home() / "perf.log")))
|
|
DEFAULT_STATE_DB = get_hermes_home() / "state.db"
|
|
|
|
# Keystroke escape sequences. Matches what xterm/VT220 send when the
|
|
# terminal has bracketed-paste disabled and the key-repeat handler fires.
|
|
KEYS = {
|
|
"page_up": b"\x1b[5~",
|
|
"page_down": b"\x1b[6~",
|
|
"wheel_up": b"\x1b[M`!!", # mouse wheel up (SGR-less) — best-effort
|
|
"shift_up": b"\x1b[1;2A",
|
|
"shift_down": b"\x1b[1;2B",
|
|
}
|
|
|
|
|
|
def pick_longest_session(db: Path) -> str:
|
|
conn = sqlite3.connect(db)
|
|
row = conn.execute(
|
|
"SELECT id FROM sessions s ORDER BY "
|
|
"(SELECT COUNT(*) FROM messages m WHERE m.session_id = s.id) DESC LIMIT 1"
|
|
).fetchone()
|
|
if not row:
|
|
sys.exit(f"no sessions in {db}")
|
|
return row[0]
|
|
|
|
|
|
def drain(fd: int, timeout: float) -> bytes:
|
|
"""Read whatever's available from fd within `timeout`, then return."""
|
|
chunks = []
|
|
end = time.monotonic() + timeout
|
|
while time.monotonic() < end:
|
|
r, _, _ = select.select([fd], [], [], max(0.0, end - time.monotonic()))
|
|
if not r:
|
|
break
|
|
try:
|
|
data = os.read(fd, 4096)
|
|
except OSError:
|
|
break
|
|
if not data:
|
|
break
|
|
chunks.append(data)
|
|
return b"".join(chunks)
|
|
|
|
|
|
def hold_key(fd: int, seq: bytes, seconds: float, rate_hz: int) -> int:
|
|
"""Write `seq` to fd at ~rate_hz for `seconds`. Returns keystrokes sent."""
|
|
interval = 1.0 / max(1, rate_hz)
|
|
end = time.monotonic() + seconds
|
|
sent = 0
|
|
while time.monotonic() < end:
|
|
try:
|
|
os.write(fd, seq)
|
|
sent += 1
|
|
except OSError:
|
|
break
|
|
# Drain stdout to keep the PTY buffer flowing; ignore content.
|
|
drain(fd, 0)
|
|
time.sleep(interval)
|
|
return sent
|
|
|
|
|
|
def summarize(log: Path, since_ts_ms: int) -> dict[str, Any]:
|
|
"""Parse perf.log, keep only events newer than since_ts_ms, return stats."""
|
|
react_events: list[dict[str, Any]] = []
|
|
frame_events: list[dict[str, Any]] = []
|
|
if not log.exists():
|
|
return {"error": f"no log at {log}", "react": [], "frame": []}
|
|
for line in log.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
row = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if int(row.get("ts", 0)) < since_ts_ms:
|
|
continue
|
|
src = row.get("src")
|
|
if src == "react":
|
|
react_events.append(row)
|
|
elif src == "frame":
|
|
frame_events.append(row)
|
|
|
|
return {
|
|
"react": react_events,
|
|
"frame": frame_events,
|
|
}
|
|
|
|
|
|
def pct(values: list[float], p: float) -> float:
|
|
if not values:
|
|
return 0.0
|
|
s = sorted(values)
|
|
idx = min(len(s) - 1, int(len(s) * p))
|
|
return s[idx]
|
|
|
|
|
|
def format_report(data: dict[str, Any]) -> str:
|
|
react = data.get("react") or []
|
|
frames = data.get("frame") or []
|
|
out = []
|
|
|
|
out.append("═══ React Profiler ═══")
|
|
if not react:
|
|
out.append(" (no react events — HERMES_DEV_PERF wired? threshold too high?)")
|
|
else:
|
|
by_id: dict[str, list[float]] = {}
|
|
for r in react:
|
|
by_id.setdefault(r["id"], []).append(r["actualMs"])
|
|
out.append(f" {'pane':<14} {'count':>6} {'p50':>8} {'p95':>8} {'p99':>8} {'max':>8}")
|
|
for pid, ms in sorted(by_id.items(), key=lambda kv: -pct(kv[1], 0.99)):
|
|
out.append(
|
|
f" {pid:<14} {len(ms):>6} {pct(ms,0.50):>8.2f} {pct(ms,0.95):>8.2f} "
|
|
f"{pct(ms,0.99):>8.2f} {max(ms):>8.2f}"
|
|
)
|
|
|
|
out.append("")
|
|
out.append("═══ Ink pipeline ═══")
|
|
if not frames:
|
|
out.append(" (no frame events — onFrame wiring broken?)")
|
|
else:
|
|
dur = [f["durationMs"] for f in frames]
|
|
phases_present = any(f.get("phases") for f in frames)
|
|
out.append(f" frames captured: {len(frames)}")
|
|
out.append(
|
|
f" durationMs p50={pct(dur,0.50):.2f} p95={pct(dur,0.95):.2f} "
|
|
f"p99={pct(dur,0.99):.2f} max={max(dur):.2f}"
|
|
)
|
|
# Effective FPS during the run: frames / elapsed seconds.
|
|
ts = sorted(f["ts"] for f in frames)
|
|
if len(ts) >= 2:
|
|
elapsed_s = (ts[-1] - ts[0]) / 1000.0
|
|
fps = len(frames) / elapsed_s if elapsed_s > 0 else float("inf")
|
|
out.append(f" throughput: {len(frames)} frames / {elapsed_s:.2f}s = {fps:.1f} fps")
|
|
|
|
if phases_present:
|
|
fields = ["yoga", "renderer", "diff", "optimize", "write", "commit"]
|
|
out.append("")
|
|
out.append(f" {'phase':<10} {'p50':>8} {'p95':>8} {'p99':>8} {'max':>8} (ms)")
|
|
for field in fields:
|
|
vals = [f["phases"][field] for f in frames if f.get("phases")]
|
|
if vals:
|
|
out.append(
|
|
f" {field:<10} {pct(vals,0.50):>8.2f} {pct(vals,0.95):>8.2f} "
|
|
f"{pct(vals,0.99):>8.2f} {max(vals):>8.2f}"
|
|
)
|
|
# Derived: sum of phases vs durationMs (reveals hidden time).
|
|
sum_ps = [
|
|
sum(f["phases"][k] for k in fields)
|
|
for f in frames if f.get("phases")
|
|
]
|
|
if sum_ps:
|
|
dur_match = [f["durationMs"] for f in frames if f.get("phases")]
|
|
deltas = [d - s for d, s in zip(dur_match, sum_ps)]
|
|
out.append(
|
|
f" {'dur-Σphases':<10} {pct(deltas,0.50):>8.2f} {pct(deltas,0.95):>8.2f} "
|
|
f"{pct(deltas,0.99):>8.2f} {max(deltas):>8.2f} (unaccounted-for time)"
|
|
)
|
|
|
|
# Yoga counters
|
|
visited = [f["phases"]["yogaVisited"] for f in frames if f.get("phases")]
|
|
measured = [f["phases"]["yogaMeasured"] for f in frames if f.get("phases")]
|
|
cache_hits = [f["phases"]["yogaCacheHits"] for f in frames if f.get("phases")]
|
|
live = [f["phases"]["yogaLive"] for f in frames if f.get("phases")]
|
|
out.append("")
|
|
out.append(" Yoga counters (per frame):")
|
|
for name, vals in (
|
|
("visited", visited),
|
|
("measured", measured),
|
|
("cacheHits", cache_hits),
|
|
("live", live),
|
|
):
|
|
if vals:
|
|
out.append(f" {name:<11} p50={pct(vals,0.5):.0f} p99={pct(vals,0.99):.0f} max={max(vals)}")
|
|
|
|
# Patch counts — proxy for "how much changed each frame"
|
|
patches = [f["phases"]["patches"] for f in frames if f.get("phases")]
|
|
if patches:
|
|
out.append(
|
|
f" patches p50={pct(patches,0.5):.0f} p99={pct(patches,0.99):.0f} "
|
|
f"max={max(patches)} total={sum(patches)}"
|
|
)
|
|
optimized = [
|
|
f["phases"].get("optimizedPatches", 0)
|
|
for f in frames if f.get("phases")
|
|
]
|
|
if any(optimized):
|
|
out.append(
|
|
f" optimized p50={pct(optimized,0.5):.0f} p99={pct(optimized,0.99):.0f} "
|
|
f"max={max(optimized)} total={sum(optimized)}"
|
|
f" (ratio: {sum(optimized)/max(1,sum(patches)):.2f})"
|
|
)
|
|
|
|
# Write bytes + drain telemetry — the outer-terminal bottleneck gauge.
|
|
bytes_written = [
|
|
f["phases"].get("writeBytes", 0)
|
|
for f in frames if f.get("phases")
|
|
]
|
|
if any(bytes_written):
|
|
total_b = sum(bytes_written)
|
|
kb = total_b / 1024
|
|
out.append(
|
|
f" writeBytes p50={pct(bytes_written,0.5):.0f}B p99={pct(bytes_written,0.99):.0f}B "
|
|
f"max={max(bytes_written)}B total={kb:.1f}KB"
|
|
)
|
|
drains = [
|
|
f["phases"].get("prevFrameDrainMs", 0)
|
|
for f in frames if f.get("phases")
|
|
]
|
|
if any(d > 0 for d in drains):
|
|
nonzero = [d for d in drains if d > 0]
|
|
out.append(
|
|
f" drainMs p50={pct(nonzero,0.5):.2f} p95={pct(nonzero,0.95):.2f} "
|
|
f"p99={pct(nonzero,0.99):.2f} max={max(nonzero):.2f} (terminal flush latency)"
|
|
)
|
|
backpressure = sum(1 for f in frames if f.get("phases", {}).get("backpressure"))
|
|
if backpressure:
|
|
out.append(
|
|
f" backpressure: {backpressure}/{len(frames)} frames "
|
|
f"({100*backpressure/len(frames):.0f}%) (Node stdout buffer full — terminal slow)"
|
|
)
|
|
|
|
# Flickers
|
|
flicker_frames = [f for f in frames if f.get("flickers")]
|
|
if flicker_frames:
|
|
out.append("")
|
|
out.append(f" ⚠ flickers detected in {len(flicker_frames)} frames")
|
|
reasons: dict[str, int] = {}
|
|
for f in flicker_frames:
|
|
for fl in f["flickers"]:
|
|
reasons[fl["reason"]] = reasons.get(fl["reason"], 0) + 1
|
|
for reason, n in sorted(reasons.items(), key=lambda kv: -kv[1]):
|
|
out.append(f" {reason}: {n}")
|
|
|
|
return "\n".join(out)
|
|
|
|
|
|
def key_metrics(data: dict[str, Any]) -> dict[str, float]:
|
|
"""Flatten the report into a dict of scalar metrics for A/B diffing."""
|
|
metrics: dict[str, float] = {}
|
|
frames = data.get("frame") or []
|
|
react = data.get("react") or []
|
|
|
|
if frames:
|
|
durs = [f["durationMs"] for f in frames]
|
|
metrics["frames"] = len(frames)
|
|
metrics["dur_p50"] = pct(durs, 0.50)
|
|
metrics["dur_p95"] = pct(durs, 0.95)
|
|
metrics["dur_p99"] = pct(durs, 0.99)
|
|
metrics["dur_max"] = max(durs)
|
|
|
|
ts = sorted(f["ts"] for f in frames)
|
|
if len(ts) >= 2:
|
|
elapsed = (ts[-1] - ts[0]) / 1000.0
|
|
metrics["fps_throughput"] = len(frames) / elapsed if elapsed > 0 else 0.0
|
|
# Interframe gaps distribution — complementary view to throughput:
|
|
gaps = [ts[i] - ts[i - 1] for i in range(1, len(ts))]
|
|
if gaps:
|
|
metrics["gap_p50_ms"] = pct(gaps, 0.50)
|
|
metrics["gap_p99_ms"] = pct(gaps, 0.99)
|
|
metrics["gaps_under_16ms"] = sum(1 for g in gaps if g < 16)
|
|
metrics["gaps_over_200ms"] = sum(1 for g in gaps if g >= 200)
|
|
|
|
for phase in ("renderer", "yoga", "diff", "write"):
|
|
vals = [f["phases"][phase] for f in frames if f.get("phases")]
|
|
if vals:
|
|
metrics[f"{phase}_p99"] = pct(vals, 0.99)
|
|
metrics[f"{phase}_max"] = max(vals)
|
|
|
|
patches = [f["phases"]["patches"] for f in frames if f.get("phases")]
|
|
if patches:
|
|
metrics["patches_total"] = sum(patches)
|
|
metrics["patches_p99"] = pct(patches, 0.99)
|
|
|
|
optimized = [
|
|
f["phases"].get("optimizedPatches", 0) for f in frames if f.get("phases")
|
|
]
|
|
if any(optimized):
|
|
metrics["optimized_total"] = sum(optimized)
|
|
|
|
bytes_list = [
|
|
f["phases"].get("writeBytes", 0) for f in frames if f.get("phases")
|
|
]
|
|
if any(bytes_list):
|
|
metrics["writeBytes_total"] = sum(bytes_list)
|
|
|
|
drains = [
|
|
f["phases"].get("prevFrameDrainMs", 0)
|
|
for f in frames if f.get("phases")
|
|
]
|
|
drain_nonzero = [d for d in drains if d > 0]
|
|
if drain_nonzero:
|
|
metrics["drain_p99"] = pct(drain_nonzero, 0.99)
|
|
metrics["drain_max"] = max(drain_nonzero)
|
|
|
|
bp = sum(1 for f in frames if f.get("phases", {}).get("backpressure"))
|
|
metrics["backpressure_frames"] = bp
|
|
|
|
if react:
|
|
for pid in {e["id"] for e in react}:
|
|
ms = [e["actualMs"] for e in react if e["id"] == pid]
|
|
metrics[f"react_{pid}_p99"] = pct(ms, 0.99)
|
|
metrics[f"react_{pid}_max"] = max(ms)
|
|
|
|
return metrics
|
|
|
|
|
|
def format_diff(before: dict[str, float], after: dict[str, float]) -> str:
|
|
"""Render a side-by-side A/B comparison table."""
|
|
keys = sorted(set(before) | set(after))
|
|
lines = [f"{'metric':<28} {'before':>12} {'after':>12} {'delta':>12} {'%':>6}"]
|
|
lines.append("─" * 76)
|
|
for k in keys:
|
|
b = before.get(k, 0.0)
|
|
a = after.get(k, 0.0)
|
|
d = a - b
|
|
pct_change = ((a / b) - 1) * 100 if b not in {0, 0.0} else float("inf") if a else 0
|
|
|
|
# Flag improvements vs regressions. For _p99 / _max / _total / gaps_over /
|
|
# patches / writeBytes / backpressure, LOWER is better. For fps / gaps_under,
|
|
# HIGHER is better.
|
|
lower_is_better = any(
|
|
token in k
|
|
for token in (
|
|
"p50",
|
|
"p95",
|
|
"p99",
|
|
"_max",
|
|
"_total",
|
|
"gaps_over",
|
|
"backpressure",
|
|
"drain",
|
|
)
|
|
)
|
|
higher_is_better = "fps_" in k or "gaps_under" in k
|
|
mark = ""
|
|
if d and not (lower_is_better or higher_is_better):
|
|
mark = ""
|
|
elif d < 0 and lower_is_better:
|
|
mark = "↓"
|
|
elif d > 0 and higher_is_better:
|
|
mark = "↑"
|
|
elif d > 0 and lower_is_better:
|
|
mark = "↑" # regression
|
|
elif d < 0 and higher_is_better:
|
|
mark = "↓" # regression
|
|
|
|
pct_str = "—" if pct_change == float("inf") else f"{pct_change:+6.1f}%"
|
|
lines.append(
|
|
f"{k:<28} {b:>12.2f} {a:>12.2f} {d:>+12.2f} {pct_str} {mark}"
|
|
)
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def run_once(args: argparse.Namespace) -> dict[str, Any]:
|
|
tui_dir = Path(args.tui_dir).resolve()
|
|
entry = tui_dir / "dist" / "entry.js"
|
|
if not entry.exists():
|
|
sys.exit(f"{entry} missing — run `npm run build` in {tui_dir} first")
|
|
|
|
sid = args.session or pick_longest_session(DEFAULT_STATE_DB)
|
|
print(f"• session: {sid}")
|
|
print(f"• hold: {args.hold} x {args.rate}Hz for {args.seconds}s after {args.warmup}s warmup")
|
|
print(f"• terminal: {args.cols}x{args.rows}")
|
|
|
|
log = Path(args.log)
|
|
if not args.keep_log and log.exists():
|
|
log.unlink()
|
|
|
|
since_ms = int(time.time() * 1000)
|
|
|
|
env = os.environ.copy()
|
|
env["HERMES_DEV_PERF"] = "1"
|
|
env["HERMES_DEV_PERF_MS"] = str(args.threshold_ms)
|
|
env["HERMES_DEV_PERF_LOG"] = str(log)
|
|
env["HERMES_TUI_RESUME"] = sid
|
|
env["COLUMNS"] = str(args.cols)
|
|
env["LINES"] = str(args.rows)
|
|
env["TERM"] = env.get("TERM", "xterm-256color")
|
|
|
|
# Pass through extra flags the TUI wrapper recognizes (e.g. --no-fullscreen).
|
|
# Stored on args as `extra_flags` list.
|
|
node = os.environ.get("HERMES_PERF_NODE", "node")
|
|
node_args = [node, str(entry), *getattr(args, "extra_flags", [])]
|
|
|
|
pid, fd = pty.fork()
|
|
if pid == 0:
|
|
os.execvpe(node, node_args, env)
|
|
|
|
try:
|
|
import fcntl, struct, termios
|
|
winsize = struct.pack("HHHH", args.rows, args.cols, 0, 0)
|
|
fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
|
|
|
|
print(f"• pid: {pid} fd: {fd}")
|
|
print(f"• warmup {args.warmup}s (drain startup output)…")
|
|
drain(fd, args.warmup)
|
|
|
|
print(f"• holding {args.hold}…")
|
|
sent = hold_key(fd, KEYS[args.hold], args.seconds, args.rate)
|
|
print(f" sent {sent} keystrokes")
|
|
|
|
drain(fd, 0.5)
|
|
finally:
|
|
try:
|
|
os.kill(pid, signal.SIGTERM)
|
|
for _ in range(10):
|
|
pid_done, _ = os.waitpid(pid, os.WNOHANG)
|
|
if pid_done == pid:
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
os.kill(pid, signal.SIGKILL) # windows-footgun: ok — POSIX-only script (imports pty at top)
|
|
os.waitpid(pid, 0)
|
|
except (ProcessLookupError, ChildProcessError):
|
|
pass
|
|
try:
|
|
os.close(fd)
|
|
except OSError:
|
|
pass
|
|
|
|
time.sleep(0.2)
|
|
return summarize(log, since_ms)
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument("--session", help="session id to resume (default: longest in db)")
|
|
p.add_argument("--hold", default="page_up", choices=sorted(KEYS.keys()), help="key to hold")
|
|
p.add_argument("--seconds", type=float, default=8.0, help="how long to hold the key")
|
|
p.add_argument("--rate", type=int, default=30, help="keystrokes per second")
|
|
p.add_argument("--warmup", type=float, default=3.0, help="seconds to wait after launch before input")
|
|
p.add_argument("--threshold-ms", type=float, default=0.0, help="HERMES_DEV_PERF_MS (0 = capture all)")
|
|
p.add_argument("--cols", type=int, default=120)
|
|
p.add_argument("--rows", type=int, default=40)
|
|
p.add_argument("--keep-log", action="store_true", help="don't wipe perf.log before run")
|
|
p.add_argument("--tui-dir", default=str(DEFAULT_TUI_DIR))
|
|
p.add_argument("--log", default=str(DEFAULT_LOG))
|
|
p.add_argument("--save", metavar="LABEL",
|
|
help="save the final metrics as <tempdir>/perf-<LABEL>.json for later --compare")
|
|
p.add_argument("--compare", metavar="LABEL",
|
|
help="diff against <tempdir>/perf-<LABEL>.json after running")
|
|
p.add_argument("--loop", action="store_true",
|
|
help="watch for source changes, rebuild, rerun, and diff vs previous run")
|
|
p.add_argument("--extra-flag", dest="extra_flags", action="append", default=[],
|
|
help="pass through to node dist/entry.js (repeatable)")
|
|
args = p.parse_args()
|
|
|
|
if args.loop:
|
|
return loop_mode(args)
|
|
|
|
# Single-shot path.
|
|
data = run_once(args)
|
|
print()
|
|
print(format_report(data))
|
|
|
|
metrics = key_metrics(data)
|
|
|
|
if args.save:
|
|
path = Path(tempfile.gettempdir()) / f"perf-{args.save}.json"
|
|
path.write_text(json.dumps(metrics, indent=2), encoding="utf-8")
|
|
print(f"\n• saved: {path}")
|
|
|
|
if args.compare:
|
|
path = Path(tempfile.gettempdir()) / f"perf-{args.compare}.json"
|
|
if not path.exists():
|
|
print(f"\n⚠ no baseline at {path} — run with --save {args.compare} first")
|
|
else:
|
|
before = json.loads(path.read_text(encoding="utf-8"))
|
|
print(f"\n═══ A/B diff vs {path} ═══")
|
|
print(format_diff(before, metrics))
|
|
|
|
if not data["react"] and not data["frame"]:
|
|
return 2
|
|
return 0
|
|
|
|
|
|
def loop_mode(args: argparse.Namespace) -> int:
|
|
"""Watch source files, rebuild, rerun, print A/B diff against previous run.
|
|
|
|
Keeps a rolling 'previous run' baseline in memory so each iteration
|
|
reports delta vs the last one — visibility into whether the last
|
|
edit moved the needle. Press Ctrl+C to stop.
|
|
"""
|
|
import subprocess
|
|
|
|
tui_dir = Path(args.tui_dir).resolve()
|
|
src_root = tui_dir / "src"
|
|
pkg_root = tui_dir / "packages" / "hermes-ink" / "src"
|
|
|
|
def collect_mtimes() -> dict[str, float]:
|
|
mtimes: dict[str, float] = {}
|
|
for root in (src_root, pkg_root):
|
|
if not root.exists():
|
|
continue
|
|
for path in root.rglob("*"):
|
|
if path.suffix in {".ts", ".tsx"} and "__tests__" not in str(path):
|
|
try:
|
|
mtimes[str(path)] = path.stat().st_mtime
|
|
except OSError:
|
|
pass
|
|
return mtimes
|
|
|
|
previous_metrics: dict[str, float] | None = None
|
|
previous_mtimes = collect_mtimes()
|
|
iteration = 0
|
|
|
|
print(f"• loop mode — watching {src_root} + {pkg_root} for *.ts(x) changes")
|
|
print("• edit any TS file, the harness rebuilds + reruns automatically")
|
|
print("• Ctrl+C to stop\n")
|
|
|
|
try:
|
|
while True:
|
|
iteration += 1
|
|
print(f"\n{'═' * 76}")
|
|
print(f"Iteration {iteration} @ {time.strftime('%H:%M:%S')}")
|
|
print("═" * 76)
|
|
|
|
if iteration > 1:
|
|
print("• rebuilding…")
|
|
result = subprocess.run(
|
|
["npm", "run", "build"],
|
|
cwd=tui_dir,
|
|
capture_output=True,
|
|
text=True, encoding='utf-8', errors='replace',
|
|
)
|
|
if result.returncode != 0:
|
|
print("✗ build failed:")
|
|
print(result.stdout[-2000:])
|
|
print(result.stderr[-2000:])
|
|
print("\n• waiting for source changes to retry…")
|
|
previous_mtimes = wait_for_change(previous_mtimes, collect_mtimes)
|
|
continue
|
|
print("✓ build ok")
|
|
|
|
data = run_once(args)
|
|
metrics = key_metrics(data)
|
|
|
|
print()
|
|
print(format_report(data))
|
|
|
|
if previous_metrics is not None:
|
|
print(f"\n═══ A/B diff vs iteration {iteration - 1} ═══")
|
|
print(format_diff(previous_metrics, metrics))
|
|
|
|
previous_metrics = metrics
|
|
|
|
print("\n• waiting for source changes…")
|
|
previous_mtimes = wait_for_change(previous_mtimes, collect_mtimes)
|
|
except KeyboardInterrupt:
|
|
print("\n• loop stopped")
|
|
return 0
|
|
|
|
|
|
def wait_for_change(prev: dict[str, float], collect) -> dict[str, float]:
|
|
"""Poll every 1s until a watched file's mtime changes. Debounced 500ms."""
|
|
while True:
|
|
time.sleep(1)
|
|
current = collect()
|
|
|
|
changed = [
|
|
path for path, mtime in current.items() if prev.get(path) != mtime
|
|
]
|
|
|
|
if changed:
|
|
print(f" ↻ {len(changed)} file(s) changed:")
|
|
for path in changed[:5]:
|
|
print(f" {path}")
|
|
# Debounce — editor save bursts can take ~500ms to settle
|
|
time.sleep(0.5)
|
|
return collect()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|