# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — SHADOW ORCHESTRATOR (FASE 4) Connects Self-Repair to real TEITAN missions in strictly isolated SHADOW mode. Observes completed missions, captures immutable MissionFailureSnapshots, detects False Success contradictions, and executes autonomous sandbox repairs without ever blocking, mutating, or becoming a dependency of the primary mission. """ from __future__ import annotations import hashlib import json import logging import os import stat import tempfile from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Tuple, Union from app.remediation.self_repair.engine import SelfRepairEngine from app.remediation.self_repair.failure_ingestor import FailureTriggerGuard from app.remediation.self_repair.mission_resume_executor import MissionResumeExecutor from app.remediation.self_repair.mission_resume_guard import ( MAX_REPAIR_ATTEMPTS, MAX_RESUME_ATTEMPTS, MissionResumeGuard, ) from app.remediation.self_repair.promotion_executor import PromotionExecutor from app.remediation.self_repair.promotion_guard import PromotionGuard from app.remediation.self_repair.schemas import ( CandidateVerdict, ComparisonResult, MissionFailureSnapshot, MissionResumeResult, PatchCandidate, PatchPlan, PatchResult, PromotionDecision, PromotionExecutionResult, PromotionProposal, ResumeDecision, SandboxManifest, ShadowRepairReport, ShadowStatus, VerificationResult, _new_id, _utc_iso, ) logger = logging.getLogger("teitan.self_repair.shadow") FROZEN_BASELINES = ( "artifacts/brain_host/capability_mission_planner.py", "artifacts/brain_host/generic_developer_executor.py", "capability_mission_planner.py", "generic_developer_executor.py", ) def _compute_sha256(path: str) -> Optional[str]: try: if not os.path.isfile(path): return None with open(path, "rb") as f: return hashlib.sha256(f.read()).hexdigest().upper() except Exception: return None def _is_secure_path_within(p: Path, allowed: List[Path]) -> bool: """ Validates physical path integrity: 1. Rejects path traversal (..) in raw path and resolved path 2. Rejects symlinks and Windows reparse points / junctions 3. Verifies path resolves strictly within the pre-authorized trusted roots """ try: raw_str = str(p).replace("\\", "/") if ".." in raw_str.split("/"): return False if p.is_symlink(): return False st = os.lstat(str(p)) if hasattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT") and (st.st_file_attributes & stat.FILE_ATTRIBUTE_REPARSE_POINT): return False resolved = p.resolve() return any(resolved == root or root in resolved.parents for root in allowed) except Exception: return False def compute_failure_fingerprint(snapshot: MissionFailureSnapshot) -> str: """ Computes a deterministic SHA256-based fingerprint from stable mission failure attributes. Does NOT include variable timestamps. """ m_id = str(snapshot.mission_id or "").strip() inst_id = str(snapshot.instance_id or "").strip() f_stage = str(snapshot.failure_stage or "").strip().lower() s_comp = str(snapshot.source_component or "").strip().lower() # Normalize symptom/error: collapse whitespace, lowercase symptom = " ".join(str(snapshot.observed_symptom or snapshot.exception_message or "").split()).lower() # Extract target if available in action_history or error_context target = "" for act in reversed(snapshot.action_history): if isinstance(act, dict) and act.get("target"): target = str(act["target"]).strip().replace("\\", "/").lower() break if not target and isinstance(snapshot.error_context, dict): sp = snapshot.error_context.get("suggested_patch") if isinstance(sp, dict) and sp.get("file_rel_path"): target = str(sp["file_rel_path"]).strip().replace("\\", "/").lower() elif snapshot.error_context.get("file_rel_path"): target = str(snapshot.error_context["file_rel_path"]).strip().replace("\\", "/").lower() stable_payload = f"{m_id}:{inst_id}:{f_stage}:{s_comp}:{symptom}:{target}" return hashlib.sha256(stable_payload.encode("utf-8")).hexdigest()[:16] def is_frozen_baseline_target( snapshot: MissionFailureSnapshot, explicit_patch: Optional[Dict[str, Any]] = None, ) -> bool: """ Checks if any targeted file belongs to the protected frozen baseline. """ targets_to_check: List[str] = [] if explicit_patch: targets_to_check.append(str(explicit_patch.get("file_rel_path") or explicit_patch.get("file") or "")) if snapshot.error_context: ec = snapshot.error_context sp = ec.get("suggested_patch") if isinstance(sp, dict): targets_to_check.append(str(sp.get("file_rel_path") or sp.get("file") or "")) targets_to_check.append(str(ec.get("file_rel_path") or ec.get("file") or "")) for act in snapshot.action_history: if isinstance(act, dict) and act.get("target"): targets_to_check.append(str(act["target"])) for src in snapshot.source_files: targets_to_check.append(str(src)) for t in targets_to_check: if not t: continue t_norm = os.path.normpath(t).replace("\\", "/") for fb in FROZEN_BASELINES: if t_norm == fb or t_norm.endswith(fb): return True return False class ShadowSelfRepairOrchestrator: """ Non-invasive, fail-safe observer orchestrating shadow self-repair cycles. Guarantees absolute isolation: - Never mutates mission state or results - Never writes to production - Never auto-promotes - Captures and isolates all internal exceptions so missions remain intact. """ SELF_REPAIR_MODE: str = "SHADOW" PRODUCTION_WRITE_AUTHORITY: bool = False CORE_WRITE_AUTHORITY: bool = False AUTO_PROMOTION: bool = False MISSION_STATE_MUTATION: bool = False MISSION_RESULT_MUTATION: bool = False MISSION_EXECUTION_DEPENDS_ON_SELF_REPAIR: bool = False MISSION_COMPLETION_DEPENDS_ON_SELF_REPAIR: bool = False MISSION_VERIFICATION_DEPENDS_ON_SELF_REPAIR: bool = False MAX_REPAIR_ATTEMPTS: int = 1 SELF_REPAIR_PROVIDER: Optional[Any] = None PROVIDER_CALLED: bool = False NETWORK_USED: bool = False def __init__( self, base_reports_dir: str = "reports/self_repair", project_root: str = ".", engine: Optional[SelfRepairEngine] = None, max_repair_attempts: int = 1, resume_guard: Optional[MissionResumeGuard] = None, resume_executor: Optional[MissionResumeExecutor] = None, promotion_guard: Optional[PromotionGuard] = None, promotion_executor: Optional[PromotionExecutor] = None, allow_auto_promotion: bool = False, ): self.base_reports_dir = os.path.abspath(base_reports_dir) self.project_root = os.path.abspath(project_root) self.engine = engine or SelfRepairEngine( base_reports_dir=self.base_reports_dir, project_root=self.project_root, ) self.max_repair_attempts = max_repair_attempts self.allow_auto_promotion = allow_auto_promotion self._seen_fingerprints: Dict[str, ShadowRepairReport] = {} self._attempts_count: Dict[str, int] = {} self._circuit_states: Dict[str, str] = {} self._first_reports: Dict[str, ShadowRepairReport] = {} self._is_handling: bool = False # Phase 5A Promotion Pipeline self.promotion_guard = promotion_guard or PromotionGuard( workspace_root=self.project_root, ) self.promotion_executor = promotion_executor or PromotionExecutor( guard=self.promotion_guard, ) # Phase 5B Live Integration Circuit: Orchestrator -> Resume Guard -> Resume Executor self.resume_guard = resume_guard or MissionResumeGuard( max_repair_attempts=self.max_repair_attempts, max_resume_attempts=1, ) self.resume_executor = resume_executor or MissionResumeExecutor( guard=self.resume_guard, evidence_dir=self.base_reports_dir, ) self._resume_results: Dict[str, MissionResumeResult] = {} self.last_promotion_result: Optional[PromotionExecutionResult] = None def compute_failure_fingerprint(self, snapshot: MissionFailureSnapshot) -> str: return compute_failure_fingerprint(snapshot) def evaluate_mission_resume( self, original_mission: Any, repair_result: Any, failure_fingerprint: str = "", attempt_number: int = 1, resumed_mission: Optional[Any] = None, extra_details: Optional[Dict[str, Any]] = None, ) -> ResumeDecision: """ Evaluates mission resumption gates strictly fail-closed via MissionResumeGuard. Enforces: - Exact identity match (mission_id, task_id, client_task_id, instance_id, workspace_root) - Repair result validity (auto_promotion, promotion_verified, rollback_ready, non-critical, confidence=HIGH) - Loop guard & attempt budgets (MAX_REPAIR_ATTEMPTS=1, MAX_RESUME_ATTEMPTS=1) - Anti-duplicate resume - Provider/network safety """ if isinstance(original_mission, MissionFailureSnapshot): fp = failure_fingerprint or compute_failure_fingerprint(original_mission) else: sn = self.create_snapshot(original_mission) fp = failure_fingerprint or compute_failure_fingerprint(sn) return self.resume_guard.build_resume_decision( original_mission=original_mission, resumed_mission=resumed_mission if resumed_mission is not None else original_mission, repair_result=repair_result, failure_fingerprint=fp, attempt_number=attempt_number, recorded_resumes=self.resume_executor.executed_resumes, extra_details=extra_details, ) # Canonical aliases evaluate_resume_decision = evaluate_mission_resume evaluate_resume = evaluate_mission_resume def resume_mission( self, original_mission: Any, repair_result: Any, final_verifier_callable: Optional[Callable[[str], bool]] = None, failure_fingerprint: str = "", attempt_number: int = 1, current_mission_state: Optional[Any] = None, mission_runner_callable: Optional[Callable[[str], bool]] = None, ) -> MissionResumeResult: """ Safely attempts mission resumption for the exact same mission. Enforces: Resume Executor CANNOT activate without Resume Guard authorization. """ if isinstance(original_mission, MissionFailureSnapshot): fp = failure_fingerprint or compute_failure_fingerprint(original_mission) m_id = str(original_mission.mission_id) i_id = str(original_mission.instance_id or "") else: sn = self.create_snapshot(original_mission) fp = failure_fingerprint or compute_failure_fingerprint(sn) m_id = str(original_mission.get("mission_id") or sn.mission_id) i_id = str(original_mission.get("instance_id") or sn.instance_id or "") # Gate check: evaluate resume eligibility via self.resume_guard decision = self.evaluate_mission_resume( original_mission=original_mission, repair_result=repair_result, failure_fingerprint=fp, attempt_number=attempt_number, resumed_mission=current_mission_state, ) if not decision.resume_allowed: logger.warning( "[ShadowOrchestrator] Resume blocked by Resume Guard: " f"reason={decision.reason_code} mission_id={m_id}" ) # ResumeExecutor.resume_mission enforces the guard decision fail-closed res = self.resume_executor.resume_mission( original_mission=original_mission, repair_result=repair_result, final_verifier_callable=final_verifier_callable if decision.resume_allowed else None, failure_fingerprint=fp, attempt_number=attempt_number, current_mission_state=current_mission_state, mission_runner_callable=mission_runner_callable if decision.resume_allowed else None, ) dup_key = f"{m_id}:{i_id}:{fp}" self._resume_results[dup_key] = res return res # Canonical aliases execute_mission_resume = resume_mission execute_resume = resume_mission def is_frozen_baseline_target( self, snapshot: MissionFailureSnapshot, explicit_patch: Optional[Dict[str, Any]] = None, ) -> bool: return is_frozen_baseline_target(snapshot, explicit_patch) def get_circuit_state( self, snapshot_or_key: Union[MissionFailureSnapshot, Dict[str, Any], str], ) -> str: if isinstance(snapshot_or_key, str): k = snapshot_or_key else: sn = self.create_snapshot(snapshot_or_key) fp = self.compute_failure_fingerprint(sn) k = f"{sn.mission_id}:{sn.instance_id or ''}:{fp}:{sn.source_component or ''}" return self._circuit_states.get(k, "CLOSED") def reconcile_telemetry( self, snapshot: MissionFailureSnapshot, repair_res: Optional[Dict[str, Any]] = None, ) -> Tuple[Optional[bool], Optional[bool], Optional[str]]: """ Reconciles provider and network telemetry across snapshot, error_context, and repair result. Returns: (prov_val, net_val, conflict_or_error_reason) where conflict_or_error_reason is one of: "CONTRADICTORY_TELEMETRY", "UNKNOWN_TELEMETRY", "PROVIDER_OR_NETWORK_CALLED", or None. """ res = repair_res or {} prov_sources: List[Any] = [] net_sources: List[Any] = [] if "provider_called" in res and res.get("provider_called") is not None: prov_sources.append(res.get("provider_called")) if snapshot.error_context and "provider_called" in snapshot.error_context and snapshot.error_context.get("provider_called") is not None: prov_sources.append(snapshot.error_context.get("provider_called")) if getattr(snapshot, "provider_called", None) is not None: prov_sources.append(getattr(snapshot, "provider_called", None)) if "network_used" in res and res.get("network_used") is not None: net_sources.append(res.get("network_used")) if snapshot.error_context and "network_used" in snapshot.error_context and snapshot.error_context.get("network_used") is not None: net_sources.append(snapshot.error_context.get("network_used")) if getattr(snapshot, "network_used", None) is not None: net_sources.append(getattr(snapshot, "network_used", None)) # Validate types: reject strings, numbers, non-booleans has_invalid_prov_type = any(type(x) is not bool for x in prov_sources) has_invalid_net_type = any(type(x) is not bool for x in net_sources) # Check for contradictory telemetry has_prov_conflict = (True in prov_sources) and (False in prov_sources) has_net_conflict = (True in net_sources) and (False in net_sources) tele_reason: Optional[str] = None if has_prov_conflict or has_net_conflict: prov_val = True if (has_prov_conflict and True in prov_sources) else None net_val = True if (has_net_conflict and True in net_sources) else None tele_reason = "CONTRADICTORY_TELEMETRY" elif has_invalid_prov_type or has_invalid_net_type: prov_val = None net_val = None tele_reason = "UNKNOWN_TELEMETRY" else: prov_val = True if (True in prov_sources) else (False if (prov_sources and all(x is False for x in prov_sources)) else None) net_val = True if (True in net_sources) else (False if (net_sources and all(x is False for x in net_sources)) else None) if not tele_reason: if prov_val is True or net_val is True: tele_reason = "PROVIDER_OR_NETWORK_CALLED" elif (prov_val is None) or (net_val is None): tele_reason = "UNKNOWN_TELEMETRY" elif (prov_val is not False) or (net_val is not False): tele_reason = "UNKNOWN_TELEMETRY" return prov_val, net_val, tele_reason def is_self_repair_recursion( self, snapshot: MissionFailureSnapshot, ) -> Tuple[bool, Optional[str]]: """ Detects if failure originated from a self-repair component or active in-flight execution. When `allow_isolated_component_repair` is explicitly set in `snapshot.error_context` for an isolated lab copy (and `self._is_handling` is False), permits non-reentrant repair of non-critical self-repair leaf modules (e.g. circuit.py, evidence_comparator.py). """ if self._is_handling: return True, "in_flight_reentrancy" allow_isolated = bool( isinstance(snapshot.error_context, dict) and snapshot.error_context.get("allow_isolated_component_repair") is True ) self_repair_components = ( "shadowselfrepairorchestrator", "selfrepairengine", "autonomouspatchreasoner", "sandboxbuilder", "patchexecutor", "candidateverifier", "candidatejudge", "remediation", "self_repair", ) comp = str(snapshot.source_component or "").strip().lower() stage = str(snapshot.failure_stage or "").strip().lower() for internal in self_repair_components: if internal in comp or internal in stage: if allow_isolated and internal in ("remediation", "self_repair"): continue return True, snapshot.source_component or snapshot.failure_stage if not allow_isolated: for tb in snapshot.tracebacks: tb_low = tb.lower() if "app/remediation/self_repair" in tb_low or "app\\remediation\\self_repair" in tb_low: return True, "traceback:app.remediation.self_repair" return False, None def create_snapshot( self, mission_data: Union[MissionFailureSnapshot, Dict[str, Any]], ) -> MissionFailureSnapshot: """ Creates an immutable MissionFailureSnapshot from mission execution output. """ if isinstance(mission_data, MissionFailureSnapshot): return mission_data m_id = str(mission_data.get("mission_id") or _new_id("mis")) inst_id = mission_data.get("instance_id") ws = str(mission_data.get("workspace_root") or self.project_root) req = str(mission_data.get("creator_request") or mission_data.get("prompt") or "") status = str(mission_data.get("mission_status") or mission_data.get("status") or "FAILED").strip().upper() f_stage = str(mission_data.get("failure_stage") or mission_data.get("source_component") or "unknown").strip() src_comp = str(mission_data.get("source_component") or f_stage or "unknown").strip() symptom = str( mission_data.get("observed_symptom") or mission_data.get("error") or mission_data.get("exception_message") or "" ).strip() timestamps = dict(mission_data.get("timestamps") or {}) if "snapshot_created" not in timestamps: timestamps["snapshot_created"] = _utc_iso() return MissionFailureSnapshot( mission_id=m_id, instance_id=inst_id, workspace_root=ws, creator_request=req, mission_status=status, failure_stage=f_stage, source_component=src_comp, planner_result=mission_data.get("planner_result"), executor_result=mission_data.get("executor_result"), effector_result=mission_data.get("effector_result"), verification_result=mission_data.get("verification_result"), final_mission_state=dict(mission_data.get("final_mission_state") or {}), action_history=list(mission_data.get("action_history") or mission_data.get("execution_history") or []), execution_evidence=dict(mission_data.get("execution_evidence") or {}), logs=list(mission_data.get("logs") or []), tracebacks=list(mission_data.get("tracebacks") or ([str(mission_data["traceback"])] if mission_data.get("traceback") else [])), source_files=list(mission_data.get("source_files") or []), timestamps=timestamps, observed_symptom=symptom, exception_type=mission_data.get("exception_type"), exception_message=mission_data.get("exception_message") or symptom, error_context=dict(mission_data.get("error_context") or {}), ) def evaluate_trigger( self, snapshot: MissionFailureSnapshot, ) -> Tuple[bool, bool, str]: """ Determines if shadow self-repair should activate. Returns: (should_trigger, is_false_success, reason) """ status = snapshot.mission_status.strip().upper() # Check for FALSE_SUCCESS contradiction: # Mission reported success, but physical verification failed or contradicts verif = snapshot.verification_result if isinstance(verif, dict): verif_failed = ( verif.get("passed") is False or str(verif.get("status") or "").upper() in ("FAIL", "FAILED", "ERROR") or bool(verif.get("failures")) or bool(verif.get("errors")) ) if verif_failed and status in ("SUCCESS", "COMPLETED_AND_E2E_VERIFIED", "COMPLETED"): return True, True, f"Triggered by FALSE_SUCCESS: Mission status '{status}' contradicted by failing verification." # Non-trigger statuses if status in ("SUCCESS", "COMPLETED_AND_E2E_VERIFIED", "USER_CANCELLED", "EXPECTED_BLOCKER"): return False, False, f"Status '{status}' does not trigger shadow repair." # Standard failure triggers if status in FailureTriggerGuard.TRIGGER_STATUSES or any(k in status for k in ("FAIL", "ERROR", "CONTRADICTION", "BLOCKER")): return True, False, f"Triggered by failure status '{status}'." return False, False, f"Status '{status}' not in trigger set." def handle_mission_finished( self, mission_input: Union[MissionFailureSnapshot, 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, promotion_result: Optional[Any] = None, final_verifier_callable: Optional[Callable[[str], bool]] = None, execute_resume_if_eligible: bool = False, allow_auto_promotion: Optional[bool] = None, mission_runner_callable: Optional[Callable[[str], bool]] = None, ) -> ShadowRepairReport: """ Post-mission shadow entrypoint. Enforces complete isolation using try/except: Self-repair failures NEVER compromise or mutate the main mission result. Enforces loop guard: max attempts per signature, circuit breaker, and no self-recursion. """ # 1. Capture Immutable Snapshot snapshot = self.create_snapshot(mission_input) # 2. Compute Deterministic Failure Fingerprint & Guard Keys fingerprint = compute_failure_fingerprint(snapshot) guard_key = f"{snapshot.mission_id}:{snapshot.instance_id or ''}:{fingerprint}:{snapshot.source_component or ''}" dup_key = f"{snapshot.mission_id}:{snapshot.instance_id or ''}:{fingerprint}" # 3. Recursion Guard: detect if failure originated from Self-Repair is_recursive, rec_source = self.is_self_repair_recursion(snapshot) if is_recursive: self._circuit_states[guard_key] = "OPEN" return ShadowRepairReport( shadow_run_id=_new_id("shd"), mission_id=snapshot.mission_id, instance_id=snapshot.instance_id, incident_id=None, original_mission_status=snapshot.mission_status, failure_fingerprint=fingerprint, shadow_status=ShadowStatus.SHADOW_INTERNAL_ERROR.value, shadow_mode=self.SELF_REPAIR_MODE, shadow_triggered=True, trigger_reason=f"SELF_REPAIR_INTERNAL_FAILURE: Failure generated internally by {rec_source}.", false_success_detected=False, candidate_repair="ERROR", sandbox_executed=False, production_modified=False, production_hash_changed=False, auto_promotion=False, creator_approval_required=True, mission_intact=True, error_message=f"Self-repair recursion blocked: {rec_source}", source_component=snapshot.source_component, attempt_number=0, max_attempts=self.max_repair_attempts, guard_decision="BLOCK", guard_reason=f"RECURSIVE_REPAIR_STARTED=False. Failure origin '{rec_source}' is internal self-repair.", recursive_source=rec_source, circuit_state="OPEN", repair_attempt_blocked=True, self_repair_internal_failure=True, recursive_repair_started=False, new_repair_started=False, provider_called=False, network_used=False, ) # 4. Evaluate Trigger should_trigger, is_false_success, reason = self.evaluate_trigger(snapshot) if not should_trigger: report = ShadowRepairReport( shadow_run_id=_new_id("shd"), mission_id=snapshot.mission_id, instance_id=snapshot.instance_id, original_mission_status=snapshot.mission_status, failure_fingerprint=fingerprint, shadow_status=ShadowStatus.NOT_TRIGGERED.value, shadow_mode=self.SELF_REPAIR_MODE, shadow_triggered=False, trigger_reason=reason, false_success_detected=is_false_success, candidate_repair="SKIPPED", mission_intact=True, new_repair_started=False, duplicate_shadow_event=False, source_component=snapshot.source_component, attempt_number=0, max_attempts=self.max_repair_attempts, guard_decision="ALLOW", guard_reason="No failure trigger condition met.", circuit_state=self._circuit_states.get(guard_key, "CLOSED"), repair_attempt_blocked=False, provider_called=False, network_used=False, ) return report # 5. Circuit Breaker & Max Attempts / Duplicate Guard Check current_attempts = self._attempts_count.get(guard_key, 0) current_circuit = self._circuit_states.get(guard_key, "CLOSED") if current_circuit == "OPEN" or current_attempts >= self.max_repair_attempts or dup_key in self._seen_fingerprints: first_report = self._first_reports.get(guard_key) or self._seen_fingerprints.get(dup_key) first_inc_id = first_report.incident_id if first_report else None cand_rep = first_report.candidate_repair if first_report else "SKIPPED" report = ShadowRepairReport( shadow_run_id=_new_id("shd"), mission_id=snapshot.mission_id, instance_id=snapshot.instance_id, incident_id=first_inc_id, original_mission_status=snapshot.mission_status, failure_fingerprint=fingerprint, shadow_status=ShadowStatus.DUPLICATE_EVENT.value, shadow_mode=self.SELF_REPAIR_MODE, shadow_triggered=True, trigger_reason=f"REPAIR_ATTEMPT_BLOCKED: Max attempts ({self.max_repair_attempts}) reached or circuit breaker OPEN for key '{guard_key}'.", false_success_detected=is_false_success, candidate_repair=cand_rep, sandbox_executed=False, production_modified=False, production_hash_changed=False, auto_promotion=False, creator_approval_required=True, mission_intact=True, duplicate_shadow_event=True, new_repair_started=False, repair_attempt_blocked=True, source_component=snapshot.source_component, attempt_number=current_attempts + 1, max_attempts=self.max_repair_attempts, guard_decision="BLOCK", guard_reason=f"REPAIR_ATTEMPT_BLOCKED=True. Circuit breaker {current_circuit}; attempts {current_attempts}>={self.max_repair_attempts}.", circuit_state="OPEN", provider_called=False, network_used=False, ) return report # 6. Frozen Baseline Target Protection if is_frozen_baseline_target(snapshot, explicit_patch_instruction): self._circuit_states[guard_key] = "OPEN" self._attempts_count[guard_key] = current_attempts + 1 report = ShadowRepairReport( shadow_run_id=_new_id("shd"), mission_id=snapshot.mission_id, instance_id=snapshot.instance_id, original_mission_status=snapshot.mission_status, failure_fingerprint=fingerprint, shadow_status=ShadowStatus.FROZEN_BASELINE_BLOCKED.value, shadow_mode=self.SELF_REPAIR_MODE, shadow_triggered=True, trigger_reason="Target file is a frozen baseline. Automated modification strictly forbidden.", false_success_detected=is_false_success, candidate_repair="REJECT", sandbox_executed=False, production_modified=False, production_hash_changed=False, auto_promotion=False, creator_approval_required=True, mission_intact=True, frozen_baseline_blocked=True, new_repair_started=False, repair_attempt_blocked=True, source_component=snapshot.source_component, attempt_number=1, max_attempts=self.max_repair_attempts, guard_decision="BLOCK", guard_reason="Target file is a protected baseline.", circuit_state="OPEN", provider_called=False, network_used=False, ) self._seen_fingerprints[dup_key] = report self._first_reports[guard_key] = report return report # Collect target file paths to monitor hash immutability target_paths_to_monitor: List[str] = [] ws_root = os.path.abspath(snapshot.workspace_root or self.project_root) for act in snapshot.action_history: if isinstance(act, dict) and act.get("target"): tp = os.path.abspath(os.path.join(ws_root, str(act["target"]))) if os.path.isfile(tp): target_paths_to_monitor.append(tp) if explicit_patch_instruction: ep = explicit_patch_instruction.get("file_rel_path") or explicit_patch_instruction.get("file") if ep: tp = os.path.abspath(os.path.join(ws_root, str(ep))) if os.path.isfile(tp): target_paths_to_monitor.append(tp) hashes_before = {tp: _compute_sha256(tp) for tp in target_paths_to_monitor} # 7. Fail-safe Shadow Execution (Mark active handling to prevent recursion) self._is_handling = True self._attempts_count[guard_key] = 1 try: effective_status = "FALSE_SUCCESS" if is_false_success else snapshot.mission_status # FIX 1 & 2: Extract explicit target_file and target_files from mission_input or snapshot t_file = None t_files = None if isinstance(mission_input, dict): t_file = mission_input.get("target_file") or mission_input.get("target_artifact") t_files = mission_input.get("target_files") or mission_input.get("target_artifacts") if not t_file and isinstance(snapshot.final_mission_state, dict): t_file = snapshot.final_mission_state.get("target_file") or snapshot.final_mission_state.get("target_artifact") if not t_files: t_files = snapshot.final_mission_state.get("target_files") if not t_file and isinstance(snapshot.error_context, dict): t_file = snapshot.error_context.get("target_file") or snapshot.error_context.get("target_artifact") if not t_files: t_files = snapshot.error_context.get("target_files") if not t_file and isinstance(snapshot.verification_result, dict): t_file = snapshot.verification_result.get("target_file") or snapshot.verification_result.get("target") # Extract verification text / assertion details if present verif_reason = "" if isinstance(snapshot.verification_result, dict): verif_reason = str( snapshot.verification_result.get("verification_reason") or snapshot.verification_result.get("reason") or "" ).strip() # Build enriched observed_symptom including verification evidence symptom_parts = [] if snapshot.observed_symptom: symptom_parts.append(str(snapshot.observed_symptom).strip()) if verif_reason and verif_reason not in symptom_parts: symptom_parts.append(verif_reason) if isinstance(snapshot.verification_result, dict): failures = snapshot.verification_result.get("failures") or [] for f in failures: f_str = str(f).strip() if f_str and f_str not in symptom_parts: symptom_parts.append(f_str) errors = snapshot.verification_result.get("errors") or [] for e in errors: e_str = str(e).strip() if e_str and e_str not in symptom_parts: symptom_parts.append(e_str) combined_symptom = " | ".join(symptom_parts) if symptom_parts else ( "Verification failed despite success status" if is_false_success else "Mission failure" ) # Build enriched error_context carrying explicit targets and verification details err_ctx = dict(snapshot.error_context or {}) if t_file: err_ctx["target_file"] = t_file if t_files: err_ctx["target_files"] = t_files if snapshot.creator_request: err_ctx["creator_request"] = snapshot.creator_request if fingerprint: err_ctx["failure_fingerprint"] = fingerprint if verif_reason: err_ctx["verification_reason"] = verif_reason mission_payload: Dict[str, Any] = { "mission_id": snapshot.mission_id, "instance_id": snapshot.instance_id, "workspace_root": snapshot.workspace_root, "creator_request": snapshot.creator_request, "failure_fingerprint": fingerprint, "target_file": t_file, "target_files": t_files, "mission_status": effective_status, "has_explicit_contradiction": is_false_success, "failure_stage": snapshot.failure_stage, "source_component": snapshot.source_component, "observed_symptom": combined_symptom, "exception_type": snapshot.exception_type or ("FalseSuccessContradiction" if is_false_success else None), "exception_message": combined_symptom if is_false_success else snapshot.exception_message, "execution_history": snapshot.action_history, "verification_result": snapshot.verification_result, "verification_reason": verif_reason, "error_context": err_ctx, "logs": snapshot.logs, "tracebacks": snapshot.tracebacks, "source_files": snapshot.source_files, "mission_state": snapshot.final_mission_state, "focused_test": err_ctx.get("focused_test"), "focused_tests": err_ctx.get("focused_tests"), "focused_test_command": err_ctx.get("focused_test_command"), "regression_tests": err_ctx.get("regression_tests"), "verification_dependencies": err_ctx.get("verification_dependencies"), } res = self.engine.auto_remediate_mission( mission_data=mission_payload, focused_test_runner=focused_test_runner, regression_test_runner=regression_test_runner, explicit_patch_instruction=explicit_patch_instruction, persist_artifacts=persist_artifacts, ) # Mark circuit state OPEN after attempt is completed self._circuit_states[guard_key] = "OPEN" # Check target files after execution hashes_after = {tp: _compute_sha256(tp) for tp in target_paths_to_monitor} prod_hash_changed = any(hashes_before.get(tp) != hashes_after.get(tp) for tp in target_paths_to_monitor) prod_modified = prod_hash_changed # Extract results inc_id = res.get("incident").incident_id if res.get("incident") else None rc_name = res.get("root_cause").root_cause if res.get("root_cause") else None rc_proven = bool(res.get("root_cause_proven")) auto_patch = bool(res.get("autonomous_patch_generated")) cand_repair = str(res.get("candidate_repair") or "NOT_STARTED") manifest = res.get("sandbox_manifest") sb_dir = manifest.sandbox_root if manifest else None frozen_blocked = bool(res.get("frozen_baseline_blocked") or res.get("status") == "FROZEN_BASELINE_BLOCKED") # Determine canonical ShadowStatus if frozen_blocked: sh_status = ShadowStatus.FROZEN_BASELINE_BLOCKED.value elif not rc_proven or res.get("status") == "UNPROVEN": sh_status = ShadowStatus.UNPROVEN.value cand_repair = "SKIPPED" elif not auto_patch: sh_status = ShadowStatus.PATCH_NOT_GENERATED.value cand_repair = "SKIPPED" elif cand_repair == "REJECT": sh_status = ShadowStatus.SANDBOX_REJECTED.value elif cand_repair == "ACCEPT": sh_status = ShadowStatus.SANDBOX_ACCEPTED.value else: sh_status = ShadowStatus.OBSERVED.value persisted: List[str] = [] if persist_artifacts and inc_id: inc_dir = os.path.join(self.base_reports_dir, inc_id) if os.path.isdir(inc_dir): persisted = sorted(os.listdir(inc_dir)) # Phase 5B Live Integration: Evaluate promotion and resume effective_repair_result = promotion_result if effective_repair_result is None: cand_verdict_obj = res.get("candidate_verdict") verif_obj = res.get("verification_result") comp_obj = res.get("comparison_result") patch_res_obj = res.get("patch_result") manifest_obj = res.get("sandbox_manifest") promo_proposal_obj = res.get("promotion_proposal") patch_plan_obj = res.get("patch_plan") ws_root = snapshot.workspace_root or self.project_root effective_allow_promo = self.allow_auto_promotion if allow_auto_promotion is None else bool(allow_auto_promotion) # Extract patch target target_file = None if patch_plan_obj and getattr(patch_plan_obj, "target_file", None): target_file = patch_plan_obj.target_file elif patch_plan_obj and getattr(patch_plan_obj, "file_rel_path", None): target_file = patch_plan_obj.file_rel_path elif patch_plan_obj and getattr(patch_plan_obj, "instructions", None) and len(patch_plan_obj.instructions) > 0: target_file = getattr(patch_plan_obj.instructions[0], "file_rel_path", None) elif promo_proposal_obj and getattr(promo_proposal_obj, "files_changed", None) and len(promo_proposal_obj.files_changed) > 0: target_file = promo_proposal_obj.files_changed[0] elif promo_proposal_obj and getattr(promo_proposal_obj, "files_to_promote", None) and len(promo_proposal_obj.files_to_promote) > 0: target_file = promo_proposal_obj.files_to_promote[0] target_file_str = str(target_file or "unknown") # Evaluate Fail-Closed Evidence Contract evidence_block_reason: Optional[str] = None # 1. Structured CandidateVerdict required (string 'ACCEPT' strictly forbidden) if not isinstance(cand_verdict_obj, CandidateVerdict): evidence_block_reason = "MISSING_OR_INVALID_CANDIDATE_VERDICT" # 2. Complete structured evidence required elif not isinstance(verif_obj, VerificationResult): evidence_block_reason = "MISSING_OR_INVALID_VERIFICATION_RESULT" elif not isinstance(comp_obj, ComparisonResult): evidence_block_reason = "MISSING_OR_INVALID_COMPARISON_RESULT" elif not isinstance(patch_res_obj, PatchResult): evidence_block_reason = "MISSING_OR_INVALID_PATCH_RESULT" elif not isinstance(manifest_obj, SandboxManifest): evidence_block_reason = "MISSING_OR_INVALID_SANDBOX_MANIFEST" elif not isinstance(promo_proposal_obj, PromotionProposal): evidence_block_reason = "MISSING_OR_INVALID_PROMOTION_PROPOSAL" elif not target_file: evidence_block_reason = "TARGET_FILE_UNRESOLVED" # 3. Consistency of incident_id, repair_id, candidate_id if not evidence_block_reason: inc_ids = { cid for cid in ( inc_id, cand_verdict_obj.incident_id, verif_obj.incident_id, comp_obj.incident_id, patch_res_obj.incident_id, manifest_obj.incident_id, promo_proposal_obj.incident_id, ) if cid } if len(inc_ids) > 1: evidence_block_reason = f"IDENTIFIER_MISMATCH: incident_id mismatch ({inc_ids})" rep_ids = { rid for rid in ( cand_verdict_obj.repair_id, verif_obj.repair_id, comp_obj.repair_id, patch_res_obj.repair_id, manifest_obj.repair_id, promo_proposal_obj.repair_id, ) if rid } if len(rep_ids) > 1: evidence_block_reason = f"IDENTIFIER_MISMATCH: repair_id mismatch ({rep_ids})" cand_ids = { did for did in ( cand_verdict_obj.candidate_id, verif_obj.candidate_id, comp_obj.candidate_id, patch_res_obj.candidate_id, manifest_obj.candidate_id, promo_proposal_obj.candidate_id, ) if did } if len(cand_ids) > 1: evidence_block_reason = f"IDENTIFIER_MISMATCH: candidate_id mismatch ({cand_ids})" # 4. Confirmation of root cause, compilation, tests, zero regressions, and baseline integrity if not evidence_block_reason: if not (rc_proven and cand_verdict_obj.root_cause_proven): evidence_block_reason = "ROOT_CAUSE_NOT_PROVEN" elif cand_verdict_obj.verdict != "ACCEPT": evidence_block_reason = f"CANDIDATE_VERDICT_NOT_ACCEPTED: {cand_verdict_obj.verdict}" elif promo_proposal_obj.candidate_verdict != "ACCEPT": evidence_block_reason = f"PROMOTION_PROPOSAL_VERDICT_NOT_ACCEPTED: {promo_proposal_obj.candidate_verdict}" elif not cand_verdict_obj.patch_applied or patch_res_obj.status != "SUCCESS" or patch_res_obj.failed_instruction is not None: evidence_block_reason = "PATCH_NOT_APPLIED_SUCCESSFULLY" elif not (cand_verdict_obj.compile_pass and verif_obj.compile_pass and len(verif_obj.compile_errors) == 0): evidence_block_reason = "COMPILE_CHECK_FAILED" elif not (cand_verdict_obj.focused_tests_pass and verif_obj.focused_tests_pass): evidence_block_reason = "FOCUSED_TESTS_FAILED" elif not (cand_verdict_obj.regression_tests_pass and verif_obj.regression_tests_pass): evidence_block_reason = "REGRESSION_TESTS_FAILED" elif cand_verdict_obj.new_failures > 0 or comp_obj.new_failures > 0 or len(comp_obj.regressions) > 0: evidence_block_reason = "EVIDENCE_COMPARISON_NEW_FAILURES_DETECTED" elif not (cand_verdict_obj.production_untouched and cand_verdict_obj.baseline_integrity): evidence_block_reason = "SANDBOX_INTEGRITY_VIOLATION" # 5. Check Creator authorization if not evidence_block_reason and not effective_allow_promo: evidence_block_reason = "PROMOTION_NOT_EXPLICITLY_AUTHORIZED" # 6. Check target critical and within workspace if not evidence_block_reason: is_crit = self.promotion_guard.is_target_critical(target_file, ws_root) is_in_ws = self.promotion_guard.is_within_workspace(target_file, ws_root) if is_crit: evidence_block_reason = "TARGET_IS_CRITICAL" elif not is_in_ws: evidence_block_reason = "TARGET_OUTSIDE_WORKSPACE" # 7. Check confidence (must be HIGH, no optimistic assumptions) rc_obj = res.get("root_cause") reasoning_obj = res.get("patch_reasoning_result") actual_confidence = getattr(rc_obj, "confidence", None) or getattr(reasoning_obj, "confidence", None) if not evidence_block_reason: if not actual_confidence or str(actual_confidence).strip().upper() != "HIGH": evidence_block_reason = f"CONFIDENCE_NOT_HIGH: {actual_confidence}" # 8. Check provider and network telemetry (strict zero external calls and proven telemetry) prov_val, net_val, tele_reason = self.reconcile_telemetry(snapshot, res) if not evidence_block_reason and tele_reason: evidence_block_reason = tele_reason # 9. Check loop guard and attempt budgets attempt_count = self._attempts_count.get(guard_key, 1) is_dup = bool(dup_key in self._seen_fingerprints or attempt_count > self.max_repair_attempts) if not evidence_block_reason and (attempt_count > self.max_repair_attempts or is_dup): evidence_block_reason = "LOOP_GUARD_BLOCKED" # 9b. Check workspace boundary consistency with project root if not evidence_block_reason and self.project_root and snapshot.workspace_root: try: p_proj = Path(self.project_root).resolve() p_ws = Path(snapshot.workspace_root).resolve() if p_proj != p_ws and p_proj not in p_ws.parents and p_ws not in p_proj.parents: evidence_block_reason = "WORKSPACE_MISMATCH" except Exception: evidence_block_reason = "WORKSPACE_MISMATCH" # 10. Check real rollback readiness (physical file, backup manifest, matching baseline hash, physical backup bytes) if not evidence_block_reason: ws_p = Path(ws_root).resolve() target_p = Path(target_file) if not target_p.is_absolute(): target_p = (ws_p / target_p).resolve() else: target_p = target_p.resolve() target_exists = target_p.is_file() and os.access(target_p, os.R_OK) and not target_p.is_symlink() has_rollback_plan = bool(promo_proposal_obj.rollback_plan and isinstance(promo_proposal_obj.rollback_plan, dict)) # Check traversal in target_file or explicit rollback_plan backup_file raw_tf = str(target_file).replace("\\", "/") if ".." in raw_tf.split("/"): evidence_block_reason = "PATH_TRAVERSAL_DETECTED" if not evidence_block_reason and has_rollback_plan and promo_proposal_obj.rollback_plan: raw_bf = promo_proposal_obj.rollback_plan.get("backup_file") or promo_proposal_obj.rollback_plan.get("backup_path") if raw_bf: raw_bf_str = str(raw_bf).replace("\\", "/") if ".." in raw_bf_str.split("/"): evidence_block_reason = "PATH_TRAVERSAL_DETECTED" elif Path(raw_bf).is_symlink(): evidence_block_reason = "ROLLBACK_NOT_READY" # Trusted roots established strictly independent of manifest_obj: trusted_roots: List[Path] = [ws_p] if self.project_root: pr_p = Path(self.project_root).resolve() if pr_p not in trusted_roots: trusted_roots.append(pr_p) if self.base_reports_dir: br_p = Path(self.base_reports_dir) resolved_br = br_p.resolve() if br_p.is_absolute() else (ws_p / br_p).resolve() if resolved_br not in trusted_roots: trusted_roots.append(resolved_br) # Dedicated container for isolated temporary tests/sandboxes: try: system_temp = Path(tempfile.gettempdir()).resolve() if system_temp in ws_p.parents and ws_p.parent != system_temp: if ws_p.parent not in trusted_roots: trusted_roots.append(ws_p.parent) except Exception: pass # Verify sandbox directories authentic provenance manifest_dirs_ok = False if ( manifest_obj.sandbox_root and os.path.isdir(manifest_obj.sandbox_root) and manifest_obj.source_dir and os.path.isdir(manifest_obj.source_dir) and manifest_obj.candidate_dir and os.path.isdir(manifest_obj.candidate_dir) ): sb_p = Path(manifest_obj.sandbox_root) src_p = Path(manifest_obj.source_dir) cand_p = Path(manifest_obj.candidate_dir) sb_ok = _is_secure_path_within(sb_p, trusted_roots) src_ok = _is_secure_path_within(src_p, trusted_roots) cand_ok = _is_secure_path_within(cand_p, trusted_roots) if sb_ok and src_ok and cand_ok: manifest_dirs_ok = True curr_hash = _compute_sha256(str(target_p)) if target_exists else None norm_tf = os.path.normpath(target_file).replace("\\", "/").lower() expected_before_hash = ( promo_proposal_obj.production_sha256_before.get(target_file) or promo_proposal_obj.production_sha256_before.get(os.path.normpath(target_file).replace("\\", "/")) ) if not expected_before_hash and manifest_obj.source_hashes: for sh in manifest_obj.source_hashes: rel_p = getattr(sh, "relative_path", "") if rel_p and os.path.normpath(rel_p).replace("\\", "/").lower() == norm_tf: expected_before_hash = getattr(sh, "source_sha256", None) break # Physical backup verification: # The backup file MUST physically exist on disk, match baseline, not be symlink, # and be strictly within authorized trusted roots (NEVER self-authorized by manifest) has_source_backup = False backup_candidate_paths: List[Path] = [] if has_rollback_plan and promo_proposal_obj.rollback_plan: bf = promo_proposal_obj.rollback_plan.get("backup_file") or promo_proposal_obj.rollback_plan.get("backup_path") if bf: backup_candidate_paths.append(Path(bf)) if manifest_obj.source_dir: backup_candidate_paths.append(Path(manifest_obj.source_dir) / target_file) if norm_tf != target_file: backup_candidate_paths.append(Path(manifest_obj.source_dir) / norm_tf) for bkp_p in backup_candidate_paths: try: if not _is_secure_path_within(bkp_p, trusted_roots): continue resolved_bkp = bkp_p.resolve() if resolved_bkp.is_file() and os.access(resolved_bkp, os.R_OK): bkp_bytes = resolved_bkp.read_bytes() bkp_sha = hashlib.sha256(bkp_bytes).hexdigest().upper() if expected_before_hash and bkp_sha.lower() == str(expected_before_hash).lower(): has_source_backup = True break elif not expected_before_hash and curr_hash and bkp_sha.lower() == str(curr_hash).lower(): has_source_backup = True break except Exception: continue hash_matches_baseline = bool(curr_hash and expected_before_hash and str(curr_hash).lower() == str(expected_before_hash).lower()) rollback_ready = bool( target_exists and has_rollback_plan and manifest_dirs_ok and has_source_backup and hash_matches_baseline ) if not rollback_ready: evidence_block_reason = "ROLLBACK_NOT_READY" # Execute or Block based on verified contract if not evidence_block_reason: logger.info( f"[ShadowOrchestrator] Executing authorized auto-promotion on verified target: {target_file}" ) # Pass verified real values, NOT optimistic constants effective_repair_result = self.promotion_executor.execute_promotion( target_file=target_file, patch_candidate=patch_plan_obj or promo_proposal_obj, workspace_root=ws_root, confidence=str(actual_confidence), root_cause_proven=True, sandbox_verdict=cand_verdict_obj, focused_tests_pass=bool(cand_verdict_obj.focused_tests_pass and verif_obj.focused_tests_pass), regression_tests_pass=bool(cand_verdict_obj.regression_tests_pass and verif_obj.regression_tests_pass), evidence_comparison_pass=bool(comp_obj.new_failures == 0 and len(comp_obj.regressions) == 0), rollback_ready=True, provider_called=prov_val, network_used=net_val, attempt_number=attempt_count, is_duplicate=is_dup, max_attempts=self.max_repair_attempts, incident_id=inc_id, rollback_plan=promo_proposal_obj.rollback_plan if promo_proposal_obj else None, ) if effective_repair_result.promotion_verified: prod_modified = True prod_hash_changed = True self.AUTO_PROMOTION = True else: logger.warning( "[ShadowOrchestrator] Auto-promotion blocked by evidence contract: reason=%s target=%s", evidence_block_reason, target_file_str, ) is_crit_flag = bool(target_file and self.promotion_guard.is_target_critical(target_file, ws_root)) is_in_ws_flag = bool(target_file and self.promotion_guard.is_within_workspace(target_file, ws_root)) effective_repair_result = PromotionExecutionResult( promotion_id=_new_id("promo_blk"), target_file=target_file_str, decision=PromotionDecision( promotion_allowed=False, decision="BLOCK", reason_code=evidence_block_reason, target_file=target_file_str, target_is_critical=is_crit_flag, outside_workspace_blocked=not is_in_ws_flag, creator_approval_required=(evidence_block_reason == "PROMOTION_NOT_EXPLICITLY_AUTHORIZED"), details={"block_reason": evidence_block_reason, "incident_id": inc_id}, ), auto_promotion=False, promotion_verified=False, production_file_changed=False, rollback_ready=False, error_message=evidence_block_reason, ) resume_decision = self.evaluate_mission_resume( original_mission=snapshot, repair_result=effective_repair_result, failure_fingerprint=fingerprint, attempt_number=1, ) mission_resumed = False resume_result_dict = None if execute_resume_if_eligible and resume_decision.resume_allowed and (final_verifier_callable or mission_runner_callable): resume_res = self.resume_mission( original_mission=snapshot, repair_result=effective_repair_result, final_verifier_callable=final_verifier_callable, failure_fingerprint=fingerprint, attempt_number=1, mission_runner_callable=mission_runner_callable, ) mission_resumed = resume_res.mission_resumed resume_result_dict = resume_res.to_dict() self.last_promotion_result = effective_repair_result auto_promo_flag = bool(getattr(effective_repair_result, "auto_promotion", False)) report = ShadowRepairReport( shadow_run_id=_new_id("shd"), mission_id=snapshot.mission_id, instance_id=snapshot.instance_id, incident_id=inc_id, original_mission_status=snapshot.mission_status, failure_fingerprint=fingerprint, shadow_status=sh_status, shadow_mode=self.SELF_REPAIR_MODE, shadow_triggered=True, trigger_reason=reason, false_success_detected=is_false_success, root_cause=rc_name, root_cause_proven=rc_proven, autonomous_patch_generated=auto_patch, candidate_repair=cand_repair, sandbox_executed=bool(manifest), sandbox_manifest_dir=sb_dir, production_modified=prod_modified, production_hash_changed=prod_hash_changed, auto_promotion=auto_promo_flag, creator_approval_required=not auto_promo_flag, mission_intact=True, artifacts_persisted=persisted, duplicate_shadow_event=False, new_repair_started=True, frozen_baseline_blocked=frozen_blocked, source_component=snapshot.source_component, attempt_number=1, max_attempts=self.max_repair_attempts, guard_decision="ALLOW", guard_reason="First repair attempt authorized under loop guard contract.", circuit_state="OPEN", repair_attempt_blocked=False, provider_called=bool(prov_val is True), network_used=bool(net_val is True), resume_decision=resume_decision.to_dict() if resume_decision else None, resume_result=resume_result_dict, mission_resumed=mission_resumed, ) # Store in caches self._seen_fingerprints[dup_key] = report self._first_reports[guard_key] = report # Persist shadow repair report if persist_artifacts and inc_id: inc_dir = os.path.join(self.base_reports_dir, inc_id) rep_path = os.path.join(inc_dir, "shadow_repair_report.json") with open(rep_path, "w", encoding="utf-8") as f: json.dump(report.to_dict(), f, indent=2, ensure_ascii=False) if "shadow_repair_report.json" not in report.artifacts_persisted: report.artifacts_persisted.append("shadow_repair_report.json") return report except Exception as e: logger.error(f"[ShadowSelfRepair] Exception during shadow execution: {e}", exc_info=True) self._circuit_states[guard_key] = "OPEN" report = ShadowRepairReport( shadow_run_id=_new_id("shd"), mission_id=snapshot.mission_id, instance_id=snapshot.instance_id, incident_id=None, original_mission_status=snapshot.mission_status, failure_fingerprint=fingerprint, shadow_status=ShadowStatus.SHADOW_INTERNAL_ERROR.value, shadow_mode=self.SELF_REPAIR_MODE, shadow_triggered=True, trigger_reason=reason, false_success_detected=is_false_success, candidate_repair="ERROR", sandbox_executed=False, production_modified=False, production_hash_changed=False, auto_promotion=False, creator_approval_required=True, mission_intact=True, error_message=f"Shadow self-repair error: {e}", duplicate_shadow_event=False, new_repair_started=False, source_component=snapshot.source_component, attempt_number=1, max_attempts=self.max_repair_attempts, guard_decision="ALLOW", guard_reason=f"Internal exception during execution: {e}", circuit_state="OPEN", repair_attempt_blocked=False, self_repair_internal_failure=True, recursive_repair_started=False, provider_called=False, network_used=False, ) self._first_reports[guard_key] = report self._seen_fingerprints[dup_key] = report return report finally: self._is_handling = False