diff --git a/src/pyob/entrance_mixins.py b/src/pyob/entrance_mixins.py index 2b5ccd5..bf9e7fa 100644 --- a/src/pyob/entrance_mixins.py +++ b/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,9 +282,15 @@ 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 = "" @@ -216,21 +298,40 @@ def execute_targeted_iteration(self: Any, iteration: int): 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: