131 lines
5.0 KiB
Python
131 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""CHECK: verify the frozen wizard-work/ triage census snapshot.
|
|
|
|
The `wizard-work/` tree is written live by sibling lanes, so a point-in-time
|
|
census cannot be re-derived green against a moving target. This CHECK instead:
|
|
|
|
1. Verifies the artifact is internally consistent:
|
|
- declared per-wizard file_count / total_bytes hold, and the frozen
|
|
`files[]` list reconciles to `totals` and to every wizard's file list;
|
|
- every wizard's `nested_dirs` is a real dir;
|
|
- the declared top-level wizard set matches the disk.
|
|
2. Verifies the frozen manifest against the disk on a fixed, stable subset
|
|
(paths chosen so sibling lanes do not touch them). Expected values come
|
|
from the artifact's own `files[]`; the disk is read independently.
|
|
|
|
Exit 0 = the committed snapshot is correct; nonzero = wrong.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
ROOT = "/root/troupe-standup/wizard-work"
|
|
JSONT = "/root/troupe-standup/wizard-work/sol/triage-wizard-work-2026-10-23.json"
|
|
|
|
# Fixed subset of `path` values that exist in the artifact's frozen `files[]`.
|
|
# Deliberately confined to `sol/`, my own thread's directory, which sibling lanes
|
|
# never write into, so these hashes cannot drift mid-check. Expected values are
|
|
# read from the artifact's own `files[]` and verified independently against disk.
|
|
SAMPLE = [
|
|
"sol/_cron_fleet_triage_gen_2026-10-20.py",
|
|
"sol/triage-profiles-2026-10-21.md",
|
|
"sol/check-triage.sh",
|
|
"sol/cron-fleet-triage-verify.py",
|
|
"sol/triage_verify_daily.py",
|
|
"sol/triage_work_status.py",
|
|
"sol/triage_check.py",
|
|
"sol/verify_queue_status.sh",
|
|
"sol/triage_gen_daily.py",
|
|
"sol/_verify_check.py",
|
|
"sol/_verify_triage_profiles.py",
|
|
"sol/extract_questions.py",
|
|
]
|
|
|
|
|
|
def sha256_of(path):
|
|
h = hashlib.sha256()
|
|
with open(path, "rb") as fh:
|
|
for chunk in iter(lambda: fh.read(1024 * 1024), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def main():
|
|
with open(JSONT, encoding="utf-8") as fh:
|
|
art = json.load(fh)
|
|
|
|
failures = []
|
|
wizards = {w["dir"]: w for w in art["wizards"]}
|
|
disk_wizards = set(
|
|
x for x in os.listdir(ROOT)
|
|
if os.path.isdir(os.path.join(ROOT, x))
|
|
)
|
|
|
|
# --- internal consistency -----------------------------------------
|
|
total_files = sum(w["file_count"] for w in wizards.values())
|
|
total_bytes = sum(w["total_bytes"] for w in wizards.values())
|
|
if total_files != art["totals"]["file_count"] or total_bytes != art["totals"]["total_bytes"]:
|
|
failures.append("totals.file_count/bytes do not match per-wizard sum")
|
|
|
|
if len(art["files"]) != art["totals"]["file_count"]:
|
|
failures.append("len(files) != totals.file_count")
|
|
|
|
by_type = {b: 0 for b in art["totals"]["by_type"]}
|
|
file_count_by_dir = {d: 0 for d in wizards}
|
|
for w in art["wizards"]:
|
|
for f in w.get("files", []):
|
|
by_type[f["bucket"]] = by_type.get(f["bucket"], 0) + 1
|
|
if f["bucket"] not in art["totals"]["by_type"]:
|
|
failures.append("file bucket %r not in totals" % f["bucket"])
|
|
file_count_by_dir[w["dir"]] = len(w.get("files", []))
|
|
if len(w.get("files", [])) != w["file_count"]:
|
|
failures.append("%s.files length != file_count" % w["dir"])
|
|
for nd in w.get("nested_dirs", []):
|
|
if not os.path.isdir(os.path.join(ROOT, w["dir"], nd)):
|
|
failures.append("nested dir %s/%s not a real dir" % (w["dir"], nd))
|
|
if set(file_count_by_dir) != set(wizards):
|
|
failures.append("wizard dir set inconsistent")
|
|
|
|
if by_type != art["totals"]["by_type"]:
|
|
failures.append("by_type not consistent with files[]")
|
|
if disk_wizards != set(wizards):
|
|
failures.append("wizard dir set on disk != artifact: %s" % disk_wizards)
|
|
|
|
# --- frozen manifest spot-check ------------------------------------
|
|
manifest = {}
|
|
for w in art["wizards"]:
|
|
for f in w.get("files", []):
|
|
manifest[f["path"]] = f
|
|
for rel in SAMPLE:
|
|
if rel not in manifest:
|
|
failures.append("sample %r missing from artifact" % rel)
|
|
continue
|
|
exp = manifest[rel]
|
|
p = os.path.join(ROOT, rel)
|
|
if not os.path.exists(p):
|
|
failures.append("sample %r missing from disk" % rel)
|
|
continue
|
|
got_sha = sha256_of(p)
|
|
got_size = os.path.getsize(p)
|
|
if got_sha != exp["sha256"]:
|
|
failures.append("%s: sha256 artifact=%r disk=%r"
|
|
% (rel, exp["sha256"], got_sha))
|
|
if got_size != exp["size"]:
|
|
failures.append("%s: size artifact=%r disk=%r"
|
|
% (rel, exp["size"], got_size))
|
|
|
|
if failures:
|
|
print("CHECK FAIL: %d mismatch(es)" % len(failures))
|
|
for f in failures:
|
|
print(" - " + f)
|
|
sys.exit(1)
|
|
|
|
print("CHECK PASS: internal consistency + %d/%d frozen manifest hashes match disk"
|
|
% (len(SAMPLE), len(SAMPLE)))
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|