"""Per-file async migration agent. Lifecycle for each file x track: 1. Read source from worktree. 2. Split into semantic chunks when file > CHUNK_SIZE_LINES. 3. Retry loop (up to MAX_RETRIES): a. Migrate (chunk-aware on iteration 0; whole-file on retries). - Whole-file path uses a multi-turn conversation: each retry appends the error feedback and the model's previous reply to the same message list, mirroring the Copilot CLI behaviour. - Chunk path: each chunk migrated in parallel with its own independent mini message list (Decision 3 - parallel execution preserved). b. Consistency pass on iteration 0 (chunks only): uses its own fresh multi-turn conversation so it is structurally prepared for future per-pass retries (Decision 1 - scope B). c. Write result to worktree. d. Fast per-file syntax check. e. On success: generate tests, full build + CTest, evaluate gates. f. On all gates passed: commit, return SUCCESS. g. On failure: collect error context, continue loop. 4. On budget exceeded or retries exhausted: quarantine + restore original. Conversation history is strictly scoped per file (Decision 4 - Option A): all message lists are local variables inside migrate_file and are discarded when the function returns. """ from __future__ import annotations import asyncio import logging import time from pathlib import Path import httpx from migration_pipeline.api_client import call_api_multiturn, select_model from migration_pipeline.build_runner import ( cmake_configure, parse_warnings, syntax_check_file, ) from migration_pipeline.chunker import CodeChunk, split_semantic from migration_pipeline.config import ( CHUNK_SIZE_LINES, HARD_BUDGET_LARGE_FILE, HARD_BUDGET_SMALL_FILE, LARGE_FILE_THRESHOLD, MAX_RETRIES, SOFT_BUDGET_PER_FILE, ) from migration_pipeline.git_manager import ( commit_generated_tests, commit_migrated_file, ) from migration_pipeline.models import ( FileResult, FileStatus, QualityGateResult, Track, ) from migration_pipeline.quality_gate import evaluate_gates from migration_pipeline.quarantine import quarantine_file from migration_pipeline.test_generator import generate_tests logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # System prompts # --------------------------------------------------------------------------- _SYSTEM_A_WHOLE = ( "Du bist ein Qt5\u2192Qt6 Migrations-Agent (Track A: konservative Kompatibilit\u00e4t). " "Migriere ausschlie\u00dflich die notwendigen Qt5-API-Aufrufe auf ihre Qt6-\u00c4quivalente. " "Behalte den vorhandenen Code-Stil und alle Kommentare unver\u00e4ndert. " "Keine Modernisierungen, keine Umstrukturierungen. " "Gib NUR den modifizierten C++ Quellcode zur\u00fcck \u2013 keine Erkl\u00e4rungen, kein Markdown." ) _SYSTEM_B_WHOLE = ( "Du bist ein Qt5\u2192Qt6 Migrations-Agent (Track B: Modernisierung). " "Migriere auf Qt6 UND modernisiere gleichzeitig auf C++20: " "nutze auto, Ranges, Concepts, std::optional, structured bindings, " "constexpr/consteval, [[nodiscard]], if constexpr, lambdas. " "Ersetze veraltete Qt5-APIs vollst\u00e4ndig durch ihre Qt6-Pendants. " "Gib NUR den modifizierten C++ Quellcode zur\u00fcck \u2013 keine Erkl\u00e4rungen, kein Markdown." ) _SYSTEM_A_CHUNK = ( "Du bist ein Qt5\u2192Qt6 Migrations-Agent (Track A) f\u00fcr einen CODE-CHUNK. " "Der Chunk ist Teil einer gr\u00f6\u00dferen Datei \u2013 erhalte alle Includes und Grenzen. " "Migriere NUR Qt5\u2192Qt6 API-\u00c4nderungen innerhalb dieses Chunks. " "Gib NUR C++ Quellcode zur\u00fcck." ) _SYSTEM_B_CHUNK = ( "Du bist ein Qt5\u2192Qt6 Modernisierungs-Agent (Track B) f\u00fcr einen CODE-CHUNK. " "Migriere auf Qt6 + C++20 innerhalb dieses Chunks. " "Behalte alle Chunk-Grenzen, Namespaces und Includes exakt bei. " "Gib NUR C++ Quellcode zur\u00fcck." ) _SYSTEM_MERGE_CHECK = ( "Du pr\u00fcfst einen zusammengesetzten C++ Quellcode, der aus mehreren migrierten Chunks " "zusammengef\u00fcgt wurde. Stelle sicher: " "1) Include-Guards / #pragma once nur einmal vorhanden. " "2) Alle Namespaces korrekt geschlossen. " "3) Keine doppelten Deklarationen oder Definitionen. " "4) Keine widerspr\u00fcchlichen Makros oder typedefs. " "Korrigiere ausschlie\u00dflich echte Merge-Fehler \u2013 keine sonstigen \u00c4nderungen. " "Gib NUR den korrigierten C++ Quellcode zur\u00fcck." ) def _whole_system(track: Track) -> str: return _SYSTEM_A_WHOLE if track == Track.A else _SYSTEM_B_WHOLE def _chunk_system(track: Track) -> str: return _SYSTEM_A_CHUNK if track == Track.A else _SYSTEM_B_CHUNK # --------------------------------------------------------------------------- # Multi-turn API helpers # --------------------------------------------------------------------------- async def _migrate_whole_mt( client: httpx.AsyncClient, model: str, code: str, track: Track, messages: list[dict], error_context: str = "", ) -> str: """Whole-file migration via a growing multi-turn conversation. *messages* is mutated in place: - On the first call (empty list) the system prompt is prepended and the initial migration request is sent. - On subsequent calls the error feedback is appended as a new user turn after the model's previous reply, so the model can learn from its own mistakes within the same conversation. Returns the migrated code (= the latest assistant reply). """ if not messages: # First call: initialise the conversation. messages.append({"role": "system", "content": _whole_system(track)}) user_msg = f"Portiere folgenden Qt5-Code nach Qt6:\n\n{code}" else: # Retry: the model sees its own previous reply + the compiler error. user_msg = ( f"Fehler im vorherigen Versuch:\n{error_context}\n\n" f"Bitte korrigiere den folgenden Code:\n\n{code}" ) messages.append({"role": "user", "content": user_msg}) reply = await call_api_multiturn(client, model, messages) messages.append({"role": "assistant", "content": reply}) return reply async def _migrate_chunk_mt( client: httpx.AsyncClient, model: str, chunk: CodeChunk, track: Track, ) -> str: """Migrate one chunk with its own independent mini messages list. Each chunk is structurally multi-turn capable (Decision 3 - Option B): the messages list is fresh per chunk, ensuring full parallel independence while being ready for per-chunk retries in future extensions. """ messages: list[dict] = [ {"role": "system", "content": _chunk_system(track)}, { "role": "user", "content": ( f"Migriere Chunk #{chunk.index} " f"(Zeilen {chunk.start_line}\u2013{chunk.end_line - 1}):\n\n{chunk.content}" ), }, ] reply = await call_api_multiturn(client, model, messages) messages.append({"role": "assistant", "content": reply}) return reply async def _consistency_pass_mt( client: httpx.AsyncClient, model: str, merged: str, ) -> str: """Consistency pass with its own fresh multi-turn conversation. Uses a dedicated message list (Decision 1 - Option B scope), independent of the whole-file retry conversation, so it can be extended with its own retry turns in future without cross-contaminating the migration history. """ messages: list[dict] = [ {"role": "system", "content": _SYSTEM_MERGE_CHECK}, {"role": "user", "content": f"Pr\u00fcfe und korrigiere:\n\n{merged}"}, ] reply = await call_api_multiturn(client, model, messages) messages.append({"role": "assistant", "content": reply}) return reply # --------------------------------------------------------------------------- # Budget helpers # --------------------------------------------------------------------------- def _hard_budget(line_count: int) -> float: return HARD_BUDGET_LARGE_FILE if line_count > LARGE_FILE_THRESHOLD else HARD_BUDGET_SMALL_FILE # --------------------------------------------------------------------------- # Core agent # --------------------------------------------------------------------------- async def migrate_file( *, client: httpx.AsyncClient, source_file: Path, # absolute path inside the worktree worktree_root: Path, build_dir: Path, track: Track, batch_id: str, ) -> FileResult: """Migrate one file and return a FileResult.""" file_str = str(source_file) original_code = source_file.read_text(errors="ignore") line_count = len(original_code.splitlines()) model = select_model(line_count) budget = _hard_budget(line_count) is_large = line_count > CHUNK_SIZE_LINES chunks = split_semantic(original_code, CHUNK_SIZE_LINES) if is_large else [] chunk_info = ( f"{len(chunks)} chunks, ~{CHUNK_SIZE_LINES} Zeilen/Chunk" if chunks else f"vollst\u00e4ndig ({line_count} Zeilen)" ) start = time.monotonic() current_code = original_code error_context = "" last_gate = QualityGateResult() iteration = 0 test_prefix: str | None = None # Multi-turn conversation for whole-file migrations. # Decision 4 (Option A): strictly per-file - this list is local and # discarded when migrate_file returns. whole_messages: list[dict] = [] while iteration < MAX_RETRIES: elapsed = time.monotonic() - start # --- Hard budget check -------------------------------------------- if elapsed >= budget: logger.warning( "[BUDGET] %s [%s] \u2013 %.0f s (budget %.0f s)", source_file.name, track.value, elapsed, budget, ) break # --- Soft budget warning ------------------------------------------ if elapsed >= SOFT_BUDGET_PER_FILE and iteration == 0: logger.warning( "[SOFT BUDGET] %s [%s] exceeded soft limit at %.0f s", source_file.name, track.value, elapsed, ) try: # ---- 1. Migrate ------------------------------------------------ if is_large and iteration == 0: # Parallel chunk migration: each chunk gets its own fresh # multi-turn messages list (Decision 3 - Option B). migrated_parts = await asyncio.gather( *[_migrate_chunk_mt(client, model, c, track) for c in chunks] ) merged = "\n".join(migrated_parts) # Consistency pass uses its own independent multi-turn # conversation (Decision 1 - Option B). migrated_code = await _consistency_pass_mt(client, model, merged) else: # Whole-file path: conversation grows per retry so the model # sees its own previous answer alongside the compiler error # (Decision 1 - Option B, Decision 4 - Option A). migrated_code = await _migrate_whole_mt( client, model, current_code, track, whole_messages, error_context ) # ---- 2. Write migrated file to worktree ----------------------- source_file.write_text(migrated_code) # ---- 3. Fast per-file syntax check ---------------------------- syntax_ok, syn_out, syn_err = await syntax_check_file( source_file, build_dir ) syntax_warnings = parse_warnings(syn_out + "\n" + syn_err) if not syntax_ok: logger.info( "[SYNTAX FAIL] %s [%s] iter %d", source_file.name, track.value, iteration, ) error_context = f"Compiler-Syntaxfehler:\n{syn_err}" current_code = migrated_code iteration += 1 continue # ---- 4. Generate tests ---------------------------------------- logger.info( "[QA] Generating tests for %s [%s]...", source_file.name, track.value, ) test_prefix = await generate_tests( client, model, file_str, migrated_code, worktree_root, track.value, ) # ---- 5. Reconfigure + full build (to include new tests) ------- await cmake_configure(worktree_root, build_dir) last_gate = await evaluate_gates(build_dir, test_prefix) # Merge syntax-check warnings into gate result. last_gate.build_warnings = syntax_warnings + last_gate.build_warnings if last_gate.all_gates_passed: elapsed_final = time.monotonic() - start logger.info( "[SUCCESS] %s [%s] in %.0f s \u2013 %d warning(s)", source_file.name, track.value, elapsed_final, last_gate.total_warnings, ) # Commit migrated file and generated tests. commit_migrated_file(source_file, worktree_root, track.value) tests_dir = worktree_root / "tests_generated" commit_generated_tests( tests_dir, worktree_root, track.value, test_prefix ) return FileResult( file_path=file_str, track=track.value, status=FileStatus.SUCCESS.value, iterations=iteration + 1, elapsed_seconds=round(elapsed_final, 1), gate=last_gate, migrated_path=file_str, test_name=test_prefix or "", ) # Gates failed - build error or test failure. error_context = ( f"Build-Fehler:\n{last_gate.build_errors}\n\n" f"Test-Ausgabe:\n{last_gate.test_output}" ) logger.info( "[GATE FAIL] %s [%s] iter %d \u2013 build_ok=%s, " "existing_ok=%s, new_ok=%s", source_file.name, track.value, iteration, last_gate.build_ok, last_gate.existing_tests_ok, last_gate.new_tests_ok, ) current_code = migrated_code iteration += 1 except Exception as exc: # noqa: BLE001 logger.error( "[ERROR] %s [%s] iter %d: %s", source_file.name, track.value, iteration, exc, ) error_context = f"API/System-Fehler: {exc}" # Restore original so next iteration starts clean. source_file.write_text(original_code) iteration += 1 # ---- Quarantine -------------------------------------------------------- elapsed_total = time.monotonic() - start logger.warning( "[QUARANTINE] %s [%s] after %d iter, %.0f s", source_file.name, track.value, iteration, elapsed_total, ) source_file.write_text(original_code) # restore q_entry = quarantine_file( file_path=file_str, track=track.value, build_errors=last_gate.build_errors, test_output=last_gate.test_output, last_code=current_code, elapsed=elapsed_total, budget=budget, retry_count=iteration, chunk_info=chunk_info, ) return FileResult( file_path=file_str, track=track.value, status=FileStatus.QUARANTINED.value, iterations=iteration, elapsed_seconds=round(elapsed_total, 1), gate=last_gate, quarantine_entry=q_entry, )