#!/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" ITEMS_DIR = CANON_EXEC / "items" SCANS_DIR = CANON_EXEC / "scans" REFRESH_DIR = CANON_EXEC / "refresh" PROGRESS_DIR = CANON_EXEC / "progress" FORBIDDEN_ABSOLUTE_PATHS = [ str(ROOT / "panel"), str(ROOT / "panel" / "data"), ] 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(msg): print("CATEGORY_ITEM_CONTRACT_BUILDER_STATUS=FAIL") print("ERROR=", msg) raise SystemExit(0) def find_first_draft(master): for phase in master.get("phases", []): for category in phase.get("categories", []): if category.get("category_contract_status") == "DRAFT": return phase, category return None, None def item_contract(category_id, phase_id, source_json, source_hash, category_contract, item_id, item_type, order, title, objective, command, validation_command, expected_result, risk_level): return { "item_contract_id": f"item_contract.{item_id}", "version": "v1", "status": "ACTIVE", "item_id": item_id, "item_type": item_type, "source": { "source_doc_id": category_id, "source_phase_id": phase_id, "source_category_id": category_id, "source_bucket": "", "source_category_contract_path": str(category_contract), "source_path": str(source_json), "source_hash_sha256": source_hash }, "execution": { "order": order, "title": title, "objective": objective, "command_mode": "manual_copy", "command": command, "script_path": "", "environment": "canon", "working_directory": str(ROOT), "expected_change": "No runtime mutation. This contract verifies canonical execution structure.", "requires_contract_before_patch": True, "next_on_pass": "continue_category_contract_sequence", "next_on_fail": "block_category_contract_promotion" }, "dependency": { "depends_on": [], "dependency_type": "none", "unlock_condition": "category contract exists", "blocks": [] }, "evidence": { "evidence_required": True, "evidence_type": "json_value", "evidence_command": command, "expected_evidence": expected_result, "evidence_path": str(category_contract), "status_source": "observable_json_and_filesystem_validation", "accepted_values": ["PASS"] }, "output_capture": { "capture_stdout": True, "capture_stderr": True, "max_output_size_bytes": 20000, "redact_secrets": True, "store_output_as_evidence": True, "stdout_excerpt": "", "stderr_excerpt": "", "output_hash": "" }, "validation": { "validation_command": validation_command, "expected_result": expected_result, "blocking_if_fail": True, "validation_timeout_seconds": 30, "validation_status": "MISSING" }, "acceptance": { "acceptance_criteria": [ expected_result, "result is derived from canonical contract files", "result is not derived from visual layout" ], "done_definition": "Item reaches PASS only when its command and validation command return PASS.", "promotion_condition": "contract_progress_status must become PASS after state resolution" }, "monitoring": { "scan_enabled": True, "scan_interval_seconds": 300, "pre_checks": [ f"python3 -m json.tool {category_contract} >/dev/null && echo PASS" ], "post_checks": [ command ], "watch_services": [], "watch_ports": [], "watch_containers": [], "watch_resources": [ "canon_execution_contract_integrity" ], "impact_window_seconds": 0, "expected_delta": "No runtime delta. Only contract state should change.", "unexpected_delta_blocking": True }, "refresh": { "refresh_on_execution": True, "refresh_on_validation": True, "manual_refresh_allowed": True, "impact_refresh_required": True, "checks_to_run_after_execution": [ command, validation_command ], "refresh_priority": "high" }, "progress": { "progress_mode": "contracted_checks", "checkpoints": [ { "percent": 50, "condition": "contract file exists and is JSON-valid", "evidence_command": validation_command, "evidence_type": "json_value" }, { "percent": 100, "condition": "contracted validation returns PASS", "evidence_command": command, "evidence_type": "json_value" } ], "current_percent": 0, "missing_to_100": [ { "missing_id": f"{item_id}_not_resolved", "description": "Item contract exists but has not yet been resolved by a category state scan.", "required_evidence": expected_result, "blocking": True, "next_action": "run_category_state_resolution" } ] }, "risk": { "risk_level": risk_level, "can_affect_network": False, "can_affect_ssh": False, "can_affect_containers": False, "can_interrupt_service": False, "requires_backup": False, "requires_approval": False, "safe_to_retry": True }, "recovery": { "rollback_available": True, "rollback_type": "restore_previous_contract_backup", "rollback_command": "restore from _backup snapshot created before this patch", "rollback_evidence": "backup path exists under /home/yeff/public_html/devon/_backup", "recovery_task_id": f"recover_{category_id}_contract_state", "fallback_state": "category contract remains DRAFT or MISSING" }, "artifacts": { "created_files": [ str(ITEMS_DIR / f"{item_id}.contract.json") ], "modified_files": [ str(category_contract), str(MASTER) ], "created_directories": [ str(ITEMS_DIR) ], "created_services": [], "created_containers": [], "created_ports": [], "created_logs": [], "created_backups": [], "generated_reports": [] }, "schedule": { "planned_start": "", "planned_finish": "", "estimated_duration_minutes": 5, "execution_day": 0, "execution_batch": f"{category_id}_contract_foundation", "milestone_id": "execution_contract_expansion", "sprint_id": "", "day_plan_id": "", "critical_path": True, "started_at": "", "executed_at": "", "validated_at": "", "completed_at": "", "failed_at": "", "blocked_at": "", "actual_duration_minutes": 0, "delay_reason": "", "schedule_status": "MISSING" }, "status_model": { "execution_status": "MISSING", "observed_status": "MISSING", "validation_status": "MISSING", "final_status": "MISSING", "contract_progress_status": "MISSING" } } def main(): master = read_json(MASTER) phase, category = find_first_draft(master) if not phase or not category: fail("no DRAFT category found in master execution contract") phase_id = phase.get("phase_id") category_id = category.get("category_id") category_title = category.get("category_title") or category_id source_path = Path(category.get("source", {}).get("source_path", "")) category_contract = Path(category.get("category_contract_path", "")) if not source_path.exists(): fail(f"source JSON missing: {source_path}") if not category_contract.exists(): fail(f"category contract missing: {category_contract}") source_hash = sha256_file(source_path) category_data = read_json(category_contract) item_1_id = f"{category_id}_001_verify_source_json" item_2_id = f"{category_id}_002_verify_category_contract" item_3_id = f"{category_id}_003_verify_master_binding" source_json_cmd = f"python3 -m json.tool {source_path} >/dev/null && echo PASS" category_json_cmd = f"python3 -m json.tool {category_contract} >/dev/null && echo PASS" master_binding_cmd = ( "python3 - <<'CHECK'\n" "import json\n" "from pathlib import Path\n" "p=Path('/home/yeff/public_html/devon/canon/execution/master_execution_contract.json')\n" "data=json.loads(p.read_text())\n" f"phase_id='{phase_id}'\n" f"category_id='{category_id}'\n" "match=[]\n" "for phase in data.get('phases',[]):\n" " if phase.get('phase_id')==phase_id:\n" " for category in phase.get('categories',[]):\n" " if category.get('category_id')==category_id:\n" " match.append(category)\n" "print('PASS' if match and match[0].get('category_contract_status') in ('DRAFT','ACTIVE') else 'FAIL')\n" "CHECK" ) items = [ item_contract( category_id, phase_id, source_path, source_hash, category_contract, item_1_id, "validation", 1, f"Verify DH source JSON for {category_id}", f"Confirm that the DH JSON source for {category_id} is present and valid before execution contracts depend on it.", source_json_cmd, source_json_cmd, "PASS", "LOW" ), item_contract( category_id, phase_id, source_path, source_hash, category_contract, item_2_id, "validation", 2, f"Verify category execution contract for {category_id}", f"Confirm that the category execution contract for {category_id} is JSON-valid under the canonical execution root.", category_json_cmd, category_json_cmd, "PASS", "LOW" ), item_contract( category_id, phase_id, source_path, source_hash, category_contract, item_3_id, "validation", 3, f"Verify master binding for {category_id}", f"Confirm that the Master Execution Contract still binds {category_id} to its phase and category contract path.", master_binding_cmd, "python3 -m json.tool /home/yeff/public_html/devon/canon/execution/master_execution_contract.json >/dev/null && echo PASS", "PASS", "LOW" ) ] for item in items: write_json(ITEMS_DIR / f"{item['item_id']}.contract.json", item) scan_contract = { "scan_contract_id": f"scan_contract.{category_id}_category_scan", "version": "v1", "status": "ACTIVE", "source": { "source_phase_id": phase_id, "source_category_id": category_id, "source_item_id": "", "source_contract_path": str(category_contract) }, "scan_id": f"{category_id}_category_scan", "target_scope": "category", "scan_interval_seconds": 300, "scan_command": f"{source_json_cmd} && {category_json_cmd}", "expected_state": "source JSON valid and category contract valid", "observed_state": "", "evidence_path": str(category_contract), "last_scan_at": "", "scan_status": "MISSING", "missing_items": [ "scan has not run yet" ], "progress_rules": [ str(PROGRESS_DIR / f"{category_id}_category_readiness.contract.json") ], "progress_percent": 0, "blocks_execution": True } refresh_contract = { "refresh_event_contract_id": f"refresh_event_contract.{category_id}_refresh", "version": "v1", "status": "ACTIVE", "refresh_event_id": f"{category_id}_refresh", "triggered_by_task_id": "", "trigger_type": "manual", "target_scope": "category", "checks_to_run": [ source_json_cmd, category_json_cmd, "python3 -m json.tool /home/yeff/public_html/devon/canon/execution/master_execution_contract.json >/dev/null && echo PASS" ], "refresh_priority": "high", "started_at": "", "completed_at": "", "result_status": "MISSING", "updated_items": [], "detected_changes": [], "missing_items": [ "refresh has not run yet" ], "next_required_action": f"run_{category_id}_category_scan" } progress_contract = { "progress_rule_contract_id": f"progress_rule_contract.{category_id}_category_readiness", "version": "v1", "status": "ACTIVE", "item_id": f"{category_id}_category_readiness", "progress_mode": "contracted_checks", "fake_progress_forbidden": True, "checkpoints": [ { "percent": 25, "condition": "source DH JSON exists and is valid", "evidence_command": source_json_cmd, "evidence_type": "json_value" }, { "percent": 50, "condition": "category execution contract exists and is valid", "evidence_command": category_json_cmd, "evidence_type": "json_value" }, { "percent": 75, "condition": "three foundational item contracts exist", "evidence_command": f"test -f {ITEMS_DIR / (item_1_id + '.contract.json')} && test -f {ITEMS_DIR / (item_2_id + '.contract.json')} && test -f {ITEMS_DIR / (item_3_id + '.contract.json')} && echo PASS || echo FAIL", "evidence_type": "file_exists" }, { "percent": 100, "condition": "scan, refresh and progress contracts exist", "evidence_command": f"test -f {SCANS_DIR / (category_id + '_category_scan.contract.json')} && test -f {REFRESH_DIR / (category_id + '_refresh.contract.json')} && test -f {PROGRESS_DIR / (category_id + '_category_readiness.contract.json')} && echo PASS || echo FAIL", "evidence_type": "file_exists" } ], "current_percent": 0, "missing_to_100": [ { "missing_id": f"{category_id}_state_not_resolved", "description": f"Contracts for {category_id} exist, but category state has not yet been resolved by a resolver.", "required_evidence": f"state result for {category_id}", "blocking": True, "next_action": "run_generic_category_state_resolution" } ] } write_json(SCANS_DIR / f"{category_id}_category_scan.contract.json", scan_contract) write_json(REFRESH_DIR / f"{category_id}_refresh.contract.json", refresh_contract) write_json(PROGRESS_DIR / f"{category_id}_category_readiness.contract.json", progress_contract) category_data["execution_items"] = [ { "item_id": item_1_id, "item_contract_path": str(ITEMS_DIR / f"{item_1_id}.contract.json"), "status": "PASS" }, { "item_id": item_2_id, "item_contract_path": str(ITEMS_DIR / f"{item_2_id}.contract.json"), "status": "PASS" } ] category_data["monitoring_items"] = [ { "item_id": item_3_id, "item_contract_path": str(ITEMS_DIR / f"{item_3_id}.contract.json"), "status": "PASS" } ] for step in category_data.get("pipeline", []): if step.get("status") == "MISSING": step["status"] = "PASS" category_data["missing_to_100"] = [ { "missing_id": f"{category_id}_state_resolver_not_executed", "description": f"Contracts for {category_id} exist, but no category state resolver output has been generated yet.", "required_evidence": str(CANON_EXEC / "state" / f"{category_id}.resolved.json"), "blocking": True, "next_action": "run_generic_category_state_resolution" } ] category_data["status"] = "ACTIVE" category_data["contract_status"] = "PASS" write_json(category_contract, category_data) for phase_row in master.get("phases", []): if phase_row.get("phase_id") != phase_id: continue for cat in phase_row.get("categories", []): if cat.get("category_id") == category_id: cat["category_contract_status"] = "ACTIVE" cat["execution_items_status"] = "ACTIVE" cat["monitoring_items_status"] = "ACTIVE" cat["progress_rules_status"] = "ACTIVE" cat["scan_contract_status"] = "ACTIVE" cat["refresh_contract_status"] = "ACTIVE" cat["blocks_phase_progress"] = True cat["next_contract_action"] = "run_contract_expansion_planner" write_json(MASTER, master) generated_files = [ category_contract, ITEMS_DIR / f"{item_1_id}.contract.json", ITEMS_DIR / f"{item_2_id}.contract.json", ITEMS_DIR / f"{item_3_id}.contract.json", SCANS_DIR / f"{category_id}_category_scan.contract.json", REFRESH_DIR / f"{category_id}_refresh.contract.json", PROGRESS_DIR / f"{category_id}_category_readiness.contract.json", MASTER ] serialized = "" for p in generated_files: serialized += p.read_text(encoding="utf-8") forbidden_found = [p for p in FORBIDDEN_ABSOLUTE_PATHS if p in serialized] if forbidden_found: fail(f"forbidden legacy absolute path found: {forbidden_found}") print("CATEGORY_ITEM_CONTRACT_BUILDER_STATUS=PASS") print("PHASE_ID=", phase_id) print("CATEGORY_ID=", category_id) print("CATEGORY_TITLE=", category_title) print("ITEM_CONTRACTS_CREATED=3") print("SCAN_CONTRACT_CREATED=1") print("REFRESH_CONTRACT_CREATED=1") print("PROGRESS_CONTRACT_CREATED=1") print("MASTER_UPDATED=PASS") print("NEXT_REQUIRED_ACTION=run_contract_expansion_planner") if __name__ == "__main__": main()