# -*- coding: utf-8 -*- """ Tests for AR88-04H: Focal Security & Continuity Corrections of InternalRepairCoordinator. VERIFICATION CRITERIA (10 FOCAL CHECKS): 1. Un FAIL manual con hash correcto no obtiene autorizacion fisica (BLOCKED_UNACCREDITED_ISOLATION). 2. Sin aislamiento acreditado no se ejecutan reparaciones dinamicas. 3. El incidente y su evidencia originales se conservan intactos (incident_id preservado). 4. Un duplicado sobrevive a una nueva instancia del coordinador (deduplicacion persistente). 5. Una reparacion fallida no queda registrada como REPAIRED en el store. 6. read_only=True no modifica candidatos ni ejecuta parches. 7. Los identificadores heredados de mision no adquieren autoridad (CONTRACT_MISMATCH si mission_id != None). 8. Una simulacion con parche suministrado no se presenta como reparacion autonoma (is_autonomous_patch=False). 9. Produccion permanece 100% intacta. 10. Ningun resultado artificial obtiene PHYSICAL_E2E_PASS. """ import json import os import sys import tempfile import pytest WS = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if WS not in sys.path: sys.path.insert(0, WS) from app.remediation.self_repair.functional_assessor import ( FunctionalAssessmentStatus, FunctionalEvaluationResult, InternalFunctionalAssessor, _AUTHORIZED_INTERNAL_CONTRACTS, ) from app.remediation.self_repair.functional_diagnostic_adapter import ( FunctionalDiagnosticAdapter, _compute_sha256, ) from app.remediation.self_repair.internal_repair_coordinator import ( InternalRepairCoordinator, InternalRepairStatus, PersistentIncidentStore, ) def _create_authentic_fail_result() -> FunctionalEvaluationResult: target_rel = "app/remediation/self_repair/circuit.py" target_abs = os.path.join(WS, target_rel) target_hash = _compute_sha256(target_abs) return FunctionalEvaluationResult( contract_id="contract_circuit_tracer_downstream_break", component_name="CircuitTracer", target_file_rel=target_rel, file_sha256=target_hash, status=FunctionalAssessmentStatus.FAIL, duration_ms=35.0, assertion_error="AssertionError: effector expected NOT_REACHED, got CircuitStatus.UNKNOWN", stack_trace="Traceback: ...", stdout="[TEST] break reproduced", ) def test_1_and_2_manual_fail_without_accredited_isolation_is_blocked(): """ Comprobaciones 1 y 2: Un FAIL manual con hash valido no obtiene autorizacion fisica para reparaciones. Sin aislamiento acreditado en el entorno de ejecucion, devuelve BLOCKED_UNACCREDITED_ISOLATION. """ eval_result = _create_authentic_fail_result() temp_dir = tempfile.mkdtemp() store_file = os.path.join(temp_dir, "test_store.json") coord = InternalRepairCoordinator( project_root=WS, base_reports_dir=os.path.join(temp_dir, "reports"), store_path=store_file, ) # Intento de ejecucion dinamica en host (donde all_barriers_satisfied=False) summary = coord.coordinate_internal_evaluation( eval_result=eval_result, read_only=False, force_accredited_isolation=False, persist_artifacts=False, ) assert summary["status"] == InternalRepairStatus.BLOCKED_UNACCREDITED_ISOLATION assert summary["internal_incident_admitted"] is True assert summary["sandbox_repair_executed"] is False assert "Unaccredited OS isolation" in summary["error"] def test_3_original_incident_and_evidence_are_preserved(): """ Comprobacion 3: El diagnostico interno conserva el incident_id y la evidencia original sin fabricar un segundo incidente ni perder la procedencia. """ eval_result = _create_authentic_fail_result() temp_dir = tempfile.mkdtemp() store_file = os.path.join(temp_dir, "test_store.json") coord = InternalRepairCoordinator( project_root=WS, base_reports_dir=os.path.join(temp_dir, "reports"), store_path=store_file, ) summary = coord.coordinate_internal_evaluation( eval_result=eval_result, read_only=True, persist_artifacts=False, ) assert summary["status"] == InternalRepairStatus.READ_ONLY_DIAGNOSTIC_COMPLETE assert summary["internal_incident_admitted"] is True assert summary["incident_id"].startswith("diag_inc_") assert summary["root_cause_proven"] is True assert summary["responsible_function"] == "CircuitTracer.trace" def test_4_persistent_deduplication_survives_new_coordinator_instance(): """ Comprobacion 4: Un incidente duplicado sobrevive al reinicio o re-instanciacion del coordinador mediante el almacén persistente atomico. """ eval_result = _create_authentic_fail_result() temp_dir = tempfile.mkdtemp() store_file = os.path.join(temp_dir, "persistent_store.json") # Instancia 1: Observa y ejecuta diagnostico de solo lectura coord1 = InternalRepairCoordinator(project_root=WS, store_path=store_file) res1 = coord1.coordinate_internal_evaluation(eval_result, read_only=True, persist_artifacts=False) assert res1["status"] == InternalRepairStatus.READ_ONLY_DIAGNOSTIC_COMPLETE # Simulamos que fue reparado en el store sig = coord1.compute_evaluation_signature(eval_result) coord1.store.record_repaired(sig, "cand_repaired_001") # Instancia 2: Se crea una instancia completamente nueva leyendo el mismo store persistente coord2 = InternalRepairCoordinator(project_root=WS, store_path=store_file) res2 = coord2.coordinate_internal_evaluation(eval_result, read_only=False, persist_artifacts=False) assert res2["status"] == InternalRepairStatus.REJECTED_DUPLICATE_INCIDENT assert "already verified as REPAIRED" in res2["error"] def test_5_failed_repair_is_not_recorded_as_repaired(): """ Comprobacion 5: Una reparacion fallida (o con regresion) no queda registrada como REPAIRED en el store. """ eval_result = _create_authentic_fail_result() temp_dir = tempfile.mkdtemp() store_file = os.path.join(temp_dir, "test_store_failed.json") coord = InternalRepairCoordinator( project_root=WS, base_reports_dir=os.path.join(temp_dir, "reports"), store_path=store_file, ) explicit_instruction = { "file_rel_path": eval_result.target_file_rel, "operation": "REPLACE_EXACT", "search_block": """ if break_reached: # After first break, subsequent nodes are NOT_REACHED unless evidence proves execution c_status = CircuitStatus.UNKNOWN""", "replace_block": """ if break_reached: # After first break, subsequent nodes are NOT_REACHED unless evidence proves execution c_status = CircuitStatus.NOT_REACHED""", } # Runner simulado que falla en regresiones def mock_failing_regression(cand_dir: str): return False, {"passed": False, "failures": ["reg_check"]} summary = coord.coordinate_internal_evaluation( eval_result=eval_result, read_only=False, force_accredited_isolation=True, # Test controlado de la logica de rechazo explicit_patch_instruction=explicit_instruction, regression_test_runner=mock_failing_regression, persist_artifacts=False, ) assert summary["status"] == InternalRepairStatus.CANDIDATE_REJECTED assert summary["candidate_verdict"] == "REJECT" # Verificar que el store persistente NO lo marco como REPAIRED sig = coord.compute_evaluation_signature(eval_result) rec = coord.store.get_record(sig) assert rec["state"] == "FAILED" assert rec["state"] != "REPAIRED" def test_6_read_only_mode_does_not_modify_candidates(): """ Comprobacion 6: read_only=True genera diagnostico y propuesta pero no ejecuta mutaciones ni sandbox. """ eval_result = _create_authentic_fail_result() temp_dir = tempfile.mkdtemp() coord = InternalRepairCoordinator(project_root=WS, store_path=os.path.join(temp_dir, "store.json")) summary = coord.coordinate_internal_evaluation(eval_result, read_only=True, persist_artifacts=False) assert summary["status"] == InternalRepairStatus.READ_ONLY_DIAGNOSTIC_COMPLETE assert summary["sandbox_repair_executed"] is False assert summary["candidate_verdict"] is None def test_7_legacy_identifiers_do_not_acquire_authority(): """ Comprobacion 7: Si un resultado intenta inyectar un mission_id de cliente en un incidente interno, es rechazado con CONTRACT_MISMATCH. """ eval_result = _create_authentic_fail_result() temp_dir = tempfile.mkdtemp() coord = InternalRepairCoordinator(project_root=WS, store_path=os.path.join(temp_dir, "store.json")) # Inyeccion de mock donde adapter devolveria un mission_id no nulo from app.remediation.self_repair.schemas import IncidentRecord, EvidenceBundle, IncidentFailureType bad_incident = IncidentRecord( incident_id="inc_bad_client_inject", mission_id="m_client_user_fake_001", failure_type=IncidentFailureType.FAILED_TEST, observed_symptom="Simulated symptom", source_component="CircuitTracer", ) bad_evidence = EvidenceBundle(incident_id="inc_bad_client_inject") # Validamos como adapter reacciona si el incidente tiene mission_id original_fn = FunctionalDiagnosticAdapter.validate_and_convert_failure try: FunctionalDiagnosticAdapter.validate_and_convert_failure = lambda res, workspace_root=None: (True, None, (None, bad_incident, bad_evidence)) summary = coord.coordinate_internal_evaluation(eval_result, read_only=True, persist_artifacts=False) assert summary["status"] == InternalRepairStatus.CONTRACT_MISMATCH assert "unauthorized client mission_id" in summary["error"] finally: FunctionalDiagnosticAdapter.validate_and_convert_failure = original_fn def test_8_explicit_patch_instruction_is_marked_non_autonomous(): """ Comprobacion 8: Una simulacion con parche suministrado no se presenta como reparacion autonoma (is_autonomous_patch=False). """ eval_result = _create_authentic_fail_result() temp_dir = tempfile.mkdtemp() coord = InternalRepairCoordinator(project_root=WS, store_path=os.path.join(temp_dir, "store.json")) explicit_instruction = { "file_rel_path": eval_result.target_file_rel, "operation": "REPLACE_EXACT", "search_block": """ if break_reached: # After first break, subsequent nodes are NOT_REACHED unless evidence proves execution c_status = CircuitStatus.UNKNOWN""", "replace_block": """ if break_reached: # After first break, subsequent nodes are NOT_REACHED unless evidence proves execution c_status = CircuitStatus.NOT_REACHED""", } def mock_focused_runner(cand_dir: str): return True, {"passed": True} def mock_regression_runner(cand_dir: str): return True, {"passed": True} summary = coord.coordinate_internal_evaluation( eval_result=eval_result, read_only=False, force_accredited_isolation=True, explicit_patch_instruction=explicit_instruction, focused_test_runner=mock_focused_runner, regression_test_runner=mock_regression_runner, persist_artifacts=False, ) assert summary["status"] == InternalRepairStatus.CANDIDATE_ACCEPTED assert summary["is_autonomous_patch"] is False # Declarado como no autonomo def test_9_and_10_production_remains_intact_and_no_physical_e2e_pass_without_isolation(): """ Comprobaciones 9 y 10: Produccion permanece intacta con 0 modificaciones. Ningun resultado artificial obtiene PHYSICAL_E2E_PASS. """ import hashlib prod_circuit = r"C:\xampp\htdocs\Digital Optimus\app\remediation\self_repair\circuit.py" with open(prod_circuit, "rb") as f: prod_h = hashlib.sha256(f.read()).hexdigest().upper() assert prod_h == "AA3999237E9AF1ACC08B533460A73826984E320B893FFFD9BAD3A5D781BAFE7E" eval_result = _create_authentic_fail_result() coord = InternalRepairCoordinator(project_root=WS) summary = coord.coordinate_internal_evaluation(eval_result, read_only=False, persist_artifacts=False) # In host environment, status is strictly BLOCKED_UNACCREDITED_ISOLATION assert summary["status"] == InternalRepairStatus.BLOCKED_UNACCREDITED_ISOLATION assert summary["status"] != "PHYSICAL_E2E_PASS"