#!/usr/bin/env python3 import json import hashlib from pathlib import Path from datetime import datetime, timezone ROOT = Path("/home/yeff/public_html/devon") CANON_EXEC = ROOT / "canon" / "execution" MASTER = CANON_EXEC / "master_execution_contract.json" PLAN = CANON_EXEC / "state" / "contract_expansion_plan.json" FORBIDDEN_ABSOLUTE_PATHS = [ str(ROOT / "panel"), str(ROOT / "panel" / "data"), ] def utc_now(): return datetime.now(timezone.utc).isoformat() def read_json(path): return json.loads(path.read_text(encoding="utf-8")) def write_json(path, data): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def sha256_file(path): h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def fail(message): print("CATEGORY_CONTRACT_BUILDER_STATUS=FAIL") print("ERROR=", message) raise SystemExit(0) def main(): plan = read_json(PLAN) if plan.get("final_status") != "PASS": fail("expansion plan is not PASS") next_category = plan.get("next_category") or {} phase_id = next_category.get("phase_id", "") category_id = next_category.get("category_id", "") category_title = next_category.get("category_title", "") or category_id source_path_raw = next_category.get("source_path", "") contract_path_raw = next_category.get("category_contract_path", "") if not phase_id or not category_id or not source_path_raw or not contract_path_raw: fail("next_category missing phase_id, category_id, source_path or category_contract_path") source_path = Path(source_path_raw) contract_path = Path(contract_path_raw) if not source_path.exists(): fail(f"source json missing: {source_path}") if str(contract_path).startswith(str(ROOT / "panel")): fail("contract path points to forbidden legacy panel root") source_hash = sha256_file(source_path) master = read_json(MASTER) contract = { "category_contract_id": f"category_execution_contract.{category_id}", "version": "v1", "status": "DRAFT", "source": { "source_type": "dh_json_category", "source_phase_id": phase_id, "source_category_id": category_id, "source_category_title": category_title, "source_path": str(source_path), "source_hash_sha256": source_hash, "source_master_contract_path": str(MASTER) }, "category": { "category_id": category_id, "category_title": category_title, "category_authority": f"Operational authority for translating the DH category {category_id} into execution, verification, progress, scan and refresh contracts before any panel renderer can use it.", "phase_id": phase_id }, "execution_scope": { "purpose": f"Create the first execution-level contract shell for {category_id}, binding the DH source to future item contracts without inventing commands, progress or runtime evidence.", "allowed_operations": [ "read_dh_json_category_source", "create_execution_items_after_category_contract", "create_monitoring_items_after_category_contract", "create_scan_contract_after_category_contract", "create_refresh_contract_after_category_contract", "create_progress_rule_after_category_contract" ], "forbidden_operations": [ "create_runtime_task_without_item_contract", "mark_pass_without_observable_evidence", "derive_panel_status_from_layout", "use_legacy_panel_directory_as_source_or_destination", "invent_phase_or_category_outside_dh" ], "requires_contract_before_patch": True }, "monitoring_scope": { "scan_enabled": True, "refresh_enabled": True, "observed_targets": [ "category_contract_presence", "category_contract_json_validity", "source_dh_json_presence", "source_hash_match", "missing_execution_items", "missing_monitoring_items", "missing_progress_rules", "missing_scan_contract", "missing_refresh_contract" ], "missing_to_100_required": True }, "contract_paths": { "category_contract_path": str(contract_path), "items_root": str(CANON_EXEC / "items"), "scans_root": str(CANON_EXEC / "scans"), "refresh_root": str(CANON_EXEC / "refresh"), "progress_root": str(CANON_EXEC / "progress") }, "execution_items": [], "monitoring_items": [], "pipeline": [ { "step_id": f"{category_id}_001_category_contract_created", "order": 1, "title": f"Create category execution contract for {category_id}", "depends_on": [], "status": "PASS" }, { "step_id": f"{category_id}_002_define_execution_items", "order": 2, "title": f"Define execution items for {category_id}", "depends_on": [ f"{category_id}_001_category_contract_created" ], "status": "MISSING" }, { "step_id": f"{category_id}_003_define_monitoring_items", "order": 3, "title": f"Define monitoring items for {category_id}", "depends_on": [ f"{category_id}_002_define_execution_items" ], "status": "MISSING" }, { "step_id": f"{category_id}_004_define_progress_rules", "order": 4, "title": f"Define progress rules for {category_id}", "depends_on": [ f"{category_id}_003_define_monitoring_items" ], "status": "MISSING" } ], "kanban_policy": { "columns": [ "Backlog", "Ready", "In Progress", "Waiting Evidence", "Validation", "Blocked", "Done" ], "pipeline_does_not_live_inside_kanban": True }, "timeline_policy": { "planned_required": True, "actual_required": True, "schedule_status_values": [ "MISSING", "PLANNED", "READY", "ON_TIME", "DELAYED", "BLOCKED", "DONE", "RESCHEDULED" ] }, "progress_rules": { "progress_mode": "contracted_checks", "fake_progress_forbidden": True, "category_progress_source": "execution_items_and_monitoring_items", "item_progress_source": "observable_evidence_and_validation" }, "scan_contract": { "scan_enabled": True, "default_interval_seconds": 300, "manual_recheck_allowed": True }, "refresh_events": { "refresh_on_execution": True, "refresh_on_validation": True, "refresh_on_manual_recheck": True, "impact_watch_required": True }, "missing_to_100": [ { "missing_id": f"{category_id}_missing_execution_items", "description": f"Execution items for {category_id} have not been created yet.", "required_evidence": str(CANON_EXEC / "items" / f"{category_id}_*.contract.json"), "blocking": True, "next_action": f"create_item_contracts_for_{category_id}_execution" }, { "missing_id": f"{category_id}_missing_monitoring_items", "description": f"Monitoring items for {category_id} have not been created yet.", "required_evidence": str(CANON_EXEC / "items" / f"{category_id}_*.contract.json"), "blocking": True, "next_action": f"create_item_contracts_for_{category_id}_monitoring" }, { "missing_id": f"{category_id}_missing_progress_rules", "description": f"Progress rules for {category_id} have not been created yet.", "required_evidence": str(CANON_EXEC / "progress" / f"{category_id}_*.contract.json"), "blocking": True, "next_action": f"create_progress_rule_contract_for_{category_id}" } ], "promotion_gate": { "promotion_allowed_when": [ "category_contract_json_valid", "source_hash_matches_dh_json", "execution_items_status_PASS", "monitoring_items_status_PASS", "progress_rules_status_PASS", "scan_contract_status_PASS", "refresh_contract_status_PASS" ], "promotion_blocked_when": [ "missing_execution_items", "missing_monitoring_items", "missing_progress_rules", "missing_scan_contract", "missing_refresh_contract", "source_hash_conflict" ] }, "contract_status": "MISSING" } serialized_contract = json.dumps(contract, ensure_ascii=False) forbidden_found = [x for x in FORBIDDEN_ABSOLUTE_PATHS if x in serialized_contract] if forbidden_found: fail(f"forbidden absolute path found in category contract: {forbidden_found}") found_in_master = False for phase in master.get("phases", []): if phase.get("phase_id") != phase_id: continue for category in phase.get("categories", []): if category.get("category_id") == category_id: found_in_master = True category["category_contract_status"] = "DRAFT" category["category_contract_path"] = str(contract_path) category["execution_items_status"] = "MISSING" category["monitoring_items_status"] = "MISSING" category["progress_rules_status"] = "MISSING" category["scan_contract_status"] = "MISSING" category["refresh_contract_status"] = "MISSING" category["blocks_phase_progress"] = True category["next_contract_action"] = "create_execution_and_monitoring_items_for_category" if not found_in_master: fail("category not found in master execution contract") write_json(contract_path, contract) write_json(MASTER, master) print("CATEGORY_CONTRACT_BUILDER_STATUS=PASS") print("PHASE_ID=", phase_id) print("CATEGORY_ID=", category_id) print("CATEGORY_TITLE=", category_title) print("SOURCE_PATH=", source_path) print("CATEGORY_CONTRACT=", contract_path) print("SOURCE_HASH=", source_hash) print("NEXT_REQUIRED_ACTION=create_item_contracts_for_category") if __name__ == "__main__": main()