# -*- coding: utf-8 -*- """ AR-76 — END-TO-END VERIFICATION & EVIDENCE HARNESS Executes strictly inside: %LOCALAPPDATA%\\OptimusRuntime\\Labs\\AR74\\06_ar76_candidate_hardening Verifies: 1. Verifier & Isolation Hardening (CREATE_SUSPENDED -> AssignProcessToJobObject -> NtResumeProcess -> IsProcessInJob=True & specific named Job=True -> PEP 578 audit hook self-test BEFORE proof -> honest os_isolation_active=False -> pre-execution block for untrusted/high-risk code). 2. Authentic pytest 9.1.1 execution (PYTEST_REAL="PASS", zero _STDLIB_PYTEST_SHIM). 3. Accurate Root Cause Diagnosis separating: A. observed_mission_stage B. defective_module C. responsible_function D. downstream_circuit_effects 4. Multi-Case Autonomous Self-Repair: - CASO A: State classification defect in app/remediation/self_repair/circuit.py (CircuitTracer.trace) -> ACCEPT - CASO B: Operator defect in app/remediation/self_repair/evidence_comparator.py (EvidenceComparator.compare) -> ACCEPT - CASO C: Complex structural defect in app/remediation/self_repair/collector.py (EvidenceCollector.collect) -> stops safely with INSUFFICIENT_DETERMINISTIC_EVIDENCE before sandbox execution 5. Spontaneous Brain Terminal Hook Integration in isolated copy (brain_host_tls.py -> dispatch_live_shadow_hook -> ShadowSelfRepairOrchestrator -> SelfRepairEngine) + Controlled Promotion & Rollback verification. 6. External Supervisor vs Live Brain Self-Repair separation (ContinuousHealthMonitor & SystemRecoveryOrchestrator). 7. Preservation of AR-74 directories (01..05) and Production (C:\\xampp\\htdocs\\Digital Optimus). """ from __future__ import annotations import difflib import hashlib import json import os import shutil import subprocess import sys import time from pathlib import Path from typing import Any, Dict, List LAB_AR74_ROOT = Path(os.environ.get("LOCALAPPDATA", r"C:\Users\intel\AppData\Local")) / "OptimusRuntime" / "Labs" / "AR74" AR76_ROOT = LAB_AR74_ROOT / "06_ar76_candidate_hardening" CANDIDATE_WS = AR76_ROOT / "candidate_workspace" REF_ORIG = LAB_AR74_ROOT / "01_reference_original" PATCHES_DIR = AR76_ROOT / "patches" EVIDENCE_DIR = AR76_ROOT / "evidence" TEST_VENV_PY = AR76_ROOT / "test_venv" / "Scripts" / "python.exe" PROD_ROOT = Path(r"C:\xampp\htdocs\Digital Optimus") # Ensure isolated candidate_workspace is first on sys.path sys.path.insert(0, str(CANDIDATE_WS)) os.environ["TEITAN_TEST_PYTHON_EXE"] = str(TEST_VENV_PY) os.environ["TEITAN_E2E_DISABLE_PROVIDER"] = "1" def sha256_file(path: Path) -> str: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(65536), b""): h.update(chunk) return h.hexdigest().upper() def prepare_isolated_copy(dest_dir: Path) -> Path: if dest_dir.exists(): shutil.rmtree(dest_dir) shutil.copytree( CANDIDATE_WS, dest_dir, ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".pytest_cache", "reports"), ) return dest_dir def write_test_suite_into_copy(copy_root: Path) -> None: """Writes focal and regression pytest suites into the isolated workspace copy.""" tests_dir = copy_root / "tests" tests_dir.mkdir(parents=True, exist_ok=True) # 1. Focal test for circuit.py (CircuitTracer.trace) (tests_dir / "test_self_repair_circuit.py").write_text( '''# -*- coding: utf-8 -*- from app.remediation.self_repair.circuit import CircuitTracer from app.remediation.self_repair.schemas import ( CircuitStatus, EvidenceBundle, IncidentFailureType, IncidentRecord, ) def test_circuit_tracer_marks_downstream_nodes_not_reached(): incident = IncidentRecord( mission_id="m_circuit_focal", failure_type=IncidentFailureType.EXCEPTION, observed_symptom="Executor failed before effector stage", source_component="executor", ) evidence = EvidenceBundle( incident_id=incident.incident_id, action_history=[{"stage": "planner", "status": "SUCCESS"}], ) tracer = CircuitTracer() trace = tracer.trace(incident=incident, evidence=evidence) node_map = {n.component: n.status for n in trace.nodes} assert trace.first_proven_break == "executor" assert node_map["planner"] == CircuitStatus.PASS assert node_map["executor"] == CircuitStatus.FAIL assert node_map["effector"] == CircuitStatus.NOT_REACHED assert node_map["verification"] == CircuitStatus.NOT_REACHED assert node_map["final_mission_state"] == CircuitStatus.NOT_REACHED assert "effector" in trace.downstream_not_reached ''', encoding="utf-8", ) # 2. Focal test for evidence_comparator.py (EvidenceComparator.compare) (tests_dir / "test_self_repair_comparator.py").write_text( '''# -*- coding: utf-8 -*- from app.remediation.self_repair.evidence_comparator import EvidenceComparator from app.remediation.self_repair.schemas import ( EvidenceBundle, IncidentFailureType, IncidentRecord, VerificationResult, ) def test_evidence_comparator_counts_regression_new_failures(): incident = IncidentRecord( mission_id="m_comp_focal", failure_type=IncidentFailureType.FAILED_TEST, observed_symptom="Primary test failed", source_component="verification", ) evidence = EvidenceBundle(incident_id=incident.incident_id) verif = VerificationResult( candidate_id="cand_01", incident_id=incident.incident_id, repair_id="rep_01", compile_pass=True, focused_tests_pass=True, regression_tests_pass=False, regression_tests_summary={"error": "1 regression test failed"}, ) comparator = EvidenceComparator() res = comparator.compare(incident=incident, evidence=evidence, verification=verif) assert res.new_failures == 1 assert res.candidate_result == "FAIL" ''', encoding="utf-8", ) # 3. Focal test for collector.py (EvidenceCollector.collect) (tests_dir / "test_self_repair_collector.py").write_text( '''# -*- coding: utf-8 -*- import os from app.remediation.self_repair.collector import EvidenceCollector from app.remediation.self_repair.schemas import IncidentFailureType, IncidentRecord def test_evidence_collector_preserves_source_excerpt_schema(tmp_path): sample_py = tmp_path / "sample_mod.py" sample_py.write_text("x = 42\\n", encoding="utf-8") incident = IncidentRecord( mission_id="m_coll_focal", failure_type=IncidentFailureType.FAILED_VERIFICATION, observed_symptom="Collector schema check", source_component="verification", ) collector = EvidenceCollector() bundle = collector.collect( incident=incident, source_files=[str(sample_py)], workspace_root=str(tmp_path), ) assert len(bundle.source_excerpts) == 1 excerpt = bundle.source_excerpts[0] assert set(excerpt.keys()) >= {"file_path", "line_count", "head_excerpt", "tail_excerpt", "sha256"} assert excerpt["sha256"] is not None and len(excerpt["sha256"]) == 64 ''', encoding="utf-8", ) # 4. Regression suite for self_repair invariants (tests_dir / "test_self_repair_regression.py").write_text( '''# -*- coding: utf-8 -*- from pathlib import Path from app.remediation.self_repair.sandbox_builder import SandboxBuilder, SandboxSecurityViolation from app.remediation.self_repair.schemas import IncidentRecord, IncidentFailureType def test_regression_sandbox_blocks_traversal(tmp_path): sb = SandboxBuilder(project_root=str(tmp_path)) try: sb._resolve_and_verify_within_root("../outside.py", tmp_path, "test") raised = False except SandboxSecurityViolation: raised = True assert raised is True def test_regression_incident_record_serialization(): inc = IncidentRecord( mission_id="m_reg_01", failure_type=IncidentFailureType.FAILED_TEST, observed_symptom="test failure", source_component="verification", ) d = inc.to_dict() assert d["mission_id"] == "m_reg_01" assert d["failure_type"] == "FAILED_TEST" ''', encoding="utf-8", ) def run_pytest_capture(copy_root: Path, test_rel: str) -> Dict[str, Any]: """Runs authentic pytest in copy_root to capture the real failure traceback and output.""" 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": str(copy_root), "PYTHONDONTWRITEBYTECODE": "1", } proc = subprocess.run(cmd, cwd=str(copy_root), env=env, capture_output=True, text=True, timeout=30) return { "returncode": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr, "combined": (proc.stdout or "") + "\n" + (proc.stderr or ""), } def verify_section_2_and_3_isolation_and_pytest() -> Dict[str, Any]: """ Verifies Section 2 (Verifier & Job Object + PEP 578 + Untrusted Code Block) and Section 3 (Authentic pytest 9.1.1). """ from app.remediation.self_repair.candidate_verifier import CandidateVerifier from app.remediation.self_repair.schemas import PatchResult, RepairProposal, SandboxManifest iso_ws = prepare_isolated_copy(AR76_ROOT / "isolation_verification_ws") write_test_suite_into_copy(iso_ws) verifier = CandidateVerifier() manifest = SandboxManifest( incident_id="inc_iso_ar76", repair_id="rep_iso_ar76", candidate_id="cand_iso_ar76", sandbox_root=str(iso_ws), source_dir=str(iso_ws), candidate_dir=str(iso_ws), ) proposal = RepairProposal( repair_id="rep_iso_ar76", incident_id="inc_iso_ar76", 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="plan_iso_ar76", incident_id="inc_iso_ar76", repair_id="rep_iso_ar76", candidate_id="cand_iso_ar76", status="SUCCESS", file_diffs={"app/remediation/self_repair/circuit.py": "# harmless comment"}, ) inc_dir = EVIDENCE_DIR / "section_2_3_verifier" 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 # Now verify that an UNTRUSTED or HIGH_RISK candidate (e.g. importing socket/ctypes in diff) # is BLOCKED BEFORE EXECUTION because full OS isolation is not active. untrusted_patch_res = PatchResult( patch_plan_id="plan_untrusted_ar76", incident_id="inc_untrusted_ar76", repair_id="rep_untrusted_ar76", candidate_id="cand_untrusted_ar76", status="SUCCESS", file_diffs={ "app/remediation/self_repair/circuit.py": "+import socket\n+import ctypes\n" }, ) untrusted_inc_dir = EVIDENCE_DIR / "section_2_untrusted_blocked" 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"), "audit_hook_registered_before_proof": f_sum.get("audit_hook_registered_before_proof"), "order_verified": (f_sum.get("isolation_proof") or {}).get("order_verified"), "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"), "containment_mechanism": f_sum.get("containment_mechanism"), "child_isolation_proof": f_sum.get("isolation_proof"), "untrusted_candidate_blocked": { "focused_tests_pass": untrusted_verif.focused_tests_pass, "status": untrusted_verif.focused_tests_summary.get("status"), "executed": untrusted_verif.focused_tests_summary.get("executed", False), "error": untrusted_verif.focused_tests_summary.get("error"), "risk_classification": untrusted_verif.focused_tests_summary.get("risk_classification"), }, } def run_case_a_state_classification() -> Dict[str, Any]: """ CASO A: Defecto de clasificación de estado en app/remediation/self_repair/circuit.py (CircuitTracer.trace: CircuitStatus.UNKNOWN en lugar de CircuitStatus.NOT_REACHED). """ from app.remediation.self_repair.shadow_orchestrator import ShadowSelfRepairOrchestrator case_ws = prepare_isolated_copy(AR76_ROOT / "case_a_state_classification") write_test_suite_into_copy(case_ws) # Introduce defect into case_ws/app/remediation/self_repair/circuit.py circuit_py = case_ws / "app" / "remediation" / "self_repair" / "circuit.py" orig_src = circuit_py.read_text(encoding="utf-8") 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, ) assert orig_src != defective_src, "Defect injection in circuit.py failed" circuit_py.write_text(defective_src, encoding="utf-8") # Run authentic pytest to capture the real failure pytest_fail = run_pytest_capture(case_ws, "tests/test_self_repair_circuit.py") assert pytest_fail["returncode"] != 0, "Expected test_self_repair_circuit.py to fail on defective circuit.py" reports_dir = EVIDENCE_DIR / "case_a_reports" if reports_dir.exists(): shutil.rmtree(reports_dir) orch = ShadowSelfRepairOrchestrator( base_reports_dir=str(reports_dir), project_root=str(case_ws), ) mission_snapshot_input = { "mission_id": "mission_ar76_case_a", "instance_id": "inst_ar76_case_a", "workspace_root": str(case_ws), "creator_request": "Verify CircuitTracer downstream node classification", "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", ], "allow_isolated_component_repair": True, "provider_called": False, "network_used": False, }, } 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")) return { "shadow_status": report.shadow_status, "candidate_repair": report.candidate_repair, "root_cause_report": rc_report, "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"), "PYTEST_REAL": verif_report.get("focused_tests_summary", {}).get("PYTEST_REAL"), "kernel_job_active": verif_report.get("focused_tests_summary", {}).get("kernel_job_active"), "specific_job_verified": verif_report.get("focused_tests_summary", {}).get("specific_job_verified"), }, "candidate_verdict": verdict_report, "production_modified": report.production_modified, } def run_case_b_operator_condition() -> Dict[str, Any]: """ CASO B: Defecto de condición u operador en otro módulo real no crítico de Optimus: app/remediation/self_repair/evidence_comparator.py (EvidenceComparator.compare: new_fails -= 1 en lugar de new_fails += 1). """ from app.remediation.self_repair.shadow_orchestrator import ShadowSelfRepairOrchestrator case_ws = prepare_isolated_copy(AR76_ROOT / "case_b_operator_condition") write_test_suite_into_copy(case_ws) 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, ) assert orig_src != defective_src, "Defect injection in evidence_comparator.py failed" comp_py.write_text(defective_src, encoding="utf-8") pytest_fail = run_pytest_capture(case_ws, "tests/test_self_repair_comparator.py") assert pytest_fail["returncode"] != 0, "Expected test_self_repair_comparator.py to fail on defective evidence_comparator.py" reports_dir = EVIDENCE_DIR / "case_b_reports" if reports_dir.exists(): shutil.rmtree(reports_dir) orch = ShadowSelfRepairOrchestrator( base_reports_dir=str(reports_dir), project_root=str(case_ws), ) mission_snapshot_input = { "mission_id": "mission_ar76_case_b", "instance_id": "inst_ar76_case_b", "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", ], "allow_isolated_component_repair": True, "provider_called": False, "network_used": False, }, } 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")) return { "shadow_status": report.shadow_status, "candidate_repair": report.candidate_repair, "root_cause_report": rc_report, "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"), "PYTEST_REAL": verif_report.get("focused_tests_summary", {}).get("PYTEST_REAL"), "kernel_job_active": verif_report.get("focused_tests_summary", {}).get("kernel_job_active"), "specific_job_verified": verif_report.get("focused_tests_summary", {}).get("specific_job_verified"), }, "candidate_verdict": verdict_report, "production_modified": report.production_modified, } def run_case_c_insufficient_confidence() -> Dict[str, Any]: """ CASO C: Defecto estructural complejo en app/remediation/self_repair/collector.py (EvidenceCollector.collect: omite campos estructurales del diccionario source_excerpts) donde la evidencia determinista no alcanza confianza alta: - Reconoce insuficiencia de evidencia (confidence="UNPROVEN") - Se detiene antes de aplicar parches o ejecutar sandbox (sandbox_executed=False) - No toca producción (production_modified=False) - Registra la causa del bloqueo """ from app.remediation.self_repair.shadow_orchestrator import ShadowSelfRepairOrchestrator case_ws = prepare_isolated_copy(AR76_ROOT / "case_c_insufficient_confidence") write_test_suite_into_copy(case_ws) coll_py = case_ws / "app" / "remediation" / "self_repair" / "collector.py" orig_src = coll_py.read_text(encoding="utf-8") defective_src = orig_src.replace( ' source_excerpts.append({\n' ' "file_path": fpath,\n' ' "line_count": len(lines),\n' ' "head_excerpt": "".join(lines[:30]),\n' ' "tail_excerpt": "".join(lines[-30:] if len(lines) > 30 else []),\n' ' "sha256": h,\n' ' })', ' source_excerpts.append({\n' ' "file_path": fpath,\n' ' "line_count": len(lines),\n' ' })', 1, ) assert orig_src != defective_src, "Defect injection in collector.py failed" coll_py.write_text(defective_src, encoding="utf-8") hash_before = sha256_file(coll_py) pytest_fail = run_pytest_capture(case_ws, "tests/test_self_repair_collector.py") assert pytest_fail["returncode"] != 0, "Expected test_self_repair_collector.py to fail on defective collector.py" reports_dir = EVIDENCE_DIR / "case_c_reports" if reports_dir.exists(): shutil.rmtree(reports_dir) orch = ShadowSelfRepairOrchestrator( base_reports_dir=str(reports_dir), project_root=str(case_ws), ) mission_snapshot_input = { "mission_id": "mission_ar76_case_c", "instance_id": "inst_ar76_case_c", "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", ], "allow_isolated_component_repair": True, "provider_called": False, "network_used": False, }, } 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 { "shadow_status": report.shadow_status, "candidate_repair": report.candidate_repair, "autonomous_patch_generated": report.autonomous_patch_generated, "sandbox_executed": report.sandbox_executed, "production_modified": report.production_modified, "target_file_hash_unchanged": (hash_before == hash_after), "root_cause_report": rc_report, "patch_reasoning": pr_report, } def verify_section_6_brain_terminal_hook_and_rollback() -> Dict[str, Any]: """ Section 6: Connects the real terminal hook of an isolated Brain copy (OPERATIONAL_TASKS[tid] -> dispatch_live_shadow_hook(_snap) -> ShadowSelfRepairOrchestrator -> SelfRepairEngine) WITHOUT the test script calling auto_remediate_mission directly, and verifies controlled promotion + rollback. """ 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 brain_ws = prepare_isolated_copy(AR76_ROOT / "brain_integration_copy") write_test_suite_into_copy(brain_ws) # Inject the state classification defect in brain_ws/app/remediation/self_repair/circuit.py circuit_py = brain_ws / "app" / "remediation" / "self_repair" / "circuit.py" orig_src = circuit_py.read_text(encoding="utf-8") 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") defective_hash = sha256_file(circuit_py) expected_repaired_hash = hashlib.sha256(orig_src.encode("utf-8")).hexdigest().upper() # Simulate an operational task running in brain_host_tls.OPERATIONAL_TASKS # whose verification step runs pytest on the workspace and records the terminal failure, # then invokes the exact terminal hook `dispatch_live_shadow_hook(_snap)` from brain_host_tls.py! pytest_fail = run_pytest_capture(brain_ws, "tests/test_self_repair_circuit.py") tid = "op_task_ar76_brain_hook_01" operational_tasks: Dict[str, Dict[str, Any]] = { tid: { "task_id": tid, "mission_id": "mission_ar76_brain_live_01", "instance_id": "inst_ar76_brain_live_01", "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", ], "allow_isolated_component_repair": True, "provider_called": False, "network_used": False, "activity": [ {"timestamp": "2026-10-27T13:00:00Z", "message": "Mission verification failed on CircuitTracer"} ], } } reports_dir = EVIDENCE_DIR / "brain_hook_reports" 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), ) # Execute the exact terminal hook pattern from brain_host_tls.py (_run_bg_operational_mission finally block) _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}" # Verify that shadow mode did NOT modify brain_ws/app/remediation/self_repair/circuit.py hash_after_shadow = sha256_file(circuit_py) assert hash_after_shadow == defective_hash, "Shadow repair mutated workspace before promotion!" # Now verify controlled promotion + rollback in the isolated copy 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 = PromotionGuard(workspace_root=str(brain_ws)) promo_exec = PromotionExecutor(guard=guard) 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"] ], ) promo_res = promo_exec.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, network_used=False, 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") # Now test physical rollback to prove original defective bytes/hash can be restored atomically rollback_art = promo_exec.execute_rollback(promo_res.rollback_artifact) hash_after_rollback = sha256_file(circuit_py) 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), "controlled_promotion": { "promotion_verified": promo_res.promotion_verified, "hash_after_promotion": hash_after_promotion, "matches_canonical_repaired_hash": (hash_after_promotion == expected_repaired_hash), "post_promotion_pytest_returncode": pytest_after_promo["returncode"], }, "controlled_rollback": { "rollback_completed": rollback_art.rollback_completed, "original_hash_restored": rollback_art.original_hash_restored, "hash_after_rollback": hash_after_rollback, "matches_initial_hash": (hash_after_rollback == defective_hash), }, } def verify_section_7_external_supervisor_vs_internal_repair() -> Dict[str, Any]: """ Section 7: Verifies the distinction between: - Live Brain with defective function -> SelfRepairEngine / ShadowSelfRepairOrchestrator - Hung or stopped Brain (PORT_DOWN / PROCESS_DEAD / HEALTH_TIMEOUT) -> ContinuousHealthMonitor + SystemRecoveryOrchestrator (with MAX_RESTARTS=2, COOLDOWN_SEC=30, Circuit Breaker). """ from app.system_health.recovery_orchestrator import SystemRecoveryOrchestrator from app.system_health.runtime_monitor import ContinuousHealthMonitor, RuntimeFailureClass sup_dir = EVIDENCE_DIR / "section_7_supervisor" if sup_dir.exists(): shutil.rmtree(sup_dir) sup_dir.mkdir(parents=True, exist_ok=True) # Probe an unused local port to deterministically observe PORT_DOWN / PROCESS_DEAD detection monitor = ContinuousHealthMonitor( host="127.0.0.1", port=59443, pid=999999, # non-existent PID timeout_sec=1.0, ) obs = monitor.observe() # Verify SystemRecoveryOrchestrator loop guards and circuit breaker without spawning production processes rec_orch = SystemRecoveryOrchestrator( reports_dir=str(sup_dir), launcher_cmd=[str(TEST_VENV_PY), "-c", "pass"], monitor=monitor, ) rec_orch.COOLDOWN_SEC = 0.01 # fast cooldown for lab test # Override _execute_controlled_restart and _verify_post_recovery in lab instance to test loop guard & circuit breaker safely restart_calls = [] def _mock_restart(inc_id: str) -> bool: restart_calls.append(inc_id) return True rec_orch._execute_controlled_restart = _mock_restart # type: ignore rec_orch._verify_post_recovery = lambda: False # simulate brain still hung after restart inc1 = rec_orch.trigger_recovery(obs) time.sleep(0.02) inc2 = rec_orch.trigger_recovery(obs) time.sleep(0.02) inc3_circuit_open = rec_orch.trigger_recovery(obs) return { "monitor_observation": obs.to_dict(), "failure_classes_detected": obs.failure_classes, "attempt_1_outcome": inc1.recovery_outcome, "attempt_2_outcome": inc2.recovery_outcome, "attempt_3_outcome": inc3_circuit_open.recovery_outcome, "circuit_breaker_opened_on_attempt_3": (inc3_circuit_open.recovery_outcome == "CIRCUIT_BREAKER_OPEN"), "max_restarts_enforced": rec_orch.MAX_RESTARTS, "restart_calls_executed": len(restart_calls), } def generate_patches_and_hashes() -> Dict[str, Any]: """Generates unified .patch files in 06_ar76_candidate_hardening/patches and SHA256 table.""" PATCHES_DIR.mkdir(parents=True, exist_ok=True) modified_rel_paths = [ "app/remediation/__init__.py", "app/remediation/self_repair/sandbox_builder.py", "app/remediation/self_repair/patch_executor.py", "app/remediation/self_repair/candidate_verifier.py", "app/remediation/self_repair/planner.py", "app/remediation/self_repair/engine.py", "app/remediation/self_repair/autonomous_patch_reasoner.py", "app/remediation/self_repair/circuit.py", "app/remediation/self_repair/schemas.py", "app/remediation/self_repair/analyzer.py", "app/remediation/self_repair/live_shadow_hook.py", "app/remediation/self_repair/shadow_orchestrator.py", "app/system_health/runtime_monitor.py", "app/system_health/recovery_orchestrator.py", ] file_records: List[Dict[str, Any]] = [] for rel in modified_rel_paths: orig_p = REF_ORIG / rel cand_p = CANDIDATE_WS / rel if not orig_p.is_file() or not cand_p.is_file(): continue orig_sha = sha256_file(orig_p) cand_sha = sha256_file(cand_p) orig_lines = orig_p.read_text(encoding="utf-8").splitlines(keepends=True) cand_lines = cand_p.read_text(encoding="utf-8").splitlines(keepends=True) diff_lines = list( difflib.unified_diff( orig_lines, cand_lines, fromfile=f"a/{rel}", tofile=f"b/{rel}", ) ) patch_name = Path(rel).stem + ".patch" patch_path = PATCHES_DIR / patch_name patch_path.write_text("".join(diff_lines), encoding="utf-8") patch_sha = sha256_file(patch_path) file_records.append({ "relative_path": rel, "original_sha256": orig_sha, "candidate_sha256": cand_sha, "changed": (orig_sha != cand_sha), "patch_file": str(patch_path), "patch_sha256": patch_sha, "diff_line_count": len(diff_lines), }) return {"files": file_records} 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(AR76_ROOT), "pytest_venv_python": str(TEST_VENV_PY), } print("[AR-76] 1/5 Verifying Section 2 & 3 (Isolation & Authentic Pytest 9.1.1)...") summary["section_2_3_isolation_and_pytest"] = verify_section_2_and_3_isolation_and_pytest() print("[AR-76] 2/5 Verifying Section 4 & 5 (Cases A, B, C)...") summary["case_a_state_classification"] = run_case_a_state_classification() summary["case_b_operator_condition"] = run_case_b_operator_condition() summary["case_c_insufficient_confidence"] = run_case_c_insufficient_confidence() print("[AR-76] 3/5 Verifying Section 6 (Brain Terminal Hook & Controlled Rollback)...") summary["section_6_brain_hook_and_rollback"] = verify_section_6_brain_terminal_hook_and_rollback() print("[AR-76] 4/5 Verifying Section 7 (External Supervisor vs Internal Repair)...") summary["section_7_external_supervisor"] = verify_section_7_external_supervisor_vs_internal_repair() print("[AR-76] 5/5 Generating unified patches & SHA256 manifest...") summary["patches_and_hashes"] = generate_patches_and_hashes() out_json = EVIDENCE_DIR / "ar76_verification_summary.json" out_json.write_text(json.dumps(summary, indent=2, ensure_ascii=False), encoding="utf-8") print(f"[AR-76] Verification complete. Summary written to: {out_json}") if __name__ == "__main__": main()