from __future__ import annotations import argparse import json from collections import Counter from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Sequence from aureon.obsidian_paths import resolve_obsidian_note_path from aureon.autonomous.aureon_capability_forge import REPO_ROOT from aureon.autonomous.aureon_frontend_evolution_queue import ( DEFAULT_OUTPUT_JSON as DEFAULT_EVOLUTION_JSON, DEFAULT_PUBLIC_JSON as DEFAULT_EVOLUTION_PUBLIC_JSON, build_frontend_evolution_queue, write_queue, ) SCHEMA_VERSION = "aureon-evolution-queue-autonomous-certification-v1" DEFAULT_STATE_PATH = Path("state/aureon_evolution_queue_autonomous_certification_last_run.json") DEFAULT_AUDIT_JSON = Path("docs/audits/aureon_evolution_queue_autonomous_certification.json") DEFAULT_AUDIT_MD = Path("docs/audits/aureon_evolution_queue_autonomous_certification.md") DEFAULT_PUBLIC_JSON = Path("frontend/public/aureon_evolution_queue_autonomous_certification.json") CODING_BRIDGE_EVIDENCE_PATHS = [ Path("state/aureon_coding_organism_last_run.json"), Path("docs/audits/aureon_coding_organism_bridge.json"), Path("frontend/public/aureon_coding_organism_bridge.json"), ] MANUAL_BOUNDARY_SAFETY = { "live_trading_boundary", "payment_or_kyc_boundary", "credential_or_auth_boundary", "admin_or_tenant_boundary", } def _default_root() -> Path: return REPO_ROOT def _utc_now() -> str: return datetime.now(timezone.utc).isoformat() def _rooted(root: Path, rel: Path) -> Path: return rel if rel.is_absolute() else root / rel def _read_json(path: Path) -> Dict[str, Any]: try: if path.exists(): payload = json.loads(path.read_text(encoding="utf-8")) return payload if isinstance(payload, dict) else {} except Exception: return {} return {} def _write_json(path: Path, payload: Dict[str, Any]) -> Dict[str, Any]: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, sort_keys=True, default=str), encoding="utf-8") return {"path": str(path), "bytes": path.stat().st_size} def _write_text(path: Path, text: str) -> Dict[str, Any]: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") return {"path": str(path), "bytes": path.stat().st_size} def _load_or_build_queue(root: Path) -> Dict[str, Any]: queue = _read_json(_rooted(root, DEFAULT_EVOLUTION_PUBLIC_JSON)) or _read_json(_rooted(root, DEFAULT_EVOLUTION_JSON)) if queue.get("work_orders"): return queue built = build_frontend_evolution_queue(root).to_dict() write_queue( build_frontend_evolution_queue(root), markdown_path=_rooted(root, Path("docs/audits/aureon_frontend_evolution_queue.md")), json_path=_rooted(root, DEFAULT_EVOLUTION_JSON), public_json_path=_rooted(root, DEFAULT_EVOLUTION_PUBLIC_JSON), vault_path=resolve_obsidian_note_path( Path(".obsidian/Aureon Self Understanding/aureon_frontend_evolution_queue.md"), repo_root=root, ), ) return built def _outcome_for(order: Dict[str, Any]) -> Dict[str, Any]: status = str(order.get("status") or "") safety = str(order.get("safety_boundary") or order.get("evidence", {}).get("safety_class") or "") if status == "blocked_security_review": return { "outcome": "read_only_blocker_card_certified", "bucket": "manual_boundary_visible", "handover_state": "boundary_visible", "description": "Unsafe interactive authority is not mounted; a read-only blocker card is the autonomous output.", } if status == "needs_safe_status_adapter": return { "outcome": "safe_status_adapter_certified", "bucket": "safe_status_adapter", "handover_state": "safe_status_visible", "description": "Credential/auth/admin state may be shown as metadata only through a safe status adapter.", } if status == "link_generated_output": return { "outcome": "evidence_link_certified", "bucket": "evidence_link", "handover_state": "evidence_link_visible", "description": "Generated accounting or document output is represented as an evidence link, not executed.", } if status == "archive_candidate": return { "outcome": "archive_decision_certified", "bucket": "archive_decision", "handover_state": "archive_decision_visible", "description": "Low-value support file is classified for archive/supporting status instead of a dashboard panel.", } if status == "ready_for_frontend_adapter": return { "outcome": "read_only_adapter_certified", "bucket": "read_only_adapter", "handover_state": "adapter_ready", "description": "A read-only adapter can be generated and mounted without importing legacy controls.", } if safety in MANUAL_BOUNDARY_SAFETY: return { "outcome": "manual_boundary_status_certified", "bucket": "manual_boundary_visible", "handover_state": "boundary_visible", "description": "Manual authority boundary is visible and does not block safe queue throughput.", } return { "outcome": "watch_only_certified", "bucket": "watch_only", "handover_state": "watch_record_visible", "description": "Work order is retained as watch-only evidence.", } def _proof(order: Dict[str, Any], outcome: Dict[str, Any]) -> List[Dict[str, Any]]: data_contract = order.get("data_contract") if isinstance(order.get("data_contract"), dict) else {} acceptance_tests = order.get("acceptance_tests") if isinstance(order.get("acceptance_tests"), list) else [] source_path = str(order.get("source_path") or "") target_screen = str(order.get("target_screen") or "") frontend_action = str(order.get("frontend_action") or "") status = str(order.get("status") or "") safety = str(order.get("safety_boundary") or order.get("evidence", {}).get("safety_class") or "") return [ { "id": "manifest_contract_present", "ok": bool(order.get("id") and source_path and target_screen and status), "evidence": f"{source_path} -> {target_screen} ({status})", }, { "id": "safe_data_contract_declared", "ok": bool(data_contract.get("safe_fields") and data_contract.get("secret_policy") == "metadata_only_hide_values"), "evidence": data_contract.get("expected_topic") or "missing", }, { "id": "legacy_execution_not_required", "ok": True, "evidence": "certification reads manifest evidence only; no legacy dashboard import or execution", }, { "id": "boundary_handled", "ok": ( outcome["bucket"] != "manual_boundary_visible" or "read-only" in frontend_action.lower() or "status adapter" in frontend_action.lower() or safety in MANUAL_BOUNDARY_SAFETY ), "evidence": frontend_action or outcome["description"], }, { "id": "acceptance_tests_present", "ok": len(acceptance_tests) >= 3, "evidence": f"{len(acceptance_tests)} acceptance test(s)", }, ] def _certify_order(order: Dict[str, Any], queue_source: str) -> Dict[str, Any]: outcome = _outcome_for(order) proof = _proof(order, outcome) passed = all(item["ok"] for item in proof) fake_pass = outcome["handover_state"] and not passed return { "id": order.get("id"), "title": order.get("title"), "source_path": order.get("source_path"), "target_screen": order.get("target_screen"), "priority": order.get("priority"), "input_status": order.get("status"), "safety_boundary": order.get("safety_boundary") or order.get("evidence", {}).get("safety_class"), "autonomous_outcome": outcome["outcome"], "outcome_bucket": outcome["bucket"], "description": outcome["description"], "proof_checklist": proof, "ok": passed and not fake_pass, "fake_pass": bool(fake_pass), "handover_state": { "state": outcome["handover_state"] if passed else "blocked_by_failed_proof", "ready": passed and not fake_pass, "manual_boundary_visible": outcome["bucket"] == "manual_boundary_visible", }, "actual_artifacts": { "queue_source": queue_source, "public_evidence": "/aureon_evolution_queue_autonomous_certification.json", "source_path": order.get("source_path"), }, "repair_attempts": [], "failure_reason": "" if passed else ", ".join(item["id"] for item in proof if not item["ok"]), } def _attach_to_coding_bridge(root: Path, compact: Dict[str, Any]) -> List[Dict[str, Any]]: writes: List[Dict[str, Any]] = [] for rel in CODING_BRIDGE_EVIDENCE_PATHS: path = _rooted(root, rel) payload = _read_json(path) if not payload: continue payload["evolution_queue_autonomous_certification"] = compact summary = payload.setdefault("summary", {}) if isinstance(summary, dict): compact_summary = compact.get("summary") or {} summary["evolution_queue_certification_status"] = compact.get("status") summary["evolution_queue_certification_processed_count"] = compact_summary.get("processed_count") summary["evolution_queue_certification_queue_count"] = compact_summary.get("queue_count") writes.append(_write_json(path, payload)) return writes def _make_markdown(report: Dict[str, Any]) -> str: summary = report.get("summary", {}) lines = [ "# Aureon Evolution Queue Autonomous Certification", "", f"- status: {report.get('status')}", f"- processed: {summary.get('processed_count')}/{summary.get('queue_count')}", f"- throughput: {summary.get('throughput_percent')}%", f"- safe autonomous outcomes: {summary.get('safe_autonomous_outcome_count')}", f"- manual boundaries visible: {summary.get('manual_boundary_visible_count')}", f"- failed: {summary.get('failed_count')}", f"- fake passes: {summary.get('fake_pass_count')}", "", "## Outcome Buckets", ] for key, value in sorted((summary.get("outcome_buckets") or {}).items()): lines.append(f"- {key}: {value}") lines.extend(["", "## First Cases"]) for case in report.get("cases", [])[:20]: lines.append( f"- {case.get('id')}: {case.get('autonomous_outcome')} -> {case.get('handover_state', {}).get('state')}" ) return "\n".join(lines) + "\n" def build_and_write_evolution_queue_autonomous_certification( *, root: Optional[Path] = None, ) -> Dict[str, Any]: root = Path(root or _default_root()).resolve() generated_at = _utc_now() queue = _load_or_build_queue(root) orders = [order for order in queue.get("work_orders", []) if isinstance(order, dict)] queue_source = str(DEFAULT_EVOLUTION_PUBLIC_JSON) cases = [_certify_order(order, queue_source) for order in orders] buckets = Counter(case.get("outcome_bucket") for case in cases) failures = [case for case in cases if not case.get("ok")] fake_passes = [case for case in cases if case.get("fake_pass")] processed_count = len(cases) queue_count = int((queue.get("summary") or {}).get("queue_count") or len(orders)) throughput_percent = round((processed_count / queue_count) * 100, 2) if queue_count else 0.0 certified = bool(queue_count and processed_count == queue_count and not failures and not fake_passes) status = ( "evolution_queue_584_autonomous_certified" if certified and queue_count == 584 else "evolution_queue_autonomous_certified" if certified else "evolution_queue_autonomous_certification_attention" ) summary = { "queue_count": queue_count, "processed_count": processed_count, "throughput_percent": throughput_percent, "safe_autonomous_outcome_count": len([case for case in cases if case.get("ok")]), "manual_boundary_visible_count": buckets.get("manual_boundary_visible", 0), "read_only_adapter_count": buckets.get("read_only_adapter", 0), "blocker_card_count": len([case for case in cases if case.get("autonomous_outcome") == "read_only_blocker_card_certified"]), "safe_status_adapter_count": buckets.get("safe_status_adapter", 0), "evidence_link_count": buckets.get("evidence_link", 0), "archive_decision_count": buckets.get("archive_decision", 0), "watch_only_count": buckets.get("watch_only", 0), "failed_count": len(failures), "fake_pass_count": len(fake_passes), "outcome_buckets": dict(buckets), } report = { "schema_version": SCHEMA_VERSION, "generated_at": generated_at, "status": status, "ok": certified, "repo_root": str(root), "source_queue": { "status": queue.get("status"), "generated_at": queue.get("generated_at"), "summary": queue.get("summary", {}), "public_json": str(DEFAULT_EVOLUTION_PUBLIC_JSON), }, "certification_contract": { "mode": "full_safe_local_autonomy", "claim": "Every evolution work order receives an autonomous safe outcome or a visible true-boundary state.", "no_legacy_execution": True, "no_external_mutation": True, "manual_authority_boundaries": sorted(MANUAL_BOUNDARY_SAFETY), "fake_pass_policy": "handover_ready is blocked when proof is missing", }, "summary": summary, "cases": cases, "failures": failures[:20], "fake_passes": fake_passes[:20], "output_files": [ str(DEFAULT_STATE_PATH), str(DEFAULT_AUDIT_JSON), str(DEFAULT_AUDIT_MD), str(DEFAULT_PUBLIC_JSON), ], } writes = [ _write_json(_rooted(root, DEFAULT_STATE_PATH), report), _write_json(_rooted(root, DEFAULT_AUDIT_JSON), report), _write_text(_rooted(root, DEFAULT_AUDIT_MD), _make_markdown(report)), _write_json(_rooted(root, DEFAULT_PUBLIC_JSON), report), ] compact = { "schema_version": SCHEMA_VERSION, "status": report["status"], "ok": report["ok"], "generated_at": report["generated_at"], "summary": report["summary"], "sample_cases": cases[:8], } report["write_info"] = { "evidence_writes": writes, "coding_bridge_evidence_writes": _attach_to_coding_bridge(root, compact), } for rel in (DEFAULT_STATE_PATH, DEFAULT_AUDIT_JSON, DEFAULT_PUBLIC_JSON): _write_json(_rooted(root, rel), report) _write_text(_rooted(root, DEFAULT_AUDIT_MD), _make_markdown(report)) return report def main(argv: Optional[Sequence[str]] = None) -> int: parser = argparse.ArgumentParser(description="Certify Aureon's full frontend evolution queue autonomous throughput.") parser.add_argument("--root", default="", help="Repository root. Defaults to current Aureon repo.") parser.add_argument("--json", action="store_true", help="Print JSON report.") args = parser.parse_args(argv) root = Path(args.root).resolve() if args.root else None report = build_and_write_evolution_queue_autonomous_certification(root=root) if args.json: print(json.dumps(report, indent=2, sort_keys=True, default=str)) else: summary = report.get("summary", {}) print( f"{report.get('status')}: processed={summary.get('processed_count')}/" f"{summary.get('queue_count')} fake_passes={summary.get('fake_pass_count')} " f"failed={summary.get('failed_count')}" ) return 0 if report.get("ok") else 1 if __name__ == "__main__": raise SystemExit(main())