Update entrance_mixins.py

src/pyob/entrance_mixins.py

@@ -8,19 +8,89 @@
 import urllib.parse
 from http.server import HTTPServer
 from pathlib import Path
-from typing import Any
+from typing import Any, Callable, Optional
 
 from .dashboard_html import OBSERVER_HTML
 from .pyob_dashboard import ObserverHandler
-from .targeted_reviewer import TargetedReviewer
 
 logger = logging.getLogger(__name__)
 
 
-# [TargetedReviewer class moved to targeted_reviewer.py]
+class EntranceMixin:
+    """
+    Mixin providing the core iteration logic and dashboard management.
+    Dummy returns are used in stubs to satisfy strict Mypy [empty-body] checks.
+    """
 
+    # --- TYPE ANNOTATIONS FOR MYPY ---
+    target_dir: str
+    pyob_dir: str
+    ENGINE_FILES: list[str]
+    llm_engine: Any
+    code_parser: Any
+    cascade_queue: list[str]
+    cascade_diffs: dict[str, str]
+    session_pr_count: int
+    self_evolved_flag: bool
+    memory_path: str
+    history_path: str
+    analysis_path: str
+    symbols_path: str
+    manual_target_file: Optional[str]
+    key_cooldowns: dict[str, float]
+
+    def pick_target_file(self) -> str:
+        return ""
+
+    def _read_file(self, path: str) -> str:
+        return ""
+
+    def _extract_path_from_llm_response(self, text: str) -> str:
+        return ""
+
+    def get_valid_llm_response(
+        self, p: str, v: Callable[[str], bool], context: str
+    ) -> str:
+        return ""
+
+    def update_analysis_for_single_file(self, abs_p: str, rel_p: str):
+        pass
+
+    def update_ledger_for_file(self, rel_p: str, code: str):
+        pass
+
+    def detect_symbolic_ripples(self, o: str, n: str, p: str) -> list[str]:
+        return []
+
+    def _run_final_verification_and_heal(self, b: dict) -> bool:
+        return False
+
+    def handle_git_librarian(self, p: str, i: int):
+        pass
+
+    def reboot_pyob(self):
+        pass
+
+    def trigger_production_build(self):
+        pass
+
+    def load_ledger(self) -> dict:
+        return {}
+
+    def append_to_history(self, p: str, o: str, n: str):
+        pass
+
+    def _run_git_command(self, cmd: list[str]) -> bool:
+        return False
+
+    def wrap_up_evolution_session(self):
+        pass
+
+    def generate_pr_summary(self, rel_path: str, diff_text: str) -> dict:
+        return {}
+
+    # ---------------------------------
 
-class EntranceMixin:
     def start_dashboard(self: Any):
         # 1. Save to the internal .pyob folder
         obs_path = os.path.join(self.pyob_dir, "observer.html")
@@ -168,9 +238,15 @@ def run_server():
         print(f"FILE: {obs_path}")
         print("=" * 60 + "\n")
 
-    def execute_targeted_iteration(self: Any, iteration: int):
+    def execute_targeted_iteration(self, iteration: int):
+        """
+        Orchestrates a single targeted evolution step.
+        Preserves engine safety pods and manages symbolic ripples.
+        """
+
         backup_state = self.llm_engine.backup_workspace()
         target_diff = ""
+
         if self.cascade_queue:
             target_rel_path = self.cascade_queue.pop(0)
             target_diff = self.cascade_diffs.get(target_rel_path, "")
@@ -185,8 +261,8 @@ def execute_targeted_iteration(self: Any, iteration: int):
         if not target_rel_path:
             return
 
+        # --- SAFETY POD LOGIC (Preserved) ---
         is_engine_file = any(Path(target_rel_path).name == f for f in self.ENGINE_FILES)
-
         if is_engine_file:
             timestamp = time.strftime("%Y%m%d_%H%M%S")
             project_name = os.path.basename(self.target_dir)
