# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — REPAIR PLANNER Formulates structured repair proposals for sandbox evaluation. Never grants production write authority, never auto-promotes, and fails closed when root cause is unproven. """ from __future__ import annotations import json import os from typing import Any, Dict, List, Optional from app.remediation.self_repair.sandbox_contract import SelfRepairSandboxContract from app.remediation.self_repair.schemas import ( IncidentRecord, RepairProposal, RootCauseClassification, _new_id, ) class RepairPlanner: """ Constructs sandbox repair proposals based on proven root cause determinations. Enforces strict safety boundaries and explicit test/rollback specifications. """ PRODUCTION_WRITE_AUTHORITY: bool = False CORE_WRITE_AUTHORITY: bool = False AUTO_PROMOTION: bool = False COMPONENT_FILE_MAPPING: Dict[str, List[str]] = { "requirement_contract": ["artifacts/brain_host/requirement_contract_parser.py"], "planner": ["app/planning/mission_planner.py"], "executor": ["app/actions/generic_executor.py"], "effector": ["app/capabilities/effector_dispatcher.py"], "verification": ["app/verification/execution_evidence.py"], "chat_gateway": ["app/web/chat_gateway.py"], "brain": ["app/agent/agent_orchestrator.py"], "circuit": ["app/remediation/self_repair/circuit.py"], "circuit_tracer": ["app/remediation/self_repair/circuit.py"], "sandbox_test": ["sandbox/test_component.py"], } def plan_repair( self, incident: IncidentRecord, root_cause_info: RootCauseClassification, custom_files_allowed: Optional[List[str]] = None, custom_focused_tests: Optional[List[str]] = None, custom_regression_tests: Optional[List[str]] = None, focused_test_command: Optional[str] = None, verification_dependencies: Optional[List[str]] = None, workspace_root: Optional[str] = None, ) -> RepairProposal: """ Synthesizes a RepairProposal with full rollback plan and sandbox contract. """ # Rule: If root cause is UNPROVEN, do not authorize or invent repairs if not root_cause_info.proven or root_cause_info.root_cause == "UNPROVEN": return RepairProposal( repair_id=_new_id("rep"), incident_id=incident.incident_id, root_cause="UNPROVEN", repair_authorized=False, files_allowed=[], files_forbidden=list(SelfRepairSandboxContract.FORBIDDEN_FILES), exact_change_scope="NONE: Root cause unproven. Repair creation blocked (Fail-Closed).", expected_effect="No modification permitted. Additional evidence collection required.", regression_risks=["N/A - No repair generated."], focused_tests=[], regression_tests=[], rollback_plan={"method": "NONE", "status": "NO_CHANGE"}, confidence=0.0, sandbox_contract=SelfRepairSandboxContract.get_contract_spec(incident.incident_id, []), production_write_authority=False, core_write_authority=False, auto_promotion=False, ) rc = root_cause_info.root_cause # Resolve candidate files allowed based on proven root cause allowed = list(custom_files_allowed or self.COMPONENT_FILE_MAPPING.get(rc, [f"app/{rc}/handler.py"])) # Also check initial evidence for explicit target file err_ctx = (incident.initial_evidence or {}).get("error_context", {}) if isinstance(err_ctx, dict): target_from_ev = ( err_ctx.get("file_rel_path") or err_ctx.get("file") or (err_ctx.get("suggested_patch", {}) if isinstance(err_ctx.get("suggested_patch"), dict) else {}).get("file_rel_path") or (err_ctx.get("suggested_patch", {}) if isinstance(err_ctx.get("suggested_patch"), dict) else {}).get("file") ) if target_from_ev: target_norm = os.path.normpath(str(target_from_ev)).replace("\\", "/") if target_norm not in [os.path.normpath(f).replace("\\", "/") for f in allowed]: allowed.append(target_norm) # Filter out any accidentally included forbidden files for forbidden in SelfRepairSandboxContract.FORBIDDEN_FILES: if forbidden in allowed: allowed.remove(forbidden) # Ensure verification dependencies are NEVER in allowed (they are read-only) v_deps = list(verification_dependencies or []) for vdep in v_deps: vdep_norm = os.path.normpath(str(vdep)).replace("\\", "/") allowed = [f for f in allowed if os.path.normpath(f).replace("\\", "/") != vdep_norm] # PARTE B & C: Determine focused tests strictly without fictitious targets focused: List[str] = [] if custom_focused_tests is not None: focused = list(custom_focused_tests) elif v_deps: focused = list(v_deps) elif focused_test_command: focused = [focused_test_command] else: cand_test = f"tests/test_remediation_{rc}.py" if os.path.isfile(cand_test): focused = [cand_test] else: focused = [] # PARTE F: Conditional regression policy (avoid irrelevant internal regressions for external disposable workspaces) ws_base = os.path.abspath(workspace_root) if workspace_root else None if custom_regression_tests is not None: regressions = list(custom_regression_tests) else: teitan_core_prefixes = ("app/", "artifacts/", "runtime_v1/", "new_mission_dispatcher.py") touches_core = any( any(os.path.normpath(f).replace("\\", "/").startswith(p) for p in teitan_core_prefixes) for f in allowed ) if touches_core: default_regressions = [ "tests/test_teitan_self_repair_c12_patch_generation.py", "tests/test_teitan_self_repair_phase5a_promotion_guard.py", "tests/test_teitan_self_repair_phase5a_rollback.py", "tests/test_teitan_semantic_requirement_route_patch_01.py", "tests/test_teitan_artifact_only_runtime_separation.py", ] focused_norm_set = {os.path.normpath(ft).replace("\\", "/") for ft in focused} regressions = [] for r in default_regressions: if r in focused_norm_set: continue if ws_base and os.path.isfile(os.path.join(ws_base, r)): regressions.append(r) elif not ws_base and os.path.isfile(r): regressions.append(r) else: regressions = [] for reg_t in regressions: reg_norm = os.path.normpath(str(reg_t)).replace("\\", "/") if " " not in reg_norm and reg_norm not in v_deps: if (ws_base and os.path.isfile(os.path.join(ws_base, reg_norm))) or os.path.isfile(reg_norm): v_deps.append(reg_norm) rollback = { "method": "RESTORE_SNAPSHOT_PRE_REPAIR", "source_origin_hashes_recorded": True, "verification_post_rollback": "HASH_MATCH_AND_SMOKE_TEST", "max_rollback_attempts": 3, } sandbox_spec = SelfRepairSandboxContract.get_contract_spec( incident_id=incident.incident_id, files_allowed=allowed, ) if focused_test_command: sandbox_spec["focused_test_command"] = focused_test_command if v_deps: sandbox_spec["verification_dependencies"] = v_deps proposal = RepairProposal( repair_id=_new_id("rep"), incident_id=incident.incident_id, root_cause=rc, repair_authorized=True, files_allowed=allowed, files_forbidden=list(SelfRepairSandboxContract.FORBIDDEN_FILES), exact_change_scope=f"Surgical patch to component '{rc}' addressing symptom: {incident.observed_symptom}", expected_effect=f"Resolve failure at '{rc}' and eliminate downstream symptom: {root_cause_info.symptom}", regression_risks=[ f"Potential behavioral shift in '{rc}' if contract boundaries are modified.", "Ensure backward compatibility across adjacent pipeline stages.", ], focused_tests=focused, regression_tests=regressions, rollback_plan=rollback, confidence=0.90, sandbox_contract=sandbox_spec, production_write_authority=False, core_write_authority=False, auto_promotion=False, ) if focused_test_command: setattr(proposal, "focused_test_command", focused_test_command) if v_deps: setattr(proposal, "verification_dependencies", v_deps) return proposal def persist(self, proposal: RepairProposal, output_dir: str) -> str: """ Persists repair_proposal.json to the specified output directory. """ os.makedirs(output_dir, exist_ok=True) path = os.path.join(output_dir, "repair_proposal.json") with open(path, "w", encoding="utf-8") as f: json.dump(proposal.to_dict(), f, indent=2, ensure_ascii=False) return path