# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — SELF REPAIR ENGINE End-to-end coordinator for observation, evidence collection, circuit tracing, root cause diagnosis, and repair proposal generation. """ from __future__ import annotations import os from typing import Any, Dict, List, Optional, Tuple, Union from app.remediation.self_repair.analyzer import RootCauseAnalyzer from app.remediation.self_repair.autonomous_patch_reasoner import AutonomousPatchReasoner from app.remediation.self_repair.candidate_judge import CandidateJudge from app.remediation.self_repair.candidate_verifier import CandidateVerifier from app.remediation.self_repair.circuit import CircuitTracer from app.remediation.self_repair.collector import EvidenceCollector from app.remediation.self_repair.evidence_comparator import EvidenceComparator from app.remediation.self_repair.failure_ingestor import FailureTriggerGuard, RealFailureIngestor from app.remediation.self_repair.observer import IncidentObserver from app.remediation.self_repair.patch_executor import PatchExecutor from app.remediation.self_repair.patch_plan_generator import PatchPlanGenerator, PatchPlanGeneratorResult from app.remediation.self_repair.planner import RepairPlanner from app.remediation.self_repair.sandbox_builder import SandboxBuilder from app.remediation.self_repair.schemas import ( CandidateVerdict, CircuitTrace, ComparisonResult, EvidenceBundle, IncidentFailureType, IncidentRecord, MissionFailureEvent, PatchPlan, PatchResult, PromotionProposal, RepairProposal, RootCauseClassification, SandboxManifest, VerificationResult, ) class SelfRepairEngine: """ Coordinator executing the complete MIRAR -> ENTENDER -> TRAZAR -> AISLAR CAUSA -> PROPONER -> SANDBOX cycle. Operates strictly read-only on production without LLM provider calls. """ PRODUCTION_WRITE_AUTHORITY: bool = False CORE_WRITE_AUTHORITY: bool = False AUTO_PROMOTION: bool = False PROVIDER_REQUIRED: bool = False def __init__( self, base_reports_dir: str = "reports/self_repair", project_root: str = ".", patch_reasoner: Optional[AutonomousPatchReasoner] = None, brain_bridge: Optional[Any] = None, ): self.base_reports_dir = base_reports_dir self.project_root = os.path.abspath(project_root) self.observer = IncidentObserver() self.collector = EvidenceCollector() self.tracer = CircuitTracer() self.analyzer = RootCauseAnalyzer() self.planner = RepairPlanner() self.sandbox_builder = SandboxBuilder(project_root=self.project_root) self.patch_executor = PatchExecutor() self.candidate_verifier = CandidateVerifier() self.evidence_comparator = EvidenceComparator() self.candidate_judge = CandidateJudge() self.failure_ingestor = RealFailureIngestor(project_root=self.project_root) self.patch_plan_generator = PatchPlanGenerator() if patch_reasoner is not None: self.patch_reasoner = patch_reasoner elif brain_bridge is not None: self.patch_reasoner = AutonomousPatchReasoner(project_root=self.project_root, brain_bridge=brain_bridge) else: self.patch_reasoner = AutonomousPatchReasoner(project_root=self.project_root) def process_incident( self, failure_type: Union[IncidentFailureType, str], observed_symptom: str, source_component: str, initial_evidence: Optional[Dict[str, Any]] = None, mission_id: Optional[str] = None, instance_id: Optional[str] = None, logs: Optional[List[str]] = None, tracebacks: Optional[List[str]] = None, mission_state: Optional[Dict[str, Any]] = None, action_history: Optional[List[Dict[str, Any]]] = None, verification_results: Optional[Dict[str, Any]] = None, relevant_config: Optional[Dict[str, Any]] = None, source_files: Optional[List[str]] = None, workspace_root: Optional[str] = None, node_status_overrides: Optional[Dict[str, Dict[str, Any]]] = None, persist_artifacts: bool = True, ) -> Tuple[IncidentRecord, EvidenceBundle, CircuitTrace, RootCauseClassification, RepairProposal]: """ Executes the Phase 1 diagnostic and proposal flow. """ # 1. OBSERVE incident = self.observer.observe( failure_type=failure_type, observed_symptom=observed_symptom, source_component=source_component, initial_evidence=initial_evidence, mission_id=mission_id, instance_id=instance_id, ) # 2. COLLECT EVIDENCE (Read-Only) evidence = self.collector.collect( incident=incident, logs=logs, tracebacks=tracebacks, mission_state=mission_state, action_history=action_history, verification_results=verification_results, relevant_config=relevant_config, source_files=source_files, workspace_root=workspace_root or self.project_root, ) # 3. TRACE CIRCUIT trace = self.tracer.trace( incident=incident, evidence=evidence, node_status_overrides=node_status_overrides, ) # 4. ANALYZE ROOT CAUSE root_cause = self.analyzer.analyze( incident=incident, evidence=evidence, trace=trace, workspace_root=workspace_root or self.project_root, ) # 5. PLAN REPAIR (Sandbox only) proposal = self.planner.plan_repair( incident=incident, root_cause_info=root_cause, ) # 6. PERSIST ARTIFACTS if persist_artifacts: out_dir = os.path.join(self.base_reports_dir, incident.incident_id) self.observer.persist(incident, out_dir) self.collector.persist(evidence, out_dir) self.tracer.persist(trace, out_dir) self.analyzer.persist(root_cause, out_dir) self.planner.persist(proposal, out_dir) return incident, evidence, trace, root_cause, proposal def execute_sandbox_repair( self, incident: IncidentRecord, evidence: EvidenceBundle, root_cause: RootCauseClassification, proposal: RepairProposal, patch_plan: PatchPlan, focused_test_runner: Optional[Any] = None, regression_test_runner: Optional[Any] = None, persist_artifacts: bool = True, workspace_root: Optional[str] = None, ) -> Tuple[SandboxManifest, PatchResult, VerificationResult, ComparisonResult, CandidateVerdict, Optional[PromotionProposal]]: """ Executes the Phase 2 autonomous sandbox repair cycle: Sandbox Builder -> Baseline Snapshot -> Candidate Patch -> Compile -> Focused Tests -> Regression Tests -> Evidence Comparator -> Candidate Verdict -> Promotion Proposal """ if workspace_root: self.sandbox_builder.project_root = os.path.abspath(workspace_root) inc_dir = os.path.join(self.base_reports_dir, incident.incident_id) if persist_artifacts else None # 1. BUILD SANDBOX & BASELINE SNAPSHOT manifest = self.sandbox_builder.build_sandbox( proposal=proposal, base_reports_dir=self.base_reports_dir, candidate_id=patch_plan.candidate_id or None, ) patch_plan.candidate_id = manifest.candidate_id # 2. APPLY CANDIDATE PATCH (Guard verification dependencies against modification) v_deps = ( getattr(proposal, "verification_dependencies", []) or getattr(manifest, "verification_dependencies", []) or proposal.sandbox_contract.get("verification_dependencies", []) ) if v_deps and patch_plan.instructions: for inst in patch_plan.instructions: inst_norm = os.path.normpath(str(inst.file_rel_path)).replace("\\", "/") for vd in v_deps: vd_norm = os.path.normpath(str(vd)).replace("\\", "/") if inst_norm == vd_norm or inst_norm.endswith(vd_norm): from app.remediation.self_repair.sandbox_builder import SandboxSecurityViolation raise SandboxSecurityViolation( f"Modification of verification dependency '{inst.file_rel_path}' strictly forbidden", "VERIFICATION_DEPENDENCY_WRITE_BLOCKED" ) patch_res = self.patch_executor.apply_patch_plan( plan=patch_plan, manifest=manifest, incident_dir=inc_dir, ) # 3. VERIFY CANDIDATE (Compile, Focused, Regression) verification = self.candidate_verifier.verify_candidate( manifest=manifest, proposal=proposal, patch_result=patch_res, focused_test_runner=focused_test_runner, regression_test_runner=regression_test_runner, incident_dir=inc_dir, ) # 4. EVIDENCE COMPARATOR comparison = self.evidence_comparator.compare( incident=incident, evidence=evidence, verification=verification, incident_dir=inc_dir, ) # 5. CANDIDATE JUDGE & PROMOTION PROPOSAL verdict, promotion = self.candidate_judge.judge( manifest=manifest, proposal=proposal, root_cause_info=root_cause, patch_result=patch_res, verification=verification, comparison=comparison, project_root=self.project_root, incident_dir=inc_dir, ) return manifest, patch_res, verification, comparison, verdict, promotion def ingest_mission_failure( self, mission_data: Dict[str, Any], has_explicit_contradiction: bool = False, ) -> Tuple[bool, Optional[MissionFailureEvent], Optional[IncidentRecord], Optional[EvidenceBundle], str]: """ Determines trigger condition and ingests failure data. """ status = str(mission_data.get("mission_status") or mission_data.get("status") or "").strip().upper() should_trigger, reason = FailureTriggerGuard.should_trigger( mission_status=status, has_explicit_contradiction=has_explicit_contradiction, ) if not should_trigger: return False, None, None, None, reason event, incident, evidence = self.failure_ingestor.ingest_mission_data(mission_data) return True, event, incident, evidence, reason def auto_remediate_mission( self, mission_data: Dict[str, Any], focused_test_runner: Optional[Any] = None, regression_test_runner: Optional[Any] = None, explicit_patch_instruction: Optional[Dict[str, Any]] = None, persist_artifacts: bool = True, ) -> Dict[str, Any]: """ Full autonomous flow: Trigger Guard -> Ingest -> Trace -> Analyze -> Plan -> PatchPlanGenerator -> Sandbox Repair -> Verdict """ # 1. Trigger Check status = str(mission_data.get("mission_status") or mission_data.get("status") or "").strip().upper() has_contradiction = bool(mission_data.get("has_explicit_contradiction") or status == "FALSE_SUCCESS") should_trigger, reason = FailureTriggerGuard.should_trigger(status, has_contradiction) if not should_trigger: return { "self_repair_triggered": False, "status": "SKIPPED_NOT_TRIGGERED", "reason": reason, } # 2. Ingest event, incident, evidence = self.failure_ingestor.ingest_mission_data(mission_data) inc_dir = os.path.join(self.base_reports_dir, incident.incident_id) if persist_artifacts else None if persist_artifacts: self.failure_ingestor.persist_failure_event(event, inc_dir) self.observer.persist(incident, inc_dir) self.collector.persist(evidence, inc_dir) # 3. Trace trace = self.tracer.trace(incident, evidence) if persist_artifacts: self.tracer.persist(trace, inc_dir) ws_root = os.path.abspath(event.workspace_root or self.project_root) # 4. Root Cause Analysis root_cause = self.analyzer.analyze(incident, evidence, trace, workspace_root=ws_root) if persist_artifacts: self.analyzer.persist(root_cause, inc_dir) result: Dict[str, Any] = { "self_repair_triggered": True, "failure_event": event, "incident": incident, "circuit_trace": trace, "root_cause": root_cause, "first_proven_break": trace.first_proven_break, "root_cause_proven": root_cause.proven, "provider_called": False, "network_used": False, } # Check if root cause is unproven if not root_cause.proven or root_cause.root_cause == "UNPROVEN": result.update({ "patch_plan_generated": False, "sandbox_repair_started": False, "status": "UNPROVEN", "auto_repair_authorized": False, }) return result self.sandbox_builder.project_root = ws_root # Check if structured patch instructions already exist (explicit or in error_context) patch_info = explicit_patch_instruction or event.error_context.get("suggested_patch") or event.error_context has_explicit_structured_patch = bool( explicit_patch_instruction or ( isinstance(patch_info, dict) and patch_info.get("before") is not None and patch_info.get("after") is not None ) ) result["use_existing_structured_patch"] = has_explicit_structured_patch target_rel = patch_info.get("file_rel_path") or patch_info.get("file") if isinstance(patch_info, dict) else None if not target_rel: disc_target, _ = self.patch_reasoner.discover_target_file(event, evidence, ws_root) target_rel = disc_target if not target_rel and getattr(root_cause, "defective_module", None): target_rel = root_cause.defective_module target_norm = os.path.normpath(str(target_rel)).replace("\\", "/") if target_rel else None # PARTE A: Extract focused verification contract following strict source priority: # 1. Explicit mission field focused_test_cmd = ( mission_data.get("focused_test_command") or mission_data.get("focused_test") or (event.error_context or {}).get("focused_test_command") or (event.error_context or {}).get("focused_test") or (event.verification_result or {}).get("command") or (event.verification_result or {}).get("focused_test_command") ) # 2. Physical execution history (last EXECUTE_COMMAND associated with verification) if not focused_test_cmd and event.execution_history: for act in reversed(event.execution_history): act_type = act.get("action_type") or act.get("type") or "" if act_type == "EXECUTE_COMMAND" or "command" in act or "action_payload" in act: payload = act.get("action_payload") or act.get("payload") or act cmd = payload.get("command") or act.get("command") if cmd and any(kw in str(cmd) for kw in ("pytest", "test_", "unittest", "python")): focused_test_cmd = str(cmd) break # Verification dependencies (test files required to execute verification) v_deps: List[str] = list( mission_data.get("verification_dependencies") or (event.error_context or {}).get("verification_dependencies") or [] ) if focused_test_cmd: import shlex try: tokens = shlex.split(focused_test_cmd) except Exception: tokens = focused_test_cmd.split() for tok in tokens: if tok.startswith("-"): continue if tok.endswith(".py") or "test_" in tok or "_test" in tok: tok_clean = os.path.normpath(tok).replace("\\", "/") if tok_clean not in v_deps: v_deps.append(tok_clean) for extra_k in ("test_file", "test_files", "target_test", "target_tests"): val = mission_data.get(extra_k) or (event.error_context or {}).get(extra_k) if val: val_list = [val] if isinstance(val, str) else list(val) for item in val_list: item_clean = os.path.normpath(str(item)).replace("\\", "/") if item_clean not in v_deps: v_deps.append(item_clean) # PARTE D: If target_norm attempts to modify a verification dependency, block immediately if target_norm and any( os.path.normpath(target_norm).replace("\\", "/") == os.path.normpath(vd).replace("\\", "/") for vd in v_deps ): result.update({ "verification_dependency_write_blocked": True, "patch_plan_generated": False, "sandbox_repair_started": False, "candidate_repair": "REJECT", "status": "VERIFICATION_DEPENDENCY_WRITE_BLOCKED", "error": f"VERIFICATION_DEPENDENCY_WRITE_BLOCKED: '{target_norm}' is a read-only verification dependency and cannot be modified.", }) return result # Immediate frozen baseline check if target_norm: for fb in ("artifacts/brain_host/capability_mission_planner.py", "artifacts/brain_host/generic_developer_executor.py"): if target_norm == fb or target_norm.endswith(fb): result.update({ "frozen_baseline_blocked": True, "patch_plan_generated": False, "sandbox_repair_started": False, "candidate_repair": "REJECT", "status": "FROZEN_BASELINE_BLOCKED", }) return result # 5. Repair Proposal custom_allowed = [target_norm] if target_norm else None custom_regressions = ( mission_data.get("regression_tests") or (event.error_context or {}).get("regression_tests") ) proposal = self.planner.plan_repair( incident=incident, root_cause_info=root_cause, custom_files_allowed=custom_allowed, custom_focused_tests=v_deps if v_deps else ([focused_test_cmd] if focused_test_cmd else None), custom_regression_tests=list(custom_regressions) if custom_regressions is not None else None, focused_test_command=focused_test_cmd, verification_dependencies=v_deps, workspace_root=ws_root, ) for reg_dep in (proposal.regression_tests or []): reg_clean = os.path.normpath(str(reg_dep)).replace("\\", "/") if " " not in reg_clean and reg_clean not in v_deps: if os.path.isfile(os.path.join(ws_root, reg_clean)): v_deps.append(reg_clean) if v_deps: setattr(proposal, "verification_dependencies", list(v_deps)) proposal.sandbox_contract["verification_dependencies"] = list(v_deps) if target_norm and target_norm not in [os.path.normpath(f).replace("\\", "/") for f in proposal.files_allowed]: if not any(os.path.normpath(target_norm).replace("\\", "/") == os.path.normpath(vd).replace("\\", "/") for vd in v_deps): proposal.files_allowed.append(target_norm) if persist_artifacts and inc_dir: self.planner.persist(proposal, inc_dir) result["repair_proposal"] = proposal # Check frozen baseline in proposal.files_allowed for fb in ("artifacts/brain_host/capability_mission_planner.py", "artifacts/brain_host/generic_developer_executor.py"): target_norm_list = [os.path.normpath(f).replace("\\", "/") for f in proposal.files_allowed] if fb in target_norm_list: result.update({ "frozen_baseline_blocked": True, "patch_plan_generated": False, "sandbox_repair_started": False, "candidate_repair": "REJECT", "status": "FROZEN_BASELINE_BLOCKED", }) return result # 6. Patch Plan Generation (Explicit Fallback or Autonomous Reasoning) with Feedback Loop max_repair_attempts = int( mission_data.get("max_repair_attempts") or (event.error_context or {}).get("max_repair_attempts") or 1 ) if max_repair_attempts < 1: max_repair_attempts = 1 # Hard upper ceiling to prevent any accidental runaway: max 3 attempts max_repair_attempts = min(max_repair_attempts, 3) attempt = 1 previous_feedback: Optional[Dict[str, Any]] = None while attempt <= max_repair_attempts: if has_explicit_structured_patch: plan_res = self.patch_plan_generator.generate_plan( event=event, root_cause_info=root_cause, proposal=proposal, workspace_root=ws_root, explicit_patch_instruction=explicit_patch_instruction, ) else: auth_ctx = mission_data.get("authorization_context") or (event.error_context or {}).get("authorization_context") reasoning_res = self.patch_reasoner.reason_patch( event=event, incident=incident, evidence=evidence, root_cause_info=root_cause, proposal=proposal, workspace_root=ws_root, previous_failure_feedback=previous_feedback, authorization_context=auth_ctx, ) if persist_artifacts and inc_dir: self.patch_reasoner.persist(reasoning_res, inc_dir) result["patch_reasoning_result"] = reasoning_res if not reasoning_res.success or reasoning_res.confidence != "HIGH": is_frozen = reasoning_res.failure_reason == "FROZEN_BASELINE_BLOCKED" is_ambiguous = reasoning_res.failure_reason == "PATCH_ANCHOR_AMBIGUOUS" is_creator_disabled = reasoning_res.failure_reason == "CREATOR_API_DISABLED" is_budget_exceeded = reasoning_res.failure_reason == "BUDGET_EXCEEDED" is_api_inaccessible = reasoning_res.failure_reason == "EXTERNAL_API_INACCESSIBLE" status = reasoning_res.failure_reason or "AUTONOMOUS_REASONING_FAILED" is_paused = is_creator_disabled or is_budget_exceeded or is_api_inaccessible if is_creator_disabled: status = "PAUSED_AWAITING_CREATOR_AUTHORIZATION" elif is_budget_exceeded: status = "PAUSED_BUDGET_EXCEEDED" elif is_api_inaccessible: status = "PAUSED_EXTERNAL_API_INACCESSIBLE" result.update({ "autonomous_patch_generated": False, "patch_plan_generated": False, "sandbox_repair_started": False, "candidate_repair": "NOT_STARTED", "frozen_baseline_blocked": is_frozen, "patch_anchor_ambiguous": is_ambiguous, "self_repair_paused": is_paused, "paused_reason": reasoning_res.failure_reason if is_paused else None, "system_fault": not is_paused, "failure_reason": reasoning_res.failure_reason, "status": status, }) return result result["autonomous_patch_generated"] = True inferred_instruction = reasoning_res.candidate_instructions[0] target_norm = inferred_instruction.file_rel_path if target_norm and any( os.path.normpath(target_norm).replace("\\", "/") == os.path.normpath(vd).replace("\\", "/") for vd in v_deps ): result.update({ "verification_dependency_write_blocked": True, "patch_plan_generated": False, "sandbox_repair_started": False, "candidate_repair": "REJECT", "status": "VERIFICATION_DEPENDENCY_WRITE_BLOCKED", "error": f"VERIFICATION_DEPENDENCY_WRITE_BLOCKED: '{target_norm}' is a read-only verification dependency and cannot be modified.", }) return result if target_norm not in [os.path.normpath(f).replace("\\", "/") for f in proposal.files_allowed]: proposal.files_allowed.append(target_norm) plan_res = self.patch_plan_generator.generate_plan( event=event, root_cause_info=root_cause, proposal=proposal, workspace_root=ws_root, explicit_patch_instruction=inferred_instruction.to_dict(), ) if plan_res.frozen_baseline_blocked: result.update({ "frozen_baseline_blocked": True, "patch_plan_generated": False, "sandbox_repair_started": False, "candidate_repair": "REJECT", "status": "FROZEN_BASELINE_BLOCKED", }) return result if plan_res.anchor_ambiguous: result.update({ "patch_anchor_ambiguous": True, "patch_plan_generated": False, "sandbox_repair_started": False, "candidate_repair": "NOT_STARTED", "status": "AMBIGUOUS_ANCHOR", }) return result if not plan_res.success: result.update({ "patch_plan_generated": False, "sandbox_repair_started": False, "candidate_repair": "NOT_STARTED", "error": plan_res.error_message, "status": "PATCH_PLAN_FAILED", }) return result # 7. Sandbox Execution (Phase 2) patch_plan = plan_res.plan result["patch_plan"] = patch_plan result["patch_plan_generated"] = True if persist_artifacts and inc_dir: self.patch_plan_generator.persist(patch_plan, inc_dir) manifest, patch_res, verif, comp, verdict, promotion = self.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=ws_root, ) result.update({ "sandbox_manifest": manifest, "patch_result": patch_res, "verification_result": verif, "comparison_result": comp, "candidate_verdict": verdict, "promotion_proposal": promotion, "candidate_repair": verdict.verdict, "real_sandbox_autorepair": "PASS" if verdict.verdict == "ACCEPT" else "REJECT", "auto_promotion": False, "creator_approval_required": True, "attempts_executed": attempt, "status": "REMEDIATION_ACCEPTED" if verdict.verdict == "ACCEPT" else "REMEDIATION_REJECTED", }) if verdict.verdict == "ACCEPT" or has_explicit_structured_patch or attempt >= max_repair_attempts: break # Collect feedback from failed sandbox verification for the reasoner retry previous_feedback = { "attempt": attempt, "candidate": { "file_rel_path": patch_plan.instructions[0].file_rel_path if patch_plan.instructions else "", "before": patch_plan.instructions[0].before if patch_plan.instructions else "", "after": patch_plan.instructions[0].after if patch_plan.instructions else "", }, "verdict": verdict.verdict, "rejection_reasons": verdict.reasons, "verification_failure": ( verif.focused_output if verif and hasattr(verif, "focused_output") else "Sandbox verification failed" ), } attempt += 1 return result