Codex
104 lines
5.2 KiB
Python
104 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate reported decision consistency, not evidence truth or live execution."""
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
|
|
def validate(record, previous=None):
|
|
errors = []
|
|
if not isinstance(record, dict):
|
|
return ["record must be an object"]
|
|
for key in ("issue_id", "baseline", "failure_class", "status"):
|
|
if not isinstance(record.get(key), str) or not record[key].strip():
|
|
errors.append(key + " must be nonempty text")
|
|
if type(record.get("schema_version")) is not int or record["schema_version"] != 1:
|
|
errors.append("schema_version must be 1")
|
|
if record.get("failure_class") not in ("environment", "authority", "contract", "implementation", "reasoning"):
|
|
errors.append("invalid failure_class")
|
|
if record.get("status") not in ("investigating", "blocked", "escalated", "resolved"):
|
|
errors.append("invalid status")
|
|
count = record.get("same_class_failures")
|
|
if type(count) is not int or count < 0:
|
|
errors.append("same_class_failures must be a nonnegative integer")
|
|
elif count >= 2 and not text(record.get("reassessment")):
|
|
errors.append("repeated failure requires reassessment")
|
|
for key in ("evidence", "unresolved"):
|
|
value = record.get(key)
|
|
if not isinstance(value, list) or any(not text(item) for item in value):
|
|
errors.append(key + " must be a list of nonempty references/issues")
|
|
escalation = record.get("escalation")
|
|
if not isinstance(escalation, dict):
|
|
errors.append("escalation must be an object")
|
|
escalation = {}
|
|
target = escalation.get("target")
|
|
if target not in ("none", "terra", "sol", "astra"):
|
|
errors.append("invalid escalation target")
|
|
if target != "none" and not text(escalation.get("reason")):
|
|
errors.append("escalation requires a reason")
|
|
if target == "astra" and record.get("failure_class") in ("environment", "authority"):
|
|
errors.append("Astra cannot repair environment or authority")
|
|
consultations = escalation.get("expert_consultations")
|
|
if type(consultations) is not int or consultations < 0:
|
|
errors.append("expert_consultations must be a nonnegative integer")
|
|
elif consultations > 1 and not text(escalation.get("followup_basis")):
|
|
errors.append("additional consultation requires new evidence or a specific omission")
|
|
if record.get("status") == "escalated" and target == "none":
|
|
errors.append("escalated status requires a target")
|
|
if record.get("status") == "resolved":
|
|
verification = record.get("verification", {})
|
|
if not isinstance(verification, dict):
|
|
verification = {}
|
|
if (verification.get("status") != "passed"
|
|
or verification.get("baseline") != record.get("baseline")
|
|
or verification.get("parent_understands") is not True):
|
|
errors.append("resolution requires understood, passed verification on current baseline")
|
|
if record.get("unresolved") != [] or not record.get("evidence"):
|
|
errors.append("resolution requires evidence and no unresolved items")
|
|
if not text(record.get("recovery_route")):
|
|
errors.append("resolution requires a recovery route")
|
|
if previous is not None:
|
|
if not isinstance(previous, dict):
|
|
errors.append("previous record must be an object")
|
|
else:
|
|
if previous.get("issue_id") != record.get("issue_id"):
|
|
errors.append("issue_id changed across continuation")
|
|
old = previous.get("same_class_failures")
|
|
if (previous.get("failure_class") == record.get("failure_class")
|
|
and type(old) is int and type(count) is int and count < old):
|
|
errors.append("failure counter decreased within the same class")
|
|
previous_escalation = previous.get("escalation")
|
|
old_consultations = (previous_escalation.get("expert_consultations")
|
|
if isinstance(previous_escalation, dict) else None)
|
|
if type(old_consultations) is not int or old_consultations < 0:
|
|
errors.append("previous expert_consultations must be a nonnegative integer")
|
|
elif (type(consultations) is int and consultations >= 0
|
|
and consultations < old_consultations):
|
|
errors.append("expert consultation counter decreased across continuation")
|
|
return errors
|
|
|
|
|
|
def text(value):
|
|
return isinstance(value, str) and bool(value.strip())
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("record")
|
|
parser.add_argument("--previous")
|
|
args = parser.parse_args()
|
|
try:
|
|
record = json.loads(Path(args.record).read_text(encoding="utf-8"))
|
|
previous = json.loads(Path(args.previous).read_text(encoding="utf-8")) if args.previous else None
|
|
errors = validate(record, previous)
|
|
except (OSError, ValueError) as exc:
|
|
print("Invalid decision record: " + str(exc), file=sys.stderr)
|
|
return 1
|
|
print("\n".join(errors) if errors else "Decision record consistency passed (truth not verified)")
|
|
return int(bool(errors))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|