# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — AUTONOMOUS PATCH REASONER (FASE 3B) Infers structured patch candidates directly from raw failure evidence, symptoms, and source inspection without requiring human-prebuilt patch context. Strictly adheres to deterministic patterns (operator mismatch, literal values, boolean inversion) and fails closed when evidence is insufficient or ambiguous. """ from __future__ import annotations import json import os import re from typing import Any, Dict, List, Optional, Tuple from app.remediation.self_repair.brain_reasoning_bridge import BrainReasoningBridge from app.remediation.self_repair.sandbox_contract import SelfRepairSandboxContract from app.remediation.self_repair.schemas import ( EvidenceBundle, IncidentRecord, MissionFailureEvent, PatchCandidate, PatchInstruction, PatchOperationType, PatchReasoningResult, RepairProposal, RootCauseClassification, _new_id, ) FROZEN_BASELINES = [ "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", ] CRITICAL_SYSTEM_PATHS = [ "app/remediation/self_repair/schemas.py", "app/remediation/self_repair/engine.py", "app/remediation/self_repair/live_shadow_hook.py", ] MAX_PATCH_CANDIDATES = 3 def _is_protected_or_critical(rel_path: str) -> bool: norm = os.path.normpath(rel_path).replace("\\", "/").lower() for protected in FROZEN_BASELINES + CRITICAL_SYSTEM_PATHS: prot_norm = os.path.normpath(protected).replace("\\", "/").lower() if norm == prot_norm or norm.endswith(prot_norm) or prot_norm.endswith(norm): return True return False def _is_test_file(path_str: str) -> bool: clean = os.path.normpath(path_str).replace("\\", "/") filename = os.path.basename(clean).lower() parts = [p.lower() for p in clean.split("/")] if filename.startswith("test_") and filename.endswith(".py"): return True if filename.endswith("_test.py"): return True if "tests" in parts or "test" in parts: if filename.endswith(".py"): return True return False class AutonomousPatchReasoner: """ Transforms failure telemetry and source code into structured, verified patch proposals using deterministic reasoning. """ CONFIDENCE_HIGH = "HIGH" CONFIDENCE_MEDIUM = "MEDIUM" CONFIDENCE_LOW = "LOW" CONFIDENCE_UNPROVEN = "UNPROVEN" DISQUALIFYING_KEYWORDS = [ "timeout", "timed out", "connectionrefused", "connection reset", "econnrefused", "database error", "deadlock", "no module named", "modulenotfounderror", "importerror", "segmentation fault", "killed", "out of memory", "oom", ] def __init__( self, project_root: str = ".", brain_bridge: Optional[BrainReasoningBridge] = None, ): self.project_root = os.path.abspath(project_root) self.brain_bridge = brain_bridge or BrainReasoningBridge() def discover_target_file( self, event: MissionFailureEvent, evidence: Optional[EvidenceBundle] = None, workspace_root: Optional[str] = None, ) -> Tuple[Optional[str], Optional[str]]: """ Determines the single proven target file from available evidence. Priority: 1. target_file explícito 2. target_artifact explícito 3. target_files explícitos con candidato único 4. inferencia desde execution_history Si existe un target_file válido: inside workspace, exists, not frozen, not critical -> seleccionarlo directamente. No permitir que un archivo de test lo invalide. Returns: (target_file_rel_path, failure_reason) """ ws = os.path.abspath(workspace_root or event.workspace_root or self.project_root) def _validate_candidate(target: Any) -> Tuple[bool, Optional[str], Optional[str]]: target_str = str(target or "").strip() if not target_str: return False, None, "TARGET_FILE=UNPROVEN: Empty target candidate." # Check if outside workspace if os.path.isabs(target_str): target_abs = os.path.abspath(target_str) try: rel = os.path.relpath(target_abs, ws).replace("\\", "/") except ValueError: return False, None, f"TARGET_FILE=UNPROVEN: Target '{target_str}' is outside workspace (different drive)." if rel.startswith("../") or ".." in rel.split("/"): return False, None, f"TARGET_FILE=UNPROVEN: Target '{target_str}' resolves outside workspace." else: norm = os.path.normpath(target_str).replace("\\", "/") if norm.startswith("../") or ".." in norm.split("/"): return False, None, f"TARGET_FILE=UNPROVEN: Target '{target_str}' resolves outside workspace." rel = norm target_abs = os.path.abspath(os.path.join(ws, rel)) # If it's a protected baseline or critical file, accept as candidate so reason_patch can fail closed with FROZEN_BASELINE_BLOCKED if _is_protected_or_critical(rel): return True, rel, None # Check exists on disk if not os.path.isfile(target_abs): return False, None, f"TARGET_FILE=UNPROVEN: Target file '{rel}' not found on disk at '{target_abs}'." return True, rel, None # ------------------------------------------------------------- # PRIORITY 1: target_file explícito # ------------------------------------------------------------- explicit_target = None if isinstance(event.error_context, dict): explicit_target = event.error_context.get("target_file") if not explicit_target: explicit_target = getattr(event, "target_file", None) if not explicit_target and evidence: explicit_target = getattr(evidence, "target_file", None) if not explicit_target and evidence and isinstance(evidence.mission_state, dict): explicit_target = evidence.mission_state.get("target_file") if not explicit_target and isinstance(event.verification_result, dict): explicit_target = event.verification_result.get("target_file") if explicit_target: valid, rel, reason = _validate_candidate(explicit_target) if valid and rel: return rel, None return None, reason # ------------------------------------------------------------- # PRIORITY 2: target_artifact explícito # ------------------------------------------------------------- explicit_art = None if isinstance(event.error_context, dict): explicit_art = event.error_context.get("target_artifact") if not explicit_art: explicit_art = getattr(event, "target_artifact", None) if not explicit_art and evidence: explicit_art = getattr(evidence, "target_artifact", None) if not explicit_art and evidence and isinstance(evidence.mission_state, dict): explicit_art = evidence.mission_state.get("target_artifact") if not explicit_art and isinstance(event.verification_result, dict): explicit_art = event.verification_result.get("target_artifact") if explicit_art: valid, rel, reason = _validate_candidate(explicit_art) if valid and rel: return rel, None return None, reason # ------------------------------------------------------------- # PRIORITY 3: target_files / target_artifacts explícitos # ------------------------------------------------------------- explicit_files = None if isinstance(event.error_context, dict): explicit_files = event.error_context.get("target_files") or event.error_context.get("target_artifacts") if not explicit_files: explicit_files = getattr(event, "target_files", None) or getattr(event, "target_artifacts", None) if not explicit_files and evidence: explicit_files = getattr(evidence, "target_files", None) or getattr(evidence, "target_artifacts", None) if not explicit_files and evidence and isinstance(evidence.mission_state, dict): explicit_files = evidence.mission_state.get("target_files") or evidence.mission_state.get("target_artifacts") if explicit_files and isinstance(explicit_files, (list, tuple)): valid_explicit = [] for f in explicit_files: v, r, _ = _validate_candidate(f) if v and r and r not in valid_explicit: valid_explicit.append(r) if len(valid_explicit) == 1: return valid_explicit[0], None elif len(valid_explicit) > 1: impl_explicit = [c for c in valid_explicit if not _is_test_file(c)] if len(impl_explicit) == 1: return impl_explicit[0], None return None, f"TARGET_FILE=UNPROVEN: Multiple conflicting explicit target candidates: {valid_explicit}" else: return None, "TARGET_FILE=UNPROVEN: Explicit target_files contained no valid existing files." # ------------------------------------------------------------- # PRIORITY 4: Inferencia desde execution_history / tracebacks / sources # ------------------------------------------------------------- raw_candidates: List[str] = [] # 4a. execution_history exec_hist = list(event.execution_history or (evidence.action_history if evidence else [])) for step in reversed(exec_hist): if isinstance(step, dict): st = str(step.get("status") or "").upper() target = step.get("target") or step.get("file") or step.get("file_rel_path") if target: norm = os.path.normpath(str(target)).replace("\\", "/") if "FAIL" in st or "ERROR" in st: raw_candidates.append(norm) elif not raw_candidates: raw_candidates.append(norm) # 4b. traceback filename tb_sources = [] if event.exception_message: tb_sources.append(event.exception_message) if evidence and evidence.tracebacks: tb_sources.extend(evidence.tracebacks) if evidence and evidence.logs: tb_sources.extend(evidence.logs) tb_regex = re.compile(r'File\s+["\']([^"\']+)["\'],\s+line\s+\d+') for text in tb_sources: for match in tb_regex.finditer(str(text)): raw_path = match.group(1) if os.path.isabs(raw_path): if raw_path.startswith(ws): rel = os.path.relpath(raw_path, ws).replace("\\", "/") raw_candidates.append(rel) else: norm = os.path.normpath(raw_path).replace("\\", "/") raw_candidates.append(norm) # 4c. source_files in event, error_context, or evidence sources_to_check: List[str] = [] if getattr(event, "source_files", None): sources_to_check.extend(event.source_files) if isinstance(event.error_context, dict) and event.error_context.get("source_files"): sources_to_check.extend(event.error_context["source_files"]) if evidence and getattr(evidence, "source_files", None): sources_to_check.extend(evidence.source_files) if evidence and isinstance(evidence.mission_state, dict) and evidence.mission_state.get("source_files"): sources_to_check.extend(evidence.mission_state["source_files"]) for sf in sources_to_check: if sf: raw_candidates.append(os.path.normpath(str(sf)).replace("\\", "/")) if evidence and evidence.source_excerpts: for exc in evidence.source_excerpts: p = exc.get("file_path") or exc.get("path") or exc.get("file") if p: raw_candidates.append(os.path.normpath(str(p)).replace("\\", "/")) # 4d. verification target verif = event.verification_result or (evidence.verification_results if evidence else {}) if isinstance(verif, dict): vt = verif.get("target") or verif.get("file") if vt: raw_candidates.append(os.path.normpath(str(vt)).replace("\\", "/")) # 4e. Fallback for internal system components: inspect registered component module metadata if not raw_candidates: cid = None if isinstance(event.error_context, dict): cid = event.error_context.get("component_id") or event.error_context.get("source_component") if not cid: cid = getattr(event, "source_component", None) if cid: try: from app.system_health.component_registry import get_component_registry reg = get_component_registry() comp = reg.get_component(str(cid)) if comp and comp.metadata and comp.metadata.get("module"): raw_candidates.append(os.path.normpath(str(comp.metadata["module"])).replace("\\", "/")) except Exception: pass # Validate all candidates valid_candidates: List[str] = [] for c in raw_candidates: v, r, _ = _validate_candidate(c) if v and r and r not in valid_candidates: valid_candidates.append(r) if not valid_candidates: return None, "TARGET_FILE=UNPROVEN: No valid existing target file found in failure evidence." # FIX 4: Discriminar tests de implementación test_candidates = [c for c in valid_candidates if _is_test_file(c)] impl_candidates = [c for c in valid_candidates if not _is_test_file(c)] # AR-74: If only a single failing test candidate is in raw_candidates, inspect its imports # within workspace_root to discover the single implementation module under test. if not impl_candidates and len(test_candidates) == 1: test_abs = os.path.join(ws, test_candidates[0]) if os.path.isfile(test_abs): try: with open(test_abs, "r", encoding="utf-8") as tf: test_src = tf.read() imported_impls: List[str] = [] for m_imp in re.finditer(r"^\s*from\s+([a-zA-Z0-9_\.]+)\s+import\s+([^\n#]+)", test_src, re.MULTILINE): mod_path = m_imp.group(1).replace(".", "/") symbols = [s.strip().split(" as ")[0].strip() for s in m_imp.group(2).strip("() ").split(",") if s.strip()] cand_py = f"{mod_path}.py" v_py, r_py, _ = _validate_candidate(cand_py) if v_py and r_py and not _is_test_file(r_py) and not _is_protected_or_critical(r_py): if r_py not in imported_impls: imported_impls.append(r_py) else: init_py = os.path.join(ws, mod_path, "__init__.py") if os.path.isfile(init_py): with open(init_py, "r", encoding="utf-8") as inf: init_src = inf.read() for sym in symbols: if not sym: continue m_reexp = re.search( rf"from\s+([a-zA-Z0-9_\.]+)\s+import\s+[^\n]*\b{re.escape(sym)}\b", init_src, ) if m_reexp: sub_py = m_reexp.group(1).replace(".", "/") + ".py" if not os.path.isfile(os.path.join(ws, sub_py)): sub_py = f"{mod_path}/{m_reexp.group(1).lstrip('.').replace('.', '/')}.py" v_sub, r_sub, _ = _validate_candidate(sub_py) if ( v_sub and r_sub and not _is_test_file(r_sub) and not _is_protected_or_critical(r_sub) and r_sub not in imported_impls ): imported_impls.append(r_sub) if len(imported_impls) == 1: impl_candidates.append(imported_impls[0]) except Exception: pass if len(impl_candidates) == 1: return impl_candidates[0], None if len(impl_candidates) >= 2: return None, f"TARGET_FILE=UNPROVEN: Multiple conflicting target candidates: {impl_candidates}" # len(impl_candidates) == 0: Solo quedaron archivos de test return None, f"TARGET_FILE=UNPROVEN: Only test candidates found: {test_candidates}; refusing to patch test files without explicit target." def reason_patch( self, event: MissionFailureEvent, incident: IncidentRecord, evidence: EvidenceBundle, root_cause_info: RootCauseClassification, proposal: Optional[RepairProposal] = None, workspace_root: Optional[str] = None, previous_failure_feedback: Optional[Dict[str, Any]] = None, authorization_context: Optional[Dict[str, Any]] = None, ) -> PatchReasoningResult: """ Main reasoning entry point. Inspects evidence and target source to synthesize deterministic patch candidates. """ evidence_used: List[str] = [] rejected_alternatives: List[str] = [] # 1. Verify root cause was proven if not root_cause_info.proven or root_cause_info.root_cause == "UNPROVEN": return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, reasoning_summary="Root cause is unproven. Autonomous reasoning refused.", failure_reason="ROOT_CAUSE_UNPROVEN", ) ws = os.path.abspath(workspace_root or event.workspace_root or self.project_root) # 2. Target File Discovery target_file, discovery_err = self.discover_target_file(event, evidence, ws) if not target_file and getattr(root_cause_info, "defective_module", None): rc_mod = os.path.normpath(str(root_cause_info.defective_module)).replace("\\", "/") if os.path.isfile(os.path.join(ws, rc_mod)): target_file = rc_mod discovery_err = None if not target_file: return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, reasoning_summary=discovery_err or "Target file unproven.", failure_reason=discovery_err or "TARGET_FILE_UNPROVEN", ) evidence_used.append(f"Target file discovered: {target_file}") # 3. Check Frozen Baselines & Critical System Files if _is_protected_or_critical(target_file): return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, target_file=target_file, reasoning_summary=f"FROZEN_BASELINE_BLOCKED: '{target_file}' is protected.", failure_reason="FROZEN_BASELINE_BLOCKED", ) # 4. Check Allowed / Forbidden Files if proposal given if proposal: for forbidden_pattern in proposal.files_forbidden: import fnmatch if fnmatch.fnmatch(target_file, forbidden_pattern) or target_file == forbidden_pattern: return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, target_file=target_file, reasoning_summary=f"FORBIDDEN_FILE_BLOCKED: '{target_file}' is in files_forbidden.", failure_reason="FORBIDDEN_FILE_BLOCKED", ) target_abs = os.path.join(ws, target_file) try: with open(target_abs, "r", encoding="utf-8") as f: source_code = f.read() except Exception as e: return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, target_file=target_file, reasoning_summary=f"Failed to read source file: {e}", failure_reason=f"CANNOT_READ_SOURCE: {e}", ) # 5. Extract failure symptoms & check disqualifying conditions symptom = (event.observed_symptom or "").strip() exc_msg = (event.exception_message or "").strip() symptom_parts = [] if symptom: symptom_parts.append(symptom) if exc_msg and exc_msg not in symptom_parts: symptom_parts.append(exc_msg) if isinstance(event.verification_result, dict): vr = event.verification_result for k in ("verification_reason", "reason", "output", "details", "message", "stdout", "stderr"): v_val = str(vr.get(k) or "").strip() if v_val and v_val not in symptom_parts: symptom_parts.append(v_val) for f in vr.get("failures", []): f_val = str(f).strip() if f_val and f_val not in symptom_parts: symptom_parts.append(f_val) for e in vr.get("errors", []): e_val = str(e).strip() if e_val and e_val not in symptom_parts: symptom_parts.append(e_val) if isinstance(event.error_context, dict): ec = event.error_context for k in ("verification_reason", "failure_fingerprint", "creator_request", "symptom", "error", "message", "exception"): ec_val = str(ec.get(k) or "").strip() if ec_val and ec_val not in symptom_parts: symptom_parts.append(ec_val) if evidence: if evidence.logs: for log_line in evidence.logs: ll_val = str(log_line).strip() if ll_val and ll_val not in symptom_parts: symptom_parts.append(ll_val) if evidence.tracebacks: for tb in evidence.tracebacks: tb_val = str(tb).strip() if tb_val and tb_val not in symptom_parts: symptom_parts.append(tb_val) if isinstance(evidence.verification_results, dict): evr = evidence.verification_results for k in ("verification_reason", "reason", "output", "details", "message", "stdout", "stderr"): ev_val = str(evr.get(k) or "").strip() if ev_val and ev_val not in symptom_parts: symptom_parts.append(ev_val) for f in evr.get("failures", []): f_val = str(f).strip() if f_val and f_val not in symptom_parts: symptom_parts.append(f_val) for e in evr.get("errors", []): e_val = str(e).strip() if e_val and e_val not in symptom_parts: symptom_parts.append(e_val) full_symptom_text = "\n".join(symptom_parts) # No guessing rule: check for network/database/timeout/system issues symptom_lower = full_symptom_text.lower() for kw in self.DISQUALIFYING_KEYWORDS: if kw in symptom_lower: return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, target_file=target_file, reasoning_summary=f"Symptom mentions '{kw}'. Environmental or non-deterministic failure.", failure_reason="INSUFFICIENT_DETERMINISTIC_EVIDENCE", ) candidates: List[PatchCandidate] = [] # 6. Deterministic Reasoning Engine # ---------------------------------------------------------------------- # PATTERN A: Operator Mismatch # E.g.: "multiply(2, 3) returned 5, expected 6" -> + vs * # or augmented assignment: "new_fails -= 1" vs "new_fails += 1" # ---------------------------------------------------------------------- op_candidates = self._infer_operator_mismatch( full_symptom_text, source_code, target_file, responsible_function=getattr(root_cause_info, "responsible_function", None), ) for c in op_candidates: candidates.append(c) # ---------------------------------------------------------------------- # PATTERN B: Literal Expected Value Mismatch # E.g.: "expected 'completed', actual 'pending'" # ---------------------------------------------------------------------- lit_candidates = self._infer_literal_mismatch(full_symptom_text, source_code, target_file) for c in lit_candidates: candidates.append(c) # ---------------------------------------------------------------------- # PATTERN C: Boolean Inversion # E.g.: "expected True, got False" # ---------------------------------------------------------------------- bool_candidates = self._infer_boolean_inversion(full_symptom_text, source_code, target_file) for c in bool_candidates: candidates.append(c) # ---------------------------------------------------------------------- # PATTERN D: Enum / Attribute Value Mismatch # E.g.: "assert CircuitStatus.UNKNOWN == CircuitStatus.NOT_REACHED" # ---------------------------------------------------------------------- enum_candidates = self._infer_enum_or_attribute_mismatch(full_symptom_text, source_code, target_file) for c in enum_candidates: candidates.append(c) # 7. Anchor Uniqueness & Validation on all Candidates valid_candidates: List[PatchCandidate] = [] for cand in candidates: # Check anchor uniqueness in source occ_count = source_code.count(cand.before) if occ_count == 0: rejected_alternatives.append(f"Anchor '{cand.before.strip()}' not found in '{target_file}'") continue if occ_count > 1: rejected_alternatives.append( f"PATCH_ANCHOR_AMBIGUOUS: Anchor '{cand.before.strip()}' appears {occ_count} times in '{target_file}'" ) continue valid_candidates.append(cand) if not valid_candidates: # Deterministic inference yielded no valid candidates. # Query the Brain Reasoning Bridge for general reasoning. det_reason = "INSUFFICIENT_DETERMINISTIC_EVIDENCE" if rejected_alternatives and any("AMBIGUOUS" in r for r in rejected_alternatives): det_reason = "PATCH_ANCHOR_AMBIGUOUS" evidence_used.append("Deterministic rules insufficient; querying Brain Reasoning Bridge.") obj = getattr(event, "creator_request", "") or (event.error_context or {}).get("creator_request") or (evidence.mission_state or {}).get("objective", "") verif_details = json.dumps(event.verification_result or (evidence.verification_results if evidence else {}) or {}, default=str) logs_and_traces = (evidence.logs if evidence else []) + (evidence.tracebacks if evidence else []) allowed_files = proposal.files_allowed if proposal else [target_file] forbidden_files = proposal.files_forbidden if proposal else [] feedback_list = [] if previous_failure_feedback: feedback_list.append(previous_failure_feedback) brain_resp = self.brain_bridge.investigate_and_propose_patch( objective=str(obj or ""), symptom=full_symptom_text, target_file=target_file, source_code=source_code, verification_details=verif_details, logs_and_traces=logs_and_traces, history_and_feedback=feedback_list, files_allowed=allowed_files, files_forbidden=forbidden_files, workspace_root=ws, task_id=getattr(event, "task_id", None), authorization_context=authorization_context, ) if not brain_resp.get("success") or not brain_resp.get("candidate"): status = brain_resp.get("status", "") failure_reason = det_reason if status == "PAUSED_CREATOR_API_DISABLED": failure_reason = "CREATOR_API_DISABLED" elif status == "PAUSED_BUDGET_EXCEEDED": failure_reason = "BUDGET_EXCEEDED" elif status == "EXTERNAL_API_INACCESSIBLE": failure_reason = "EXTERNAL_API_INACCESSIBLE" elif det_reason == "INSUFFICIENT_DETERMINISTIC_EVIDENCE" and status == "EXTERNAL_REASONING_UNAVAILABLE": failure_reason = "EXTERNAL_REASONING_UNAVAILABLE" summary = brain_resp.get("reasoning_summary") or f"Deterministic rules yielded no candidate ({det_reason}). Brain reasoning: {status}" return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, target_file=target_file, reasoning_summary=summary, rejected_alternatives=rejected_alternatives, failure_reason=failure_reason, ) # Brain produced a structured candidate - validate it strictly cand_dict = brain_resp["candidate"] cand_file = os.path.normpath(str(cand_dict.get("file_rel_path") or target_file)).replace("\\", "/") cand_op = cand_dict.get("operation") or "REPLACE_EXACT" cand_before = str(cand_dict.get("before") or "") cand_after = str(cand_dict.get("after") or "") cand_reason = cand_dict.get("reason") or "Brain reasoning candidate" # Check target file match target_norm = os.path.normpath(target_file).replace("\\", "/") allowed_norm = [os.path.normpath(f).replace("\\", "/") for f in allowed_files] if cand_file != target_norm and cand_file not in allowed_norm: return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, target_file=target_file, reasoning_summary=f"FORBIDDEN_FILE_BLOCKED: Brain proposed modification for unauthorized file '{cand_file}'.", failure_reason="FORBIDDEN_FILE_BLOCKED", ) # Check frozen baselines & critical if _is_protected_or_critical(cand_file): return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, target_file=target_file, reasoning_summary=f"FROZEN_BASELINE_BLOCKED: Brain proposed modification for protected file '{cand_file}'.", failure_reason="FROZEN_BASELINE_BLOCKED", ) # Check anchor uniqueness in source_code occ = source_code.count(cand_before) if occ == 0: rejected_alternatives.append(f"Brain anchor '{cand_before.strip()}' not found in '{target_file}'") return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, target_file=target_file, reasoning_summary=f"Brain patch anchor not found in '{target_file}'.", rejected_alternatives=rejected_alternatives, failure_reason="PATCH_ANCHOR_NOT_FOUND", ) if occ > 1: rejected_alternatives.append( f"PATCH_ANCHOR_AMBIGUOUS: Brain anchor appears {occ} times in '{target_file}'" ) return PatchReasoningResult( success=False, confidence=self.CONFIDENCE_UNPROVEN, target_file=target_file, reasoning_summary=f"PATCH_ANCHOR_AMBIGUOUS: Brain anchor appears {occ} times in '{target_file}'.", rejected_alternatives=rejected_alternatives, failure_reason="PATCH_ANCHOR_AMBIGUOUS", ) op_enum = PatchOperationType.REPLACE_EXACT try: op_enum = PatchOperationType(cand_op) except Exception: pass brain_candidate = PatchCandidate( candidate_patch_id=_new_id("cand"), file_rel_path=cand_file, operation=op_enum, before=cand_before, after=cand_after, confidence=self.CONFIDENCE_HIGH, reason=cand_reason, ) brain_instruction = PatchInstruction( file_rel_path=cand_file, operation=op_enum, before=cand_before, after=cand_after, reason=cand_reason, ) if brain_resp.get("analysis"): evidence_used.append(f"Brain Analysis: {brain_resp['analysis']}") if brain_resp.get("hypothesis"): evidence_used.append(f"Brain Hypothesis: {brain_resp['hypothesis']}") if brain_resp.get("justification"): evidence_used.append(f"Brain Justification: {brain_resp['justification']}") prov_model = str(brain_resp.get("provider_model") or "").strip() prov_used = str(brain_resp.get("provider_used") or "").strip() if prov_model or prov_used: evidence_used.append(f"Brain Provider: {prov_used or 'openai.responses'} | Model: {prov_model or 'default'}") brain_candidate.evidence_refs.append(f"provider={prov_used or 'openai.responses'}") brain_candidate.evidence_refs.append(f"model={prov_model or 'default'}") return PatchReasoningResult( success=True, confidence=self.CONFIDENCE_HIGH, reasoning_summary=f"Brain reasoned fix in '{target_file}': {cand_reason}", target_file=target_file, candidate_instructions=[brain_instruction], candidates=[brain_candidate], evidence_used=evidence_used, rejected_alternatives=rejected_alternatives, ) # 8. Sort and limit to MAX_PATCH_CANDIDATES # Order by confidence: HIGH > MEDIUM > LOW conf_order = {self.CONFIDENCE_HIGH: 0, self.CONFIDENCE_MEDIUM: 1, self.CONFIDENCE_LOW: 2} valid_candidates.sort(key=lambda c: conf_order.get(c.confidence, 99)) selected_candidates = valid_candidates[:MAX_PATCH_CANDIDATES] best_cand = selected_candidates[0] if best_cand.confidence != self.CONFIDENCE_HIGH: return PatchReasoningResult( success=False, confidence=best_cand.confidence, target_file=target_file, candidates=selected_candidates, reasoning_summary=f"Candidate generated with {best_cand.confidence} confidence; auto-sandbox requires HIGH.", rejected_alternatives=rejected_alternatives, failure_reason="AUTO_SANDBOX_REQUIRES_HIGH_CONFIDENCE", ) best_instruction = PatchInstruction( file_rel_path=best_cand.file_rel_path, operation=best_cand.operation, before=best_cand.before, after=best_cand.after, reason=best_cand.reason, ) evidence_used.append(f"Symptom analyzed: {symptom or full_symptom_text[:80]}") evidence_used.append(f"Inferred patch: {best_cand.reason}") return PatchReasoningResult( success=True, confidence=self.CONFIDENCE_HIGH, reasoning_summary=f"Deterministically inferred fix in '{target_file}': {best_cand.reason}", target_file=target_file, candidate_instructions=[best_instruction], candidates=selected_candidates, evidence_used=evidence_used, rejected_alternatives=rejected_alternatives, ) def _infer_operator_mismatch( self, symptom_text: str, source_code: str, target_file: str, responsible_function: Optional[str] = None, ) -> List[PatchCandidate]: """ Pattern A: Expected vs actual operator mismatch. E.g.: multiply(2, 3) returned 5, expected 6 -> + replaced with * or: assert -1 == 5 -> - replaced with + or: assert -1 == 1 -> -= replaced with += """ candidates: List[PatchCandidate] = [] act_val: Optional[float] = None exp_val: Optional[float] = None # Matches: multiply(2, 3) returned 5, expected 6 # or: multiply(2, 3) == 5, expected 6 # or: sumar(2,3) devuelve -1; se esperaba 5 pattern = re.compile( r"(?:(\w+)\s*\(([^)]*)\)\s*)?(?:returned|devuelve|retorna|==)\s*(-?\d+(?:\.\d+)?)[;,]?\s*(?:expected|se esperaba|esperado)\s*(-?\d+(?:\.\d+)?)" ) match = pattern.search(symptom_text) fn_name = None if match: fn_name = match.group(1) act_val = float(match.group(3)) exp_val = float(match.group(4)) else: # Inverse: expected 6, got 5 / returned 5 / se esperaba 5, devuelve -1 pattern_inv = re.compile( r"(?:expected|se esperaba|esperado)\s*(-?\d+(?:\.\d+)?)[;,]?\s*(?:but\s*)?(?:got|returned|actual|devuelve|retorna)\s*(-?\d+(?:\.\d+)?)" ) match_inv = pattern_inv.search(symptom_text) if match_inv: exp_val = float(match_inv.group(1)) act_val = float(match_inv.group(2)) else: # Pytest assertion: assert -1 == 5 pattern_assert = re.compile( r"assert\s+(-?\d+(?:\.\d+)?)\s*==\s*(-?\d+(?:\.\d+)?)" ) match_assert = pattern_assert.search(symptom_text) if match_assert: act_val = float(match_assert.group(1)) exp_val = float(match_assert.group(2)) else: # Pytest multiline: expected 5 ... got -1 / actual -1 m_exp = re.search(r"(?:expected|se esperaba|esperado)\s*[:=]?\s*(-?\d+(?:\.\d+)?)", symptom_text, re.IGNORECASE) m_got = re.search(r"(?:got|actual|returned|devuelve|retorna)\s*[:=]?\s*(-?\d+(?:\.\d+)?)", symptom_text, re.IGNORECASE) if m_exp and m_got: exp_val = float(m_exp.group(1)) act_val = float(m_got.group(1)) else: m_assert_fn = re.search(r"assert\s+(\w+)\s*\(([^)]*)\)\s*==\s*(-?\d+(?:\.\d+)?)", symptom_text) if m_assert_fn: fn_name = m_assert_fn.group(1) exp_val = float(m_assert_fn.group(3)) m_act = re.search(r"(?:where|got|actual|devuelve|retorna)\s*[:=]?\s*(-?\d+(?:\.\d+)?)", symptom_text, re.IGNORECASE) act_val = float(m_act.group(1)) if m_act else None else: return candidates # Determine target function name in source code if not fn_name and responsible_function: short_fn = responsible_function.split(".")[-1] if re.search(rf"^\s*def\s+{re.escape(short_fn)}\s*\(", source_code, re.MULTILINE): fn_name = short_fn if not fn_name or not re.search(rf"^\s*def\s+{re.escape(fn_name)}\s*\(", source_code, re.MULTILINE): candidate_fns = re.findall(r"\b([a-zA-Z_]\w*)\s*\(", symptom_text) fn_name = None for f in candidate_fns: if f.startswith("test_") or f in ("assert", "where", "print", "len", "range"): continue if re.search(rf"^\s*def\s+{re.escape(f)}\s*\(", source_code, re.MULTILINE): fn_name = f break # Inspect source code line by line lines = source_code.splitlines(keepends=True) in_target_fn = (fn_name is None) # Direction-aware operator maps if act_val is not None and exp_val is not None: if act_val < exp_val: aug_op_map = [("-=", "+="), ("/=", "*=")] bin_op_map = [("-", "+"), ("+", "*"), ("/", "*")] elif act_val > exp_val: aug_op_map = [("+=", "-="), ("*=", "+=")] bin_op_map = [("+", "-"), ("*", "+")] else: aug_op_map = [("-=", "+="), ("+=", "-=")] bin_op_map = [("+", "*"), ("*", "+"), ("-", "+"), ("/", "*")] else: aug_op_map = [("-=", "+="), ("+=", "-=")] bin_op_map = [("+", "*"), ("*", "+"), ("-", "+"), ("/", "*")] for line in lines: if fn_name and re.match(rf"^\s*def\s+{fn_name}\s*\(", line): in_target_fn = True continue if fn_name and in_target_fn and re.match(r"^\s*def\s+", line): # exited function in_target_fn = False if in_target_fn and ("return " in line or "=" in line): # Strip string literals before checking for arithmetic operators so "utf-8" is not matched line_no_strings = re.sub(r"(['\"])(?:\\.|(?!\1).)*\1", "''", line) # 1. Check augmented assignment operators first (-= <-> +=, etc.) matched_aug = False for cur_aug, target_aug in aug_op_map: aug_pattern = rf"(\b[a-zA-Z_]\w*\s*){re.escape(cur_aug)}(\s*[^\n#]+)" if re.search(aug_pattern, line_no_strings): new_line = line.replace(cur_aug, target_aug, 1) if new_line != line: matched_aug = True candidates.append( PatchCandidate( candidate_patch_id=_new_id("cand"), file_rel_path=target_file, operation=PatchOperationType.REPLACE_EXACT, before=line, after=new_line, reason=f"Operator mismatch: replacing '{cur_aug}' with '{target_aug}' in {fn_name or 'function'}", confidence=self.CONFIDENCE_HIGH, evidence_refs=[symptom_text], ) ) if matched_aug: continue # 2. Check standard binary operators for cur_op, target_op in bin_op_map: op_pattern = rf"(\b[a-zA-Z_]\w*\s*)\{cur_op}(\s*[a-zA-Z0-9_]\w*\b)" if re.search(op_pattern, line_no_strings): new_line = re.sub(op_pattern, rf"\1{target_op}\2", line) if new_line != line: candidates.append( PatchCandidate( candidate_patch_id=_new_id("cand"), file_rel_path=target_file, operation=PatchOperationType.REPLACE_EXACT, before=line, after=new_line, reason=f"Operator mismatch: replacing '{cur_op}' with '{target_op}' in {fn_name or 'function'}", confidence=self.CONFIDENCE_HIGH, evidence_refs=[symptom_text], ) ) return candidates def _infer_literal_mismatch( self, symptom_text: str, source_code: str, target_file: str, ) -> List[PatchCandidate]: """ Pattern B: Literal expected value mismatch. E.g.: expected 'completed', actual 'pending' or: 'pending' != 'completed' """ candidates: List[PatchCandidate] = [] patterns = [ re.compile(r"expected\s+['\"]([^'\"]+)['\"]\s*,?\s*(?:but\s+got|got|actual)\s+['\"]([^'\"]+)['\"]"), re.compile(r"(?:returned|actual|got)\s+['\"]([^'\"]+)['\"]\s*,?\s*expected\s+['\"]([^'\"]+)['\"]"), re.compile(r"['\"]([^'\"]+)['\"]\s*!=\s*['\"]([^'\"]+)['\"]"), ] exp_val = None act_val = None for p in patterns: m = p.search(symptom_text) if m: if "expected" in p.pattern[:15]: exp_val = m.group(1) act_val = m.group(2) else: act_val = m.group(1) exp_val = m.group(2) break if not exp_val or not act_val: return candidates lines = source_code.splitlines(keepends=True) for line in lines: if f'"{act_val}"' in line or f"'{act_val}'" in line: new_line = line.replace(f'"{act_val}"', f'"{exp_val}"').replace(f"'{act_val}'", f"'{exp_val}'") if new_line != line: candidates.append( PatchCandidate( candidate_patch_id=_new_id("cand"), file_rel_path=target_file, operation=PatchOperationType.REPLACE_EXACT, before=line, after=new_line, reason=f"Literal mismatch: replacing '{act_val}' with '{exp_val}'", confidence=self.CONFIDENCE_HIGH, evidence_refs=[symptom_text], ) ) return candidates def _infer_boolean_inversion( self, symptom_text: str, source_code: str, target_file: str, ) -> List[PatchCandidate]: """ Pattern C: Boolean inversion. E.g.: expected True, got False """ candidates: List[PatchCandidate] = [] pattern = re.compile(r"expected\s+(True|False)\s*,?\s*(?:but\s+got|got|actual)\s+(True|False)") match = pattern.search(symptom_text) if not match: pattern2 = re.compile(r"(?:returned|actual|got)\s+(True|False)\s*,?\s*expected\s+(True|False)") match2 = pattern2.search(symptom_text) if match2: act_bool = match2.group(1) exp_bool = match2.group(2) else: pattern3 = re.compile(r"expected\s+(True|False)") match3 = pattern3.search(symptom_text) if match3: exp_bool = match3.group(1) act_bool = "False" if exp_bool == "True" else "True" else: return candidates else: exp_bool = match.group(1) act_bool = match.group(2) lines = source_code.splitlines(keepends=True) for line in lines: if re.search(rf"\breturn\s+{act_bool}\b", line) or re.search(rf"=\s*{act_bool}\b", line): new_line = re.sub(rf"\b{act_bool}\b", exp_bool, line) if new_line != line: candidates.append( PatchCandidate( candidate_patch_id=_new_id("cand"), file_rel_path=target_file, operation=PatchOperationType.REPLACE_EXACT, before=line, after=new_line, reason=f"Boolean inversion: replacing '{act_bool}' with '{exp_bool}'", confidence=self.CONFIDENCE_HIGH, evidence_refs=[symptom_text], ) ) return candidates def _infer_enum_or_attribute_mismatch( self, symptom_text: str, source_code: str, target_file: str, ) -> List[PatchCandidate]: """ Pattern D: Enum or dotted attribute value mismatch in assertions. E.g.: - assert CircuitStatus.UNKNOWN == CircuitStatus.NOT_REACHED - AssertionError: assert == - expected CircuitStatus.NOT_REACHED, got CircuitStatus.UNKNOWN """ candidates: List[PatchCandidate] = [] patterns = [ re.compile( r"(?:expected|se esperaba)\s+?\s*,?\s*(?:but\s+got|got|actual|returned|devuelve)\s+?\s*,?\s*(?:expected|se esperaba)\s+?\s*==\s*= 16: score += 2 prev_window = "".join(lines[max(0, i - 3):i]) if exp_member in prev_window: score += 5 if re.search(r"^\s*(if|elif)\s+", prev_window, re.MULTILINE): score += 3 scored_matches.append((score, line, new_line)) if not scored_matches: return candidates scored_matches.sort(key=lambda item: item[0], reverse=True) top_score = scored_matches[0][0] top_matches = [m for m in scored_matches if m[0] == top_score] for _, before_line, after_line in top_matches: candidates.append( PatchCandidate( candidate_patch_id=_new_id("cand"), file_rel_path=target_file, operation=PatchOperationType.REPLACE_EXACT, before=before_line, after=after_line, reason=f"Enum/attribute mismatch: replacing '{act_sym}' with '{exp_sym}'", confidence=self.CONFIDENCE_HIGH, evidence_refs=[symptom_text], ) ) return candidates def persist(self, result: PatchReasoningResult, out_dir: str) -> str: """Persists patch_reasoning_result.json to incident directory.""" os.makedirs(out_dir, exist_ok=True) path = os.path.join(out_dir, "patch_reasoning_result.json") with open(path, "w", encoding="utf-8") as f: json.dump(result.to_dict(), f, indent=2, ensure_ascii=False) return path