80 lines
4.1 KiB
Python
80 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Summarize per-agent supervised proposals; never grant or execute authority."""
|
|
import argparse
|
|
import json
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
|
|
def summarize(record):
|
|
if record.get("mode") not in {"supervised", "autopilot"}:
|
|
raise ValueError("invalid agent mode")
|
|
if not record.get("agent_id") or not record.get("scope"):
|
|
raise ValueError("agent_id and scope required")
|
|
seen = set()
|
|
counts = Counter()
|
|
for proposal in record["proposals"]:
|
|
identity = proposal["proposal_id"]
|
|
if identity in seen:
|
|
raise ValueError("duplicate original proposal; revisions are not new trials")
|
|
seen.add(identity)
|
|
decision = proposal["disposition"]
|
|
outcome = proposal["outcome"]
|
|
if decision not in {"pending", "withdrawn", "accepted_unchanged", "accepted_revised", "rejected", "unscored"}:
|
|
raise ValueError("invalid disposition")
|
|
if outcome not in {"not_executed", "unverified", "succeeded", "failed"}:
|
|
raise ValueError("invalid outcome")
|
|
counts[decision] += 1
|
|
counts["total"] += 1
|
|
counts["outcome_" + outcome] += 1
|
|
if proposal.get("refinement_or_rescue") is True:
|
|
counts["refinement_or_rescue"] += 1
|
|
if decision == "unscored":
|
|
# Historic/broad conversation approval lacks an exact submitted
|
|
# revision. Preserve it without fabricating promotion evidence.
|
|
continue
|
|
if decision in {"accepted_unchanged", "accepted_revised", "rejected"}:
|
|
counts["adjudicated"] += 1
|
|
if outcome != "not_executed" and decision not in {"accepted_unchanged", "accepted_revised"}:
|
|
raise ValueError("executed trial requires an accepted proposal")
|
|
if decision in {"accepted_unchanged", "accepted_revised"}:
|
|
if not proposal.get("approval_ref") or not proposal.get("original_digest") or not proposal.get("approved_digest"):
|
|
raise ValueError("scored acceptance requires exact proposal and approval references")
|
|
equal = proposal["original_digest"] == proposal["approved_digest"]
|
|
if equal != (decision == "accepted_unchanged"):
|
|
raise ValueError("disposition disagrees with approved revision")
|
|
if outcome == "unverified":
|
|
counts["unverified"] += 1
|
|
if outcome in {"succeeded", "failed"}:
|
|
if not proposal.get("verification_ref") or not proposal.get("executed_digest"):
|
|
raise ValueError("verified execution requires receipt and exact executed revision")
|
|
if proposal["executed_digest"] != proposal["approved_digest"]:
|
|
raise ValueError("execution differs from approved revision")
|
|
if type(proposal.get("refinement_or_rescue")) is not bool:
|
|
raise ValueError("execution refinement/rescue must be explicit")
|
|
counts["verified_executions"] += 1
|
|
if outcome == "succeeded" and decision == "accepted_unchanged" and not proposal["refinement_or_rescue"]:
|
|
counts["unchanged_successes"] += 1
|
|
def rate(numerator, denominator):
|
|
return counts[numerator] / counts[denominator] if counts[denominator] else None
|
|
return {"agent_id": record["agent_id"], "scope": record["scope"],
|
|
"mode": record["mode"], "counts": dict(counts),
|
|
"unchanged_acceptance_rate": rate("accepted_unchanged", "adjudicated"),
|
|
"unchanged_execution_success_rate": rate("unchanged_successes", "verified_executions"),
|
|
"authority_granted": False,
|
|
"note": "Descriptive evidence only; null rates mean no eligible sample. Never promotes an agent."}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("record", type=Path)
|
|
args = parser.parse_args()
|
|
try:
|
|
report = summarize(json.loads(args.record.read_text()))
|
|
except (ValueError, KeyError, TypeError, OSError):
|
|
parser.exit(2, "Invalid supervision record; no report produced.\n")
|
|
print(json.dumps(report, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|