swe_agent_loop.py

#!/usr/bin/env python3
"""
swe_agent_loop.py — ATYPIK v24.0
Boucle agentique complete : plan -> search -> edit -> test -> iterate.
"""

import os
import sys
import json
import time
import textwrap
import tempfile
import shutil
from pathlib import Path
from typing import Dict, List, Optional, Callable
import logging

logger = logging.getLogger("atypik.swe")

class SWEAgentLoop:
    def __init__(self, generate_patch_fn: Callable, repo_path: Optional[Path] = None):
        self.generate_patch = generate_patch_fn
        self.repo_path = repo_path
        self.history: List[Dict] = []
        self.max_iterations = 5

    def solve(self, issue: str, repo_url: str, branch: str = "main") -> Dict:
        from github_api import download_repo
        from repo_navigator import RepoNavigator
        from multi_file_editor import MultiFileEditor
        from test_runner import TestRunner

        start_time = time.time()
        workspace = Path(tempfile.mkdtemp(prefix="swe_task_"))

        try:
            logger.info("[SWE] Telechargement du repo...")
            self.repo_path = download_repo(repo_url, branch, workspace / "repo")
            logger.info(f"[SWE] Repo telecharge: {self.repo_path}")

            navigator = RepoNavigator(self.repo_path)
            editor = MultiFileEditor(self.repo_path)
            runner = TestRunner(self.repo_path)

            logger.info("[SWE] Installation des dependances...")
            ok, msg = runner.install_dependencies()
            if not ok:
                logger.warning(f"[SWE] Dependances: {msg}")

            logger.info("[SWE] Tests baseline...")
            baseline = runner.run_tests(timeout=30)

            context = {
                "issue": issue,
                "repo_url": repo_url,
                "structure": navigator.get_structure(),
                "baseline_tests": baseline,
                "files_explored": [],
                "edits_made": [],
            }

            for iteration in range(1, self.max_iterations + 1):
                logger.info(f"[SWE] Iteration {iteration}/{self.max_iterations}")

                if iteration == 1:
                    search_results = navigator.grep(issue[:50].split()[0])
                    context["search_results"] = search_results[:10]

                prompt = self._build_agent_prompt(context, iteration)
                patch = self.generate_patch(prompt, max_new_tokens=2048, temperature=0.2)
                results = editor.apply_patch_unidiff(patch)
                successful_edits = [r for r in results if r.success]

                if not successful_edits:
                    context["last_error"] = "Aucune edition n'a pu etre appliquee"
                    context["files_explored"].append(f"iteration_{iteration}_failed")
                    continue

                context["edits_made"].extend([r.file for r in successful_edits])
                test_result = runner.run_tests(timeout=60)
                context["last_test_result"] = test_result

                if test_result["success"]:
                    if not baseline["success"]:
                        return {
                            "status": "success",
                            "patch": patch,
                            "iterations": iteration,
                            "duration_seconds": time.time() - start_time,
                            "files_modified": editor.get_summary()["files_modified"],
                            "tests_passed": test_result["passed"],
                            "agent_logs": self.history,
                        }
                    else:
                        context["last_error"] = "Les tests passaient deja avant le patch"
                else:
                    context["last_error"] = test_result["stderr"][:1000]
                    self.history.append({
                        "iteration": iteration,
                        "error": context["last_error"],
                        "edits": [r.file for r in successful_edits]
                    })

            return {
                "status": "max_iterations_reached",
                "patch": patch,
                "iterations": self.max_iterations,
                "duration_seconds": time.time() - start_time,
                "last_error": context.get("last_error", "Unknown"),
                "files_modified": editor.get_summary()["files_modified"],
            }

        except Exception as e:
            logger.exception("[SWE] Erreur dans l'agent")
            return {
                "status": "error",
                "error": str(e),
                "duration_seconds": time.time() - start_time,
            }
        finally:
            pass  # shutil.rmtree(workspace, ignore_errors=True)

    def _build_agent_prompt(self, context: Dict, iteration: int) -> str:
        issue = context["issue"]
        search_results = context.get("search_results", [])
        last_error = context.get("last_error", "")
        edits_made = context.get("edits_made", [])
        search_str = "\n".join([
            f"- {r['file']}:{r['line']} : {r['content'][:100]}"
            for r in search_results[:10]
        ])
        error_str = f"\n\nERREUR PRECEDENTE (iteration {iteration-1}):\n{last_error}" if last_error and iteration > 1 else ""
        edits_str = f"\n\nFICHIERS DEJA MODIFIES:\n" + "\n".join(edits_made) if edits_made else ""
        return textwrap.dedent(f"""
        Tu es ATYPIK, un agent de reparation de code expert.

        ISSUE A RESOUDRE:
        {issue}

        RESULTATS DE RECHERCHE:
        {search_str}
        {error_str}
        {edits_str}

        INSTRUCTIONS:
        1. Analyse l'issue et les resultats de recherche.
        2. Identifie le(s) fichier(s) a modifier.
        3. Genere un patch au format `git diff` (unified diff).
        4. Le patch doit etre MINIMAL — modifie le moins de lignes possible.
        5. Ne modifie JAMAIS les fichiers de test.
        6. Assure-toi que la syntaxe Python est valide.

        Reponds UNIQUEMENT avec le patch git. Pas d'explications.
        """)