105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Compile experience receipts into non-enforcing candidate skill rules."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import json
|
||
|
|
from collections import defaultdict
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
DEFAULT_RECEIPTS = Path(__file__).resolve().parents[1] / "references" / "experience-receipts"
|
||
|
|
REQUIRED_FIELDS = {
|
||
|
|
"schema",
|
||
|
|
"receipt_id",
|
||
|
|
"intent",
|
||
|
|
"scope",
|
||
|
|
"input",
|
||
|
|
"evidence",
|
||
|
|
"decision",
|
||
|
|
"action",
|
||
|
|
"observed_result",
|
||
|
|
"correction",
|
||
|
|
"invariant_id",
|
||
|
|
"invariant",
|
||
|
|
"verification",
|
||
|
|
"receipt_path",
|
||
|
|
"promotion_state",
|
||
|
|
"recorded_at",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def load_receipts(directory: Path = DEFAULT_RECEIPTS) -> list[dict]:
|
||
|
|
receipts = []
|
||
|
|
for path in sorted(directory.glob("*.json")):
|
||
|
|
with path.open(encoding="utf-8") as handle:
|
||
|
|
receipt = json.load(handle)
|
||
|
|
missing = REQUIRED_FIELDS.difference(receipt)
|
||
|
|
if missing:
|
||
|
|
raise ValueError(f"{path.name} missing fields: {sorted(missing)}")
|
||
|
|
if receipt["schema"] != "guanghu.ops-experience-receipt/v1":
|
||
|
|
raise ValueError(f"{path.name} has unsupported schema")
|
||
|
|
if not receipt["evidence"] or not receipt["verification"]:
|
||
|
|
raise ValueError(f"{path.name} must contain evidence and verification")
|
||
|
|
receipts.append(receipt)
|
||
|
|
return receipts
|
||
|
|
|
||
|
|
|
||
|
|
def compile_candidates(receipts: list[dict]) -> dict:
|
||
|
|
grouped: dict[str, list[dict]] = defaultdict(list)
|
||
|
|
for receipt in receipts:
|
||
|
|
grouped[receipt["invariant_id"]].append(receipt)
|
||
|
|
|
||
|
|
candidates = []
|
||
|
|
for invariant_id, items in sorted(grouped.items()):
|
||
|
|
invariant_texts = {item["invariant"] for item in items}
|
||
|
|
if len(invariant_texts) != 1:
|
||
|
|
raise ValueError(f"{invariant_id} has conflicting invariant text")
|
||
|
|
candidates.append(
|
||
|
|
{
|
||
|
|
"candidate_id": f"CANDIDATE-{invariant_id}",
|
||
|
|
"invariant_id": invariant_id,
|
||
|
|
"invariant": items[0]["invariant"],
|
||
|
|
"scopes": sorted({item["scope"] for item in items}),
|
||
|
|
"source_receipts": [item["receipt_id"] for item in items],
|
||
|
|
"corrections": sorted({item["correction"] for item in items}),
|
||
|
|
"verification": sorted({step for item in items for step in item["verification"]}),
|
||
|
|
"enforcement_level": "CANDIDATE_ONLY",
|
||
|
|
"hard_block_allowed": False,
|
||
|
|
"promotion_requirements": [
|
||
|
|
"current fact-source review",
|
||
|
|
"applicability and counterexample review",
|
||
|
|
"automated or reproducible tests",
|
||
|
|
"registry update with recovery route",
|
||
|
|
"explicit governance approval for hard enforcement",
|
||
|
|
],
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
return {
|
||
|
|
"schema": "guanghu.emergent-skill-candidates/v1",
|
||
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
||
|
|
"source_receipt_count": len(receipts),
|
||
|
|
"candidate_count": len(candidates),
|
||
|
|
"enforcement_policy": "Compilation never promotes or enforces a rule.",
|
||
|
|
"candidates": candidates,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
parser = argparse.ArgumentParser()
|
||
|
|
parser.add_argument("--receipts", type=Path, default=DEFAULT_RECEIPTS)
|
||
|
|
parser.add_argument("--output", type=Path)
|
||
|
|
args = parser.parse_args()
|
||
|
|
result = compile_candidates(load_receipts(args.receipts))
|
||
|
|
payload = json.dumps(result, ensure_ascii=False, indent=2) + "\n"
|
||
|
|
if args.output:
|
||
|
|
args.output.write_text(payload, encoding="utf-8")
|
||
|
|
else:
|
||
|
|
print(payload, end="")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|