# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — CIRCUIT TRACER Reconstructs the full end-to-end highway of execution for an incident, evaluates node statuses (PASS, FAIL, NOT_REACHED, UNKNOWN), detects contradictions, and locates the FIRST_PROVEN_BREAK. """ from __future__ import annotations import json import os from typing import Any, Dict, List, Optional from app.remediation.self_repair.schemas import ( CANONICAL_CIRCUIT_HIGHWAY, CircuitNode, CircuitStatus, CircuitTrace, EvidenceBundle, IncidentRecord, ) class CircuitTracer: """ Traces execution through the complete highway: creator_input -> chat_gateway -> brain -> requirement_contract -> planner -> executor -> effector -> filesystem_runtime_browser -> verification -> final_mission_state """ def trace( self, incident: IncidentRecord, evidence: EvidenceBundle, node_status_overrides: Optional[Dict[str, Dict[str, Any]]] = None, ) -> CircuitTrace: """ Reconstructs each node on the canonical highway. Correlates incident data, logs, action history, and verification results. """ overrides = dict(node_status_overrides or {}) nodes: List[CircuitNode] = [] contradictions: List[str] = [] first_break: Optional[str] = None break_reached: bool = False downstream_not_reached: List[str] = [] # Analyze evidence to deduce node statuses if not explicitly overridden deduced_statuses: Dict[str, CircuitStatus] = {} deduced_evidence: Dict[str, List[Dict[str, Any]]] = {} deduced_details: Dict[str, str] = {} # 1. Mission / Action History analysis actions = evidence.action_history verif = evidence.verification_results m_state = evidence.mission_state init_ev = incident.initial_evidence # Let's inspect initial evidence and verification evidence for clues circuit_evidence = init_ev.get("circuit_evidence") or {} # Reconstruct along highway for comp in CANONICAL_CIRCUIT_HIGHWAY: c_status = CircuitStatus.UNKNOWN c_ev_list: List[Dict[str, Any]] = [] c_details = "" # Check explicit override if comp in overrides: ov = overrides[comp] st_val = ov.get("status") c_status = ( st_val if isinstance(st_val, CircuitStatus) else CircuitStatus(st_val.upper()) if st_val else CircuitStatus.UNKNOWN ) c_ev_list = list(ov.get("evidence") or []) c_details = str(ov.get("details") or "") # Check circuit_evidence from incident elif comp in circuit_evidence: cev = circuit_evidence[comp] st_val = cev.get("status") c_status = ( st_val if isinstance(st_val, CircuitStatus) else CircuitStatus(st_val.upper()) if st_val else CircuitStatus.UNKNOWN ) c_ev_list = list(cev.get("evidence") or []) c_details = str(cev.get("details") or "") else: # 1. Check if action_history has explicit record for this component comp_actions = [ a for a in actions if str(a.get("component") or a.get("stage") or "").lower() == comp.lower() ] # Check if failure_stage matches is_failed_stage = ( init_ev.get("failure_stage") == comp or incident.source_component == comp ) if break_reached: # After first break, subsequent nodes are NOT_REACHED unless evidence proves execution c_status = CircuitStatus.UNKNOWN c_details = f"Not reached due to upstream break at {first_break}" elif any(str(a.get("status") or "").upper() in ("FAIL", "FAILED", "ERROR") for a in comp_actions): c_status = CircuitStatus.FAIL c_ev_list.extend([a for a in comp_actions if str(a.get("status") or "").upper() in ("FAIL", "FAILED", "ERROR")]) c_details = f"Failure occurred during {comp} execution in action history" elif is_failed_stage: c_status = CircuitStatus.FAIL c_ev_list.append({"incident_symptom": incident.observed_symptom, "source": comp}) c_details = f"Failure originated at {comp}: {incident.observed_symptom}" elif any(str(a.get("status") or "").upper() in ("PASS", "SUCCESS", "OK") for a in comp_actions): c_status = CircuitStatus.PASS c_ev_list.extend([a for a in comp_actions if str(a.get("status") or "").upper() in ("PASS", "SUCCESS", "OK")]) c_details = f"Component {comp} succeeded in action history" elif comp == "creator_input" and (incident.observed_symptom or actions or m_state): c_status = CircuitStatus.PASS c_details = "Creator input received" elif comp == "chat_gateway" and (actions or m_state or incident.mission_id): c_status = CircuitStatus.PASS c_details = "Dispatched through gateway" elif comp == "brain" and (incident.mission_id or actions or m_state): c_status = CircuitStatus.PASS c_details = "Brain orchestrated mission" elif comp == "requirement_contract" and (actions or m_state.get("spec") or m_state.get("contract")): c_status = CircuitStatus.PASS c_details = "Requirement contract verified" elif comp == "planner" and (actions or m_state.get("plan") or m_state.get("developer_plan")): c_status = CircuitStatus.PASS c_details = "Planner generated mission plan" elif comp == "verification" and verif: if verif.get("passed") is False or verif.get("error"): c_status = CircuitStatus.FAIL c_ev_list.append({"verification": verif}) c_details = f"Verification check failed: {verif.get('error') or verif.get('summary')}" elif verif.get("passed") is True: c_status = CircuitStatus.PASS c_details = "Verification check passed" # Check for contradiction: e.g. verifier passed but mission failed if comp == "verification" and c_status == CircuitStatus.PASS and incident.failure_type.value == "FALSE_SUCCESS": contradictions.append("Verification reported PASS but actual state is broken (FALSE_SUCCESS)") # Record first proven break if c_status == CircuitStatus.FAIL and not first_break: first_break = comp break_reached = True if break_reached and comp != first_break and c_status == CircuitStatus.NOT_REACHED: downstream_not_reached.append(comp) node = CircuitNode( component=comp, status=c_status, evidence=c_ev_list, details=c_details, ) nodes.append(node) trace_complete = all(n.status != CircuitStatus.UNKNOWN for n in nodes) return CircuitTrace( incident_id=incident.incident_id, nodes=nodes, trace_complete=trace_complete, first_proven_break=first_break, downstream_not_reached=downstream_not_reached, contradictions=contradictions, ) def persist(self, trace: CircuitTrace, output_dir: str) -> str: """ Persists circuit_trace.json to the specified output directory. """ os.makedirs(output_dir, exist_ok=True) path = os.path.join(output_dir, "circuit_trace.json") with open(path, "w", encoding="utf-8") as f: json.dump(trace.to_dict(), f, indent=2, ensure_ascii=False) return path