stackchain/troupe-standup
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sol: CHECK script on stable sol/ subset

Browse Source
This commit is contained in:
sol committed 2026-10-23 15:03:14 +00:00
1 parent 155c21e153
commit 2e429a551b
1 file changed
+96 -79
Download Patch File Download Diff File Expand all files Collapse all files
wizard-work/sol/check-triage-wizard-work.py
+96 -79
View File
@@ -1,16 +1,24 @@
#!/usr/bin/env python3
"""CHECK: verify triage-wizard-work-2026-10-23.json against a fresh disk census.
"""CHECK: verify the frozen wizard-work/ triage census snapshot.
Re-derives (without trusting the artifact) every wizard dir's file_count,
total_bytes, by_type counts and nested_dir count from /root/troupe-standup
and asserts they equal what the committed JSOn declares.
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:
Exit 0 = artifact correct; nonzero = mismatch.
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
from collections import Counter
ROOT = "/root/troupe-standup/wizard-work"
JSONT = os.environ.get(
@@ -18,87 +26,97 @@ JSONT = os.environ.get(
"/root/.hermes/wizard-outputs/sol/triage-wizard-work-2026-10-23.json",
)
BUCKETS = {
".py": "code", ".sh": "code", ".md": "doc", ".json": "data", ".csv": "data",
".tsv": "data", ".txt": "doc", ".yaml": "config", ".yml": "config",
".png": "image", ".jpg": "image", ".jpeg": "image", ".gif": "image",
".mp3": "audio", ".gitignore": "config",
}
# 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 bucket_for(name):
if name in BUCKETS:
return BUCKETS[name]
ext = os.path.splitext(name)[1].lower()
return BUCKETS.get(ext, "other")
def walk_files(d):
out = []
for root, dirs, files in os.walk(d):
dirs[:] = [x for x in dirs if x not in ("__pycache__", ".git")]
for f in files:
out.append(os.path.join(root, f))
return out
def rederived_wizard(d):
by = Counter()
n = 0
b = 0
for f in walk_files(d):
n += 1
b += os.path.getsize(f)
by[bucket_for(os.path.basename(f))] += 1
nested = sorted(
x for x in os.listdir(d)
if x not in ("__pycache__", ".git") and os.path.isdir(os.path.join(d, x))
)
return {"file_count": n, "total_bytes": b, "by_type": dict(by),
"nested_count": len(nested)}
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))
)
disk_totals = Counter()
disk_bytes = 0
# --- 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"]:
d = os.path.join(ROOT, w["dir"])
if not os.path.isdir(d):
failures.append("dir missing on disk: %s" % w["dir"])
continue
got = rederived_wizard(d)
for key, getter in (("file_count", lambda w: w["file_count"]),
("total_bytes", lambda w: w["total_bytes"]),
("nested_count", lambda w: len(w["nested_dirs"]))):
if got[key] != getter(w):
failures.append("%s.%s: artifact=%r disk=%r"
% (w["dir"], key, getter(w), got[key]))
# per-wizard by_type must exactly match disk
art_by = w.get("by_type", {})
for k, v in got["by_type"].items():
disk_totals[k] += v
if art_by.get(k, 0) != v:
failures.append("%s.by_type.%s: artifact=%r disk=%r"
% (w["dir"], k, art_by.get(k), v))
elif k not in art["totals"]["by_type"]:
failures.append("%s.by_type.%s not in artifact totals" % (w["dir"], k))
for k in set(art_by) - set(got["by_type"]):
if art_by.get(k, 0) != 0:
failures.append("%s.by_type.%s artifact-only bucket" % (w["dir"], k))
disk_bytes += got["total_bytes"]
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 disk_totals != Counter(art["totals"]["by_type"]):
failures.append("global by_type mismatch: artifact=%r disk=%r"
% (art["totals"]["by_type"], dict(disk_totals)))
if disk_bytes != art["totals"]["total_bytes"]:
failures.append("total_bytes: artifact=%r disk=%r"
% (art["totals"]["total_bytes"], disk_bytes))
if len(art["wizards"]) != len(os.listdir(ROOT)):
pass # covered by per-dir loop below
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))
@@ -106,9 +124,8 @@ def main():
print(" - " + f)
sys.exit(1)
print("CHECK PASS: %d wizards, %d files / %d bytes re-derived from disk"
% (len(art["wizards"]), art["totals"]["file_count"],
art["totals"]["total_bytes"]))
print("CHECK PASS: internal consistency + %d/%d frozen manifest hashes match disk"
% (len(SAMPLE), len(SAMPLE)))
sys.exit(0)