# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR — INTERNAL AUTONOMOUS SELF-REPAIR COORDINATOR (AR88-04H) Orchestrates the internal autonomous recovery loop for TEITAN's own components. DEFECT CORRECTIONS AND SECURITY HARDENING (AR88-04H): 1. PUERTA OBLIGATORIA DE AISLAMIENTO (FAIL-CLOSED): Verifies that execution provenance is accredited by physical OS containment (all_barriers_satisfied=True or explicit execution provenance environment). If unaccredited, dynamic execution is strictly BLOCKED (BLOCKED_UNACCREDITED_ISOLATION). 2. PRESERVACION DEL INCIDENTE ORIGINAL: Reuses IncidentRecord and EvidenceBundle directly without reconstructing a second incident or overwriting the original incident_id. Uses engine subcomponents directly (tracer, analyzer, planner) without passing through observer.observe() or mission_data. 3. DEDUPLICACION Y CHECKPOINTS PERSISTENTES: Maintains a persistent JSON checkpoint store on local disk with atomic writes. Distinguishes states: OBSERVED, IN_PROGRESS, FAILED, REPAIRED (only upon ACCEPT verdict). Survives coordinator restarts and enforces attempt limits. 4. SEPARACION DE DIAGNOSTICO DE EJECUCION AUTORIZADA: read_only=True generates diagnostic and proposal artifacts without modifying candidates. Dynamic repair execution requires authorized entry with accredited isolation. 5. AUTONOMIA VERDADERA: Explicit patch instructions are restricted to explicit simulation / test overrides. If autonomous reasoning fails to synthesize a high-confidence patch, returns PATCH_PLAN_UNAVAILABLE without fabricating candidate code. 6. VEREDICTOS RIGUROSOS: Distinguishes CANDIDATE_ACCEPTED, CANDIDATE_REJECTED, ROOT_CAUSE_UNPROVEN, BLOCKED_UNACCREDITED_ISOLATION, and PATCH_PLAN_UNAVAILABLE. """ from __future__ import annotations import hashlib import json import logging import os import tempfile from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Set, Tuple from app.remediation.self_repair.engine import SelfRepairEngine 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.schemas import ( CandidateVerdict, CircuitTrace, ComparisonResult, EvidenceBundle, IncidentRecord, PatchInstruction, PatchOperationType, PatchPlan, PatchResult, PromotionProposal, RepairProposal, RootCauseClassification, SandboxManifest, VerificationResult, _new_id, _utc_iso, ) logger = logging.getLogger("teitan.self_repair.internal_coordinator") class InternalRepairStatus: CANDIDATE_ACCEPTED = "CANDIDATE_ACCEPTED" CANDIDATE_REJECTED = "CANDIDATE_REJECTED" ROOT_CAUSE_UNPROVEN = "ROOT_CAUSE_UNPROVEN" BLOCKED_UNACCREDITED_ISOLATION = "BLOCKED_UNACCREDITED_ISOLATION" PATCH_PLAN_UNAVAILABLE = "PATCH_PLAN_UNAVAILABLE" READ_ONLY_DIAGNOSTIC_COMPLETE = "READ_ONLY_DIAGNOSTIC_COMPLETE" REJECTED_DUPLICATE_INCIDENT = "REJECTED_DUPLICATE_INCIDENT" REJECTED_STATUS = "REJECTED_STATUS" REJECTED_PROVENANCE_OR_INTEGRITY = "REJECTED_PROVENANCE_OR_INTEGRITY" CONTRACT_MISMATCH = "CONTRACT_MISMATCH" class PersistentIncidentStore: """ Persistently tracks incident lifecycle states in local workspace storage. Enforces atomic writes and tracks states: OBSERVED, IN_PROGRESS, FAILED, REPAIRED. """ MAX_REPAIR_ATTEMPTS = 3 def __init__(self, store_path: str): self.store_path = os.path.abspath(store_path) self.store_dir = os.path.dirname(self.store_path) os.makedirs(self.store_dir, exist_ok=True) self._cache: Dict[str, Dict[str, Any]] = {} self._load() def _load(self) -> None: if os.path.isfile(self.store_path): try: with open(self.store_path, "r", encoding="utf-8") as f: self._cache = json.load(f) except Exception as e: logger.warning(f"Error loading persistent incident store: {e}") self._cache = {} else: self._cache = {} def _save_atomic(self) -> None: temp_fd, temp_path = tempfile.mkstemp(dir=self.store_dir, prefix="inc_store_", suffix=".tmp") try: with open(temp_fd, "w", encoding="utf-8") as f: json.dump(self._cache, f, indent=2) if os.path.exists(self.store_path): os.replace(temp_path, self.store_path) else: os.rename(temp_path, self.store_path) except Exception as e: if os.path.exists(temp_path): os.remove(temp_path) raise e def get_record(self, signature: str) -> Optional[Dict[str, Any]]: self._load() return self._cache.get(signature) def record_observed(self, signature: str, incident_id: str, contract_id: str) -> None: rec = self._cache.get(signature, { "signature": signature, "incident_id": incident_id, "contract_id": contract_id, "state": "OBSERVED", "attempts": 0, "first_observed": _utc_iso(), "last_updated": _utc_iso(), }) rec["state"] = "OBSERVED" rec["last_updated"] = _utc_iso() self._cache[signature] = rec self._save_atomic() def record_in_progress(self, signature: str) -> bool: """Returns False if max attempts exceeded, else True.""" rec = self._cache.get(signature, {}) attempts = rec.get("attempts", 0) + 1 if attempts > self.MAX_REPAIR_ATTEMPTS: rec["state"] = "MAX_ATTEMPTS_EXCEEDED" rec["last_updated"] = _utc_iso() self._cache[signature] = rec self._save_atomic() return False rec["attempts"] = attempts rec["state"] = "IN_PROGRESS" rec["last_updated"] = _utc_iso() self._cache[signature] = rec self._save_atomic() return True def record_failed(self, signature: str, reason: str) -> None: rec = self._cache.get(signature, {}) rec["state"] = "FAILED" rec["failure_reason"] = reason rec["last_updated"] = _utc_iso() self._cache[signature] = rec self._save_atomic() def record_repaired(self, signature: str, candidate_id: str) -> None: rec = self._cache.get(signature, {}) rec["state"] = "REPAIRED" rec["candidate_id"] = candidate_id rec["repaired_at"] = _utc_iso() rec["last_updated"] = _utc_iso() self._cache[signature] = rec self._save_atomic() class InternalRepairCoordinator: """ Dedicated coordinator for TEITAN internal component self-repair. """ def __init__( self, project_root: str, engine: Optional[SelfRepairEngine] = None, base_reports_dir: str = "reports/self_repair", store_path: Optional[str] = None, ): self.project_root = os.path.abspath(project_root) self.base_reports_dir = base_reports_dir self.engine = engine or SelfRepairEngine( base_reports_dir=self.base_reports_dir, project_root=self.project_root, ) resolved_store = store_path or os.path.join(self.project_root, "reports", "self_repair", "incident_store.json") self.store = PersistentIncidentStore(resolved_store) def compute_evaluation_signature(self, eval_result: FunctionalEvaluationResult) -> str: """Computes an immutable content signature to detect duplicate or replayed results.""" raw = f"{eval_result.contract_id}:{eval_result.component_name}:{eval_result.target_file_rel}:{eval_result.file_sha256}:{eval_result.status}:{eval_result.assertion_error}" return hashlib.sha256(raw.encode("utf-8")).hexdigest() def audit_execution_isolation( self, eval_result: FunctionalEvaluationResult, ) -> Tuple[bool, str]: """ PUERTA OBLIGATORIA DE AISLAMIENTO (FAIL-CLOSED) Audits whether physical OS containment barriers are satisfied. A manually constructed FunctionalEvaluationResult without physical isolation accreditation cannot authorize dynamic repairs. Caller cannot inject fake assessor or bypass check. """ ass = InternalFunctionalAssessor(workspace_root=self.project_root) barrier_audit = ass.audit_os_isolation_barriers() if not barrier_audit.get("all_barriers_satisfied", False): unmet = [k for k, v in barrier_audit.items() if k != "all_barriers_satisfied" and not v] return False, f"Unaccredited OS isolation in execution environment. Unmet barriers: {', '.join(unmet)}." # Verify execution context in result if provided exec_ctx = getattr(eval_result, "execution_context", None) if isinstance(exec_ctx, dict): if not exec_ctx.get("sandbox_accredited", False): return False, "Execution context lacks accredited sandbox provenance." return True, "OS isolation accredited." def execute_internal_diagnosis( self, incident: IncidentRecord, evidence: EvidenceBundle, persist_artifacts: bool = True, ) -> Tuple[CircuitTrace, RootCauseClassification, RepairProposal]: """ DEFECT B: PRESERVACION DEL INCIDENTE ORIGINAL Executes diagnosis directly using existing subcomponents (tracer, analyzer, planner) preserving the exact incident_id, evidence, contract_id, and provenance. """ # 1. TRACE CIRCUIT trace = self.engine.tracer.trace(incident=incident, evidence=evidence) # 2. ANALYZE ROOT CAUSE root_cause = self.engine.analyzer.analyze( incident=incident, evidence=evidence, trace=trace, workspace_root=self.project_root, ) # 3. PLAN REPAIR (Sandbox only) proposal = self.engine.planner.plan_repair( incident=incident, root_cause_info=root_cause, workspace_root=self.project_root, ) # 4. PERSIST ARTIFACTS if persist_artifacts: out_dir = os.path.join(self.base_reports_dir, incident.incident_id) os.makedirs(out_dir, exist_ok=True) self.engine.tracer.persist(trace, out_dir) self.engine.analyzer.persist(root_cause, out_dir) self.engine.planner.persist(proposal, out_dir) return trace, root_cause, proposal def coordinate_internal_evaluation( self, eval_result: FunctionalEvaluationResult, read_only: bool = False, explicit_patch_instruction: Optional[Dict[str, Any]] = None, focused_test_runner: Optional[Any] = None, regression_test_runner: Optional[Any] = None, persist_artifacts: bool = True, ) -> Dict[str, Any]: """ Coordinates the complete internal recovery flow: Validation -> Persistent De-duplication -> Incident Delivery -> Direct Diagnosis -> Isolation Gate -> Patch Plan -> Sandbox Build -> Candidate Patch -> Verify -> Compare -> Judge. """ result: Dict[str, Any] = { "coordinator": "InternalRepairCoordinator", "contract_id": eval_result.contract_id, "component_name": eval_result.component_name, "status": "INITIATED", "internal_incident_admitted": False, "trace_complete": False, "root_cause_proven": False, "sandbox_repair_executed": False, "candidate_verdict": None, "promotion_proposed": False, "production_write_authority": False, "auto_promotion": False, "error": None, "is_autonomous_patch": False, } # 1. State Filter (FAIL ONLY: Reject BLOCKED, PASS, UNPROVEN) raw_status = str(eval_result.status or "").strip().upper() if raw_status != FunctionalAssessmentStatus.FAIL: result.update({ "status": InternalRepairStatus.REJECTED_STATUS, "error": f"Evaluation status '{raw_status}' cannot trigger internal repair.", }) return result # 2. Persistent De-duplication & State Check sig = self.compute_evaluation_signature(eval_result) existing_record = self.store.get_record(sig) if existing_record: st = existing_record.get("state") if st == "REPAIRED": result.update({ "status": InternalRepairStatus.REJECTED_DUPLICATE_INCIDENT, "error": f"Incident signature '{sig[:12]}...' is already verified as REPAIRED.", }) return result elif st == "MAX_ATTEMPTS_EXCEEDED": result.update({ "status": InternalRepairStatus.REJECTED_DUPLICATE_INCIDENT, "error": f"Incident signature '{sig[:12]}...' has exceeded maximum repair attempts ({self.store.MAX_REPAIR_ATTEMPTS}).", }) return result # 3. Provenance & Schema Validation via FunctionalDiagnosticAdapter valid, rejection_reason, result_tuple = FunctionalDiagnosticAdapter.validate_and_convert_failure( eval_result, workspace_root=self.project_root, ) if not valid or not result_tuple: result.update({ "status": InternalRepairStatus.REJECTED_PROVENANCE_OR_INTEGRITY, "error": rejection_reason or "Unknown provenance validation failure.", }) return result event, incident, evidence = result_tuple # Strict identity check: mission_id must be None (internal identity separation) if incident.mission_id is not None: result.update({ "status": InternalRepairStatus.CONTRACT_MISMATCH, "error": "IncidentRecord contains an unauthorized client mission_id.", }) return result # Register in persistent store as OBSERVED self.store.record_observed(sig, incident.incident_id, eval_result.contract_id) result["internal_incident_admitted"] = True result["incident_id"] = incident.incident_id # 4. Direct Internal Diagnosis (Preserving original incident and evidence) trace, root_cause, proposal = self.execute_internal_diagnosis( incident=incident, evidence=evidence, persist_artifacts=persist_artifacts, ) result["trace_complete"] = trace.trace_complete if trace else False result["root_cause_proven"] = root_cause.proven result["responsible_function"] = root_cause.responsible_function # 5. Root Cause Confidence Gate if not root_cause.proven or root_cause.root_cause == "UNPROVEN": result.update({ "status": InternalRepairStatus.ROOT_CAUSE_UNPROVEN, "error": "Root cause analysis failed to prove a definitive defect. Patching blocked.", }) self.store.record_failed(sig, "ROOT_CAUSE_UNPROVEN") return result # 6. Read-Only Gate (OptimusSelfDiagnostic operates strictly here) if read_only: result.update({ "status": InternalRepairStatus.READ_ONLY_DIAGNOSTIC_COMPLETE, "proposal_generated": True, }) return result # 7. PUERTA OBLIGATORIA DE AISLAMIENTO (FAIL-CLOSED) # Dynamic execution strictly prohibited unless physical OS isolation is accredited isolation_ok, iso_reason = self.audit_execution_isolation(eval_result) if not isolation_ok: result.update({ "status": InternalRepairStatus.BLOCKED_UNACCREDITED_ISOLATION, "error": f"Dynamic repair blocked: {iso_reason}", }) self.store.record_failed(sig, "BLOCKED_UNACCREDITED_ISOLATION") return result # Advance attempt in persistent store if not self.store.record_in_progress(sig): result.update({ "status": InternalRepairStatus.REJECTED_DUPLICATE_INCIDENT, "error": f"Incident signature '{sig[:12]}...' exceeded maximum repair attempts.", }) return result # 8. Patch Plan Acquisition (Autonomy vs Explicit Instruction) patch_plan = None if explicit_patch_instruction: # Explicit simulation instruction (for controlled unit testing only) result["is_autonomous_patch"] = False op_raw = str(explicit_patch_instruction.get("operation") or "REPLACE_EXACT").strip().upper() if op_raw in ("SAFE_REPLACE", "REPLACE_EXACT"): op = PatchOperationType.REPLACE_EXACT elif op_raw in ("INSERT_AFTER",): op = PatchOperationType.INSERT_AFTER elif op_raw in ("INSERT_BEFORE",): op = PatchOperationType.INSERT_BEFORE else: op = PatchOperationType.REPLACE_EXACT before_str = str(explicit_patch_instruction.get("before") or explicit_patch_instruction.get("search_block") or "") after_str = str(explicit_patch_instruction.get("after") or explicit_patch_instruction.get("replace_block") or "") inst = PatchInstruction( file_rel_path=str(explicit_patch_instruction.get("file_rel_path") or eval_result.target_file_rel), operation=op, before=before_str, after=after_str, reason=str(explicit_patch_instruction.get("reason") or "Manual test patch"), ) patch_plan = PatchPlan( incident_id=incident.incident_id, repair_id=proposal.repair_id, instructions=[inst], ) else: # Autonomous Reasoning via TEITAN engine reasoner reasoning_res = self.engine.patch_reasoner.reason_patch( event=event, incident=incident, evidence=evidence, root_cause_info=root_cause, proposal=proposal, workspace_root=self.project_root, ) if reasoning_res.success and reasoning_res.confidence == "HIGH" and reasoning_res.candidate_instructions: plan_res = self.engine.patch_plan_generator.generate_plan( event=event, root_cause_info=root_cause, proposal=proposal, workspace_root=self.project_root, explicit_patch_instruction=reasoning_res.candidate_instructions[0].to_dict(), ) if plan_res.success and plan_res.plan: patch_plan = plan_res.plan result["is_autonomous_patch"] = True if not patch_plan: result.update({ "status": InternalRepairStatus.PATCH_PLAN_UNAVAILABLE, "error": "Autonomous patch reasoner could not synthesize a high-confidence plan, and no manual instruction provided.", }) self.store.record_failed(sig, "PATCH_PLAN_UNAVAILABLE") return result # 9. Sandbox Repair & Independent Verification Execution manifest, patch_res, verif, comp, verdict, promotion = self.engine.execute_sandbox_repair( incident=incident, evidence=evidence, root_cause=root_cause, proposal=proposal, patch_plan=patch_plan, focused_test_runner=focused_test_runner, regression_test_runner=regression_test_runner, persist_artifacts=persist_artifacts, workspace_root=self.project_root, ) verdict_status = verdict.verdict if verdict else "REJECT" if verdict_status == "ACCEPT": final_status = InternalRepairStatus.CANDIDATE_ACCEPTED self.store.record_repaired(sig, manifest.candidate_id) else: final_status = InternalRepairStatus.CANDIDATE_REJECTED reasons_str = "; ".join(verdict.reasons) if (verdict and verdict.reasons) else "Verification criteria unmet" self.store.record_failed(sig, f"Candidate rejected by judge: {reasons_str}") result.update({ "status": final_status, "sandbox_repair_executed": True, "patch_status": patch_res.status, "candidate_verdict": verdict_status, "promotion_proposed": promotion is not None, "candidate_dir": manifest.candidate_dir, "candidate_id": manifest.candidate_id, "production_write_authority": False, "auto_promotion": False, }) return result