90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent re-check for sol's daily-ledger triage artifact.
|
|
|
|
Recomputes the classification from the raw daily logs under
|
|
/root/.hermes/wizard-outputs and diffs it against
|
|
triage-daily-2026-10-22.json. Exit 0 = artifact is correct, nonzero = wrong.
|
|
|
|
Independent of the generator: it never reads the artifact during recompute,
|
|
and it validates the 7-lane id set, the latest-log name, the mtime, log count,
|
|
and the AGE_CLASSIFY status, so a wrong classifier is caught.
|
|
"""
|
|
import glob
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
BASE = "/root/.hermes/wizard-outputs"
|
|
ARTIFACT = os.path.join(BASE, "sol", "triage-daily-2026-10-22.json")
|
|
PRIMARY = ["forge", "ledger", "null", "quill", "sphinx", "sol", "spark"]
|
|
|
|
|
|
def classify(age_sec):
|
|
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 PRIMARY:
|
|
files = sorted(glob.glob(os.path.join(BASE, pid + "*-*.txt")))
|
|
if not files:
|
|
raise SystemExit("no raw logs for lane %s" % pid)
|
|
latest = max(files, key=os.path.getmtime)
|
|
mtime = os.path.getmtime(latest)
|
|
age_sec = (datetime.now(timezone.utc)
|
|
- datetime.fromtimestamp(mtime, tz=timezone.utc)).total_seconds()
|
|
out.append({
|
|
"id": pid,
|
|
"latest_log": os.path.basename(latest),
|
|
"latest_mtime_utc": datetime.fromtimestamp(mtime, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"daily_log_count": len(files),
|
|
"age_hours": round(age_sec / 3600.0, 1),
|
|
"status": classify(age_sec),
|
|
})
|
|
return out
|
|
|
|
|
|
def main():
|
|
expected = recompute()
|
|
with open(ARTIFACT, encoding="utf-8") as fh:
|
|
actual = json.load(fh).get("lanes", [])
|
|
|
|
errors = []
|
|
if len(expected) != len(actual):
|
|
errors.append("lane count mismatch: expected %d got %d" % (len(expected), len(actual)))
|
|
|
|
by_id = {a.get("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", "age_hours", "status"):
|
|
if a.get(k) != e[k]:
|
|
errors.append("%s.%s: expected %r got %r" % (e["id"], k, e[k], a.get(k)))
|
|
|
|
if sorted(by_id) != sorted(PRIMARY):
|
|
errors.append("lane id set mismatch: %s" % sorted(by_id))
|
|
|
|
if errors:
|
|
print("VERIFY FAIL")
|
|
for err in errors:
|
|
print(" -", err)
|
|
sys.exit(1)
|
|
|
|
print("VERIFY OK: %d/%d lanes match" % (len(expected), len(expected)))
|
|
for e in expected:
|
|
print(" %-6s %-24s %s age=%-6.1fh %s" % (
|
|
e["id"], e["latest_log"], e["latest_mtime_utc"], e["age_hours"], e["status"]))
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|