# -*- coding: utf-8 -*- """ AR-80 Focal Test Suite: 1. Recursion lock tests (Scenarios A, B, C, D) 2. Authentic telemetry trinary contract tests (5 scenarios) """ from __future__ import annotations import os import shutil import tempfile import pytest from app.remediation.self_repair.schemas import ( MissionFailureSnapshot, ShadowStatus, ) from app.remediation.self_repair.shadow_orchestrator import ShadowSelfRepairOrchestrator from app.remediation.self_repair.telemetry_contract import ( AuthenticTelemetryEvaluator, SCOPE_RAW_SOCKET, SCOPE_URLLIB_HTTP, ) @pytest.fixture def temp_env(): ws = tempfile.mkdtemp(prefix="ar80_ws_") rep = tempfile.mkdtemp(prefix="ar80_rep_") yield ws, rep shutil.rmtree(ws, ignore_errors=True) shutil.rmtree(rep, ignore_errors=True) # ============================================================================== # 1. RECURSION LOCK TESTS (AR-80) # ============================================================================== def test_recursion_scenario_a_normal_incident_without_permission(temp_env): """ Scenario A: Incident originating in self-repair component without internal authority must be strictly blocked. """ ws, rep = temp_env orch = ShadowSelfRepairOrchestrator(base_reports_dir=rep, project_root=ws) snapshot = MissionFailureSnapshot( mission_id="mis_rec_a", instance_id="inst_rec_a", workspace_root=ws, mission_status="FAILED", source_component="remediation", observed_symptom="Internal circuit tracer failure", ) is_rec, rec_source = orch.is_self_repair_recursion(snapshot) assert is_rec is True assert rec_source == "remediation" report = orch.handle_mission_finished(snapshot) assert report.SELF_REPAIR_INTERNAL_FAILURE is True assert report.RECURSIVE_REPAIR_STARTED is False assert report.shadow_status == ShadowStatus.SHADOW_INTERNAL_ERROR.value assert report.guard_decision == "BLOCK" assert report.circuit_state == "OPEN" def test_recursion_scenario_b_incident_injecting_boolean_manually(temp_env): """ Scenario B: Incident trying to smuggle allow_isolated_component_repair=True inside mission error_context / payload must be IGNORED and blocked. """ ws, rep = temp_env # Orchestrator does NOT grant internal authority orch = ShadowSelfRepairOrchestrator(base_reports_dir=rep, project_root=ws) snapshot = MissionFailureSnapshot( mission_id="mis_rec_b", instance_id="inst_rec_b", workspace_root=ws, mission_status="FAILED", source_component="self_repair", observed_symptom="Smuggled permission in payload", error_context={"allow_isolated_component_repair": True}, ) is_rec, rec_source = orch.is_self_repair_recursion(snapshot) assert is_rec is True assert rec_source == "self_repair" report = orch.handle_mission_finished(snapshot) assert report.SELF_REPAIR_INTERNAL_FAILURE is True assert report.RECURSIVE_REPAIR_STARTED is False assert report.shadow_status == ShadowStatus.SHADOW_INTERNAL_ERROR.value def test_recursion_scenario_c_incident_with_internal_traceback(temp_env): """ Scenario C: Incident containing internal self-repair tracebacks must be blocked, unless verifiable internal orchestrator authority is configured. """ ws, rep = temp_env orch = ShadowSelfRepairOrchestrator(base_reports_dir=rep, project_root=ws) snapshot = MissionFailureSnapshot( mission_id="mis_rec_c", instance_id="inst_rec_c", workspace_root=ws, mission_status="FAILED", source_component="generic_executor", observed_symptom="Traceback contains self repair", tracebacks=[ "Traceback (most recent call last):\n File \"app/remediation/self_repair/circuit.py\", line 42, in trace\n raise ValueError()" ], ) is_rec, rec_source = orch.is_self_repair_recursion(snapshot) assert is_rec is True assert "app.remediation.self_repair" in str(rec_source) report = orch.handle_mission_finished(snapshot) assert report.SELF_REPAIR_INTERNAL_FAILURE is True assert report.RECURSIVE_REPAIR_STARTED is False def test_recursion_scenario_d_in_flight_reentrancy_blocked(temp_env): """ Scenario D: Recursive repair invocation while handle_mission_finished is active (in_flight_reentrancy) must ALWAYS be blocked immediately, even if internal authority is set. """ ws, rep = temp_env orch = ShadowSelfRepairOrchestrator( base_reports_dir=rep, project_root=ws, allow_isolated_component_repair=True, ) sub_dir = os.path.join(ws, "app") os.makedirs(sub_dir, exist_ok=True) prod_file = os.path.join(sub_dir, "reentrant.py") with open(prod_file, "w", encoding="utf-8") as f: f.write("def f():\n return 'pending'\n") reentrant_snapshot = MissionFailureSnapshot( mission_id="mis_recursive_004b", instance_id="inst_recursive_004b", workspace_root=ws, mission_status="FAILED", failure_stage="executor", source_component="executor", observed_symptom="AssertionError: expected 'completed', actual 'pending'", action_history=[ {"step": 1, "component": "executor", "status": "FAIL", "target": "app/reentrant.py"}, ], ) reentrant_report = None def sneaky_runner(cand_dir): nonlocal reentrant_report # Attempt to recursively invoke orchestrator during active execution nested_snap = MissionFailureSnapshot( mission_id="mis_nested_reentrant", workspace_root=ws, mission_status="FAILED", observed_symptom="Nested crash", ) is_rec, rec_reason = orch.is_self_repair_recursion(nested_snap) assert is_rec is True assert rec_reason == "in_flight_reentrancy" reentrant_report = orch.handle_mission_finished(nested_snap) return True, {} orch.handle_mission_finished(reentrant_snapshot, focused_test_runner=sneaky_runner) assert reentrant_report is not None assert reentrant_report.SELF_REPAIR_INTERNAL_FAILURE is True assert reentrant_report.RECURSIVE_REPAIR_STARTED is False assert reentrant_report.shadow_status == ShadowStatus.SHADOW_INTERNAL_ERROR.value assert reentrant_report.error_message == "Self-repair recursion blocked: in_flight_reentrancy" # ============================================================================== # 2. TELEMETRY CONTRACT & TRINARY STATE TESTS (AR-80) # ============================================================================== def test_telemetry_scenario_1_monitor_absent_evaluates_unknown(temp_env): """ Scenario 1: Monitor absent / inactive evaluates to UNKNOWN and blocks promotion. """ rec = AuthenticTelemetryEvaluator.evaluate_telemetry( mission_id="m_tel_1", instance_id="inst_tel_1", workspace_root="/tmp", observer_active=False, monitored_scopes=[], detected_attempts=[], ) assert rec.provider_status == "UNKNOWN" assert rec.network_status == "UNKNOWN" assert rec.promotion_eligible is False assert rec.audit_verdict == "BLOCKED" # Orchestrator reconciliation test ws, rep = temp_env orch = ShadowSelfRepairOrchestrator(base_reports_dir=rep, project_root=ws) snapshot = MissionFailureSnapshot(mission_id="m_tel_1", workspace_root=ws, mission_status="FAILED") prov, net, reason = orch.reconcile_telemetry(snapshot, {"provider_called": "UNKNOWN", "network_used": "UNKNOWN"}) assert reason == "UNKNOWN_TELEMETRY" def test_telemetry_scenario_2_monitor_active_zero_traffic_evaluates_false_verified(temp_env): """ Scenario 2: Monitor active covering mandatory scopes with 0 events evaluates to FALSE_VERIFIED. """ rec = AuthenticTelemetryEvaluator.evaluate_telemetry( mission_id="m_tel_2", instance_id="inst_tel_2", workspace_root="/tmp", observer_active=True, monitored_scopes=[SCOPE_URLLIB_HTTP, SCOPE_RAW_SOCKET], detected_attempts=[], required_scopes={SCOPE_URLLIB_HTTP, SCOPE_RAW_SOCKET}, ) assert rec.provider_status == "FALSE_VERIFIED" assert rec.network_status == "FALSE_VERIFIED" assert rec.promotion_eligible is True assert rec.audit_verdict == "PASS" # Orchestrator reconciliation test ws, rep = temp_env orch = ShadowSelfRepairOrchestrator(base_reports_dir=rep, project_root=ws) snapshot = MissionFailureSnapshot(mission_id="m_tel_2", workspace_root=ws, mission_status="FAILED") prov, net, reason = orch.reconcile_telemetry( snapshot, {"provider_called": "FALSE_VERIFIED", "network_used": "FALSE_VERIFIED"}, ) assert prov is False assert net is False assert reason is None def test_telemetry_scenario_3_traffic_actually_detected(temp_env): """ Scenario 3: Outbound traffic detected evaluates to TRUE and blocks promotion. """ rec = AuthenticTelemetryEvaluator.evaluate_telemetry( mission_id="m_tel_3", instance_id="inst_tel_3", workspace_root="/tmp", observer_active=True, monitored_scopes=[SCOPE_URLLIB_HTTP, SCOPE_RAW_SOCKET], detected_attempts=[{"url": "https://api.openai.com", "action": "BLOCKED"}], ) assert rec.provider_status == "TRUE" assert rec.network_status == "TRUE" assert rec.promotion_eligible is False assert rec.audit_verdict == "FAIL" # Orchestrator reconciliation test ws, rep = temp_env orch = ShadowSelfRepairOrchestrator(base_reports_dir=rep, project_root=ws) snapshot = MissionFailureSnapshot(mission_id="m_tel_3", workspace_root=ws, mission_status="FAILED") prov, net, reason = orch.reconcile_telemetry( snapshot, {"provider_called": "TRUE", "network_used": "TRUE"}, ) assert prov is True assert net is True assert reason == "PROVIDER_OR_NETWORK_CALLED" def test_telemetry_scenario_4_contradictory_sources(temp_env): """ Scenario 4: Contradictory telemetry between result and error_context evaluates to CONTRADICTORY_TELEMETRY. """ ws, rep = temp_env orch = ShadowSelfRepairOrchestrator(base_reports_dir=rep, project_root=ws) snapshot = MissionFailureSnapshot( mission_id="m_tel_4", workspace_root=ws, mission_status="FAILED", error_context={"provider_called": "FALSE_VERIFIED", "network_used": "FALSE_VERIFIED"}, ) prov, net, reason = orch.reconcile_telemetry( snapshot, {"provider_called": "TRUE", "network_used": "FALSE_VERIFIED"}, ) assert prov is True assert reason == "CONTRADICTORY_TELEMETRY" def test_telemetry_scenario_5_omitted_telemetry(temp_env): """ Scenario 5: Omitted telemetry evaluates to UNKNOWN_TELEMETRY (fail-closed). """ ws, rep = temp_env orch = ShadowSelfRepairOrchestrator(base_reports_dir=rep, project_root=ws) snapshot = MissionFailureSnapshot(mission_id="m_tel_5", workspace_root=ws, mission_status="FAILED") prov, net, reason = orch.reconcile_telemetry(snapshot, {}) assert reason == "UNKNOWN_TELEMETRY"