# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — OPTIMUS AUTONOMOUS SELF-DIAGNOSTIC (LABORATORY) Allows Optimus to autonomously inspect the readiness and integrity of the self-repair, telemetry, and safe-resume infrastructure without human intervention. Guarantees: 1. Strictly read-only: Never writes to disk, mutates state, or starts services. 2. Safe isolation: Does NOT restart or terminate the Brain process on port 8443. 3. No external calls: Zero network or LLM provider consumption. 4. Exact categorization: Outputs READY, BLOCKED, DEGRADED, UNKNOWN, or NOT_TESTED. 5. Precise root-cause pointing: Reports the exact component, reason, and evidence. """ from __future__ import annotations import hashlib import importlib import logging from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple logger = logging.getLogger("teitan.self_repair.self_diagnostic") def _utc_iso() -> str: return datetime.now(timezone.utc).isoformat() # 12 Protected baselines and expected SHA256 hashes FROZEN_12_BASELINES: Dict[str, Tuple[str, ...]] = { "artifacts/brain_host/capability_mission_planner.py": ( "0B4E6A6E2A7BDD426C559206392C424C1C8981D757602C804CF98354078E6AE5", ), "artifacts/brain_host/generic_developer_executor.py": ( "F8E15BFF32AB2D3F5003CC3392FAF729F242AAFBD904B9B5EF2445FC5B6207CD", ), "artifacts/brain_host/brain_host_tls.py": ( "4E1AFF4F0B33ACC1191A69F1FCC73FE72936252F5C56EBA0BEBCE657D8988FD0", "AC42E6B735FF1BC9440A87FF63BC4B91C267E3577FB8AF444773EF86AFCBD47E", ), "artifacts/brain_host/brain_adapter.py": ( "D18FDC8B8ADCE182409B5FDF28E35C4458467E25C2960D564B5AF0696D8B3D22", "56C3CBD81783B06F9F87CEC2940CD5F65774290AF7573678EA5FDB442E74E762", ), "new_mission_dispatcher.py": ( "7507AA722BB47F96B28093198FB7361CD826D07419B76F86E6C734564EE062F2", "2D082677D812961AF34FD05536DEFA7B96063ACDA6C3EFF394AF7C8D42A5D695", ), "app/remediation/self_repair/shadow_orchestrator.py": ( "CFB51B8851C265036E8ACA9AE38257A7AAC32DAD89AF35847E105B451DEC0FBB", ), "app/remediation/self_repair/promotion_guard.py": ( "7CB2FB7F378C905D8516B4923A8C57EDC73265A46DA16A60E592D40477289BDB", ), "app/remediation/self_repair/promotion_executor.py": ( "362C808F3B1DF8BBFFE0A6131000AD58AE1E96E8797927951855163D82590843", ), "app/remediation/self_repair/engine.py": ( "6589B7781A83BA5C7182D2CDC0E2DC0C7D54530DF9DC2733E1026DC8118EF48D", ), "app/remediation/self_repair/live_shadow_hook.py": ( "A84A8BD7611F9BF77DDD5E1AB8F38077C29066B44C04BE74152A4425F722A67D", ), "app/remediation/self_repair/mission_resume_guard.py": ( "24F65DEC4990ACE863F6D1FCFB00D908FE441C50DFB227B090E94C74E0A1107A", ), "app/remediation/self_repair/mission_resume_executor.py": ( "11864BB7C62FC365D5554BB427EF9FF8616577579A638F996EB3E8E8D3651F8F", ), } @dataclass class DiagnosticCheckResult: """Individual diagnostic item result.""" name: str component: str status: str # "READY", "BLOCKED", "DEGRADED", "UNKNOWN", "NOT_TESTED" details: Dict[str, Any] = field(default_factory=dict) error_message: Optional[str] = None @dataclass class SelfRepairDiagnosticReport: """Consolidated diagnostic report generated for Optimus.""" overall_status: str # "READY", "BLOCKED", "DEGRADED", "UNKNOWN", "NOT_TESTED" project_root: str timestamp: str = field(default_factory=_utc_iso) checks: List[DiagnosticCheckResult] = field(default_factory=list) components_ready: int = 0 components_blocked: int = 0 components_degraded: int = 0 blocking_reasons: List[str] = field(default_factory=list) actionable_diagnostics: List[Dict[str, str]] = field(default_factory=list) mission_specific_status: Optional[str] = None internal_incidents: List[Dict[str, Any]] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: return asdict(self) def summary_text(self) -> str: lines = [ f"=== TEITAN SELF-REPAIR AUTONOMOUS DIAGNOSTIC REPORT ===", f"Overall Status: {self.overall_status}", f"Evaluated At: {self.timestamp}", f"Ready: {self.components_ready} | Blocked: {self.components_blocked} | Degraded: {self.components_degraded}", ] if self.blocking_reasons: lines.append("Blocking Issues:") for b in self.blocking_reasons: lines.append(f" - {b}") if self.actionable_diagnostics: lines.append("Actionable Diagnoses:") for act in self.actionable_diagnostics: lines.append(f" * [{act.get('component')}] {act.get('diagnosis')}") return "\n".join(lines) class OptimusSelfDiagnostic: """ Lightweight, read-only diagnostic evaluator for TEITAN self-repair. """ def __init__(self, project_root: Optional[str] = None): self.project_root = Path(project_root or ".").resolve() def run_diagnostic( self, target_mission_context: Optional[Dict[str, Any]] = None, read_only: bool = True, ) -> SelfRepairDiagnosticReport: """ Executes autonomous diagnostic across all self-repair components. """ checks: List[DiagnosticCheckResult] = [] blocking_reasons: List[str] = [] actionable_diagnostics: List[Dict[str, str]] = [] # 1. Protected Baselines Check base_res = self._check_protected_baselines() checks.append(base_res) if base_res.status == "BLOCKED": blocking_reasons.append("PROTECTED_BASELINES_CORRUPTED") actionable_diagnostics.append({ "component": "BaselineIntegrityGuard", "diagnosis": base_res.error_message or "Protected file hash mismatch", }) # 2. Core Self-Repair Components Importability core_res = self._check_core_components() checks.extend(core_res) for c in core_res: if c.status == "BLOCKED": blocking_reasons.append(f"MISSING_COMPONENT_{c.name}") actionable_diagnostics.append({ "component": c.component, "diagnosis": f"Cannot import {c.name}: {c.error_message}", }) # 3. Live Brain Host Runtime Resolution runtime_res = self._check_brain_runtime() checks.append(runtime_res) if runtime_res.status == "DEGRADED": actionable_diagnostics.append({ "component": "LiveShadowHook", "diagnosis": "Live Brain host process is not mapped in current thread/sys.modules. " "Sandbox self-repair is operational; live resume requires runtime bridge.", }) # 4. Scoped Bridge Registry Check scoped_res = self._check_scoped_bridge_registry() checks.append(scoped_res) # 5. Telemetry & Safety Fence Check telemetry_res = self._check_telemetry_fence() checks.append(telemetry_res) # 6. Internal Functional Contracts Check (Independent Behavioral Assessment) functional_checks, functional_diagnostics, internal_incidents = self._check_internal_functional_contracts() checks.extend(functional_checks) actionable_diagnostics.extend(functional_diagnostics) # 7. Mission-Specific Context Check (if provided) mission_status = None if target_mission_context: m_res = self._check_mission_context(target_mission_context) checks.append(m_res) mission_status = m_res.status if m_res.status == "BLOCKED": blocking_reasons.append(f"MISSION_CONTEXT_REJECTED: {m_res.error_message}") actionable_diagnostics.append({ "component": "PromotionGuard", "diagnosis": m_res.error_message or "Mission context invalid", }) # Compute overall status ready_count = sum(1 for c in checks if c.status == "READY") blocked_count = sum(1 for c in checks if c.status == "BLOCKED") degraded_count = sum(1 for c in checks if c.status == "DEGRADED") if blocked_count > 0: overall = "BLOCKED" elif degraded_count > 0: overall = "DEGRADED" elif ready_count == len(checks): overall = "READY" else: overall = "UNKNOWN" return SelfRepairDiagnosticReport( overall_status=overall, project_root=str(self.project_root), checks=checks, components_ready=ready_count, components_blocked=blocked_count, components_degraded=degraded_count, blocking_reasons=blocking_reasons, actionable_diagnostics=actionable_diagnostics, mission_specific_status=mission_status, internal_incidents=internal_incidents, ) def _check_protected_baselines(self) -> DiagnosticCheckResult: mismatches = [] missing = [] for rel_path, allowed_hashes in FROZEN_12_BASELINES.items(): full_path = self.project_root / rel_path if not full_path.is_file(): missing.append(rel_path) continue try: cur_h = hashlib.sha256(full_path.read_bytes()).hexdigest().upper() if cur_h not in allowed_hashes: mismatches.append(f"{rel_path} (got {cur_h[:8]}...)") except Exception as e: mismatches.append(f"{rel_path} (read error: {e})") if missing or mismatches: err = [] if missing: err.append(f"Missing: {', '.join(missing)}") if mismatches: err.append(f"Mismatched: {', '.join(mismatches)}") return DiagnosticCheckResult( name="ProtectedBaselines", component="BaselineIntegrityGuard", status="BLOCKED", details={"missing": missing, "mismatches": mismatches}, error_message="; ".join(err), ) return DiagnosticCheckResult( name="ProtectedBaselines", component="BaselineIntegrityGuard", status="READY", details={"verified_count": len(FROZEN_12_BASELINES)}, ) def _check_core_components(self) -> List[DiagnosticCheckResult]: components = [ ("ShadowOrchestrator", "app.remediation.self_repair.shadow_orchestrator", "ShadowSelfRepairOrchestrator"), ("SandboxBuilder", "app.remediation.self_repair.sandbox_builder", "SandboxBuilder"), ("PromotionGuard", "app.remediation.self_repair.promotion_guard", "PromotionGuard"), ("PromotionExecutor", "app.remediation.self_repair.promotion_executor", "PromotionExecutor"), ("MissionResumeExecutor", "app.remediation.self_repair.mission_resume_executor", "MissionResumeExecutor"), ("LiveShadowHook", "app.remediation.self_repair.live_shadow_hook", "dispatch_live_shadow_hook"), ] results = [] for name, mod_path, symbol in components: try: mod = importlib.import_module(mod_path) if not hasattr(mod, symbol): results.append(DiagnosticCheckResult( name=name, component=name, status="BLOCKED", error_message=f"Module {mod_path} does not export {symbol}", )) else: results.append(DiagnosticCheckResult( name=name, component=name, status="READY", details={"module": mod_path, "symbol": symbol}, )) except Exception as e: results.append(DiagnosticCheckResult( name=name, component=name, status="BLOCKED", error_message=str(e), )) return results def _check_brain_runtime(self) -> DiagnosticCheckResult: try: from app.remediation.self_repair.live_shadow_hook import resolve_live_brain_runtime res = resolve_live_brain_runtime(project_root=str(self.project_root)) if res.resolved: return DiagnosticCheckResult( name="BrainRuntimeResolution", component="LiveShadowHook", status="READY", details={"live_module_found": True}, ) else: return DiagnosticCheckResult( name="BrainRuntimeResolution", component="LiveShadowHook", status="DEGRADED", details={"reason": res.failure_reason or "OPERATIONAL_TASKS not in memory"}, error_message="Live Brain runtime not attached. Offline sandbox self-repair available.", ) except Exception as e: return DiagnosticCheckResult( name="BrainRuntimeResolution", component="LiveShadowHook", status="DEGRADED", error_message=f"Runtime resolution check failed: {e}", ) def _check_scoped_bridge_registry(self) -> DiagnosticCheckResult: try: from app.remediation.self_repair.scoped_bridge_registry import ( ScopedBridgeRegistry, get_scoped_bridge_registry, ) reg = get_scoped_bridge_registry() return DiagnosticCheckResult( name="ScopedBridgeRegistry", component="ScopedBridgeRegistry", status="READY", details={"active_missions": reg.active_mission_count()}, ) except Exception as e: return DiagnosticCheckResult( name="ScopedBridgeRegistry", component="ScopedBridgeRegistry", status="BLOCKED", error_message=f"Scoped registry unavailable: {e}", ) def _check_telemetry_fence(self) -> DiagnosticCheckResult: try: from app.remediation.self_repair.telemetry_contract import AuthenticTelemetryEvaluator from app.remediation.self_repair.provider_safety_fence import ProviderSafetyFence return DiagnosticCheckResult( name="TelemetrySafetyFence", component="ProviderSafetyFence", status="READY", details={"evaluator_ready": True, "fence_ready": True}, ) except Exception as e: return DiagnosticCheckResult( name="TelemetrySafetyFence", component="ProviderSafetyFence", status="BLOCKED", error_message=f"Telemetry evaluator or fence error: {e}", ) def _check_mission_context(self, ctx: Dict[str, Any]) -> DiagnosticCheckResult: from app.remediation.self_repair.live_shadow_hook import evaluate_c12_scoped_authorization creator_req = str(ctx.get("creator_request") or "") ws = str(ctx.get("workspace_root") or "") auth, details = evaluate_c12_scoped_authorization(creator_req, ws, terminal_rec=ctx) if not details.get("c12_token_found", False): return DiagnosticCheckResult( name="MissionAuthorization", component="PromotionGuard", status="DEGRADED", details=details, error_message="No C12 authorization token present in creator request. Auto-promotion disabled.", ) if not auth: return DiagnosticCheckResult( name="MissionAuthorization", component="PromotionGuard", status="BLOCKED", details=details, error_message=f"C12 authorization check failed: {details.get('rejection_reason', 'UNAUTHORIZED')}", ) return DiagnosticCheckResult( name="MissionAuthorization", component="PromotionGuard", status="READY", details=details, ) def _check_internal_functional_contracts(self) -> Tuple[List[DiagnosticCheckResult], List[Dict[str, str]], List[Dict[str, Any]]]: """ Executes registered functional contracts on internal components in an isolated subprocess with time and memory boundaries. Operates independently of integrity status. Converts authentic FAIL results into internal incidents via FunctionalDiagnosticAdapter. """ checks: List[DiagnosticCheckResult] = [] actionable: List[Dict[str, str]] = [] internal_incidents: List[Dict[str, Any]] = [] try: from app.remediation.self_repair.functional_assessor import ( FunctionalAssessmentStatus, InternalFunctionalAssessor, ) from app.remediation.self_repair.functional_diagnostic_adapter import ( FunctionalDiagnosticAdapter, ) assessor = InternalFunctionalAssessor( workspace_root=str(self.project_root), python_executable=None, ) eval_results = assessor.assess_all_components() for res in eval_results: diag_status = "READY" err_msg = None if res.status == FunctionalAssessmentStatus.PASS: diag_status = "READY" elif res.status == FunctionalAssessmentStatus.FAIL: diag_status = "DEGRADED" err_msg = f"Behavioral assertion failure: {res.assertion_error}" actionable.append({ "component": res.component_name, "diagnosis": f"Functional defect in {res.target_file_rel}: {res.assertion_error} (SHA256: {res.file_sha256[:8] if res.file_sha256 else 'N/A'})", }) # Connection B: Safely validate and convert authentic FAIL result valid, reason, result_tuple = FunctionalDiagnosticAdapter.validate_and_convert_failure( res, workspace_root=str(self.project_root) ) if valid and result_tuple: ev_event, ev_incident, ev_bundle = result_tuple internal_incidents.append({ "incident_id": ev_incident.incident_id, "component_name": res.component_name, "contract_id": res.contract_id, "target_file_rel": res.target_file_rel, "file_sha256": res.file_sha256, "incident": ev_incident, "evidence": ev_bundle, "event": ev_event, }) else: actionable.append({ "component": res.component_name, "diagnosis": f"Internal incident creation rejected: {reason}", }) elif res.status == FunctionalAssessmentStatus.BLOCKED: diag_status = "BLOCKED" err_msg = res.assertion_error or res.inconclusive_reason or "Functional assessment blocked" actionable.append({ "component": res.component_name, "diagnosis": f"Assessment blocked for {res.target_file_rel}: {err_msg}", }) else: # UNPROVEN diag_status = "UNKNOWN" err_msg = res.inconclusive_reason or "Functional contract execution inconclusive" checks.append(DiagnosticCheckResult( name=f"FunctionalContract_{res.contract_id}", component=res.component_name, status=diag_status, details={ "contract_id": res.contract_id, "target_file_rel": res.target_file_rel, "file_sha256": res.file_sha256, "duration_ms": res.duration_ms, "raw_status": res.status, "assertion_error": res.assertion_error, }, error_message=err_msg, )) except Exception as e: checks.append(DiagnosticCheckResult( name="InternalFunctionalContracts", component="InternalFunctionalAssessor", status="BLOCKED", error_message=f"Failed to execute functional assessor: {e}", )) return checks, actionable, internal_incidents def run_self_repair_diagnostic( project_root: Optional[str] = None, target_mission_context: Optional[Dict[str, Any]] = None, read_only: bool = True, ) -> SelfRepairDiagnosticReport: """Convenience functional interface for Optimus.""" diag = OptimusSelfDiagnostic(project_root=project_root) return diag.run_diagnostic( target_mission_context=target_mission_context, read_only=read_only, )