electron-rare/Aperant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

improve context sending to merge agent

Browse Source
This commit is contained in:
AndyMik90
2025-12-16 14:35:45 +01:00
parent 01e801aace
commit e409ae825d
2 changed files with 426 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
auto-claude/merge/prompts.py
+250
View File
@@ -227,6 +227,256 @@ merged code here
return prompt
def build_conflict_only_prompt(
file_path: str,
conflicts: list[dict],
spec_name: str,
language: str,
task_intent: dict | None = None,
) -> str:
"""
Build a focused prompt that only asks AI to resolve specific conflict regions.
This is MUCH more efficient than sending entire files - the AI only needs
to resolve the actual conflicting lines, not regenerate thousands of lines.
Args:
file_path: Path to the file being merged
conflicts: List of conflict dicts with keys:
- id: Unique conflict identifier (e.g., "CONFLICT_1")
- main_lines: Lines from main branch (the <<<<<<< section)
- worktree_lines: Lines from feature branch (the >>>>>>> section)
- context_before: Few lines before the conflict for context
- context_after: Few lines after the conflict for context
spec_name: Name of the feature branch/spec
language: Programming language
task_intent: Optional dict with title, description, spec_summary
Returns:
Focused prompt asking AI to resolve only the conflict regions
"""
intent_section = ""
if task_intent:
intent_section = f"""
FEATURE INTENT: {task_intent.get('title', spec_name)}
{task_intent.get('description', '')}
"""
conflict_sections = []
for i, conflict in enumerate(conflicts, 1):
context_before = conflict.get('context_before', '')
context_after = conflict.get('context_after', '')
main_lines = conflict.get('main_lines', '')
worktree_lines = conflict.get('worktree_lines', '')
conflict_id = conflict.get('id', f'CONFLICT_{i}')
section = f"""
--- {conflict_id} ---
{f"CONTEXT BEFORE:{chr(10)}{context_before}{chr(10)}" if context_before else ""}
MAIN BRANCH VERSION:
```{language}
{main_lines}
```
FEATURE BRANCH VERSION ({spec_name}):
```{language}
{worktree_lines}
```
{f"{chr(10)}CONTEXT AFTER:{chr(10)}{context_after}" if context_after else ""}
"""
conflict_sections.append(section)
all_conflicts = "\n".join(conflict_sections)
prompt = f'''You are a code merge expert. Resolve the following {len(conflicts)} conflict(s) in {file_path}.
{intent_section}
FILE: {file_path}
LANGUAGE: {language}
{all_conflicts}
MERGE RULES:
1. Keep ALL necessary code from both versions
2. Combine changes logically - don't lose functionality from either branch
3. If both branches add different things, include both
4. If both branches modify the same thing differently, prefer the feature branch but include main's additions
5. Output MUST be syntactically valid {language}
For EACH conflict, output the resolved code in this exact format:
--- {conflicts[0].get('id', 'CONFLICT_1')} RESOLVED ---
```{language}
resolved code here
```
{f"--- CONFLICT_2 RESOLVED ---" if len(conflicts) > 1 else ""}
{f"```{language}" if len(conflicts) > 1 else ""}
{f"resolved code here" if len(conflicts) > 1 else ""}
{f"```" if len(conflicts) > 1 else ""}
(continue for each conflict)
'''
return prompt
def parse_conflict_markers(content: str) -> tuple[list[dict], list[str]]:
"""
Parse a file with git conflict markers and extract conflict regions.
Args:
content: File content with git conflict markers
Returns:
Tuple of (conflicts, clean_sections) where:
- conflicts: List of conflict dicts with main_lines, worktree_lines, etc.
- clean_sections: List of non-conflicting parts of the file (for reassembly)
"""
import re
conflicts = []
clean_sections = []
# Pattern to match git conflict markers
# <<<<<<< HEAD or <<<<<<< branch_name
# content from current branch
# =======
# content from incoming branch
# >>>>>>> branch_name or commit_hash
conflict_pattern = re.compile(
r'<<<<<<<[^\n]*\n' # Start marker
r'(.*?)' # Main/HEAD content (group 1)
r'=======\n' # Separator
r'(.*?)' # Incoming/feature content (group 2)
r'>>>>>>>[^\n]*\n?', # End marker
re.DOTALL
)
last_end = 0
for i, match in enumerate(conflict_pattern.finditer(content), 1):
# Get the clean section before this conflict
clean_before = content[last_end:match.start()]
clean_sections.append(clean_before)
# Extract context (last 3 lines before conflict)
before_lines = clean_before.rstrip().split('\n')
context_before = '\n'.join(before_lines[-3:]) if len(before_lines) >= 3 else clean_before.rstrip()
# Extract the conflict content
main_lines = match.group(1).rstrip('\n')
worktree_lines = match.group(2).rstrip('\n')
# Get context after (first 3 lines after conflict)
after_start = match.end()
after_content = content[after_start:after_start + 500] # Look ahead 500 chars
after_lines = after_content.split('\n')[:3]
context_after = '\n'.join(after_lines)
conflicts.append({
'id': f'CONFLICT_{i}',
'start': match.start(),
'end': match.end(),
'main_lines': main_lines,
'worktree_lines': worktree_lines,
'context_before': context_before,
'context_after': context_after,
})
last_end = match.end()
# Add the final clean section after last conflict
if last_end < len(content):
clean_sections.append(content[last_end:])
return conflicts, clean_sections
def reassemble_with_resolutions(
original_content: str,
conflicts: list[dict],
resolutions: dict[str, str],
) -> str:
"""
Reassemble a file by replacing conflict regions with AI resolutions.
Args:
original_content: File content with conflict markers
conflicts: List of conflict dicts from parse_conflict_markers
resolutions: Dict mapping conflict_id to resolved code
Returns:
Clean file with conflicts resolved
"""
# Sort conflicts by start position (should already be sorted, but ensure it)
sorted_conflicts = sorted(conflicts, key=lambda c: c['start'])
result_parts = []
last_end = 0
for conflict in sorted_conflicts:
# Add clean content before this conflict
result_parts.append(original_content[last_end:conflict['start']])
# Add the resolution (or keep conflict if no resolution)
conflict_id = conflict['id']
if conflict_id in resolutions:
result_parts.append(resolutions[conflict_id])
else:
# Fallback: prefer feature branch version if no resolution
result_parts.append(conflict['worktree_lines'])
last_end = conflict['end']
# Add remaining content after last conflict
result_parts.append(original_content[last_end:])
return ''.join(result_parts)
def extract_conflict_resolutions(response: str, conflicts: list[dict], language: str) -> dict[str, str]:
"""
Extract resolved code for each conflict from AI response.
Args:
response: AI response with resolved code blocks
conflicts: List of conflict dicts (to get the IDs)
language: Programming language for code block detection
Returns:
Dict mapping conflict_id to resolved code
"""
import re
resolutions = {}
# Pattern to match resolution blocks
# --- CONFLICT_1 RESOLVED --- or similar variations
resolution_pattern = re.compile(
r'---\s*(CONFLICT_\d+)\s*RESOLVED\s*---\s*\n'
r'```(?:\w+)?\n'
r'(.*?)'
r'```',
re.DOTALL | re.IGNORECASE
)
for match in resolution_pattern.finditer(response):
conflict_id = match.group(1).upper()
resolved_code = match.group(2).rstrip('\n')
resolutions[conflict_id] = resolved_code
# Fallback: if only one conflict and we can find a single code block
if len(conflicts) == 1 and not resolutions:
code_block_pattern = re.compile(
r'```(?:\w+)?\n(.*?)```',
re.DOTALL
)
matches = list(code_block_pattern.finditer(response))
if matches:
# Use the first (or only) code block
resolutions[conflicts[0]['id']] = matches[0].group(1).rstrip('\n')
return resolutions
def optimize_prompt_for_length(
context: "MergeContext",
max_content_chars: int = 50000,
auto-claude/workspace.py
+176 -19
View File
@@ -1694,9 +1694,15 @@ def _validate_merged_syntax(file_path: str, content: str, project_dir: Path) ->
timeout=30,
)
if result.returncode != 0:
# Extract first error line
error_lines = result.stderr.strip().split('\n')[:3]
return False, '\n'.join(error_lines)
# Filter out npm warnings (they go to stderr but aren't errors)
error_lines = [
line for line in result.stderr.strip().split('\n')
if line and not line.startswith('npm warn') and not line.startswith('npm WARN')
]
# Only treat as error if there are actual TypeScript errors
if error_lines:
return False, '\n'.join(error_lines[:3])
# No actual errors, just npm warnings - syntax is valid
# Try eslint for all JS/TS
result = subprocess.run(
@@ -1744,6 +1750,71 @@ def _validate_merged_syntax(file_path: str, content: str, project_dir: Path) ->
return True, ""
def _create_conflict_file_with_git(
main_content: str,
worktree_content: str,
base_content: Optional[str],
project_dir: Path,
) -> Optional[str]:
"""
Use git merge-file to create a file with conflict markers.
This produces a file with standard git conflict markers that can be
parsed to extract only the conflicting regions.
Returns the merged content with conflict markers, or None if no conflicts.
"""
import subprocess
import tempfile
if not base_content:
# Without a base, we can't do proper 3-way merge with markers
return None
try:
# Create temp files for git merge-file
with tempfile.NamedTemporaryFile(mode='w', suffix='.main', delete=False) as main_f:
main_f.write(main_content)
main_path = main_f.name
with tempfile.NamedTemporaryFile(mode='w', suffix='.base', delete=False) as base_f:
base_f.write(base_content)
base_path = base_f.name
with tempfile.NamedTemporaryFile(mode='w', suffix='.worktree', delete=False) as worktree_f:
worktree_f.write(worktree_content)
worktree_path = worktree_f.name
try:
# git merge-file modifies the first file in place
# Returns 0 if no conflicts, >0 if conflicts
result = subprocess.run(
['git', 'merge-file', '-p', main_path, base_path, worktree_path],
cwd=project_dir,
capture_output=True,
text=True,
timeout=30,
)
# Return code > 0 means conflicts exist
if result.returncode > 0 and '<<<<<<' in result.stdout:
return result.stdout
elif result.returncode == 0:
# Clean merge, no conflicts
return None
finally:
# Cleanup temp files
Path(main_path).unlink(missing_ok=True)
Path(base_path).unlink(missing_ok=True)
Path(worktree_path).unlink(missing_ok=True)
except Exception as e:
debug_warning(MODULE, f"git merge-file failed: {e}")
return None
def _merge_file_with_ai(
file_path: str,
main_content: str,
@@ -1756,7 +1827,12 @@ def _merge_file_with_ai(
"""
Use AI to merge a conflicting file.
OPTIMIZED: First tries to identify specific conflict regions and only
sends those to the AI, rather than regenerating the entire file.
This enhanced version includes:
- Conflict-region-only merging (FAST - only sends conflict lines)
- Fallback to full-file merge for complex cases
- FileTimelineTracker context for full situational awareness
- Task intent from implementation_plan.json
- Binary file detection
@@ -1766,7 +1842,14 @@ def _merge_file_with_ai(
Returns merged content, or None if AI couldn't resolve.
"""
from merge import create_claude_resolver
from merge.prompts import build_timeline_merge_prompt, build_simple_merge_prompt
from merge.prompts import (
build_timeline_merge_prompt,
build_simple_merge_prompt,
build_conflict_only_prompt,
parse_conflict_markers,
reassemble_with_resolutions,
extract_conflict_resolutions,
)
debug(MODULE, f"AI merge starting for: {file_path}",
spec_name=spec_name,
@@ -1795,8 +1878,6 @@ def _merge_file_with_ai(
print(warning(f" File too large ({max_lines} lines > {MAX_FILE_LINES_FOR_AI}), skipping AI merge"))
return None
print(muted(f" Using AI to merge changes..."))
# Create an AI resolver
resolver = create_claude_resolver()
@@ -1805,6 +1886,90 @@ def _merge_file_with_ai(
print(muted(f" AI not available, trying heuristic merge..."))
return _heuristic_merge(main_content, worktree_content, base_content)
# Determine language
language = resolver._infer_language(file_path)
debug(MODULE, "Detected language", language=language)
# Get task intent for context
task_intent = None
if project_dir:
task_intent = _get_task_intent(project_dir, spec_name)
# OPTIMIZATION: Try conflict-region-only merge first
# This is MUCH faster than sending entire files to AI
if project_dir and base_content:
conflict_file = _create_conflict_file_with_git(
main_content, worktree_content, base_content, project_dir
)
if conflict_file:
conflicts, _ = parse_conflict_markers(conflict_file)
if conflicts:
# Calculate how much smaller this approach is
total_conflict_lines = sum(
len(c['main_lines'].split('\n')) + len(c['worktree_lines'].split('\n'))
for c in conflicts
)
savings_pct = 100 - (total_conflict_lines * 100 // max(max_lines, 1))
debug(MODULE, "Using conflict-only merge (optimized)",
num_conflicts=len(conflicts),
conflict_lines=total_conflict_lines,
file_lines=max_lines,
savings_pct=savings_pct)
print(muted(f" Found {len(conflicts)} conflict region(s) ({total_conflict_lines} lines vs {max_lines} total - {savings_pct}% smaller prompt)"))
# Build focused prompt with only conflict regions
prompt = build_conflict_only_prompt(
file_path=file_path,
conflicts=conflicts,
spec_name=spec_name,
language=language,
task_intent=task_intent,
)
try:
response = resolver.ai_call_fn(
"You are an expert code merge assistant. Resolve ONLY the specific conflicts shown. Output the resolved code for each conflict.",
prompt,
)
if response:
# Extract resolutions for each conflict
resolutions = extract_conflict_resolutions(response, conflicts, language)
if resolutions:
debug(MODULE, "Extracted conflict resolutions",
num_resolutions=len(resolutions),
expected=len(conflicts))
# Reassemble the file with resolved conflicts
merged = reassemble_with_resolutions(conflict_file, conflicts, resolutions)
# Validate syntax
if project_dir:
is_valid, error_msg = _validate_merged_syntax(file_path, merged, project_dir)
if is_valid:
debug_success(MODULE, "Conflict-only merge succeeded",
file_path=file_path,
conflicts_resolved=len(resolutions))
print(success(f" ✓ Resolved {len(resolutions)} conflict(s)"))
return merged
else:
debug_warning(MODULE, "Conflict-only merge produced invalid syntax, falling back",
error=error_msg)
print(muted(f" Conflict-only merge had syntax issues, trying full-file merge..."))
else:
return merged
except Exception as e:
debug_warning(MODULE, f"Conflict-only merge failed: {e}, falling back to full-file")
print(muted(f" Conflict-only merge failed, trying full-file merge..."))
# FALLBACK: Full-file merge approach (slower but more comprehensive)
print(muted(f" Using full-file AI merge..."))
# Try to get timeline context for richer merge prompt
timeline_context = None
if project_dir:
@@ -1819,10 +1984,6 @@ def _merge_file_with_ai(
except Exception as e:
debug_warning(MODULE, f"Could not get timeline context: {e}")
# Determine language
language = resolver._infer_language(file_path)
debug(MODULE, "Detected language", language=language)
# Build prompt - use timeline context if available, fallback to simple prompt
if timeline_context and timeline_context.total_commits_behind > 0:
# Use the rich timeline-based prompt with full situational awareness
@@ -1836,14 +1997,10 @@ def _merge_file_with_ai(
# Fallback to simple three-way merge prompt
debug(MODULE, "Building simple merge prompt (no timeline context)")
# Get task intent for basic context
task_intent = None
if project_dir:
task_intent = _get_task_intent(project_dir, spec_name)
if task_intent:
debug(MODULE, "Loaded task intent",
title=task_intent.get('title'),
num_subtasks=len(task_intent.get('subtasks', [])))
if task_intent:
debug(MODULE, "Loaded task intent",
title=task_intent.get('title'),
num_subtasks=len(task_intent.get('subtasks', [])))
prompt = build_simple_merge_prompt(
file_path=file_path,
@@ -1856,7 +2013,7 @@ def _merge_file_with_ai(
)
try:
debug(MODULE, "Calling AI for merge",
debug(MODULE, "Calling AI for full-file merge",
file_path=file_path,
has_timeline_context=timeline_context is not None)