# -*- coding: utf-8 -*- """ Tests for AR88-INTERNAL-AUTONOMY-03: Execution Provenance, Internal Identity, and Safe Internal Delivery. FOCAL TESTS: 1. Rejection of artificial FAIL fixture without accredited execution provenance. 2. Rejection of unauthorized contracts. 3. Rejection of synthetic/fictitious mission IDs masquerading as internal identity. 4. Safe handling of internal incidents: mission_id is strictly None (decoupled from client missions). 5. Safe rejection of BLOCKED, PASS, and UNPROVEN results. 6. Execution of diagnosis without patch execution, candidate mutation, or promotion. 7. Zero writes to production directory. 8. Preservation of FAIL_CLOSED and OBS-1 invariants. """ import os import sys 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.self_diagnostic import OptimusSelfDiagnostic from app.remediation.self_repair.engine import SelfRepairEngine from app.remediation.self_repair.schemas import IncidentFailureType # ------------------------------------------------------------------------------ # 1. PROCEDENCIA DE EJECUCION Y RECHAZOS SEGUROS # ------------------------------------------------------------------------------ def test_assessor_host_execution_produces_blocked_due_to_unaccredited_isolation(): """ Demonstrates Category 2 vs Category 3: When InternalFunctionalAssessor runs on host without accredited OS isolation barriers, it returns BLOCKED. This ensures no uncontained dynamic execution occurs. """ assessor = InternalFunctionalAssessor(workspace_root=WS) barrier_audit = assessor.audit_os_isolation_barriers() assert barrier_audit["all_barriers_satisfied"] is False results = assessor.assess_all_components() assert len(results) == len(_AUTHORIZED_INTERNAL_CONTRACTS) for res in results: assert res.status == FunctionalAssessmentStatus.BLOCKED assert "FAIL_CLOSED" in (res.inconclusive_reason or "") def test_adapter_rejects_blocked_pass_and_unproven(): """ Ensures that non-FAIL outcomes (BLOCKED, PASS, UNPROVEN) are strictly rejected and never converted into diagnostic incidents. """ target_rel = "app/remediation/self_repair/circuit.py" target_hash = _compute_sha256(os.path.join(WS, target_rel)) for non_fail_status in [ FunctionalAssessmentStatus.BLOCKED, FunctionalAssessmentStatus.PASS, FunctionalAssessmentStatus.UNPROVEN, ]: res = FunctionalEvaluationResult( contract_id="contract_circuit_tracer_downstream_break", component_name="CircuitTracer", target_file_rel=target_rel, file_sha256=target_hash, status=non_fail_status, duration_ms=10.0, assertion_error=None, ) valid, reason, result_tuple = FunctionalDiagnosticAdapter.validate_and_convert_failure(res, WS) assert not valid assert "REJECTED_STATUS" in reason assert result_tuple is None def test_adapter_rejects_unauthorized_contract(): """ Ensures that any contract not registered in the authorized internal catalog is immediately rejected. """ res = FunctionalEvaluationResult( contract_id="contract_synthetic_external_test", component_name="CircuitTracer", target_file_rel="app/remediation/self_repair/circuit.py", file_sha256=_compute_sha256(os.path.join(WS, "app/remediation/self_repair/circuit.py")), status=FunctionalAssessmentStatus.FAIL, duration_ms=10.0, assertion_error="AssertionError: synthetic", ) valid, reason, result_tuple = FunctionalDiagnosticAdapter.validate_and_convert_failure(res, WS) assert not valid assert "REJECTED_UNAUTHORIZED_CONTRACT" in reason assert result_tuple is None def test_adapter_rejects_hash_mismatch(): """ Ensures that evaluation results with a recorded hash different from physical disk are rejected. """ res = FunctionalEvaluationResult( contract_id="contract_circuit_tracer_downstream_break", component_name="CircuitTracer", target_file_rel="app/remediation/self_repair/circuit.py", file_sha256="0000000000000000000000000000000000000000000000000000000000000000", status=FunctionalAssessmentStatus.FAIL, duration_ms=10.0, assertion_error="AssertionError: simulated", ) valid, reason, result_tuple = FunctionalDiagnosticAdapter.validate_and_convert_failure(res, WS) assert not valid assert "REJECTED_HASH_MISMATCH" in reason assert result_tuple is None # ------------------------------------------------------------------------------ # 2. IDENTIDAD INTERNA SEPARADA DE MISIONES DE USUARIO / CLIENTES # ------------------------------------------------------------------------------ def test_internal_incident_has_no_synthetic_mission_id(): """ FASE B: Internal self-repair incidents must NOT fabricate client mission IDs. IncidentRecord.mission_id must be None, establishing distinct internal identity. """ target_rel = "app/remediation/self_repair/circuit.py" target_hash = _compute_sha256(os.path.join(WS, target_rel)) res = 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=30.0, assertion_error="AssertionError: effector expected NOT_REACHED, got CircuitStatus.UNKNOWN", stack_trace="Traceback: ...", ) valid, reason, result_tuple = FunctionalDiagnosticAdapter.validate_and_convert_failure(res, WS) assert valid, f"Conversion rejected: {reason}" assert result_tuple is not None event, incident, evidence = result_tuple # Strict identity separation: assert incident.mission_id is None, "Internal self-repair incident must not possess a synthetic client mission_id" assert incident.instance_id is None assert incident.initial_evidence.get("incident_category") == "INTERNAL_FUNCTIONAL_CONTRACT_FAIL" assert incident.initial_evidence.get("provenance_origin") == "INTERNAL_BEHAVIORAL_CONTRACT" # ------------------------------------------------------------------------------ # 3. CONEXION Y FLUJO DE DIAGNOSTICO EN MODO LECTURA SIN PROMOCION # ------------------------------------------------------------------------------ def test_internal_diagnostic_flow_executes_in_read_only_mode_without_promotion(): """ FASE C: Traces the delivery of an internal incident to SelfRepairEngine. Verifies that: - Root cause analysis correctly identifies CircuitTracer.trace. - Proposal has repair_authorized=True (for sandbox), but production_write_authority=False and auto_promotion=False. - No patch execution or candidate promotion occurs. """ target_rel = "app/remediation/self_repair/circuit.py" target_hash = _compute_sha256(os.path.join(WS, target_rel)) res = 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: ...", ) valid, reason, result_tuple = FunctionalDiagnosticAdapter.validate_and_convert_failure(res, WS) assert valid _, incident, evidence = result_tuple engine = SelfRepairEngine(project_root=WS, base_reports_dir=os.path.join(WS, "reports", "test_runs")) inc, ev, trace, root_cause, proposal = engine.process_incident( failure_type=incident.failure_type, observed_symptom=incident.observed_symptom, source_component=incident.source_component, initial_evidence=incident.initial_evidence, source_files=[target_rel], mission_id=incident.mission_id, # None instance_id=incident.instance_id, # None workspace_root=WS, persist_artifacts=False, ) # Verification of diagnosis assert inc.mission_id is None assert root_cause.responsible_function == "CircuitTracer.trace" assert root_cause.confidence == "HIGH" assert root_cause.proven is True # Absolute safety fence: assert proposal.production_write_authority is False assert proposal.core_write_authority is False assert proposal.auto_promotion is False assert proposal.files_allowed == [target_rel] def test_optimus_self_diagnostic_host_run_safely_reports_zero_admitted_incidents(): """ Verifies that running OptimusSelfDiagnostic in the active host environment produces 0 internal incidents because assessor fails closed with BLOCKED. """ diag = OptimusSelfDiagnostic(project_root=WS) report = diag.run_diagnostic() assert report.overall_status == "BLOCKED" assert len(report.internal_incidents) == 0 # Ensure functional contract checks are reported as BLOCKED functional_checks = [c for c in report.checks if c.name.startswith("FunctionalContract_")] assert len(functional_checks) == len(_AUTHORIZED_INTERNAL_CONTRACTS) for fc in functional_checks: assert fc.status == "BLOCKED" assert "FAIL_CLOSED" in (fc.error_message or "")