147 lines
4.8 KiB
Python
147 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent re-check for the Sol profile directory triage artifact (schema-B).
|
|
|
|
Re-reads the profile directory from disk, re-applies the classification rules
|
|
(schema-B: the lane's canonical rule table), and compares against the generated
|
|
artifact. Exit code 0 = artifact is correct; nonzero = artifact is wrong,
|
|
stale, or missing.
|
|
|
|
Usage: python3 check_sol_profile_triage.py [artifact.json]
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
PROFILE = "/root/.hermes/profiles/sol"
|
|
DEFAULT_ARTIFACT = "/root/.hermes/wizard-outputs/sol/sol-profile-triage-2026-11-04.json"
|
|
ARTIFACT = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_ARTIFACT
|
|
|
|
CATEGORIES = [
|
|
"identity", "config", "state_db", "cache", "sessions", "skills", "plugins",
|
|
"logs", "backups", "vault", "extension_dir", "script", "config_data",
|
|
"other",
|
|
]
|
|
|
|
IDENTITY = {".env", ".env.bak-20261014T165715Z"}
|
|
CONFIG = {".env.lock", "config.yaml", "config.yaml.bak-20261014T165715Z"}
|
|
STATE_DB = {"auth.json", "auth.lock", ".skills_prompt_snapshot.json",
|
|
"models_dev_cache.json", "models_dev_cache.etag",
|
|
"state.db", "state.db-shm", "state.db-wal",
|
|
"state.db.auto-maintenance.lock", "state.db.fts_rebuild.lock",
|
|
"state.db.quarantine.lock"}
|
|
CACHE = {"cache", "audio_cache", "image_cache", "context_length_cache.yaml"}
|
|
SESSIONS = {"sessions"}
|
|
SKILLS = {"skills"}
|
|
PLUGINS = {"plugins", "plugin-update-checks"}
|
|
LOGS = {"logs"}
|
|
BACKUPS = {"backups"}
|
|
VAULT = {"vault"}
|
|
|
|
|
|
def classify(name, is_dir):
|
|
# identity (exact top-level names)
|
|
if name in IDENTITY:
|
|
return "identity"
|
|
# config (exact top-level names)
|
|
if name in CONFIG:
|
|
return "config"
|
|
# state_db (exact names + state.db* + state.db lock files)
|
|
if name in STATE_DB:
|
|
return "state_db"
|
|
# cache (exact names + context_length_cache.yaml)
|
|
if name in CACHE:
|
|
return "cache"
|
|
if name in SESSIONS:
|
|
return "sessions"
|
|
if name in SKILLS:
|
|
return "skills"
|
|
if name in PLUGINS:
|
|
return "plugins"
|
|
if name in LOGS:
|
|
return "logs"
|
|
if name in BACKUPS:
|
|
return "backups"
|
|
if name in VAULT:
|
|
return "vault"
|
|
# extension_dir: any directory
|
|
if is_dir:
|
|
return "extension_dir"
|
|
# script: .py
|
|
if name.endswith(".py"):
|
|
return "script"
|
|
# config_data: .yaml or .json (not matched above)
|
|
if name.endswith(".yaml") or name.endswith(".json"):
|
|
return "config_data"
|
|
return "other"
|
|
|
|
|
|
def main():
|
|
# Fresh disk read
|
|
try:
|
|
names = sorted(os.listdir(PROFILE))
|
|
except Exception as e:
|
|
print("FAIL: cannot list profile: %s" % e)
|
|
sys.exit(1)
|
|
|
|
expected = []
|
|
for name in names:
|
|
path = os.path.join(PROFILE, name)
|
|
is_dir = os.path.isdir(path)
|
|
expected.append({"name": name, "type": "dir" if is_dir else "file",
|
|
"category": classify(name, is_dir)})
|
|
|
|
# Load artifact
|
|
try:
|
|
with open(ARTIFACT, encoding="utf-8") as fh:
|
|
art = json.load(fh)
|
|
except Exception as e:
|
|
print("FAIL: cannot load artifact: %s" % e)
|
|
sys.exit(1)
|
|
|
|
errs = []
|
|
if art.get("total_entries") != len(expected):
|
|
errs.append("total_entries %r != fresh count %d"
|
|
% (art.get("total_entries"), len(expected)))
|
|
|
|
disk_names = [e["name"] for e in expected]
|
|
recorded = {e["name"]: e for e in art.get("entries", [])}
|
|
|
|
if sorted(disk_names) != sorted(recorded.keys()):
|
|
errs.append("entry name sets differ: missing=%s extra=%s" % (
|
|
sorted(set(disk_names) - set(recorded)),
|
|
sorted(set(recorded) - set(disk_names))))
|
|
|
|
counts = {}
|
|
for e in expected:
|
|
counts[e["category"]] = counts.get(e["category"], 0) + 1
|
|
if art.get("categories") != counts:
|
|
errs.append("categories mismatch: artifact=%r fresh=%r"
|
|
% (art.get("categories"), counts))
|
|
|
|
for e in expected:
|
|
r = recorded.get(e["name"])
|
|
if r is None:
|
|
errs.append("artifact missing entry %s" % e["name"])
|
|
continue
|
|
if r.get("type") != e["type"]:
|
|
errs.append("%s type %r != fresh %r" % (e["name"], r.get("type"), e["type"]))
|
|
if r.get("category") != e["category"]:
|
|
errs.append("%s category %r != fresh %r" % (e["name"], r.get("category"), e["type"]))
|
|
|
|
for e in art.get("entries", []):
|
|
if e.get("category") not in CATEGORIES:
|
|
errs.append("unknown category %r for %s" % (e.get("category"), e.get("name")))
|
|
|
|
if errs:
|
|
print("FAIL: triage artifact incorrect")
|
|
for f in errs:
|
|
print(" - " + f)
|
|
sys.exit(1)
|
|
|
|
print("PASS: triage artifact verified against fresh disk read")
|
|
print(" total_entries=%d categories=%s" % (len(expected), json.dumps(counts)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|