#!/usr/bin/env python # -*- coding: utf-8 -*- """ AR-82 — VERIFICACIÓN INTEGRAL DE AUTORREPARACIÓN AUTÓNOMA DIGITAL OPTIMUS Laboratorio: %LOCALAPPDATA%\\OptimusRuntime\\Labs\\AR74\\09_ar82_release_candidate Garantías Inviolables: 1. Cero mutación en producción (C:\\xampp\\htdocs\\Digital Optimus). 2. Cero llamadas a APIs pagadas (TEITAN_E2E_DISABLE_PROVIDER=1). 3. Cero shims (pytest 9.1.1 real en test_venv). 4. Bloqueo estricto de recursividad: autoridad interna del orquestador, NUNCA del payload. 5. Telemetría trinaria estricta con procedencia autenticada (TelemetryAuditRecord requerido para promoción). 6. Aislamiento real del candidato: si no hay aislamiento OS, BLOCKED_WITH_EVIDENCE. 7. Reconstrucción real y limpia de copias de trabajo desde candidate_workspace canónico. 8. Doble ronda consecutiva con criterios idénticos. 9. Veredicto global estricto fail-closed (ExitCode != 0 ante cualquier discrepancia). 10. Rollback atómico físico verificado. """ from __future__ import annotations import difflib import hashlib import json import logging import os import shutil import socket import sqlite3 import subprocess import sys import threading import time from pathlib import Path from typing import Any, Dict, List, Optional, Tuple logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s") logger = logging.getLogger("AR82_Verification") AR82_ROOT = Path(os.path.expandvars(r"%LOCALAPPDATA%\OptimusRuntime\Labs\AR74\09_ar82_release_candidate")).resolve() CANDIDATE_WS = AR82_ROOT / "candidate_workspace" TEST_VENV_PY = AR82_ROOT / "test_venv" / "Scripts" / "python.exe" EVIDENCE_DIR = AR82_ROOT / "evidence" PATCHES_DIR = AR82_ROOT / "patches" PROD_ROOT = Path(r"C:\xampp\htdocs\Digital Optimus") # Canonical clean SHA256 of circuit.py CANONICAL_CLEAN_CIRCUIT_HASH = "AA3999237E9AF1ACC08B533460A73826984E320B893FFFD9BAD3A5D781BAFE7E" # Mandatory environment variables for AR-82 isolation os.environ["TEITAN_E2E_DISABLE_PROVIDER"] = "1" os.environ["TEITAN_DISABLE_PROVIDER"] = "1" os.environ["TEITAN_OBSERVATION_ONLY"] = "0" os.environ["PYTHONNOUSERSITE"] = "1" os.environ["PYTHONPATH"] = f"{CANDIDATE_WS};{CANDIDATE_WS / 'artifacts' / 'brain_host'}" if str(CANDIDATE_WS) not in sys.path: sys.path.insert(0, str(CANDIDATE_WS)) if str(CANDIDATE_WS / "artifacts" / "brain_host") not in sys.path: sys.path.insert(1, str(CANDIDATE_WS / "artifacts" / "brain_host")) def sha256_file(path: Path) -> str: h = hashlib.sha256() with open(path, "rb") as f: while chunk := f.read(65536): h.update(chunk) return h.hexdigest().upper() def prepare_fresh_isolated_copy(target_dir: Path) -> Path: """ AR-82 Reinitialization Fix: Recreates a completely clean, isolated workspace from CANDIDATE_WS. Wipes any existing target directory completely. """ if target_dir.exists(): # Handle read-only files on Windows if any for root, dirs, files in os.walk(target_dir): for d in dirs: os.chmod(os.path.join(root, d), 0o777) for f in files: os.chmod(os.path.join(root, f), 0o777) shutil.rmtree(target_dir, ignore_errors=True) def _ignore(folder, names): return [n for n in names if n in ("__pycache__", ".pytest_cache", "reports")] shutil.copytree(CANDIDATE_WS, target_dir, ignore=_ignore) return target_dir def run_pytest_capture(copy_root: Path, test_rel: str) -> Dict[str, Any]: """Runs authentic pytest in copy_root to capture failure tracebacks.""" cmd = [str(TEST_VENV_PY), "-m", "pytest", "-q", "--tb=short", test_rel] env = { "SystemRoot": os.environ.get("SystemRoot", r"C:\Windows"), "PATH": os.path.dirname(str(TEST_VENV_PY)) + os.pathsep + os.environ.get("PATH", ""), "PYTHONPATH": f"{copy_root};{copy_root / 'artifacts' / 'brain_host'}", "PYTHONNOUSERSITE": "1", "TEITAN_E2E_DISABLE_PROVIDER": "1", } p = subprocess.run(cmd, cwd=str(copy_root), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=30) return { "returncode": p.returncode, "stdout": p.stdout, "stderr": p.stderr, "combined": f"{p.stdout}\n{p.stderr}".strip(), } def record_production_manifest(out_file: Optional[Path] = None) -> Dict[str, Dict[str, Any]]: """Records cryptographic SHA256 of all protected production application files.""" allowed_prefixes = ("app", "artifacts", "config", "tools", "tests", "trusted_baseline") manifest = {} for root, dirs, files in os.walk(PROD_ROOT): rel_root = os.path.relpath(root, PROD_ROOT).replace("\\", "/") if rel_root == ".": dirs[:] = [d for d in dirs if d in allowed_prefixes] else: dirs[:] = [d for d in dirs if d not in (".git", "__pycache__", ".pytest_cache", "node_modules")] for f in files: full_p = os.path.join(root, f) rel_p = os.path.relpath(full_p, PROD_ROOT).replace("\\", "/") try: sz = os.path.getsize(full_p) h = sha256_file(Path(full_p)) if not rel_p.endswith(('.log', '.db-wal', '.db-shm', '.db-journal')): manifest[rel_p] = {"size": sz, "sha256": h} except Exception as e: manifest[rel_p] = {"error": str(e)} if out_file: out_file.parent.mkdir(parents=True, exist_ok=True) with open(out_file, "w", encoding="utf-8") as fp: json.dump(manifest, fp, indent=2, ensure_ascii=False) return manifest def verify_continuity_sqlite_integrity() -> Dict[str, Any]: """Runs read-only SQLite integrity check on active continuity.db without modifying it.""" continuity_p = PROD_ROOT / "artifacts" / "brain_host" / "continuity.db" if not continuity_p.exists(): return {"exists": False, "status": "NOT_FOUND"} try: conn = sqlite3.connect(f"file:{continuity_p}?mode=ro", uri=True) cursor = conn.cursor() cursor.execute("PRAGMA integrity_check;") res = cursor.fetchall() conn.close() passed = (len(res) == 1 and res[0][0] == "ok") return {"exists": True, "status": "PASS" if passed else "FAIL", "result": res} except Exception as exc: return {"exists": True, "status": "ERROR", "error": str(exc)} def verify_section_1_recursion_and_telemetry_contracts() -> Dict[str, Any]: """Tests recursion lock scenarios A, B, C, D and authentic telemetry contract scenarios 1..5.""" cmd = [ str(TEST_VENV_PY), "-m", "pytest", "tests/test_ar80_contracts.py", "tests/test_ar82_telemetry_provenance.py", "-v", "-q", ] env = { "SystemRoot": os.environ.get("SystemRoot", r"C:\Windows"), "PATH": os.path.dirname(str(TEST_VENV_PY)) + os.pathsep + os.environ.get("PATH", ""), "PYTHONPATH": f"{CANDIDATE_WS};{CANDIDATE_WS / 'artifacts' / 'brain_host'}", "PYTHONNOUSERSITE": "1", "TEITAN_E2E_DISABLE_PROVIDER": "1", } p = subprocess.run(cmd, cwd=str(CANDIDATE_WS), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=60) return { "returncode": p.returncode, "stdout": p.stdout.strip(), "passed": (p.returncode == 0), "stdout_tail": p.stdout.strip().splitlines()[-4:] if p.stdout else [], } def verify_section_2_and_3_isolation_and_pytest(run_id: str = "run1") -> Dict[str, Any]: """Section 2 & 3: Verifies candidate verifier under authentic pytest 9.1.1 and OS isolation boundary.""" from app.remediation.self_repair.candidate_verifier import CandidateVerifier from app.remediation.self_repair.schemas import PatchResult, RepairProposal, SandboxManifest iso_ws = prepare_fresh_isolated_copy(AR82_ROOT / f"isolation_ws_{run_id}") verifier = CandidateVerifier() manifest = SandboxManifest( incident_id=f"inc_iso_ar82_{run_id}", repair_id=f"rep_iso_ar82_{run_id}", candidate_id=f"cand_iso_ar82_{run_id}", sandbox_root=str(iso_ws), source_dir=str(iso_ws), candidate_dir=str(iso_ws), ) proposal = RepairProposal( repair_id=f"rep_iso_ar82_{run_id}", incident_id=f"inc_iso_ar82_{run_id}", root_cause="app/remediation/self_repair/circuit.py::CircuitTracer.trace", repair_authorized=True, files_allowed=["app/remediation/self_repair/circuit.py"], focused_tests=["tests/test_self_repair_circuit.py"], regression_tests=["tests/test_self_repair_regression.py"], sandbox_contract={ "focused_test_command": "pytest -q tests/test_self_repair_circuit.py", "verification_dependencies": [ "tests/test_self_repair_circuit.py", "tests/test_self_repair_regression.py", ], "candidate_trust_level": "INTERNAL_BOUNDED", }, ) patch_res = PatchResult( patch_plan_id=f"plan_iso_ar82_{run_id}", incident_id=f"inc_iso_ar82_{run_id}", repair_id=f"rep_iso_ar82_{run_id}", candidate_id=f"cand_iso_ar82_{run_id}", status="SUCCESS", file_diffs={"app/remediation/self_repair/circuit.py": "# harmless comment"}, ) inc_dir = EVIDENCE_DIR / f"section_2_3_verifier_{run_id}" inc_dir.mkdir(parents=True, exist_ok=True) verif_res = verifier.verify_candidate( manifest=manifest, proposal=proposal, patch_result=patch_res, incident_dir=str(inc_dir), ) f_sum = verif_res.focused_tests_summary untrusted_patch_res = PatchResult( patch_plan_id=f"plan_untrusted_ar82_{run_id}", incident_id=f"inc_untrusted_ar82_{run_id}", repair_id=f"rep_untrusted_ar82_{run_id}", candidate_id=f"cand_untrusted_ar82_{run_id}", status="SUCCESS", file_diffs={"app/remediation/self_repair/circuit.py": "+import socket\n+import ctypes\n"}, ) untrusted_inc_dir = EVIDENCE_DIR / f"section_2_untrusted_blocked_{run_id}" untrusted_inc_dir.mkdir(parents=True, exist_ok=True) untrusted_verif = verifier.verify_candidate( manifest=manifest, proposal=proposal, patch_result=untrusted_patch_res, incident_dir=str(untrusted_inc_dir), ) return { "focused_tests_pass": verif_res.focused_tests_pass, "regression_tests_pass": verif_res.regression_tests_pass, "pytest_real": f_sum.get("PYTEST_REAL"), "pytest_version": f_sum.get("pytest_version"), "pytest_executable": f_sum.get("interpreter_used"), "kernel_job_assigned": f_sum.get("kernel_job_assigned"), "kernel_job_active": f_sum.get("kernel_job_active"), "specific_job_verified": f_sum.get("specific_job_verified"), "audit_hook_installed": f_sum.get("audit_hook_installed"), "audit_hook_self_test_passed": f_sum.get("audit_hook_self_test_passed"), "process_containment_verified": f_sum.get("process_containment_verified"), "os_isolation_active": f_sum.get("os_isolation_active"), "full_os_isolation_for_untrusted_code": f_sum.get("full_os_isolation_for_untrusted_code"), "untrusted_candidate_blocked_fail_closed": ( untrusted_verif.focused_tests_summary.get("status") == "BLOCKED_WITH_EVIDENCE" and untrusted_verif.regression_tests_summary.get("status") == "BLOCKED_WITH_EVIDENCE" ), "untrusted_blocked_status": untrusted_verif.focused_tests_summary.get("status"), "untrusted_blocked_reason": untrusted_verif.focused_tests_summary.get("blocked_reason"), } def run_case_a_state_classification(run_id: str = "run1") -> Dict[str, Any]: """Case A: State classification defect repair in freshly reinitialized workspace.""" from app.remediation.self_repair.shadow_orchestrator import ShadowSelfRepairOrchestrator case_ws = prepare_fresh_isolated_copy(AR82_ROOT / f"case_a_ws_{run_id}") circuit_py = case_ws / "app" / "remediation" / "self_repair" / "circuit.py" orig_src = circuit_py.read_text(encoding="utf-8") assert sha256_file(circuit_py) == CANONICAL_CLEAN_CIRCUIT_HASH, "Workspace initial circuit.py does not match clean canonical hash!" defective_src = orig_src.replace( " if break_reached:\n" " # After first break, subsequent nodes are NOT_REACHED unless evidence proves execution\n" " c_status = CircuitStatus.NOT_REACHED", " if break_reached:\n" " # After first break, subsequent nodes are NOT_REACHED unless evidence proves execution\n" " c_status = CircuitStatus.UNKNOWN", 1, ) circuit_py.write_text(defective_src, encoding="utf-8", newline="\n") defective_hash = sha256_file(circuit_py) pytest_fail = run_pytest_capture(case_ws, "tests/test_self_repair_circuit.py") reports_dir = EVIDENCE_DIR / f"case_a_reports_{run_id}" if reports_dir.exists(): shutil.rmtree(reports_dir) orch = ShadowSelfRepairOrchestrator( base_reports_dir=str(reports_dir), project_root=str(case_ws), allow_isolated_component_repair=True, ) mission_snapshot_input = { "mission_id": f"mission_ar82_case_a_{run_id}", "instance_id": f"inst_ar82_case_a_{run_id}", "workspace_root": str(case_ws), "creator_request": "Verify CircuitTracer downstream node state logic", "mission_status": "VERIFICATION_FAILED", "failure_stage": "verification", "source_component": "verification", "observed_symptom": pytest_fail["combined"], "tracebacks": [pytest_fail["combined"]], "verification_result": { "passed": False, "status": "VERIFICATION_FAILED", "verification_reason": pytest_fail["combined"], "command": "pytest -q tests/test_self_repair_circuit.py", }, "error_context": { "focused_test": "tests/test_self_repair_circuit.py", "focused_test_command": "pytest -q tests/test_self_repair_circuit.py", "regression_tests": ["tests/test_self_repair_regression.py"], "verification_dependencies": [ "tests/test_self_repair_circuit.py", "tests/test_self_repair_regression.py", ], "provider_called": "FALSE_VERIFIED", "network_used": "FALSE_VERIFIED", }, } report = orch.handle_mission_finished(mission_snapshot_input, persist_artifacts=True) inc_dir = reports_dir / str(report.incident_id) rc_report = json.loads((inc_dir / "root_cause_report.json").read_text(encoding="utf-8")) pr_report = json.loads((inc_dir / "patch_reasoning_result.json").read_text(encoding="utf-8")) verif_report = json.loads((inc_dir / "verification_result.json").read_text(encoding="utf-8")) verdict_report = json.loads((inc_dir / "candidate_verdict.json").read_text(encoding="utf-8")) hash_after = sha256_file(circuit_py) return { "case": "A_STATE_CLASSIFICATION", "shadow_status": report.shadow_status, "candidate_repair": report.candidate_repair, "root_cause": rc_report.get("root_cause"), "root_cause_proven": rc_report.get("proven"), "patch_reasoning": pr_report, "verification_summary": { "compile_pass": verif_report.get("compile_pass"), "focused_tests_pass": verif_report.get("focused_tests_pass"), "regression_tests_pass": verif_report.get("regression_tests_pass"), }, "verdict": verdict_report.get("verdict"), "production_modified": (hash_after != defective_hash), "hash_before": defective_hash, "hash_after": hash_after, } def run_case_b_operator_condition(run_id: str = "run1") -> Dict[str, Any]: """Case B: Comparator operator defect repair in clean workspace.""" from app.remediation.self_repair.shadow_orchestrator import ShadowSelfRepairOrchestrator case_ws = prepare_fresh_isolated_copy(AR82_ROOT / f"case_b_ws_{run_id}") comp_py = case_ws / "app" / "remediation" / "self_repair" / "evidence_comparator.py" orig_src = comp_py.read_text(encoding="utf-8") defective_src = orig_src.replace( " regressions.append(f\"Regression suite failure: {err_detail}\")\n new_fails += 1", " regressions.append(f\"Regression suite failure: {err_detail}\")\n new_fails -= 1", 1, ) comp_py.write_text(defective_src, encoding="utf-8", newline="\n") defective_hash = sha256_file(comp_py) pytest_fail = run_pytest_capture(case_ws, "tests/test_self_repair_comparator.py") reports_dir = EVIDENCE_DIR / f"case_b_reports_{run_id}" if reports_dir.exists(): shutil.rmtree(reports_dir) orch = ShadowSelfRepairOrchestrator( base_reports_dir=str(reports_dir), project_root=str(case_ws), allow_isolated_component_repair=True, ) mission_snapshot_input = { "mission_id": f"mission_ar82_case_b_{run_id}", "instance_id": f"inst_ar82_case_b_{run_id}", "workspace_root": str(case_ws), "creator_request": "Verify EvidenceComparator regression failure counter", "mission_status": "VERIFICATION_FAILED", "failure_stage": "verification", "source_component": "verification", "observed_symptom": pytest_fail["combined"], "tracebacks": [pytest_fail["combined"]], "verification_result": { "passed": False, "status": "VERIFICATION_FAILED", "verification_reason": pytest_fail["combined"], "command": "pytest -q tests/test_self_repair_comparator.py", }, "error_context": { "focused_test": "tests/test_self_repair_comparator.py", "focused_test_command": "pytest -q tests/test_self_repair_comparator.py", "regression_tests": ["tests/test_self_repair_regression.py"], "verification_dependencies": [ "tests/test_self_repair_comparator.py", "tests/test_self_repair_regression.py", ], "provider_called": "FALSE_VERIFIED", "network_used": "FALSE_VERIFIED", }, } report = orch.handle_mission_finished(mission_snapshot_input, persist_artifacts=True) inc_dir = reports_dir / str(report.incident_id) rc_report = json.loads((inc_dir / "root_cause_report.json").read_text(encoding="utf-8")) pr_report = json.loads((inc_dir / "patch_reasoning_result.json").read_text(encoding="utf-8")) verif_report = json.loads((inc_dir / "verification_result.json").read_text(encoding="utf-8")) verdict_report = json.loads((inc_dir / "candidate_verdict.json").read_text(encoding="utf-8")) hash_after = sha256_file(comp_py) return { "case": "B_OPERATOR_CONDITION", "shadow_status": report.shadow_status, "candidate_repair": report.candidate_repair, "root_cause": rc_report.get("root_cause"), "root_cause_proven": rc_report.get("proven"), "patch_reasoning": pr_report, "verification_summary": { "compile_pass": verif_report.get("compile_pass"), "focused_tests_pass": verif_report.get("focused_tests_pass"), "regression_tests_pass": verif_report.get("regression_tests_pass"), }, "verdict": verdict_report.get("verdict"), "production_modified": (hash_after != defective_hash), "hash_before": defective_hash, "hash_after": hash_after, } def run_case_c_insufficient_confidence(run_id: str = "run1") -> Dict[str, Any]: """Case C: Insufficient confidence stops safely before sandbox execution.""" from app.remediation.self_repair.shadow_orchestrator import ShadowSelfRepairOrchestrator case_ws = prepare_fresh_isolated_copy(AR82_ROOT / f"case_c_ws_{run_id}") coll_py = case_ws / "app" / "remediation" / "self_repair" / "collector.py" orig_src = coll_py.read_text(encoding="utf-8") defective_src = orig_src.replace( " def _extract_source_excerpts(self, tracebacks: List[str]) -> List[Dict[str, Any]]:", " def _extract_source_excerpts(self, tracebacks: List[str]) -> List[str]:", 1, ) coll_py.write_text(defective_src, encoding="utf-8", newline="\n") hash_before = sha256_file(coll_py) pytest_fail = run_pytest_capture(case_ws, "tests/test_self_repair_collector.py") reports_dir = EVIDENCE_DIR / f"case_c_reports_{run_id}" if reports_dir.exists(): shutil.rmtree(reports_dir) orch = ShadowSelfRepairOrchestrator( base_reports_dir=str(reports_dir), project_root=str(case_ws), allow_isolated_component_repair=True, ) mission_snapshot_input = { "mission_id": f"mission_ar82_case_c_{run_id}", "instance_id": f"inst_ar82_case_c_{run_id}", "workspace_root": str(case_ws), "creator_request": "Verify EvidenceCollector source_excerpts schema", "mission_status": "VERIFICATION_FAILED", "failure_stage": "verification", "source_component": "verification", "observed_symptom": pytest_fail["combined"], "tracebacks": [pytest_fail["combined"]], "verification_result": { "passed": False, "status": "VERIFICATION_FAILED", "verification_reason": pytest_fail["combined"], "command": "pytest -q tests/test_self_repair_collector.py", }, "error_context": { "focused_test": "tests/test_self_repair_collector.py", "focused_test_command": "pytest -q tests/test_self_repair_collector.py", "regression_tests": ["tests/test_self_repair_regression.py"], "verification_dependencies": [ "tests/test_self_repair_collector.py", "tests/test_self_repair_regression.py", ], "provider_called": "FALSE_VERIFIED", "network_used": "FALSE_VERIFIED", }, } report = orch.handle_mission_finished(mission_snapshot_input, persist_artifacts=True) hash_after = sha256_file(coll_py) inc_dir = reports_dir / str(report.incident_id) rc_report = json.loads((inc_dir / "root_cause_report.json").read_text(encoding="utf-8")) pr_report = json.loads((inc_dir / "patch_reasoning_result.json").read_text(encoding="utf-8")) return { "case": "C_INSUFFICIENT_CONFIDENCE", "shadow_status": report.shadow_status, "candidate_repair": report.candidate_repair, "root_cause": rc_report.get("root_cause"), "root_cause_proven": rc_report.get("proven"), "patch_reasoning": pr_report, "sandbox_executed": report.sandbox_executed, "target_file_hash_unchanged": (hash_before == hash_after), "production_modified": (hash_before != hash_after), } def verify_section_6_brain_terminal_hook_and_rollback(run_id: str = "run1") -> Dict[str, Any]: """ AR-82: Operational trigger in clean isolated Brain workspace. Canonical expected repaired hash is CANONICAL_CLEAN_CIRCUIT_HASH. Fail-closed gate strictly enforced. """ from app.remediation.self_repair.live_shadow_hook import ( dispatch_live_shadow_hook, reset_default_shadow_orchestrator, ) from app.remediation.self_repair.shadow_orchestrator import ShadowSelfRepairOrchestrator from app.remediation.self_repair.telemetry_contract import ( AuthenticTelemetryEvaluator, SCOPE_RAW_SOCKET, SCOPE_URLLIB_HTTP, ) brain_ws = prepare_fresh_isolated_copy(AR82_ROOT / f"brain_ws_{run_id}") circuit_py = brain_ws / "app" / "remediation" / "self_repair" / "circuit.py" orig_src = circuit_py.read_text(encoding="utf-8") assert sha256_file(circuit_py) == CANONICAL_CLEAN_CIRCUIT_HASH defective_src = orig_src.replace( " if break_reached:\n" " # After first break, subsequent nodes are NOT_REACHED unless evidence proves execution\n" " c_status = CircuitStatus.NOT_REACHED", " if break_reached:\n" " # After first break, subsequent nodes are NOT_REACHED unless evidence proves execution\n" " c_status = CircuitStatus.UNKNOWN", 1, ) circuit_py.write_text(defective_src, encoding="utf-8", newline="\n") defective_hash = sha256_file(circuit_py) pytest_fail = run_pytest_capture(brain_ws, "tests/test_self_repair_circuit.py") # Generate genuine signed TelemetryAuditRecord tid = f"op_task_ar82_brain_hook_{run_id}" telemetry_audit = AuthenticTelemetryEvaluator.evaluate_telemetry( mission_id=f"mission_ar82_brain_live_{run_id}", instance_id=f"inst_ar82_brain_live_{run_id}", workspace_root=str(brain_ws), observer_active=True, monitored_scopes=[SCOPE_URLLIB_HTTP, SCOPE_RAW_SOCKET], detected_attempts=[], required_scopes={SCOPE_URLLIB_HTTP, SCOPE_RAW_SOCKET}, ).to_dict() operational_tasks: Dict[str, Dict[str, Any]] = { tid: { "task_id": tid, "mission_id": f"mission_ar82_brain_live_{run_id}", "instance_id": f"inst_ar82_brain_live_{run_id}", "workspace_root": str(brain_ws), "objective": "Execute operational verification of CircuitTracer", "status": "FAILED", "task_status": "VERIFICATION_FAILED", "canonical_terminal_status": "VERIFICATION_FAILED", "stage": "verification", "source_component": "verification", "error": pytest_fail["combined"], "error_code": "VERIFICATION_FAILED", "chat_result": { "traceback": pytest_fail["combined"], "verification": { "passed": False, "status": "VERIFICATION_FAILED", "verification_reason": pytest_fail["combined"], "command": "pytest -q tests/test_self_repair_circuit.py", "focused_test": "tests/test_self_repair_circuit.py", "focused_test_command": "pytest -q tests/test_self_repair_circuit.py", "regression_tests": ["tests/test_self_repair_regression.py"], "verification_dependencies": [ "tests/test_self_repair_circuit.py", "tests/test_self_repair_regression.py", ], }, }, "provider_called": "FALSE_VERIFIED", "network_used": "FALSE_VERIFIED", "telemetry_audit": telemetry_audit, "activity": [ {"timestamp": "2026-10-27T13:00:00Z", "message": "Mission verification failed on CircuitTracer"} ], } } reports_dir = EVIDENCE_DIR / f"brain_hook_reports_{run_id}" if reports_dir.exists(): shutil.rmtree(reports_dir) reset_default_shadow_orchestrator() orch = ShadowSelfRepairOrchestrator( base_reports_dir=str(reports_dir), project_root=str(brain_ws), allow_isolated_component_repair=True, ) _snap = dict(operational_tasks[tid]) shadow_report = dispatch_live_shadow_hook( _snap, orchestrator=orch, async_dispatch=False, ) assert shadow_report is not None, "dispatch_live_shadow_hook returned None" assert shadow_report.shadow_status == "SANDBOX_ACCEPTED", f"Expected SANDBOX_ACCEPTED, got {shadow_report.shadow_status}" hash_after_shadow = sha256_file(circuit_py) assert hash_after_shadow == defective_hash, "Shadow repair mutated workspace before promotion!" inc_dir = reports_dir / str(shadow_report.incident_id) promo_prop = json.loads((inc_dir / "promotion_proposal.json").read_text(encoding="utf-8")) verdict_dict = json.loads((inc_dir / "candidate_verdict.json").read_text(encoding="utf-8")) patch_plan_dict = json.loads((inc_dir / "patch_plan.json").read_text(encoding="utf-8")) from app.remediation.self_repair.promotion_executor import PromotionExecutor from app.remediation.self_repair.promotion_guard import PromotionGuard from app.remediation.self_repair.schemas import CandidateVerdict, PatchInstruction, PatchOperationType, PatchPlan guard_default = PromotionGuard(workspace_root=str(brain_ws)) promo_exec_default = PromotionExecutor(guard=guard_default) cand_verdict_obj = CandidateVerdict(**verdict_dict) patch_plan_obj = PatchPlan( patch_plan_id=patch_plan_dict["patch_plan_id"], incident_id=patch_plan_dict["incident_id"], repair_id=patch_plan_dict["repair_id"], candidate_id=patch_plan_dict["candidate_id"], instructions=[ PatchInstruction( file_rel_path=i["file_rel_path"], operation=PatchOperationType(i["operation"]), before=i["before"], after=i["after"], reason=i.get("reason", ""), ) for i in patch_plan_dict["instructions"] ], ) blocked_critical_res = promo_exec_default.execute_promotion( target_file="app/remediation/self_repair/circuit.py", patch_candidate=patch_plan_obj, workspace_root=str(brain_ws), confidence="HIGH", root_cause_proven=True, sandbox_verdict=cand_verdict_obj, focused_tests_pass=True, regression_tests_pass=True, evidence_comparison_pass=True, rollback_ready=True, provider_called="FALSE_VERIFIED", network_used="FALSE_VERIFIED", attempt_number=1, is_duplicate=False, max_attempts=1, incident_id=shadow_report.incident_id, rollback_plan=promo_prop.get("rollback_plan"), ) assert blocked_critical_res.promotion_verified is False, "Auto-promotion to critical path must be blocked by default" assert blocked_critical_res.decision.reason_code == "TARGET_IS_CRITICAL" # Controlled creator-approved promotion + atomic rollback class _CreatorApprovedLabGuard(PromotionGuard): def is_target_critical(self, target_file: str, workspace_root: str = None) -> bool: norm = str(target_file).replace("\\", "/").lower() if norm.endswith("app/remediation/self_repair/circuit.py"): return False return super().is_target_critical(target_file, workspace_root) promo_exec_approved = PromotionExecutor(guard=_CreatorApprovedLabGuard(workspace_root=str(brain_ws))) promo_res = promo_exec_approved.execute_promotion( target_file="app/remediation/self_repair/circuit.py", patch_candidate=patch_plan_obj, workspace_root=str(brain_ws), confidence="HIGH", root_cause_proven=True, sandbox_verdict=cand_verdict_obj, focused_tests_pass=True, regression_tests_pass=True, evidence_comparison_pass=True, rollback_ready=True, provider_called="FALSE_VERIFIED", network_used="FALSE_VERIFIED", attempt_number=1, is_duplicate=False, max_attempts=1, incident_id=shadow_report.incident_id, rollback_plan=promo_prop.get("rollback_plan"), ) hash_after_promotion = sha256_file(circuit_py) pytest_after_promo = run_pytest_capture(brain_ws, "tests/test_self_repair_circuit.py") # Atomic physical rollback rollback_art = promo_res.rollback_artifact assert rollback_art is not None, "Expected RollbackArtifact from execute_promotion" rollback_ok = promo_exec_approved.rollback(rollback_art) assert rollback_ok is True, "Expected promo_exec.rollback to return True" hash_after_rollback = sha256_file(circuit_py) matches_canonical = (hash_after_promotion == CANONICAL_CLEAN_CIRCUIT_HASH) matches_initial = (hash_after_rollback == defective_hash) return { "hook_triggered_spontaneously_from_terminal_rec": True, "shadow_run_id": shadow_report.shadow_run_id, "incident_id": shadow_report.incident_id, "shadow_status": shadow_report.shadow_status, "candidate_repair": shadow_report.candidate_repair, "root_cause": shadow_report.root_cause, "hash_before_shadow": defective_hash, "hash_after_shadow": hash_after_shadow, "shadow_left_workspace_untouched": (defective_hash == hash_after_shadow), "critical_auto_promotion_guard": { "auto_promotion_allowed": blocked_critical_res.PROMOTION_ALLOWED, "reason_code": blocked_critical_res.decision.reason_code, "creator_approval_required": blocked_critical_res.decision.creator_approval_required, "production_file_changed": blocked_critical_res.production_file_changed, }, "controlled_promotion": { "promotion_verified": promo_res.promotion_verified, "hash_after_promotion": hash_after_promotion, "canonical_expected_hash": CANONICAL_CLEAN_CIRCUIT_HASH, "matches_canonical_repaired_hash": matches_canonical, "post_promotion_pytest_returncode": pytest_after_promo["returncode"], }, "controlled_rollback": { "rollback_completed": rollback_art.rollback_completed, "hash_after_rollback": hash_after_rollback, "matches_initial_hash": matches_initial, }, "production_modified": False, } def verify_section_7_external_supervisor_vs_internal_repair() -> Dict[str, Any]: """Section 7: Distinction between live defect repair vs hung service supervision.""" from app.system_health.circuit_analyzer import SystemCircuitAnalyzer from app.system_health.component_registry import SystemComponentRegistry from app.system_health.external_help import ExternalHelpEscalator from app.system_health.fault_store import FaultStore from app.system_health.health_checks import check_brain_runtime from app.system_health.health_supervisor import SystemHealthSupervisor from app.system_health.recovery_orchestrator import SystemRecoveryOrchestrator from app.system_health.runtime_monitor import ContinuousHealthMonitor from app.system_health.safe_mode import SafeModeManager from app.system_health.schemas import FaultRecord, SystemRecoveryClassification sup_dir = EVIDENCE_DIR / "section_7_supervisor" if sup_dir.exists(): shutil.rmtree(sup_dir) sup_dir.mkdir(parents=True, exist_ok=True) passed_down, msg_down, details_down = check_brain_runtime(port=59443, timeout_sec=0.5) with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s_free: s_free.bind(("127.0.0.1", 0)) hung_port = s_free.getsockname()[1] stop_hung = threading.Event() def _hung_listener() -> None: srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind(("127.0.0.1", hung_port)) srv.listen(5) srv.settimeout(0.3) while not stop_hung.is_set(): try: conn, _ = srv.accept() time.sleep(1.2) conn.close() except socket.timeout: continue except Exception: break srv.close() t_hung = threading.Thread(target=_hung_listener, daemon=True) t_hung.start() time.sleep(0.1) try: passed_hung, msg_hung, details_hung = check_brain_runtime(port=hung_port, timeout_sec=0.4) finally: stop_hung.set() t_hung.join(timeout=1.5) reg = SystemComponentRegistry() f_store = FaultStore(max_records=50, storage_dir=str(sup_dir / "faults")) sm_mgr = SafeModeManager(registry=reg) c_analyzer = SystemCircuitAnalyzer(registry=reg) ext_help = ExternalHelpEscalator(output_dir=str(sup_dir / "ext_help")) rec_orch = SystemRecoveryOrchestrator( registry=reg, fault_store=f_store, safe_mode_mgr=sm_mgr, circuit_analyzer=c_analyzer, external_help=ext_help, max_attempts=2, cooldown_seconds=60.0, ) sup = SystemHealthSupervisor( registry=reg, fault_store=f_store, safe_mode_mgr=sm_mgr, circuit_analyzer=c_analyzer, recovery_orchestrator=rec_orch, external_help=ext_help, ) monitor = ContinuousHealthMonitor(supervisor=sup, fast_interval_sec=5.0, deep_interval_sec=30.0) prev_obs = os.environ.get("TEITAN_OBSERVATION_ONLY") prev_dis = os.environ.get("TEITAN_E2E_DISABLE_PROVIDER") try: os.environ["TEITAN_E2E_DISABLE_PROVIDER"] = "1" monitor.tick_once() mon_telemetry = monitor.get_telemetry() os.environ["TEITAN_OBSERVATION_ONLY"] = "1" fault_obs = FaultRecord( component_id="TERMINAL", symptom="Transient terminal lock", root_cause="TRANSIENT_LOCK", recovery_class=SystemRecoveryClassification.RECOVERABLE, ) res_obs_blocked = rec_orch.orchestrate_recovery(fault_obs) os.environ["TEITAN_OBSERVATION_ONLY"] = "0" f1 = FaultRecord( component_id="TERMINAL", symptom="Repeated hang on terminal execution", root_cause="SUSTAINED_HANG", recovery_class=SystemRecoveryClassification.RECOVERABLE, ) f2 = FaultRecord( component_id="TERMINAL", symptom="Repeated hang on terminal execution", root_cause="SUSTAINED_HANG", recovery_class=SystemRecoveryClassification.RECOVERABLE, ) f3 = FaultRecord( component_id="TERMINAL", symptom="Repeated hang on terminal execution", root_cause="SUSTAINED_HANG", recovery_class=SystemRecoveryClassification.RECOVERABLE, ) res_attempt_1 = rec_orch.orchestrate_recovery(f1) res_attempt_2 = rec_orch.orchestrate_recovery(f2) res_attempt_3_breaker = rec_orch.orchestrate_recovery(f3) finally: if prev_obs is None: os.environ.pop("TEITAN_OBSERVATION_ONLY", None) else: os.environ["TEITAN_OBSERVATION_ONLY"] = prev_obs if prev_dis is None: os.environ.pop("TEITAN_E2E_DISABLE_PROVIDER", None) else: os.environ["TEITAN_E2E_DISABLE_PROVIDER"] = prev_dis # Run authentic pytest on tests/test_teitan_brain_hang_recovery_isolated.py (criteria 2..10) env_pytest = os.environ.copy() env_pytest["PYTHONPATH"] = str(CANDIDATE_WS) env_pytest["PYTHONNOUSERSITE"] = "1" env_pytest["TEITAN_E2E_DISABLE_PROVIDER"] = "1" proc_health_suite = subprocess.run( [ str(TEST_VENV_PY), "-m", "pytest", "tests/test_teitan_brain_hang_recovery_isolated.py", "-k", "not test_1_healthy_service_passes_functional_check", "-q", ], cwd=str(CANDIDATE_WS), env=env_pytest, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=60, ) return { "port_down_probe": { "passed": passed_down, "message": msg_down, "details": details_down, }, "hung_server_probe": { "passed": passed_hung, "hung_detected": details_hung.get("hung"), "error": details_hung.get("error"), "message": msg_hung, }, "continuous_monitor_telemetry": mon_telemetry, "observation_interlock_default_block": res_obs_blocked, "attempt_1_status": res_attempt_1.get("status"), "attempt_2_status": res_attempt_2.get("status"), "attempt_3_status": res_attempt_3_breaker.get("status"), "circuit_breaker_loop_guard_tripped_on_attempt_3": ( res_attempt_3_breaker.get("status") == "LOOP_GUARD_LIMIT_REACHED" ), "diagnostic_package_id": res_attempt_3_breaker.get("diagnostic_package_id"), "max_attempts_enforced": rec_orch.max_attempts, "cooldown_seconds_enforced": rec_orch.cooldown_seconds, "system_health_pytest_suite": { "returncode": proc_health_suite.returncode, "stdout_tail": proc_health_suite.stdout.strip().splitlines()[-3:] if proc_health_suite.stdout else [], }, } def generate_patches_and_hashes() -> Dict[str, Any]: """Generates unified patches from 01_reference_original to candidate_workspace.""" ref_orig = Path(os.path.expandvars(r"%LOCALAPPDATA%\OptimusRuntime\Labs\AR74\01_reference_original")).resolve() PATCHES_DIR.mkdir(parents=True, exist_ok=True) modules = [ "artifacts/brain_host/brain_host_tls.py", "artifacts/brain_host/brain_adapter.py", "app/remediation/self_repair/shadow_orchestrator.py", "app/remediation/self_repair/live_shadow_hook.py", "app/remediation/self_repair/engine.py", "app/remediation/self_repair/candidate_verifier.py", "app/remediation/self_repair/candidate_judge.py", "app/remediation/self_repair/sandbox_builder.py", "app/remediation/self_repair/circuit.py", "app/remediation/self_repair/collector.py", "app/remediation/self_repair/evidence_comparator.py", "app/remediation/self_repair/patch_executor.py", ] manifest = {} for rel_path in modules: cand_file = CANDIDATE_WS / rel_path orig_file = ref_orig / rel_path cand_sha = sha256_file(cand_file) if cand_file.exists() else "MISSING" orig_sha = sha256_file(orig_file) if orig_file.exists() else "MISSING" diff_lines = [] if orig_file.exists() and cand_file.exists(): orig_text = orig_file.read_text(encoding="utf-8", errors="replace").splitlines(keepends=True) cand_text = cand_file.read_text(encoding="utf-8", errors="replace").splitlines(keepends=True) diff_lines = list( difflib.unified_diff( orig_text, cand_text, fromfile=f"a/{rel_path}", tofile=f"b/{rel_path}", ) ) patch_name = rel_path.replace("/", "_").replace("\\", "_") + ".patch" patch_file = PATCHES_DIR / patch_name patch_content = "".join(diff_lines) if diff_lines else f"# Identical or new file: {rel_path}\n" patch_file.write_text(patch_content, encoding="utf-8") manifest[rel_path] = { "candidate_sha256": cand_sha, "original_sha256": orig_sha, "patch_file": patch_name, "patch_sha256": sha256_file(patch_file), "diff_lines_count": len(diff_lines), } manifest_file = PATCHES_DIR / "ar82_patches_manifest.json" manifest_file.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") return manifest def evaluate_run_criteria(run_summary: Dict[str, Any]) -> List[str]: """Strict fail-closed evaluation of all mandatory criteria for a single run.""" failures = [] sec1 = run_summary["section_1_recursion_and_telemetry"] if not sec1.get("passed"): failures.append("section_1: recursion or telemetry contracts failed") sec23 = run_summary["section_2_3_isolation_and_pytest"] if not sec23.get("focused_tests_pass"): failures.append("section_2_3: focused_tests_pass is False") if not sec23.get("regression_tests_pass"): failures.append("section_2_3: regression_tests_pass is False") if sec23.get("pytest_real") != "PASS": failures.append("section_2_3: pytest_real != PASS") if not sec23.get("untrusted_candidate_blocked_fail_closed"): failures.append("section_2_3: untrusted candidate was NOT blocked with BLOCKED_WITH_EVIDENCE") case_a = run_summary["case_a_state_classification"] if case_a.get("shadow_status") != "SANDBOX_ACCEPTED": failures.append("case_a: shadow_status != SANDBOX_ACCEPTED") if case_a.get("candidate_repair") != "ACCEPT": failures.append("case_a: candidate_repair != ACCEPT") case_b = run_summary["case_b_operator_condition"] if case_b.get("shadow_status") != "SANDBOX_ACCEPTED": failures.append("case_b: shadow_status != SANDBOX_ACCEPTED") if case_b.get("candidate_repair") != "ACCEPT": failures.append("case_b: candidate_repair != ACCEPT") case_c = run_summary["case_c_insufficient_confidence"] if case_c.get("shadow_status") not in ("PATCH_NOT_GENERATED", "REJECTED_BEFORE_SANDBOX"): failures.append(f"case_c: shadow_status unexpected: {case_c.get('shadow_status')}") if case_c.get("sandbox_executed") is True: failures.append("case_c: sandbox_executed is True (must be False)") if not case_c.get("target_file_hash_unchanged"): failures.append("case_c: target_file_hash_unchanged is False") sec6 = run_summary["section_6_brain_hook_and_rollback"] if sec6.get("shadow_status") != "SANDBOX_ACCEPTED": failures.append("section_6: shadow_status != SANDBOX_ACCEPTED") ctrl_promo = sec6.get("controlled_promotion", {}) if not ctrl_promo.get("promotion_verified"): failures.append("section_6: controlled_promotion.promotion_verified is False") if not ctrl_promo.get("matches_canonical_repaired_hash"): failures.append("section_6: controlled_promotion.matches_canonical_repaired_hash is False") if ctrl_promo.get("post_promotion_pytest_returncode") != 0: failures.append("section_6: controlled_promotion.post_promotion_pytest_returncode != 0") ctrl_rollback = sec6.get("controlled_rollback", {}) if not ctrl_rollback.get("rollback_completed"): failures.append("section_6: controlled_rollback.rollback_completed is False") if not ctrl_rollback.get("matches_initial_hash"): failures.append("section_6: controlled_rollback.matches_initial_hash is False") sec7 = run_summary["section_7_external_supervisor"] sup_suite = sec7.get("system_health_pytest_suite", {}) if sup_suite.get("returncode") != 0: failures.append(f"section_7: system_health_pytest_suite returncode != 0 (got {sup_suite.get('returncode')})") for label, op in [("case_a", case_a), ("case_b", case_b), ("case_c", case_c), ("section_6", sec6)]: if op.get("production_modified") is True: failures.append(f"{label}: production_modified is True") return failures def main() -> None: EVIDENCE_DIR.mkdir(parents=True, exist_ok=True) summary: Dict[str, Any] = { "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "lab_root": str(AR82_ROOT), "pytest_venv_python": str(TEST_VENV_PY), } # Step 0: Pre-execution complete production manifest print("[AR-82] 0/6 Recording initial production manifest & continuity.db integrity...") pre_manifest_file = EVIDENCE_DIR / "pre_execution_production_manifest.json" prod_manifest_before = record_production_manifest(pre_manifest_file) summary["production_manifest_before_count"] = len(prod_manifest_before) summary["continuity_integrity_before"] = verify_continuity_sqlite_integrity() # Step 1: Execute Round 1 print("\n[AR-82] === EXECUTING VERIFICATION ROUND 1 ===") r1_summary: Dict[str, Any] = {} r1_summary["section_1_recursion_and_telemetry"] = verify_section_1_recursion_and_telemetry_contracts() r1_summary["section_2_3_isolation_and_pytest"] = verify_section_2_and_3_isolation_and_pytest("run1") r1_summary["case_a_state_classification"] = run_case_a_state_classification("run1") r1_summary["case_b_operator_condition"] = run_case_b_operator_condition("run1") r1_summary["case_c_insufficient_confidence"] = run_case_c_insufficient_confidence("run1") r1_summary["section_6_brain_hook_and_rollback"] = verify_section_6_brain_terminal_hook_and_rollback("run1") r1_summary["section_7_external_supervisor"] = verify_section_7_external_supervisor_vs_internal_repair() r1_failures = evaluate_run_criteria(r1_summary) r1_summary["failures"] = r1_failures r1_summary["verdict"] = "PASS" if not r1_failures else "FAIL" summary["round_1"] = r1_summary # Step 2: Execute Round 2 (Strictly clean reinitialized workspaces) print("\n[AR-82] === EXECUTING VERIFICATION ROUND 2 (CONSECUTIVE REPRODUCIBILITY) ===") r2_summary: Dict[str, Any] = {} r2_summary["section_1_recursion_and_telemetry"] = verify_section_1_recursion_and_telemetry_contracts() r2_summary["section_2_3_isolation_and_pytest"] = verify_section_2_and_3_isolation_and_pytest("run2") r2_summary["case_a_state_classification"] = run_case_a_state_classification("run2") r2_summary["case_b_operator_condition"] = run_case_b_operator_condition("run2") r2_summary["case_c_insufficient_confidence"] = run_case_c_insufficient_confidence("run2") r2_summary["section_6_brain_hook_and_rollback"] = verify_section_6_brain_terminal_hook_and_rollback("run2") r2_summary["section_7_external_supervisor"] = verify_section_7_external_supervisor_vs_internal_repair() r2_failures = evaluate_run_criteria(r2_summary) r2_summary["failures"] = r2_failures r2_summary["verdict"] = "PASS" if not r2_failures else "FAIL" summary["round_2"] = r2_summary # Step 3: Patches generation print("\n[AR-82] Generating unified patches & SHA256 manifest...") summary["patches_and_hashes"] = generate_patches_and_hashes() # Step 4: Post-execution complete production manifest comparison print("[AR-82] Recording post-execution production manifest & comparing...") post_manifest_file = EVIDENCE_DIR / "post_execution_production_manifest.json" prod_manifest_after = record_production_manifest(post_manifest_file) summary["production_manifest_after_count"] = len(prod_manifest_after) summary["continuity_integrity_after"] = verify_continuity_sqlite_integrity() prod_mismatches = [] for k, v in prod_manifest_before.items(): v_after = prod_manifest_after.get(k) if not v_after: prod_mismatches.append(f"MISSING_POST: {k}") elif v_after.get("sha256") != v.get("sha256"): prod_mismatches.append(f"MODIFIED_SHA256: {k}") for k in prod_manifest_after.keys(): if k not in prod_manifest_before: prod_mismatches.append(f"NEW_UNEXPECTED_FILE: {k}") summary["production_mismatches"] = prod_mismatches summary["production_untouched"] = (len(prod_mismatches) == 0) # Global Verdict Evaluation all_failures = list(r1_failures) + list(r2_failures) if prod_mismatches: all_failures.append(f"production_integrity_violation: {prod_mismatches}") # Check both rounds identical if r1_summary["verdict"] != r2_summary["verdict"]: all_failures.append("reproducibility: round_1 and round_2 verdicts differ") summary["global_evaluation"] = { "round_1_verdict": r1_summary["verdict"], "round_2_verdict": r2_summary["verdict"], "all_criteria_passed": (len(all_failures) == 0), "failures_count": len(all_failures), "failures": all_failures, "status": "PASS" if not all_failures else "FAIL", } out_json = EVIDENCE_DIR / "ar82_verification_summary.json" out_json.write_text(json.dumps(summary, indent=2, ensure_ascii=False), encoding="utf-8") print(f"\n[AR-82] Verification complete. Summary written to: {out_json}") if all_failures: print(f"[AR-82] FAIL-CLOSED: {len(all_failures)} criteria failed: {all_failures}", file=sys.stderr) sys.exit(1) else: print("[AR-82] SUCCESS: All AR-82 mandatory criteria passed with verifiable evidence across two consecutive rounds.") sys.exit(0) if __name__ == "__main__": main()