88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent re-verification of the daily triage artifact.
|
|
|
|
Re-computes each primary lane's latest daily log status from the raw logs
|
|
(AGE_CLASSIFY: ACTIVE <= 24h, WATCH <= 168h, else STALE) and diffs the
|
|
result against /root/.hermes/wizard-outputs/sol/triage.json. Exit code 0
|
|
iff the regenerated artifact matches the independent recomputation exactly.
|
|
"""
|
|
import glob
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
BASE = "/root/.hermes/wizard-outputs"
|
|
ARTIFACT = "/root/.hermes/wizard-outputs/sol/triage-daily-2026-10-22.json"
|
|
LANES = ["forge", "ledger", "null", "quill", "sphinx", "sol", "spark"]
|
|
|
|
|
|
def classify(ts):
|
|
age_sec = (datetime.now(timezone.utc) - ts).total_seconds()
|
|
age_h = age_sec / 3600.0
|
|
if age_h <= 24.0:
|
|
return "ACTIVE"
|
|
if age_h <= 168.0:
|
|
return "WATCH"
|
|
return "STALE"
|
|
|
|
|
|
def recompute():
|
|
out = []
|
|
for pid in LANES:
|
|
files = sorted(glob.glob(os.path.join(BASE, pid + "*-*.txt")))
|
|
if not files:
|
|
raise SystemExit("no logs for lane %s" % pid)
|
|
latest = max(files, key=os.path.getmtime)
|
|
ts = datetime.fromtimestamp(os.path.getmtime(latest), tz=timezone.utc)
|
|
out.append({
|
|
"id": pid,
|
|
"latest_log": os.path.basename(latest),
|
|
"latest_mtime_utc": ts.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"daily_log_count": len(files),
|
|
"status": classify(ts),
|
|
})
|
|
return out
|
|
|
|
|
|
def main():
|
|
expected = recompute()
|
|
with open(ARTIFACT, encoding="utf-8") as fh:
|
|
actual = json.load(fh)["lanes"]
|
|
|
|
errors = []
|
|
if len(expected) != len(actual):
|
|
errors.append("lane count mismatch: expected %d got %d"
|
|
% (len(expected), len(actual)))
|
|
|
|
by_id = {a["id"]: a for a in actual}
|
|
for e in expected:
|
|
a = by_id.get(e["id"])
|
|
if a is None:
|
|
errors.append("lane %s missing from artifact" % e["id"])
|
|
continue
|
|
for k in ("latest_log", "latest_mtime_utc", "daily_log_count", "status"):
|
|
if a.get(k) != e[k]:
|
|
errors.append("%s.%s: expected %r got %r"
|
|
% (e["id"], k, e[k], a.get(k)))
|
|
|
|
if errors:
|
|
print("VERIFY FAIL")
|
|
for err in errors:
|
|
print(" -", err)
|
|
sys.exit(1)
|
|
|
|
print("VERIFY OK: 7/7 lanes match")
|
|
for e in expected:
|
|
print(" %-6s %-24s %s age=%-6.1fh status=%-6s (log_count=%d)"
|
|
% (e["id"], e["latest_log"], e["latest_mtime_utc"],
|
|
(datetime.now(timezone.utc) - datetime.fromtimestamp(
|
|
os.path.getmtime(os.path.join(BASE, e["latest_log"])),
|
|
tz=timezone.utc)).total_seconds() / 3600.0,
|
|
e["status"], e["daily_log_count"]))
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|