# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR - FUNCTIONAL DIAGNOSTIC ADAPTER (AR88-JOINT-FIX-01) Converts an InternalFunctionalAssessor evaluation failure into a canonical IncidentRecord and EvidenceBundle for root-cause diagnosis. SAFETY AND STATE CONTRACT: - FAIL: ONLY an authentically executed and failed functional evaluation with genuine assertion error / failure evidence may produce an actionable incident. - BLOCKED: Represents an unexecuted evaluation or an isolation/security boundary. NEVER produces an actionable incident or FAILED test event (returns None). - UNPROVEN: Inconclusive evaluation (e.g. timeout, execution error). NEVER produces an actionable incident (returns None). - PASS: Successful contract evaluation. NEVER produces an actionable incident (returns None). """ from __future__ import annotations import hashlib import os from typing import Any, Dict, List, Optional, Tuple from app.remediation.self_repair.functional_assessor import ( FunctionalAssessmentStatus, FunctionalEvaluationResult, ) from app.remediation.self_repair.schemas import ( EvidenceBundle, IncidentFailureType, IncidentRecord, MissionFailureEvent, _new_id, _utc_iso, ) # Protected system paths and baselines that must never be targeted by internal incidents PROTECTED_TARGET_PATHS = [ "artifacts/brain_host/brain_host_tls.py", "artifacts/brain_host/brain_adapter.py", "new_mission_dispatcher.py", "artifacts/brain_host/capability_mission_planner.py", "artifacts/brain_host/generic_developer_executor.py", "app/remediation/self_repair/schemas.py", "app/remediation/self_repair/engine.py", "app/remediation/self_repair/live_shadow_hook.py", "DIGITAL_OPTIMUS_NORTE_MAESTRO.md", ] def _compute_sha256(path: str) -> Optional[str]: try: if not os.path.isfile(path): return None with open(path, "rb") as f: return hashlib.sha256(f.read()).hexdigest().upper() except Exception: return None def _is_canonically_within(parent_dir: str, target_path: str) -> bool: try: p_real = os.path.realpath(os.path.abspath(parent_dir)) t_real = os.path.realpath(os.path.abspath(target_path)) return os.path.commonpath([p_real, t_real]) == p_real except Exception: return False def _is_protected_target(rel_path: str) -> bool: norm = os.path.normpath(rel_path).replace("\\", "/").lower() for prot in PROTECTED_TARGET_PATHS: prot_norm = os.path.normpath(prot).replace("\\", "/").lower() if norm == prot_norm or norm.endswith(prot_norm) or prot_norm.endswith(norm): return True return False class FunctionalDiagnosticAdapter: """ Transforms internal functional evaluation failures into canonical IncidentRecord and EvidenceBundle for root-cause diagnosis. Enforces strict state rejection: BLOCKED, UNPROVEN, and PASS return None. """ @classmethod def validate_and_convert_failure( cls, eval_result: FunctionalEvaluationResult, workspace_root: str, ) -> Tuple[bool, Optional[str], Optional[Tuple[MissionFailureEvent, IncidentRecord, EvidenceBundle]]]: """ Validates the entry conditions and safely converts a functional evaluation result. Returns: (is_admissible, rejection_reason, result_tuple) """ # Condition 1 & 2: Status check (FAIL only, no BLOCKED/UNPROVEN/PASS) raw_status = str(eval_result.status or "").strip().upper() if raw_status != FunctionalAssessmentStatus.FAIL: return False, f"REJECTED_STATUS: Evaluation status is '{raw_status}'. Only authentic FAIL can generate an actionable incident.", None # Condition 3: Evidence of real execution (must have assertion_error or stack_trace) if not eval_result.assertion_error and not eval_result.stack_trace: return False, "REJECTED_INSUFFICIENT_EVIDENCE: FAIL result contains no assertion error or traceback evidence.", None # Condition 4: Target file must be specified target_rel = str(eval_result.target_file_rel or "").strip() if not target_rel: return False, "REJECTED_MISSING_TARGET: Evaluation result contains no target_file_rel.", None ws_abs = os.path.realpath(os.path.abspath(workspace_root)) target_abs = os.path.realpath(os.path.abspath(os.path.join(ws_abs, target_rel))) # Condition 5: Boundary check (strictly within workspace_root, no escape) if not _is_canonically_within(ws_abs, target_abs): return False, f"REJECTED_BOUNDARY_ESCAPE: Target file '{target_rel}' resolves outside workspace boundary.", None # Condition 6: Protected target check (cannot target engine, schemas, baselines, etc.) if _is_protected_target(target_rel): return False, f"REJECTED_PROTECTED_TARGET: Target file '{target_rel}' is a protected baseline or critical system path.", None # Condition 7: Physical existence on disk if not os.path.isfile(target_abs): return False, f"REJECTED_FILE_NOT_FOUND: Target file '{target_rel}' does not exist on disk at '{target_abs}'.", None # Condition 8: Hash verification current_sha256 = _compute_sha256(target_abs) if not current_sha256: return False, f"REJECTED_HASH_FAILURE: Unable to compute physical SHA256 of target file '{target_rel}'.", None if eval_result.file_sha256 and eval_result.file_sha256.upper() != current_sha256.upper(): return False, f"REJECTED_HASH_MISMATCH: Recorded hash {eval_result.file_sha256} does not match current physical hash {current_sha256}.", None # Construct canonical internal incident and evidence diag_id = _new_id("diag_inc") now_ts = _utc_iso() symptom = ( f"Internal functional contract failure: {eval_result.contract_id} " f"on component {eval_result.component_name}. Error: {eval_result.assertion_error}" ) err_ctx = { "component_id": eval_result.component_name, "contract_id": eval_result.contract_id, "target_file": target_rel, "file_sha256": current_sha256, "assertion_error": eval_result.assertion_error, "duration_ms": eval_result.duration_ms, "incident_category": "INTERNAL_FUNCTIONAL_CONTRACT_FAIL", } exec_history = [ { "stage": "functional_assessment", "component": eval_result.component_name, "target": target_rel, "status": "FAIL", "error": eval_result.assertion_error, "duration_ms": eval_result.duration_ms, } ] tracebacks = [eval_result.stack_trace] if eval_result.stack_trace else [] logs = [f"Functional contract execution output:\n{eval_result.stdout}"] if eval_result.stdout else [] event = MissionFailureEvent( mission_id=f"m_internal_eval_{eval_result.contract_id}", instance_id="inst_internal_assessor", task_id=f"task_eval_{eval_result.contract_id}", timestamp=now_ts, workspace_root=ws_abs, failure_stage=eval_result.component_name, observed_symptom=symptom, exception_type="AssertionError", exception_message=eval_result.assertion_error or symptom, source_component=eval_result.component_name, execution_history=exec_history, verification_result={"target": target_rel, "status": "FAIL", "error": eval_result.assertion_error}, evidence_locations={"target_file": target_rel}, mission_status="FAILED", error_context=err_ctx, target_file=target_rel, target_files=[target_rel], source_files=[target_rel], ) incident = IncidentRecord( incident_id=diag_id, mission_id=event.mission_id, failure_type=IncidentFailureType.FAILED_TEST, observed_symptom=symptom, source_component=eval_result.component_name, timestamp=now_ts, initial_evidence=err_ctx, ) source_excerpts = [] try: with open(target_abs, "r", encoding="utf-8", errors="replace") as f: lines = f.readlines() source_excerpts.append({ "file_path": target_rel, "line_count": len(lines), "head_excerpt": "".join(lines[:30]), "tail_excerpt": "".join(lines[-30:]), "sha256": current_sha256, }) except Exception: pass evidence = EvidenceBundle( incident_id=diag_id, logs=logs, tracebacks=tracebacks, action_history=exec_history, target_file=target_rel, source_files=[target_rel], source_excerpts=source_excerpts, relevant_config={"assessor_version": "ar88_hardened_1.0"}, ) return True, None, (event, incident, evidence) @classmethod def failure_result_to_incident_and_evidence( cls, eval_result: FunctionalEvaluationResult, workspace_root: str, ) -> Optional[Tuple[MissionFailureEvent, IncidentRecord, EvidenceBundle]]: """ Public conversion interface preserving backward-compatible signature. Rejects non-FAIL status safely without relying solely on assert. Returns None if result is BLOCKED, UNPROVEN, PASS, or invalid. """ valid, reason, result_tuple = cls.validate_and_convert_failure(eval_result, workspace_root) if not valid: return None return result_tuple