@@ -206,31 +282,56 @@ def execute_targeted_iteration(self: Any, iteration: int):
                 logger.error(f"Failed to create external safety pod: {e}")
 
         target_abs_path = os.path.join(self.target_dir, target_rel_path)
+
+        # --- PREPARE REVIEWER ---
         self.llm_engine.session_context = []
         if is_cascade and target_diff:
-            msg = f"CRITICAL SYMBOLIC RIPPLE: This file depends on code that was just modified. Ensure this file is updated to support these changes:\n\n### DEPDENDENCY CHANGE DIFF:\n{target_diff}"
+            msg = (
+                f"CRITICAL SYMBOLIC RIPPLE: This file depends on code that was just modified. "
+                f"Ensure this file is updated to support these changes:\n\n"
+                f"### DEPENDENCY CHANGE DIFF:\n{target_diff}"
+            )
             self.llm_engine.session_context.append(msg)
 
         old_content = ""
         if os.path.exists(target_abs_path):
             with open(target_abs_path, "r", encoding="utf-8", errors="ignore") as f:
                 old_content = f.read()
 
+        # Initialize the reviewer with the same state as the controller
+        from pyob.targeted_reviewer import TargetedReviewer
+
         reviewer = TargetedReviewer(self.target_dir, target_abs_path)
+
+        # SYNC STATE: Ensure reviewer uses the same memory and cooldowns
         reviewer.session_context = self.llm_engine.session_context[:]
+        if hasattr(self, "key_cooldowns"):
+            reviewer.key_cooldowns = self.key_cooldowns
+        if hasattr(self, "session_pr_count"):
+            reviewer.session_pr_count = self.session_pr_count
+
+        # Execute the reviewer pipeline
         reviewer.run_pipeline(iteration)
 
+        # Sync context back after review
         self.llm_engine.session_context = reviewer.session_context[:]
+        if hasattr(reviewer, "session_pr_count"):
+            self.session_pr_count = reviewer.session_pr_count
 
+        # Capture the results of the edit
         new_content = ""
         if os.path.exists(target_abs_path):
             with open(target_abs_path, "r", encoding="utf-8", errors="ignore") as f:
                 new_content = f.read()
 
+        # Update symbolic maps immediately
         logger.info(f"Refreshing metadata for `{target_rel_path}`...")
-        self.update_analysis_for_single_file(target_abs_path, target_rel_path)
-        self.update_ledger_for_file(target_rel_path, new_content)
+        if hasattr(self, "update_analysis_for_single_file"):
+            self.update_analysis_for_single_file(target_abs_path, target_rel_path)
+        if hasattr(self, "update_ledger_for_file"):
+            self.update_ledger_for_file(target_rel_path, new_content)
 
+        # --- FINAL VERIFICATION GATE ---
         if old_content != new_content:
             logger.info(
                 f"Edit successful. Checking ripples and running final verification for {target_rel_path}..."
@@ -244,6 +345,7 @@ def execute_targeted_iteration(self: Any, iteration: int):
                 )
             )
 
+            # Detect if this change impacts other files
             ripples = self.detect_symbolic_ripples(
                 old_content, new_content, target_rel_path
             )
@@ -257,12 +359,16 @@ def execute_targeted_iteration(self: Any, iteration: int):
                         self.cascade_diffs[r] = current_diff
 
             logger.info("\n" + "=" * 20 + " FINAL VERIFICATION " + "=" * 20)
+
+            # THE ABSOLUTE GATE: We only push the PR if the app is stable
             if not self._run_final_verification_and_heal(backup_state):
                 logger.error(
-                    "Final verification failed and could not be auto-repaired. Iteration changes have been rolled back."
+                    f"Final verification failed for `{target_rel_path}` and could not be auto-repaired. "
+                    "Iteration changes have been rolled back to protect the branch."
                 )
             else:
                 logger.info("Final verification successful. Application is stable.")
+                # OPEN PR ONLY ON SUCCESS
                 self.handle_git_librarian(target_rel_path, iteration)
 
                 if is_engine_file: