104 lines
3.7 KiB
Python
Executable File
104 lines
3.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""CHECK (Sol daily-lane triage): independent recompute of ACTIVE/WATCH/STALE from the
|
|
raw daily ledger logs under /root/.hermes/wizard-outputs/<lane>-*.txt, compared against
|
|
the committed artifact sol-lane-status-2026-10-27.json.
|
|
|
|
Independent recompute: derives status from raw mtimes only. The committed JSON's
|
|
statuses/latests are then compared against this recompute. A PASS = 0 proves the
|
|
committed artifact matches reality; any mismatch = nonzero.
|
|
|
|
Exit code IS the verdict: 0 = correct, nonzero = wrong.
|
|
"""
|
|
import glob
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
BASE = "/root/.hermes/wizard-outputs"
|
|
JSON_PATH = "/root/.hermes/wizard-outputs/sol/sol-lane-status-2026-10-27.json"
|
|
LANES = ["forge", "ledger", "null", "quill", "sphinx", "sol", "spark"]
|
|
|
|
|
|
def recompute():
|
|
now = datetime.now(timezone.utc)
|
|
out = {}
|
|
for pid in LANES:
|
|
files = glob.glob(os.path.join(BASE, pid + "-*.txt"))
|
|
if not files:
|
|
raise SystemExit("no daily logs for lane " + pid)
|
|
latest = max(files, key=os.path.getmtime)
|
|
mt = os.path.getmtime(latest)
|
|
mtime_utc = datetime.fromtimestamp(mt, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
age_h = (now - datetime.fromtimestamp(mt, tz=timezone.utc)).total_seconds() / 3600.0
|
|
if age_h <= 24.0:
|
|
status = "ACTIVE"
|
|
elif age_h <= 168.0:
|
|
status = "WATCH"
|
|
else:
|
|
status = "STALE"
|
|
out[pid] = {"latest_log": os.path.basename(latest), "latest_mtime_utc": mtime_utc,
|
|
"age_h": round(age_h, 2), "status": status}
|
|
return out
|
|
|
|
|
|
def main():
|
|
failures = []
|
|
|
|
# Load committed artifact.
|
|
if not os.path.exists(JSON_PATH):
|
|
print("FAIL: artifact missing: " + JSON_PATH)
|
|
sys.exit(1)
|
|
with open(JSON_PATH, encoding="utf-8") as fh:
|
|
art = json.load(fh)
|
|
|
|
rec = recompute()
|
|
|
|
# Compare lane by lane.
|
|
for pid in LANES:
|
|
r = rec[pid]
|
|
hit = art.get("lanes", [])
|
|
row = next((x for x in hit if x.get("id") == pid), None)
|
|
if row is None:
|
|
failures.append("lane " + pid + " missing from artifact lanes")
|
|
print("FAIL: " + pid + " missing")
|
|
continue
|
|
if row.get("latest_log") != r["latest_log"]:
|
|
failures.append("%s: latest_log %s != %s" % (pid, row.get("latest_log"), r["latest_log"]))
|
|
print("FAIL: " + pid + " latest_log")
|
|
if row.get("latest_mtime_utc") != r["latest_mtime_utc"]:
|
|
failures.append("%s: mtime %s != %s" % (pid, row.get("latest_mtime_utc"), r["latest_mtime_utc"]))
|
|
print("FAIL: " + pid + " mtime")
|
|
if row.get("status") != r["status"]:
|
|
failures.append("%s: status %s != %s" % (pid, row.get("status"), r["status"]))
|
|
print("FAIL: " + pid + " status")
|
|
else:
|
|
print("PASS: %s latest=%s age=%.2f status=%s" % (pid, r["latest_log"], r["age_h"], r["status"]))
|
|
|
|
# Compare status_counts.
|
|
counts = {}
|
|
for x in art.get("lanes", []):
|
|
counts[x.get("status")] = counts.get(x.get("status"), 0) + 1
|
|
if counts != rec_counts(rec):
|
|
failures.append("counts %s != recompute %s" % (counts, rec_counts(rec)))
|
|
|
|
if failures:
|
|
print("FAIL: triage artifact does not match independent recompute")
|
|
for f in failures:
|
|
print(" - " + f)
|
|
sys.exit(1)
|
|
|
|
print("PASS: 7/7 lanes match independent recompute from raw daily logs")
|
|
print(" status_counts=%s" % json.dumps(rec_counts(rec)))
|
|
|
|
|
|
def rec_counts(rec):
|
|
c = {}
|
|
for v in rec.values():
|
|
c[v["status"]] = c.get(v["status"], 0) + 1
|
|
return c
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|