From 458d4bb97adcef27b72295c4c764b622e9c49d12 Mon Sep 17 00:00:00 2001 From: AndyMik90 Date: Sat, 20 Dec 2025 13:19:57 +0100 Subject: [PATCH] feat: implement parallel AI merge functionality MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the ability to perform parallel merges using AI for conflict resolution. This includes the implementation of the `_run_parallel_merges` function, which processes multiple merge tasks concurrently, and the `_merge_file_with_ai_async` function for handling individual file merges. Key changes: - Introduced AI-based merging logic with a system prompt for 3-way merges. - Added helper functions for inferring file types and building merge prompts. - Updated tests to cover various scenarios for the new merging functionality. This enhancement allows for more efficient handling of merge conflicts, leveraging AI to ensure accurate and context-aware resolutions. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- auto-claude/core/workspace.py | 342 ++++++++++++++++++++++++- auto-claude/core/workspace/__init__.py | 5 +- tests/test_merge_parallel.py | 84 +++++- 3 files changed, 415 insertions(+), 16 deletions(-) diff --git a/auto-claude/core/workspace.py b/auto-claude/core/workspace.py index 0349a821..51403384 100644 --- a/auto-claude/core/workspace.py +++ b/auto-claude/core/workspace.py @@ -98,7 +98,9 @@ from core.workspace.models import ( MergeLock, MergeLockError, ParallelMergeTask, + ParallelMergeResult, ) +from core.workspace.git_utils import MAX_PARALLEL_AI_MERGES from merge import ( FileTimelineTracker, MergeOrchestrator, @@ -858,13 +860,11 @@ def _resolve_git_conflicts_with_ai( start_time = time.time() # Run parallel merges - # TODO: _run_parallel_merges not yet implemented - see line 140 - # parallel_results = asyncio.run(_run_parallel_merges( - # tasks=files_needing_ai_merge, - # project_dir=project_dir, - # max_concurrent=MAX_PARALLEL_AI_MERGES, - # )) - parallel_results = [] # Placeholder until function is implemented + parallel_results = asyncio.run(_run_parallel_merges( + tasks=files_needing_ai_merge, + project_dir=project_dir, + max_concurrent=MAX_PARALLEL_AI_MERGES, + )) elapsed = time.time() - start_time @@ -996,3 +996,331 @@ def _resolve_git_conflicts_with_ai( # - Git utilities from workspace/git_utils.py # - Display functions from workspace/display.py # - Finalization functions from workspace/finalization.py + + +# ============================================================================= +# Parallel AI Merge Implementation +# ============================================================================= + +import asyncio +import os +import re +import logging + +_merge_logger = logging.getLogger(__name__) + +# System prompt for AI file merging +AI_MERGE_SYSTEM_PROMPT = """You are an expert code merge assistant. Your task is to perform a 3-way merge of code files. + +RULES: +1. Preserve all functional changes from both versions (ours and theirs) +2. Maintain code style consistency +3. Resolve conflicts by understanding the semantic purpose of each change +4. When changes are independent (different functions/sections), include both +5. When changes overlap, combine them logically or prefer the more complete version +6. Preserve all imports from both versions +7. Output ONLY the merged code - no explanations, no markdown, no code fences + +IMPORTANT: Output the raw merged file content only. Do not wrap in code blocks.""" + + +def _infer_language_from_path(file_path: str) -> str: + """Infer programming language from file extension.""" + ext_map = { + ".py": "python", + ".js": "javascript", + ".jsx": "javascript", + ".ts": "typescript", + ".tsx": "typescript", + ".rs": "rust", + ".go": "go", + ".java": "java", + ".cpp": "cpp", + ".c": "c", + ".h": "c", + ".hpp": "cpp", + ".rb": "ruby", + ".php": "php", + ".swift": "swift", + ".kt": "kotlin", + ".scala": "scala", + ".json": "json", + ".yaml": "yaml", + ".yml": "yaml", + ".toml": "toml", + ".md": "markdown", + ".html": "html", + ".css": "css", + ".scss": "scss", + ".sql": "sql", + } + ext = os.path.splitext(file_path)[1].lower() + return ext_map.get(ext, "text") + + +def _try_simple_3way_merge( + base: str | None, + ours: str, + theirs: str, +) -> tuple[bool, str | None]: + """ + Attempt a simple 3-way merge without AI. + + Returns: + (success, merged_content) - if success is True, merged_content is the result + """ + # If base is None, we can't do a proper 3-way merge + if base is None: + # If both are identical, no conflict + if ours == theirs: + return True, ours + # Otherwise, we need AI to decide + return False, None + + # If ours equals base, theirs is the only change - take theirs + if ours == base: + return True, theirs + + # If theirs equals base, ours is the only change - take ours + if theirs == base: + return True, ours + + # If ours equals theirs, both made same change - take either + if ours == theirs: + return True, ours + + # Both changed differently from base - need AI merge + # We could try a line-by-line merge here, but for safety let's use AI + return False, None + + +def _build_merge_prompt( + file_path: str, + base_content: str | None, + main_content: str, + worktree_content: str, + spec_name: str, +) -> str: + """Build the prompt for AI file merge.""" + language = _infer_language_from_path(file_path) + + base_section = "" + if base_content: + # Truncate very large files + if len(base_content) > 10000: + base_content = base_content[:10000] + "\n... (truncated)" + base_section = f""" +BASE (common ancestor): +```{language} +{base_content} +``` +""" + + # Truncate large content + if len(main_content) > 15000: + main_content = main_content[:15000] + "\n... (truncated)" + if len(worktree_content) > 15000: + worktree_content = worktree_content[:15000] + "\n... (truncated)" + + prompt = f"""Perform a 3-way merge for file: {file_path} +Task being merged: {spec_name} +{base_section} +OURS (current main branch): +```{language} +{main_content} +``` + +THEIRS (changes from task worktree): +```{language} +{worktree_content} +``` + +Merge these versions, preserving all meaningful changes from both. Output only the merged file content, no explanations.""" + + return prompt + + +def _strip_code_fences(content: str) -> str: + """Remove markdown code fences if present.""" + # Check if content starts with code fence + lines = content.strip().split('\n') + if lines and lines[0].startswith('```'): + # Remove first and last line if they're code fences + if lines[-1].strip() == '```': + return '\n'.join(lines[1:-1]) + else: + return '\n'.join(lines[1:]) + return content + + +async def _merge_file_with_ai_async( + task: ParallelMergeTask, + semaphore: asyncio.Semaphore, +) -> ParallelMergeResult: + """ + Merge a single file using AI. + + Args: + task: The merge task with file contents + semaphore: Semaphore for concurrency control + + Returns: + ParallelMergeResult with merged content or error + """ + async with semaphore: + try: + # First try simple 3-way merge + success, merged = _try_simple_3way_merge( + task.base_content, + task.main_content, + task.worktree_content, + ) + + if success and merged is not None: + debug(MODULE, f"Auto-merged {task.file_path} without AI") + return ParallelMergeResult( + file_path=task.file_path, + merged_content=merged, + success=True, + was_auto_merged=True, + ) + + # Need AI merge + debug(MODULE, f"Using AI to merge {task.file_path}") + + # Import auth utilities + from core.auth import ensure_claude_code_oauth_token, get_auth_token + + if not get_auth_token(): + return ParallelMergeResult( + file_path=task.file_path, + merged_content=None, + success=False, + error="No authentication token available", + ) + + ensure_claude_code_oauth_token() + + # Build prompt + prompt = _build_merge_prompt( + task.file_path, + task.base_content, + task.main_content, + task.worktree_content, + task.spec_name, + ) + + # Call Claude Haiku for fast merge + try: + from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient + except ImportError: + return ParallelMergeResult( + file_path=task.file_path, + merged_content=None, + success=False, + error="claude_agent_sdk not installed", + ) + + client = ClaudeSDKClient( + options=ClaudeAgentOptions( + model="claude-haiku-4-5-20251001", + system_prompt=AI_MERGE_SYSTEM_PROMPT, + allowed_tools=[], + max_turns=1, + max_thinking_tokens=1024, # Low thinking for speed + ) + ) + + response_text = "" + async with client: + await client.query(prompt) + + async for msg in client.receive_response(): + msg_type = type(msg).__name__ + if msg_type == "AssistantMessage" and hasattr(msg, "content"): + for block in msg.content: + if hasattr(block, "text"): + response_text += block.text + + if response_text: + # Strip any code fences the model might have added + merged_content = _strip_code_fences(response_text.strip()) + + debug(MODULE, f"AI merged {task.file_path} successfully") + return ParallelMergeResult( + file_path=task.file_path, + merged_content=merged_content, + success=True, + was_auto_merged=False, + ) + else: + return ParallelMergeResult( + file_path=task.file_path, + merged_content=None, + success=False, + error="AI returned empty response", + ) + + except Exception as e: + _merge_logger.error(f"Failed to merge {task.file_path}: {e}") + return ParallelMergeResult( + file_path=task.file_path, + merged_content=None, + success=False, + error=str(e), + ) + + +async def _run_parallel_merges( + tasks: list[ParallelMergeTask], + project_dir: Path, + max_concurrent: int = MAX_PARALLEL_AI_MERGES, +) -> list[ParallelMergeResult]: + """ + Run file merges in parallel with concurrency control. + + Args: + tasks: List of merge tasks to process + project_dir: Project directory (for context, not currently used) + max_concurrent: Maximum number of concurrent merge operations + + Returns: + List of ParallelMergeResult for each task + """ + if not tasks: + return [] + + debug(MODULE, f"Starting parallel merge of {len(tasks)} files (max concurrent: {max_concurrent})") + + # Create semaphore for concurrency control + semaphore = asyncio.Semaphore(max_concurrent) + + # Create tasks + merge_coroutines = [ + _merge_file_with_ai_async(task, semaphore) + for task in tasks + ] + + # Run all merges concurrently + results = await asyncio.gather(*merge_coroutines, return_exceptions=True) + + # Process results, converting exceptions to error results + final_results: list[ParallelMergeResult] = [] + for i, result in enumerate(results): + if isinstance(result, Exception): + final_results.append(ParallelMergeResult( + file_path=tasks[i].file_path, + merged_content=None, + success=False, + error=str(result), + )) + else: + final_results.append(result) + + debug( + MODULE, + f"Parallel merge complete: {sum(1 for r in final_results if r.success)} succeeded, " + f"{sum(1 for r in final_results if not r.success)} failed" + ) + + return final_results diff --git a/auto-claude/core/workspace/__init__.py b/auto-claude/core/workspace/__init__.py index 43a9e23e..e5b5ac71 100644 --- a/auto-claude/core/workspace/__init__.py +++ b/auto-claude/core/workspace/__init__.py @@ -27,8 +27,7 @@ _spec = importlib.util.spec_from_file_location("workspace_module", _workspace_fi _workspace_module = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(_workspace_module) merge_existing_build = _workspace_module.merge_existing_build -# TODO: _run_parallel_merges not yet implemented in workspace.py -# _run_parallel_merges = _workspace_module._run_parallel_merges +_run_parallel_merges = _workspace_module._run_parallel_merges # Models and Enums # Display Functions @@ -105,7 +104,7 @@ from .setup import ( __all__ = [ # Merge Operations (from workspace.py) "merge_existing_build", - # '_run_parallel_merges', # TODO: not yet implemented - Private but used by tests + "_run_parallel_merges", # Private but used internally # Models "WorkspaceMode", "WorkspaceChoice", diff --git a/tests/test_merge_parallel.py b/tests/test_merge_parallel.py index 41face5d..be2f9159 100644 --- a/tests/test_merge_parallel.py +++ b/tests/test_merge_parallel.py @@ -21,9 +21,7 @@ import pytest sys.path.insert(0, str(Path(__file__).parent.parent / "auto-claude")) from workspace import ParallelMergeTask, ParallelMergeResult - -# _run_parallel_merges is not yet implemented - tests that use it are skipped -_run_parallel_merges = None +from core.workspace import _run_parallel_merges class TestParallelMergeDataclasses: @@ -101,11 +99,10 @@ class TestParallelMergeDataclasses: class TestParallelMergeRunner: """Tests for the parallel merge runner.""" - @pytest.mark.skip(reason="_run_parallel_merges not yet implemented") - def test_run_parallel_merges_empty_list(self, project_dir): + def test_run_parallel_merges_empty_list(self, tmp_path): """Running with empty task list returns empty results.""" import asyncio - results = asyncio.run(_run_parallel_merges([], project_dir)) + results = asyncio.run(_run_parallel_merges([], tmp_path)) assert results == [] def test_parallel_merge_task_with_data(self): @@ -123,6 +120,81 @@ class TestParallelMergeRunner: assert task.spec_name == "001-feature" +class TestSimple3WayMerge: + """Tests for the simple 3-way merge logic.""" + + def test_identical_files_merge(self, tmp_path): + """When both versions are identical, return that version.""" + import asyncio + + task = ParallelMergeTask( + file_path="src/test.py", + main_content="def main(): pass", + worktree_content="def main(): pass", # Same as main + base_content="def main(): pass", # Same as both + spec_name="001-no-change", + ) + + results = asyncio.run(_run_parallel_merges([task], tmp_path)) + assert len(results) == 1 + assert results[0].success is True + assert results[0].was_auto_merged is True + assert results[0].merged_content == "def main(): pass" + + def test_only_worktree_changed(self, tmp_path): + """When only worktree changed, take worktree version.""" + import asyncio + + task = ParallelMergeTask( + file_path="src/test.py", + main_content="def main(): pass", # Same as base + worktree_content="def main():\n print('new')", # Changed + base_content="def main(): pass", + spec_name="001-worktree-only", + ) + + results = asyncio.run(_run_parallel_merges([task], tmp_path)) + assert len(results) == 1 + assert results[0].success is True + assert results[0].was_auto_merged is True + assert "print('new')" in results[0].merged_content + + def test_only_main_changed(self, tmp_path): + """When only main changed, take main version.""" + import asyncio + + task = ParallelMergeTask( + file_path="src/test.py", + main_content="def main():\n print('main')", # Changed + worktree_content="def main(): pass", # Same as base + base_content="def main(): pass", + spec_name="001-main-only", + ) + + results = asyncio.run(_run_parallel_merges([task], tmp_path)) + assert len(results) == 1 + assert results[0].success is True + assert results[0].was_auto_merged is True + assert "print('main')" in results[0].merged_content + + def test_no_base_but_identical(self, tmp_path): + """When no base and both identical, return that version.""" + import asyncio + + task = ParallelMergeTask( + file_path="src/new.py", + main_content="# Same content", + worktree_content="# Same content", + base_content=None, # New file, no base + spec_name="001-new-identical", + ) + + results = asyncio.run(_run_parallel_merges([task], tmp_path)) + assert len(results) == 1 + assert results[0].success is True + assert results[0].was_auto_merged is True + + class TestParallelMergeIntegration: """Integration tests for parallel merge flow."""