Merge-orchestrator

This commit is contained in:
AndyMik90
2025-12-16 10:14:59 +01:00
parent 488bbfa2c9
commit d8ba532beb
22 changed files with 7251 additions and 26 deletions
@@ -1248,7 +1248,16 @@ export function registerTaskHandlers(
ipcMain.handle(
IPC_CHANNELS.TASK_WORKTREE_MERGE,
async (_, taskId: string, options?: { noCommit?: boolean }): Promise<IPCResult<import('../../shared/types').WorktreeMergeResult>> => {
// Always log merge operations for debugging
const DEBUG_MERGE = true; // TODO: Change back to: process.env.DEBUG_MERGE === 'true' || process.env.DEBUG === 'true';
const debug = (...args: unknown[]) => {
if (DEBUG_MERGE) console.log('[MERGE DEBUG]', ...args);
};
try {
console.log('[MERGE] Handler called with taskId:', taskId, 'options:', options);
debug('Starting merge for taskId:', taskId, 'options:', options);
// Ensure Python environment is ready
if (!pythonEnvManager.isEnvReady()) {
const autoBuildSource = getEffectiveSourcePath();
@@ -1264,9 +1273,12 @@ export function registerTaskHandlers(
const { task, project } = findTaskAndProject(taskId);
if (!task || !project) {
debug('Task or project not found');
return { success: false, error: 'Task not found' };
}
debug('Found task:', task.specId, 'project:', project.path);
// Use run.py --merge to handle the merge
const sourcePath = getEffectiveSourcePath();
if (!sourcePath) {
@@ -1277,9 +1289,26 @@ export function registerTaskHandlers(
const specDir = path.join(project.path, project.autoBuildPath || '.auto-claude', 'specs', task.specId);
if (!existsSync(specDir)) {
debug('Spec directory not found:', specDir);
return { success: false, error: 'Spec directory not found' };
}
// Check worktree exists before merge
const worktreePath = path.join(project.path, '.worktrees', task.specId);
debug('Worktree path:', worktreePath, 'exists:', existsSync(worktreePath));
// Get git status before merge
if (DEBUG_MERGE) {
try {
const gitStatusBefore = execSync('git status --short', { cwd: project.path, encoding: 'utf-8' });
debug('Git status BEFORE merge in main project:\n', gitStatusBefore || '(clean)');
const gitBranch = execSync('git branch --show-current', { cwd: project.path, encoding: 'utf-8' }).trim();
debug('Current branch:', gitBranch);
} catch (e) {
debug('Failed to get git status before:', e);
}
}
const args = [
runScript,
'--spec', task.specId,
@@ -1292,8 +1321,11 @@ export function registerTaskHandlers(
args.push('--no-commit');
}
const pythonPath = pythonEnvManager.getPythonPath() || 'python3';
debug('Running command:', pythonPath, args.join(' '));
debug('Working directory:', sourcePath);
return new Promise((resolve) => {
const pythonPath = pythonEnvManager.getPythonPath() || 'python3';
const mergeProcess = spawn(pythonPath, args, {
cwd: sourcePath,
env: {
@@ -1306,24 +1338,57 @@ export function registerTaskHandlers(
let stderr = '';
mergeProcess.stdout.on('data', (data: Buffer) => {
stdout += data.toString();
const chunk = data.toString();
stdout += chunk;
debug('STDOUT:', chunk);
});
mergeProcess.stderr.on('data', (data: Buffer) => {
stderr += data.toString();
const chunk = data.toString();
stderr += chunk;
debug('STDERR:', chunk);
});
mergeProcess.on('close', (code: number) => {
debug('Process exited with code:', code);
debug('Full stdout:', stdout);
debug('Full stderr:', stderr);
// Get git status after merge
if (DEBUG_MERGE) {
try {
const gitStatusAfter = execSync('git status --short', { cwd: project.path, encoding: 'utf-8' });
debug('Git status AFTER merge in main project:\n', gitStatusAfter || '(clean)');
const gitDiffStaged = execSync('git diff --staged --stat', { cwd: project.path, encoding: 'utf-8' });
debug('Staged changes:\n', gitDiffStaged || '(none)');
} catch (e) {
debug('Failed to get git status after:', e);
}
}
if (code === 0) {
const isStageOnly = options?.noCommit === true;
// For stage-only: keep in human_review so user commits manually
// For full merge: mark as done
const newStatus = isStageOnly ? 'human_review' : 'done';
const planStatus = isStageOnly ? 'review' : 'completed';
debug('Merge successful. isStageOnly:', isStageOnly, 'newStatus:', newStatus);
// Persist the status change to implementation_plan.json
const planPath = path.join(specDir, AUTO_BUILD_PATHS.IMPLEMENTATION_PLAN);
try {
if (existsSync(planPath)) {
const planContent = readFileSync(planPath, 'utf-8');
const plan = JSON.parse(planContent);
plan.status = 'done';
plan.planStatus = 'completed';
plan.status = newStatus;
plan.planStatus = planStatus;
plan.updated_at = new Date().toISOString();
if (isStageOnly) {
plan.stagedAt = new Date().toISOString();
plan.stagedInMainProject = true;
}
writeFileSync(planPath, JSON.stringify(plan, null, 2));
}
} catch (persistError) {
@@ -1332,19 +1397,26 @@ export function registerTaskHandlers(
const mainWindow = getMainWindow();
if (mainWindow) {
mainWindow.webContents.send(IPC_CHANNELS.TASK_STATUS_CHANGE, taskId, 'done');
mainWindow.webContents.send(IPC_CHANNELS.TASK_STATUS_CHANGE, taskId, newStatus as TaskStatus);
}
const message = isStageOnly
? 'Changes staged in main project. Review with git status and commit when ready.'
: 'Changes merged successfully';
resolve({
success: true,
data: {
success: true,
message: 'Changes merged successfully'
message,
staged: isStageOnly,
projectPath: isStageOnly ? project.path : undefined
}
});
} else {
// Check if there were conflicts
const hasConflicts = stdout.includes('conflict') || stderr.includes('conflict');
debug('Merge failed. hasConflicts:', hasConflicts);
resolve({
success: true,
@@ -1358,6 +1430,7 @@ export function registerTaskHandlers(
});
mergeProcess.on('error', (err: Error) => {
console.error('[MERGE] Process spawn error:', err);
resolve({
success: false,
error: `Failed to run merge: ${err.message}`
@@ -1365,7 +1438,7 @@ export function registerTaskHandlers(
});
});
} catch (error) {
console.error('Failed to merge worktree:', error);
console.error('[MERGE] Exception in merge handler:', error);
return {
success: false,
error: error instanceof Error ? error.message : 'Failed to merge worktree'
@@ -1648,6 +1721,138 @@ export function registerTaskHandlers(
}
);
/**
* Preview merge conflicts before actually merging
* Uses the smart merge system to analyze potential conflicts
*/
ipcMain.handle(
IPC_CHANNELS.TASK_WORKTREE_MERGE_PREVIEW,
async (_, taskId: string): Promise<IPCResult<import('../../shared/types').WorktreeMergeResult>> => {
console.log('[IPC] TASK_WORKTREE_MERGE_PREVIEW called with taskId:', taskId);
try {
// Ensure Python environment is ready
if (!pythonEnvManager.isEnvReady()) {
console.log('[IPC] Python environment not ready, initializing...');
const autoBuildSource = getEffectiveSourcePath();
if (autoBuildSource) {
const status = await pythonEnvManager.initialize(autoBuildSource);
if (!status.ready) {
console.error('[IPC] Python environment failed to initialize:', status.error);
return { success: false, error: `Python environment not ready: ${status.error || 'Unknown error'}` };
}
} else {
console.error('[IPC] Auto Claude source not found');
return { success: false, error: 'Python environment not ready and Auto Claude source not found' };
}
}
const { task, project } = findTaskAndProject(taskId);
if (!task || !project) {
console.error('[IPC] Task not found:', taskId);
return { success: false, error: 'Task not found' };
}
console.log('[IPC] Found task:', task.specId, 'project:', project.name);
const sourcePath = getEffectiveSourcePath();
if (!sourcePath) {
console.error('[IPC] Auto Claude source not found');
return { success: false, error: 'Auto Claude source not found' };
}
const runScript = path.join(sourcePath, 'run.py');
const args = [
runScript,
'--spec', task.specId,
'--project-dir', project.path,
'--merge-preview'
];
const pythonPath = pythonEnvManager.getPythonPath() || 'python3';
console.log('[IPC] Running merge preview:', pythonPath, args.join(' '));
return new Promise((resolve) => {
const previewProcess = spawn(pythonPath, args, {
cwd: sourcePath,
env: { ...process.env, PYTHONUNBUFFERED: '1', DEBUG: 'true' }
});
let stdout = '';
let stderr = '';
previewProcess.stdout.on('data', (data: Buffer) => {
const chunk = data.toString();
stdout += chunk;
console.log('[IPC] merge-preview stdout:', chunk);
});
previewProcess.stderr.on('data', (data: Buffer) => {
const chunk = data.toString();
stderr += chunk;
console.log('[IPC] merge-preview stderr:', chunk);
});
previewProcess.on('close', (code: number) => {
console.log('[IPC] merge-preview process exited with code:', code);
if (code === 0) {
try {
// Parse JSON output from Python
const result = JSON.parse(stdout.trim());
console.log('[IPC] merge-preview result:', JSON.stringify(result, null, 2));
resolve({
success: true,
data: {
success: result.success,
message: result.error || 'Preview completed',
preview: {
files: result.files || [],
conflicts: result.conflicts || [],
summary: result.summary || {
totalFiles: 0,
conflictFiles: 0,
totalConflicts: 0,
autoMergeable: 0
}
}
}
});
} catch (parseError) {
console.error('[IPC] Failed to parse preview result:', parseError);
console.error('[IPC] stdout:', stdout);
console.error('[IPC] stderr:', stderr);
resolve({
success: false,
error: `Failed to parse preview result: ${stderr || stdout}`
});
}
} else {
console.error('[IPC] Preview failed with exit code:', code);
console.error('[IPC] stderr:', stderr);
console.error('[IPC] stdout:', stdout);
resolve({
success: false,
error: `Preview failed: ${stderr || stdout}`
});
}
});
previewProcess.on('error', (err: Error) => {
console.error('[IPC] merge-preview spawn error:', err);
resolve({
success: false,
error: `Failed to run preview: ${err.message}`
});
});
});
} catch (error) {
console.error('[IPC] TASK_WORKTREE_MERGE_PREVIEW error:', error);
return {
success: false,
error: error instanceof Error ? error.message : 'Failed to preview merge'
};
}
}
);
// Setup task log service event forwarding to renderer
taskLogService.on('logs-changed', (specId: string, logs: import('../../shared/types').TaskLogs) => {
const mainWindow = getMainWindow();
@@ -47,6 +47,7 @@ export interface TaskAPI {
getWorktreeStatus: (taskId: string) => Promise<IPCResult<import('../../shared/types').WorktreeStatus>>;
getWorktreeDiff: (taskId: string) => Promise<IPCResult<import('../../shared/types').WorktreeDiff>>;
mergeWorktree: (taskId: string, options?: { noCommit?: boolean }) => Promise<IPCResult<import('../../shared/types').WorktreeMergeResult>>;
mergeWorktreePreview: (taskId: string) => Promise<IPCResult<import('../../shared/types').WorktreeMergeResult>>;
discardWorktree: (taskId: string) => Promise<IPCResult<import('../../shared/types').WorktreeDiscardResult>>;
listWorktrees: (projectId: string) => Promise<IPCResult<import('../../shared/types').WorktreeListResult>>;
archiveTasks: (projectId: string, taskIds: string[], version?: string) => Promise<IPCResult<boolean>>;
@@ -129,6 +130,9 @@ export const createTaskAPI = (): TaskAPI => ({
mergeWorktree: (taskId: string, options?: { noCommit?: boolean }): Promise<IPCResult<import('../../shared/types').WorktreeMergeResult>> =>
ipcRenderer.invoke(IPC_CHANNELS.TASK_WORKTREE_MERGE, taskId, options),
mergeWorktreePreview: (taskId: string): Promise<IPCResult<import('../../shared/types').WorktreeMergeResult>> =>
ipcRenderer.invoke(IPC_CHANNELS.TASK_WORKTREE_MERGE_PREVIEW, taskId),
discardWorktree: (taskId: string): Promise<IPCResult<import('../../shared/types').WorktreeDiscardResult>> =>
ipcRenderer.invoke(IPC_CHANNELS.TASK_WORKTREE_DISCARD, taskId),
@@ -72,7 +72,16 @@ export function TaskDetailPanel({ task, onClose }: TaskDetailPanelProps) {
state.setWorkspaceError(null);
const result = await window.electronAPI.mergeWorktree(task.id, { noCommit: state.stageOnly });
if (result.success && result.data?.success) {
onClose();
// For stage-only: don't close the panel, show success message
// For full merge: close the panel
if (state.stageOnly && result.data.staged) {
// Changes are staged in main project - show success but keep panel open
state.setWorkspaceError(null);
state.setStagedSuccess(result.data.message || 'Changes staged in main project');
state.setStagedProjectPath(result.data.projectPath);
} else {
onClose();
}
} else {
state.setWorkspaceError(result.data?.message || result.error || 'Failed to merge changes');
}
@@ -172,6 +181,11 @@ export function TaskDetailPanel({ task, onClose }: TaskDetailPanelProps) {
showDiffDialog={state.showDiffDialog}
workspaceError={state.workspaceError}
stageOnly={state.stageOnly}
stagedSuccess={state.stagedSuccess}
stagedProjectPath={state.stagedProjectPath}
mergePreview={state.mergePreview}
isLoadingPreview={state.isLoadingPreview}
showConflictDialog={state.showConflictDialog}
onFeedbackChange={state.setFeedback}
onReject={handleReject}
onMerge={handleMerge}
@@ -179,6 +193,8 @@ export function TaskDetailPanel({ task, onClose }: TaskDetailPanelProps) {
onShowDiscardDialog={state.setShowDiscardDialog}
onShowDiffDialog={state.setShowDiffDialog}
onStageOnlyChange={state.setStageOnly}
onShowConflictDialog={state.setShowConflictDialog}
onLoadMergePreview={state.loadMergePreview}
/>
)}
</div>
@@ -9,7 +9,11 @@ import {
FolderX,
Loader2,
AlertCircle,
RotateCcw
RotateCcw,
Search,
CheckCircle,
AlertTriangle,
XCircle
} from 'lucide-react';
import { Button } from '../ui/button';
import { Textarea } from '../ui/textarea';
@@ -25,7 +29,7 @@ import {
} from '../ui/alert-dialog';
import { Badge } from '../ui/badge';
import { cn } from '../../lib/utils';
import type { Task, WorktreeStatus, WorktreeDiff } from '../../../shared/types';
import type { Task, WorktreeStatus, WorktreeDiff, MergeConflict, MergeStats } from '../../../shared/types';
interface TaskReviewProps {
task: Task;
@@ -40,6 +44,11 @@ interface TaskReviewProps {
showDiffDialog: boolean;
workspaceError: string | null;
stageOnly: boolean;
stagedSuccess: string | null;
stagedProjectPath: string | undefined;
mergePreview: { files: string[]; conflicts: MergeConflict[]; summary: MergeStats } | null;
isLoadingPreview: boolean;
showConflictDialog: boolean;
onFeedbackChange: (value: string) => void;
onReject: () => void;
onMerge: () => void;
@@ -47,6 +56,8 @@ interface TaskReviewProps {
onShowDiscardDialog: (show: boolean) => void;
onShowDiffDialog: (show: boolean) => void;
onStageOnlyChange: (value: boolean) => void;
onShowConflictDialog: (show: boolean) => void;
onLoadMergePreview: () => void;
}
export function TaskReview({
@@ -62,20 +73,95 @@ export function TaskReview({
showDiffDialog,
workspaceError,
stageOnly,
stagedSuccess,
stagedProjectPath,
mergePreview,
isLoadingPreview,
showConflictDialog,
onFeedbackChange,
onReject,
onMerge,
onDiscard,
onShowDiscardDialog,
onShowDiffDialog,
onStageOnlyChange
onStageOnlyChange,
onShowConflictDialog,
onLoadMergePreview
}: TaskReviewProps) {
// Helper function to get severity icon
const getSeverityIcon = (severity: string) => {
switch (severity) {
case 'none':
case 'low':
return <CheckCircle className="h-4 w-4 text-success" />;
case 'medium':
return <AlertTriangle className="h-4 w-4 text-warning" />;
case 'high':
case 'critical':
return <XCircle className="h-4 w-4 text-destructive" />;
default:
return <AlertCircle className="h-4 w-4 text-muted-foreground" />;
}
};
// Helper to get severity badge variant
const getSeverityVariant = (severity: string) => {
switch (severity) {
case 'none':
case 'low':
return 'bg-success/10 text-success';
case 'medium':
return 'bg-warning/10 text-warning';
case 'high':
case 'critical':
return 'bg-destructive/10 text-destructive';
default:
return 'bg-muted text-muted-foreground';
}
};
return (
<div className="space-y-4">
{/* Section divider */}
<div className="section-divider-gradient" />
{/* Workspace Status */}
{/* Staged Success Message */}
{stagedSuccess && (
<div className="rounded-xl border border-success/30 bg-success/10 p-4">
<h3 className="font-medium text-sm text-foreground mb-2 flex items-center gap-2">
<GitMerge className="h-4 w-4 text-success" />
Changes Staged Successfully
</h3>
<p className="text-sm text-muted-foreground mb-3">
{stagedSuccess}
</p>
<div className="bg-background/50 rounded-lg p-3 mb-3">
<p className="text-xs text-muted-foreground mb-2">Next steps:</p>
<ol className="text-xs text-muted-foreground space-y-1 list-decimal list-inside">
<li>Open your project in your IDE or terminal</li>
<li>Review the staged changes with <code className="bg-background px-1 rounded">git status</code> and <code className="bg-background px-1 rounded">git diff --staged</code></li>
<li>Commit when ready: <code className="bg-background px-1 rounded">git commit -m "your message"</code></li>
</ol>
</div>
{stagedProjectPath && (
<Button
variant="outline"
size="sm"
onClick={() => {
window.electronAPI.createTerminal({
id: `project-${task.id}`,
cwd: stagedProjectPath
});
}}
className="w-full"
>
<ExternalLink className="mr-2 h-4 w-4" />
Open Project in Terminal
</Button>
)}
</div>
)}
{/* Workspace Status - hide if staging was successful (worktree is deleted after staging) */}
{isLoadingWorktree ? (
<div className="rounded-xl border border-border bg-secondary/30 p-4">
<div className="flex items-center gap-2 text-muted-foreground">
@@ -83,7 +169,7 @@ export function TaskReview({
<span className="text-sm">Loading workspace info...</span>
</div>
</div>
) : worktreeStatus?.exists ? (
) : worktreeStatus?.exists && !stagedSuccess ? (
<div className="review-section-highlight">
<h3 className="font-medium text-sm text-foreground mb-3 flex items-center gap-2">
<GitBranch className="h-4 w-4 text-purple-400" />
@@ -141,6 +227,28 @@ export function TaskReview({
<Eye className="mr-2 h-4 w-4" />
View Changes
</Button>
<Button
variant="outline"
size="sm"
onClick={() => {
console.log('[TaskReview] Check Conflicts button clicked');
onLoadMergePreview();
}}
disabled={isLoadingPreview}
className="flex-1"
>
{isLoadingPreview ? (
<>
<Loader2 className="mr-2 h-4 w-4 animate-spin" />
Checking...
</>
) : (
<>
<Search className="mr-2 h-4 w-4" />
Check Conflicts
</>
)}
</Button>
{worktreeStatus.worktreePath && (
<Button
variant="outline"
@@ -158,6 +266,60 @@ export function TaskReview({
)}
</div>
{/* Merge Preview Summary */}
{mergePreview && (
<div className={cn(
"rounded-lg p-3 mb-3 border",
mergePreview.conflicts.length === 0
? "bg-success/10 border-success/30"
: mergePreview.conflicts.some(c => c.severity === 'high' || c.severity === 'critical')
? "bg-destructive/10 border-destructive/30"
: "bg-warning/10 border-warning/30"
)}>
<div className="flex items-center justify-between mb-2">
<span className="text-sm font-medium flex items-center gap-2">
{mergePreview.conflicts.length === 0 ? (
<>
<CheckCircle className="h-4 w-4 text-success" />
No Conflicts Detected
</>
) : (
<>
<AlertTriangle className="h-4 w-4 text-warning" />
{mergePreview.conflicts.length} Conflict{mergePreview.conflicts.length !== 1 ? 's' : ''} Found
</>
)}
</span>
{mergePreview.conflicts.length > 0 && (
<Button
variant="ghost"
size="sm"
onClick={() => onShowConflictDialog(true)}
className="h-7 text-xs"
>
View Details
</Button>
)}
</div>
<div className="grid grid-cols-2 gap-2 text-xs text-muted-foreground">
<div>Files to merge: {mergePreview.summary.totalFiles}</div>
{mergePreview.conflicts.length > 0 ? (
<>
<div>Auto-mergeable: {mergePreview.summary.autoMergeable}</div>
{mergePreview.summary.aiResolved !== undefined && (
<div>AI resolved: {mergePreview.summary.aiResolved}</div>
)}
{mergePreview.summary.humanRequired !== undefined && mergePreview.summary.humanRequired > 0 && (
<div className="text-warning">Manual review: {mergePreview.summary.humanRequired}</div>
)}
</>
) : (
<div className="text-success">Ready to merge</div>
)}
</div>
</div>
)}
{/* Stage Only Option */}
<label className="flex items-center gap-2 text-sm text-muted-foreground cursor-pointer select-none">
<input
@@ -366,6 +528,94 @@ export function TaskReview({
</AlertDialogFooter>
</AlertDialogContent>
</AlertDialog>
{/* Conflict Details Dialog */}
<AlertDialog open={showConflictDialog} onOpenChange={onShowConflictDialog}>
<AlertDialogContent className="max-w-2xl max-h-[80vh] overflow-hidden flex flex-col">
<AlertDialogHeader>
<AlertDialogTitle className="flex items-center gap-2">
<AlertTriangle className="h-5 w-5 text-warning" />
Merge Conflicts Preview
</AlertDialogTitle>
<AlertDialogDescription>
{mergePreview?.conflicts.length || 0} potential conflict{(mergePreview?.conflicts.length || 0) !== 1 ? 's' : ''} detected.
{mergePreview && mergePreview.summary.autoMergeable > 0 && (
<span className="text-success ml-1">
{mergePreview.summary.autoMergeable} can be auto-merged.
</span>
)}
</AlertDialogDescription>
</AlertDialogHeader>
<div className="flex-1 overflow-auto min-h-0 -mx-6 px-6">
{mergePreview?.conflicts && mergePreview.conflicts.length > 0 ? (
<div className="space-y-3">
{mergePreview.conflicts.map((conflict, idx) => (
<div
key={idx}
className={cn(
"p-3 rounded-lg border",
conflict.canAutoMerge
? "bg-secondary/30 border-border"
: conflict.severity === 'high' || conflict.severity === 'critical'
? "bg-destructive/10 border-destructive/30"
: "bg-warning/10 border-warning/30"
)}
>
<div className="flex items-start justify-between gap-2 mb-2">
<div className="flex items-center gap-2 min-w-0 flex-1">
{getSeverityIcon(conflict.severity)}
<span className="text-sm font-mono truncate">{conflict.file}</span>
</div>
<div className="flex items-center gap-2 shrink-0">
<Badge
variant="secondary"
className={cn('text-xs', getSeverityVariant(conflict.severity))}
>
{conflict.severity}
</Badge>
{conflict.canAutoMerge && (
<Badge variant="secondary" className="text-xs bg-success/10 text-success">
auto-merge
</Badge>
)}
</div>
</div>
<div className="text-xs text-muted-foreground space-y-1">
{conflict.location && (
<div><span className="text-foreground/70">Location:</span> {conflict.location}</div>
)}
{conflict.reason && (
<div><span className="text-foreground/70">Reason:</span> {conflict.reason}</div>
)}
{conflict.strategy && (
<div><span className="text-foreground/70">Strategy:</span> {conflict.strategy}</div>
)}
</div>
</div>
))}
</div>
) : (
<div className="text-center py-8 text-muted-foreground">
No conflicts detected
</div>
)}
</div>
<AlertDialogFooter className="mt-4">
<AlertDialogCancel>Close</AlertDialogCancel>
<AlertDialogAction
onClick={(e) => {
e.preventDefault();
onShowConflictDialog(false);
onMerge();
}}
className="bg-primary"
>
<GitMerge className="mr-2 h-4 w-4" />
{stageOnly ? 'Stage Anyway' : 'Merge Anyway'}
</AlertDialogAction>
</AlertDialogFooter>
</AlertDialogContent>
</AlertDialog>
</div>
);
}
@@ -1,7 +1,7 @@
import { useState, useRef, useEffect, useCallback } from 'react';
import { useProjectStore } from '../../../stores/project-store';
import { checkTaskRunning, isIncompleteHumanReview, getTaskProgress } from '../../../stores/task-store';
import type { Task, TaskLogs, TaskLogPhase, WorktreeStatus, WorktreeDiff } from '../../../../shared/types';
import type { Task, TaskLogs, TaskLogPhase, WorktreeStatus, WorktreeDiff, MergeConflict, MergeStats } from '../../../../shared/types';
export interface UseTaskDetailOptions {
task: Task;
@@ -28,12 +28,23 @@ export function useTaskDetail({ task }: UseTaskDetailOptions) {
const [workspaceError, setWorkspaceError] = useState<string | null>(null);
const [showDiffDialog, setShowDiffDialog] = useState(false);
const [stageOnly, setStageOnly] = useState(task.status === 'human_review');
const [stagedSuccess, setStagedSuccess] = useState<string | null>(null);
const [stagedProjectPath, setStagedProjectPath] = useState<string | undefined>(undefined);
const [phaseLogs, setPhaseLogs] = useState<TaskLogs | null>(null);
const [isLoadingLogs, setIsLoadingLogs] = useState(false);
const [expandedPhases, setExpandedPhases] = useState<Set<TaskLogPhase>>(new Set());
const logsEndRef = useRef<HTMLDivElement>(null);
const logsContainerRef = useRef<HTMLDivElement>(null);
// Merge preview state
const [mergePreview, setMergePreview] = useState<{
files: string[];
conflicts: MergeConflict[];
summary: MergeStats;
} | null>(null);
const [isLoadingPreview, setIsLoadingPreview] = useState(false);
const [showConflictDialog, setShowConflictDialog] = useState(false);
const selectedProject = useProjectStore((state) => state.getSelectedProject());
const isRunning = task.status === 'in_progress';
const needsReview = task.status === 'human_review';
@@ -170,6 +181,39 @@ export function useTaskDetail({ task }: UseTaskDetailOptions) {
});
}, []);
// Load merge preview (conflict detection)
const loadMergePreview = useCallback(async () => {
console.log('%c[useTaskDetail] loadMergePreview called for task:', 'color: cyan; font-weight: bold;', task.id);
setIsLoadingPreview(true);
try {
console.log('[useTaskDetail] Calling mergeWorktreePreview...');
const result = await window.electronAPI.mergeWorktreePreview(task.id);
console.log('%c[useTaskDetail] mergeWorktreePreview result:', 'color: lime; font-weight: bold;', JSON.stringify(result, null, 2));
if (result.success && result.data?.preview) {
const previewData = result.data.preview;
console.log('%c[useTaskDetail] Setting merge preview:', 'color: lime; font-weight: bold;', previewData);
console.log(' - files:', previewData.files);
console.log(' - conflicts:', previewData.conflicts);
console.log(' - summary:', previewData.summary);
setMergePreview(previewData);
// Show conflict dialog if there are conflicts that need attention
if (previewData.conflicts.length > 0) {
setShowConflictDialog(true);
}
} else {
console.warn('%c[useTaskDetail] Preview not successful or no preview data:', 'color: orange;', result);
console.warn(' - success:', result.success);
console.warn(' - data:', result.data);
console.warn(' - error:', result.error);
}
} catch (err) {
console.error('%c[useTaskDetail] Failed to load merge preview:', 'color: red; font-weight: bold;', err);
} finally {
console.log('[useTaskDetail] Setting isLoadingPreview to false');
setIsLoadingPreview(false);
}
}, [task.id]);
return {
// State
feedback,
@@ -192,6 +236,8 @@ export function useTaskDetail({ task }: UseTaskDetailOptions) {
workspaceError,
showDiffDialog,
stageOnly,
stagedSuccess,
stagedProjectPath,
phaseLogs,
isLoadingLogs,
expandedPhases,
@@ -204,6 +250,9 @@ export function useTaskDetail({ task }: UseTaskDetailOptions) {
hasActiveExecution,
isIncomplete,
taskProgress,
mergePreview,
isLoadingPreview,
showConflictDialog,
// Setters
setFeedback,
@@ -226,12 +275,18 @@ export function useTaskDetail({ task }: UseTaskDetailOptions) {
setWorkspaceError,
setShowDiffDialog,
setStageOnly,
setStagedSuccess,
setStagedProjectPath,
setPhaseLogs,
setIsLoadingLogs,
setExpandedPhases,
setMergePreview,
setIsLoadingPreview,
setShowConflictDialog,
// Handlers
handleLogsScroll,
togglePhase,
loadMergePreview,
};
}
@@ -246,6 +246,34 @@ const browserMockAPI: ElectronAPI = {
}
}),
mergeWorktreePreview: async () => ({
success: true,
data: {
success: true,
message: 'Preview generated',
preview: {
files: ['src/index.ts', 'src/utils.ts'],
conflicts: [
{
file: 'src/utils.ts',
location: 'lines 10-15',
tasks: ['task-001'],
severity: 'low' as const,
canAutoMerge: true,
strategy: 'append',
reason: 'Non-overlapping additions'
}
],
summary: {
totalFiles: 2,
conflictFiles: 1,
totalConflicts: 1,
autoMergeable: 1
}
}
}
}),
discardWorktree: async () => ({
success: true,
data: {
@@ -47,11 +47,17 @@ export const useTaskStore = create<TaskState>((set, get) => ({
updateTaskStatus: (taskId, status) =>
set((state) => ({
tasks: state.tasks.map((t) =>
t.id === taskId || t.specId === taskId
? { ...t, status, updatedAt: new Date() }
: t
)
tasks: state.tasks.map((t) => {
if (t.id !== taskId && t.specId !== taskId) return t;
// When status goes to backlog, reset execution progress to idle
// This ensures the planning/coding animation stops when task is stopped
const executionProgress = status === 'backlog'
? { phase: 'idle' as ExecutionPhase, phaseProgress: 0, overallProgress: 0 }
: t.executionProgress;
return { ...t, status, executionProgress, updatedAt: new Date() };
})
})),
updateTaskFromPlan: (taskId, plan) =>
+1
View File
@@ -146,6 +146,7 @@ export const IPC_CHANNELS = {
TASK_WORKTREE_STATUS: 'task:worktreeStatus',
TASK_WORKTREE_DIFF: 'task:worktreeDiff',
TASK_WORKTREE_MERGE: 'task:worktreeMerge',
TASK_WORKTREE_MERGE_PREVIEW: 'task:worktreeMergePreview', // Preview merge conflicts before merging
TASK_WORKTREE_DISCARD: 'task:worktreeDiscard',
TASK_LIST_WORKTREES: 'task:listWorktrees',
TASK_ARCHIVE: 'task:archive',
+1
View File
@@ -133,6 +133,7 @@ export interface ElectronAPI {
getWorktreeStatus: (taskId: string) => Promise<IPCResult<WorktreeStatus>>;
getWorktreeDiff: (taskId: string) => Promise<IPCResult<WorktreeDiff>>;
mergeWorktree: (taskId: string, options?: { noCommit?: boolean }) => Promise<IPCResult<WorktreeMergeResult>>;
mergeWorktreePreview: (taskId: string) => Promise<IPCResult<WorktreeMergeResult>>;
discardWorktree: (taskId: string) => Promise<IPCResult<WorktreeDiscardResult>>;
listWorktrees: (projectId: string) => Promise<IPCResult<WorktreeListResult>>;
+35
View File
@@ -271,10 +271,45 @@ export interface WorktreeDiffFile {
deletions: number;
}
// Conflict severity levels from merge system
export type ConflictSeverity = 'none' | 'low' | 'medium' | 'high' | 'critical';
// Information about a detected conflict
export interface MergeConflict {
file: string;
location: string;
tasks: string[];
severity: ConflictSeverity;
canAutoMerge: boolean;
strategy?: string;
reason: string;
}
// Summary statistics from merge preview/execution
export interface MergeStats {
totalFiles: number;
conflictFiles: number;
totalConflicts: number;
autoMergeable: number;
aiResolved?: number;
humanRequired?: number;
}
export interface WorktreeMergeResult {
success: boolean;
message: string;
conflictFiles?: string[];
staged?: boolean;
projectPath?: string;
// New conflict info from smart merge
conflicts?: MergeConflict[];
stats?: MergeStats;
// Preview mode results
preview?: {
files: string[];
conflicts: MergeConflict[];
summary: MergeStats;
};
}
export interface WorktreeDiscardResult {
+16 -1
View File
@@ -161,6 +161,11 @@ Environment Variables:
action="store_true",
help="With --merge: stage changes but don't commit (review in IDE first)",
)
parser.add_argument(
"--merge-preview",
action="store_true",
help="Preview merge conflicts without actually merging (returns JSON)",
)
# QA options
parser.add_argument(
@@ -297,8 +302,18 @@ def main() -> None:
debug_success("run.py", "Spec found", spec_dir=str(spec_dir))
# Handle build management commands
if args.merge_preview:
from cli.workspace_commands import handle_merge_preview_command
result = handle_merge_preview_command(project_dir, spec_dir.name)
# Output as JSON for the UI to parse
import json
print(json.dumps(result))
return
if args.merge:
handle_merge_command(project_dir, spec_dir.name, no_commit=args.no_commit)
success = handle_merge_command(project_dir, spec_dir.name, no_commit=args.no_commit)
if not success:
sys.exit(1)
return
if args.review:
+134 -2
View File
@@ -27,10 +27,24 @@ from workspace import (
from .utils import print_banner
# Import debug utilities
try:
from debug import debug, debug_detailed, debug_verbose, debug_success, debug_error, debug_section, is_debug_enabled
except ImportError:
def debug(*args, **kwargs): pass
def debug_detailed(*args, **kwargs): pass
def debug_verbose(*args, **kwargs): pass
def debug_success(*args, **kwargs): pass
def debug_error(*args, **kwargs): pass
def debug_section(*args, **kwargs): pass
def is_debug_enabled(): return False
MODULE = "cli.workspace_commands"
def handle_merge_command(
project_dir: Path, spec_name: str, no_commit: bool = False
) -> None:
) -> bool:
"""
Handle the --merge command.
@@ -38,8 +52,11 @@ def handle_merge_command(
project_dir: Project root directory
spec_name: Name of the spec
no_commit: If True, stage changes but don't commit
Returns:
True if merge succeeded, False otherwise
"""
merge_existing_build(project_dir, spec_name, no_commit=no_commit)
return merge_existing_build(project_dir, spec_name, no_commit=no_commit)
def handle_review_command(project_dir: Path, spec_name: str) -> None:
@@ -111,3 +128,118 @@ def handle_cleanup_worktrees_command(project_dir: Path) -> None:
"""
print_banner()
cleanup_all_worktrees(project_dir, confirm=True)
def handle_merge_preview_command(project_dir: Path, spec_name: str) -> dict:
"""
Handle the --merge-preview command.
Returns a JSON-serializable preview of merge conflicts without
actually performing the merge. This is used by the UI to show
potential conflicts before the user clicks "Stage Changes".
Args:
project_dir: Project root directory
spec_name: Name of the spec
Returns:
Dictionary with preview information
"""
debug_section(MODULE, "Merge Preview Command")
debug(MODULE, "handle_merge_preview_command() called",
project_dir=str(project_dir),
spec_name=spec_name)
from workspace import get_existing_build_worktree
from merge import MergeOrchestrator
worktree_path = get_existing_build_worktree(project_dir, spec_name)
debug(MODULE, f"Worktree lookup result",
worktree_path=str(worktree_path) if worktree_path else None)
if not worktree_path:
debug_error(MODULE, f"No existing build found for '{spec_name}'")
return {
"success": False,
"error": f"No existing build found for '{spec_name}'",
"files": [],
"conflicts": [],
"summary": {
"totalFiles": 0,
"conflictFiles": 0,
"totalConflicts": 0,
"autoMergeable": 0,
}
}
try:
debug(MODULE, "Initializing MergeOrchestrator for preview...")
# Initialize the orchestrator
orchestrator = MergeOrchestrator(
project_dir,
enable_ai=False, # Don't use AI for preview
dry_run=True, # Don't write anything
)
# Refresh evolution data from the worktree
debug(MODULE, f"Refreshing evolution data from worktree: {worktree_path}")
orchestrator.evolution_tracker.refresh_from_git(spec_name, worktree_path)
# Get merge preview
debug(MODULE, "Generating merge preview...")
preview = orchestrator.preview_merge([spec_name])
# Transform to UI-friendly format
conflicts = []
for c in preview.get("conflicts", []):
debug_verbose(MODULE, f"Processing conflict",
file=c.get("file", ""),
severity=c.get("severity", "unknown"))
conflicts.append({
"file": c.get("file", ""),
"location": c.get("location", ""),
"tasks": c.get("tasks", []),
"severity": c.get("severity", "unknown"),
"canAutoMerge": c.get("can_auto_merge", False),
"strategy": c.get("strategy"),
"reason": c.get("reason", ""),
})
summary = preview.get("summary", {})
result = {
"success": True,
"files": preview.get("files_to_merge", []),
"conflicts": conflicts,
"summary": {
"totalFiles": summary.get("total_files", 0),
"conflictFiles": summary.get("conflict_files", 0),
"totalConflicts": summary.get("total_conflicts", 0),
"autoMergeable": summary.get("auto_mergeable", 0),
}
}
debug_success(MODULE, "Merge preview complete",
total_files=result["summary"]["totalFiles"],
total_conflicts=result["summary"]["totalConflicts"],
auto_mergeable=result["summary"]["autoMergeable"])
return result
except Exception as e:
debug_error(MODULE, f"Merge preview failed", error=str(e))
import traceback
debug_verbose(MODULE, "Exception traceback", traceback=traceback.format_exc())
return {
"success": False,
"error": str(e),
"files": [],
"conflicts": [],
"summary": {
"totalFiles": 0,
"conflictFiles": 0,
"totalConflicts": 0,
"autoMergeable": 0,
}
}
+64
View File
@@ -0,0 +1,64 @@
"""
Merge AI System
===============
Intent-aware merge system for multi-agent collaborative development.
This module provides semantic understanding of code changes and intelligent
conflict resolution, enabling multiple AI agents to work in parallel without
traditional merge conflicts.
Components:
- SemanticAnalyzer: Tree-sitter based semantic change extraction
- ConflictDetector: Rule-based conflict detection and compatibility analysis
- AutoMerger: Deterministic merge strategies (no AI needed)
- AIResolver: Minimal-context AI resolution for ambiguous conflicts
- FileEvolutionTracker: Baseline capture and change tracking
- MergeOrchestrator: Main pipeline coordinator
Usage:
from merge import MergeOrchestrator
orchestrator = MergeOrchestrator(project_dir)
result = orchestrator.merge_task("task-001-feature")
"""
from .types import (
ChangeType,
SemanticChange,
FileAnalysis,
ConflictRegion,
ConflictSeverity,
MergeStrategy,
MergeResult,
MergeDecision,
TaskSnapshot,
FileEvolution,
)
from .semantic_analyzer import SemanticAnalyzer
from .conflict_detector import ConflictDetector
from .auto_merger import AutoMerger
from .file_evolution import FileEvolutionTracker
from .ai_resolver import AIResolver
from .orchestrator import MergeOrchestrator
__all__ = [
# Types
"ChangeType",
"SemanticChange",
"FileAnalysis",
"ConflictRegion",
"ConflictSeverity",
"MergeStrategy",
"MergeResult",
"MergeDecision",
"TaskSnapshot",
"FileEvolution",
# Components
"SemanticAnalyzer",
"ConflictDetector",
"AutoMerger",
"FileEvolutionTracker",
"AIResolver",
"MergeOrchestrator",
]
+633
View File
@@ -0,0 +1,633 @@
"""
AI Resolver
===========
Handles conflicts that cannot be resolved by deterministic rules.
This component is called ONLY when the AutoMerger cannot handle a conflict.
It uses minimal context to reduce token usage:
1. Only the conflict region, not the entire file
2. Task intents (1 sentence each)
3. Semantic change descriptions
4. The baseline code for reference
The AI is given a focused task: merge these specific changes.
No file exploration, no open-ended questions.
"""
from __future__ import annotations
import json
import logging
import re
from dataclasses import dataclass
from typing import Any, Callable, Optional
from .types import (
ChangeType,
ConflictRegion,
ConflictSeverity,
MergeDecision,
MergeResult,
MergeStrategy,
SemanticChange,
TaskSnapshot,
)
logger = logging.getLogger(__name__)
@dataclass
class ConflictContext:
"""
Minimal context needed to resolve a conflict.
This is what gets sent to the AI - optimized for minimal tokens.
"""
file_path: str
location: str
baseline_code: str # The code before any task modified it
task_changes: list[tuple[str, str, list[SemanticChange]]] # (task_id, intent, changes)
conflict_description: str
language: str = "unknown"
def to_prompt_context(self) -> str:
"""Format as context for the AI prompt."""
lines = [
f"File: {self.file_path}",
f"Location: {self.location}",
f"Language: {self.language}",
"",
"--- BASELINE CODE (before any changes) ---",
self.baseline_code,
"--- END BASELINE ---",
"",
"CHANGES FROM EACH TASK:",
]
for task_id, intent, changes in self.task_changes:
lines.append(f"\n[Task: {task_id}]")
lines.append(f"Intent: {intent}")
lines.append("Changes:")
for change in changes:
lines.append(f" - {change.change_type.value}: {change.target}")
if change.content_after:
# Truncate long content
content = change.content_after
if len(content) > 500:
content = content[:500] + "... (truncated)"
lines.append(f" Code: {content}")
lines.extend([
"",
f"CONFLICT: {self.conflict_description}",
])
return "\n".join(lines)
@property
def estimated_tokens(self) -> int:
"""Rough estimate of tokens in this context."""
text = self.to_prompt_context()
# Rough estimate: 4 chars per token for code
return len(text) // 4
# Type for the AI call function
AICallFunction = Callable[[str, str], str]
class AIResolver:
"""
Resolves conflicts using AI with minimal context.
This class:
1. Builds minimal conflict context
2. Creates focused prompts
3. Calls AI and parses response
4. Returns MergeResult with merged code
Usage:
resolver = AIResolver(ai_call_fn)
result = resolver.resolve_conflict(conflict, context)
"""
# Maximum tokens to send to AI (keeps costs down)
MAX_CONTEXT_TOKENS = 4000
# Prompt template for merge resolution
MERGE_PROMPT = '''You are a code merge assistant. Your task is to merge changes from multiple development tasks into a single coherent result.
CONTEXT:
{context}
INSTRUCTIONS:
1. Analyze what each task intended to accomplish
2. Merge the changes so that ALL task intents are preserved
3. Resolve any conflicts by understanding the semantic purpose
4. Output ONLY the merged code - no explanations
RULES:
- All imports from all tasks should be included
- All hook calls should be preserved (order matters: earlier tasks first)
- If tasks modify the same function, combine their changes logically
- If tasks wrap JSX differently, apply wrappings from outside-in (earlier task = outer)
- Preserve code style consistency
OUTPUT FORMAT:
Return only the merged code block, wrapped in triple backticks with the language:
```{language}
merged code here
```
Merge the code now:'''
def __init__(
self,
ai_call_fn: Optional[AICallFunction] = None,
max_context_tokens: int = MAX_CONTEXT_TOKENS,
):
"""
Initialize the AI resolver.
Args:
ai_call_fn: Function that calls AI. Signature: (system_prompt, user_prompt) -> response
If None, uses a stub that requires explicit calls.
max_context_tokens: Maximum tokens to include in context
"""
self.ai_call_fn = ai_call_fn
self.max_context_tokens = max_context_tokens
self._call_count = 0
self._total_tokens = 0
def set_ai_function(self, ai_call_fn: AICallFunction) -> None:
"""Set the AI call function after initialization."""
self.ai_call_fn = ai_call_fn
@property
def stats(self) -> dict[str, int]:
"""Get usage statistics."""
return {
"calls_made": self._call_count,
"estimated_tokens_used": self._total_tokens,
}
def reset_stats(self) -> None:
"""Reset usage statistics."""
self._call_count = 0
self._total_tokens = 0
def build_context(
self,
conflict: ConflictRegion,
baseline_code: str,
task_snapshots: list[TaskSnapshot],
) -> ConflictContext:
"""
Build minimal context for a conflict.
Args:
conflict: The conflict to resolve
baseline_code: Original code before any changes
task_snapshots: Snapshots from each involved task
Returns:
ConflictContext with minimal data for AI
"""
# Filter to only changes at the conflict location
task_changes: list[tuple[str, str, list[SemanticChange]]] = []
for snapshot in task_snapshots:
if snapshot.task_id not in conflict.tasks_involved:
continue
relevant_changes = [
c for c in snapshot.semantic_changes
if c.location == conflict.location or self._locations_overlap(c.location, conflict.location)
]
if relevant_changes:
task_changes.append((
snapshot.task_id,
snapshot.task_intent or "No intent specified",
relevant_changes,
))
# Determine language from file extension
language = self._infer_language(conflict.file_path)
# Build description
change_types = [ct.value for ct in conflict.change_types]
description = (
f"Tasks {', '.join(conflict.tasks_involved)} made conflicting changes: "
f"{', '.join(change_types)}. "
f"Severity: {conflict.severity.value}. "
f"{conflict.reason}"
)
return ConflictContext(
file_path=conflict.file_path,
location=conflict.location,
baseline_code=baseline_code,
task_changes=task_changes,
conflict_description=description,
language=language,
)
def _locations_overlap(self, loc1: str, loc2: str) -> bool:
"""Check if two locations might overlap."""
# Simple heuristic: if one contains the other or they share a prefix
if loc1 == loc2:
return True
if loc1.startswith(loc2) or loc2.startswith(loc1):
return True
# Check for function/class containment
if loc1.startswith("function:") and loc2.startswith("function:"):
return loc1.split(":")[1] == loc2.split(":")[1]
return False
def _infer_language(self, file_path: str) -> str:
"""Infer programming language from file path."""
ext_map = {
".py": "python",
".js": "javascript",
".ts": "typescript",
".tsx": "tsx",
".jsx": "jsx",
".go": "go",
".rs": "rust",
".java": "java",
".kt": "kotlin",
".swift": "swift",
".rb": "ruby",
".php": "php",
".css": "css",
".html": "html",
".json": "json",
".yaml": "yaml",
".yml": "yaml",
".md": "markdown",
}
for ext, lang in ext_map.items():
if file_path.endswith(ext):
return lang
return "text"
def resolve_conflict(
self,
conflict: ConflictRegion,
baseline_code: str,
task_snapshots: list[TaskSnapshot],
) -> MergeResult:
"""
Resolve a conflict using AI.
Args:
conflict: The conflict to resolve
baseline_code: Original code at the conflict location
task_snapshots: Snapshots from involved tasks
Returns:
MergeResult with the resolution
"""
if not self.ai_call_fn:
return MergeResult(
decision=MergeDecision.NEEDS_HUMAN_REVIEW,
file_path=conflict.file_path,
explanation="No AI function configured",
conflicts_remaining=[conflict],
)
# Build context
context = self.build_context(conflict, baseline_code, task_snapshots)
# Check token limit
if context.estimated_tokens > self.max_context_tokens:
logger.warning(
f"Context too large ({context.estimated_tokens} tokens), "
"flagging for human review"
)
return MergeResult(
decision=MergeDecision.NEEDS_HUMAN_REVIEW,
file_path=conflict.file_path,
explanation=f"Context too large for AI ({context.estimated_tokens} tokens)",
conflicts_remaining=[conflict],
)
# Build prompt
prompt_context = context.to_prompt_context()
prompt = self.MERGE_PROMPT.format(
context=prompt_context,
language=context.language,
)
# Call AI
try:
logger.info(f"Calling AI to resolve conflict in {conflict.file_path}")
response = self.ai_call_fn(
"You are an expert code merge assistant. Be concise and precise.",
prompt,
)
self._call_count += 1
self._total_tokens += context.estimated_tokens + len(response) // 4
# Parse response
merged_code = self._extract_code_block(response, context.language)
if merged_code:
return MergeResult(
decision=MergeDecision.AI_MERGED,
file_path=conflict.file_path,
merged_content=merged_code,
conflicts_resolved=[conflict],
ai_calls_made=1,
tokens_used=context.estimated_tokens,
explanation=f"AI resolved conflict at {conflict.location}",
)
else:
logger.warning("Could not parse AI response")
return MergeResult(
decision=MergeDecision.NEEDS_HUMAN_REVIEW,
file_path=conflict.file_path,
explanation="Could not parse AI merge response",
conflicts_remaining=[conflict],
ai_calls_made=1,
tokens_used=context.estimated_tokens,
)
except Exception as e:
logger.error(f"AI call failed: {e}")
return MergeResult(
decision=MergeDecision.FAILED,
file_path=conflict.file_path,
error=str(e),
conflicts_remaining=[conflict],
)
def _extract_code_block(self, response: str, language: str) -> Optional[str]:
"""Extract code block from AI response."""
# Try to find fenced code block
patterns = [
rf"```{language}\n(.*?)```",
rf"```{language.lower()}\n(.*?)```",
r"```\n(.*?)```",
r"```(.*?)```",
]
for pattern in patterns:
match = re.search(pattern, response, re.DOTALL)
if match:
return match.group(1).strip()
# If no code block, check if the entire response looks like code
lines = response.strip().split("\n")
if lines and not lines[0].startswith("```"):
# Assume entire response is code if it looks like it
if self._looks_like_code(response, language):
return response.strip()
return None
def _looks_like_code(self, text: str, language: str) -> bool:
"""Heuristic to check if text looks like code."""
indicators = {
"python": ["def ", "import ", "class ", "if ", "for "],
"javascript": ["function", "const ", "let ", "var ", "import ", "export "],
"typescript": ["function", "const ", "let ", "interface ", "type ", "import "],
"tsx": ["function", "const ", "return ", "import ", "export ", "<"],
"jsx": ["function", "const ", "return ", "import ", "export ", "<"],
}
lang_indicators = indicators.get(language.lower(), [])
if lang_indicators:
return any(ind in text for ind in lang_indicators)
# Generic code indicators
return any(ind in text for ind in ["=", "(", ")", "{", "}", "import", "def", "function"])
def resolve_multiple_conflicts(
self,
conflicts: list[ConflictRegion],
baseline_codes: dict[str, str],
task_snapshots: list[TaskSnapshot],
batch: bool = True,
) -> list[MergeResult]:
"""
Resolve multiple conflicts.
Args:
conflicts: List of conflicts to resolve
baseline_codes: Map of location -> baseline code
task_snapshots: All task snapshots
batch: Whether to batch conflicts (reduces API calls)
Returns:
List of MergeResults
"""
results = []
if batch and len(conflicts) > 1:
# Try to batch conflicts from the same file
by_file: dict[str, list[ConflictRegion]] = {}
for conflict in conflicts:
if conflict.file_path not in by_file:
by_file[conflict.file_path] = []
by_file[conflict.file_path].append(conflict)
for file_path, file_conflicts in by_file.items():
if len(file_conflicts) == 1:
# Single conflict, resolve individually
baseline = baseline_codes.get(file_conflicts[0].location, "")
results.append(self.resolve_conflict(
file_conflicts[0], baseline, task_snapshots
))
else:
# Multiple conflicts in same file - batch resolve
result = self._resolve_file_batch(
file_path, file_conflicts, baseline_codes, task_snapshots
)
results.append(result)
else:
# Resolve each individually
for conflict in conflicts:
baseline = baseline_codes.get(conflict.location, "")
results.append(self.resolve_conflict(
conflict, baseline, task_snapshots
))
return results
def _resolve_file_batch(
self,
file_path: str,
conflicts: list[ConflictRegion],
baseline_codes: dict[str, str],
task_snapshots: list[TaskSnapshot],
) -> MergeResult:
"""
Resolve multiple conflicts in the same file with a single AI call.
This is more efficient but may be less precise.
"""
if not self.ai_call_fn:
return MergeResult(
decision=MergeDecision.NEEDS_HUMAN_REVIEW,
file_path=file_path,
explanation="No AI function configured",
conflicts_remaining=conflicts,
)
# Combine contexts
all_contexts = []
for conflict in conflicts:
baseline = baseline_codes.get(conflict.location, "")
ctx = self.build_context(conflict, baseline, task_snapshots)
all_contexts.append(ctx)
# Check combined token limit
total_tokens = sum(ctx.estimated_tokens for ctx in all_contexts)
if total_tokens > self.max_context_tokens:
# Too big to batch, fall back to individual resolution
results = []
for conflict in conflicts:
baseline = baseline_codes.get(conflict.location, "")
results.append(self.resolve_conflict(conflict, baseline, task_snapshots))
# Combine results
merged = results[0]
for r in results[1:]:
merged.conflicts_resolved.extend(r.conflicts_resolved)
merged.conflicts_remaining.extend(r.conflicts_remaining)
merged.ai_calls_made += r.ai_calls_made
merged.tokens_used += r.tokens_used
return merged
# Build combined prompt
combined_context = "\n\n---\n\n".join(
ctx.to_prompt_context() for ctx in all_contexts
)
language = all_contexts[0].language if all_contexts else "text"
batch_prompt = f'''You are a code merge assistant. Your task is to merge changes from multiple development tasks.
There are {len(conflicts)} conflict regions in {file_path}. Resolve each one.
{combined_context}
For each conflict region, output the merged code in a separate code block labeled with the location:
## Location: <location>
```{language}
merged code
```
Resolve all conflicts now:'''
try:
response = self.ai_call_fn(
"You are an expert code merge assistant. Be concise and precise.",
batch_prompt,
)
self._call_count += 1
self._total_tokens += total_tokens + len(response) // 4
# Parse batch response
# This is a simplified parser - production would be more robust
resolved = []
remaining = []
for conflict in conflicts:
# Try to find the resolution for this location
pattern = rf"## Location: {re.escape(conflict.location)}.*?```{language}\n(.*?)```"
match = re.search(pattern, response, re.DOTALL)
if match:
resolved.append(conflict)
else:
remaining.append(conflict)
# Return combined result
if resolved:
return MergeResult(
decision=MergeDecision.AI_MERGED if not remaining else MergeDecision.NEEDS_HUMAN_REVIEW,
file_path=file_path,
merged_content=response, # Full response for manual extraction
conflicts_resolved=resolved,
conflicts_remaining=remaining,
ai_calls_made=1,
tokens_used=total_tokens,
explanation=f"Batch resolved {len(resolved)}/{len(conflicts)} conflicts",
)
else:
return MergeResult(
decision=MergeDecision.NEEDS_HUMAN_REVIEW,
file_path=file_path,
explanation="Could not parse batch AI response",
conflicts_remaining=conflicts,
ai_calls_made=1,
tokens_used=total_tokens,
)
except Exception as e:
logger.error(f"Batch AI call failed: {e}")
return MergeResult(
decision=MergeDecision.FAILED,
file_path=file_path,
error=str(e),
conflicts_remaining=conflicts,
)
def can_resolve(self, conflict: ConflictRegion) -> bool:
"""
Check if this resolver should handle a conflict.
Only handles conflicts that need AI resolution.
"""
return (
conflict.merge_strategy in {MergeStrategy.AI_REQUIRED, None}
and conflict.severity in {ConflictSeverity.MEDIUM, ConflictSeverity.HIGH}
and self.ai_call_fn is not None
)
def create_claude_resolver(
api_key: Optional[str] = None,
) -> AIResolver:
"""
Create an AIResolver configured to use Claude.
Args:
api_key: Optional API key. If None, reads from ANTHROPIC_API_KEY env var.
Returns:
Configured AIResolver
"""
import os
try:
import anthropic
except ImportError:
logger.warning("anthropic package not installed, AI resolution unavailable")
return AIResolver()
api_key = api_key or os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
logger.warning("No Anthropic API key found, AI resolution unavailable")
return AIResolver()
client = anthropic.Anthropic(api_key=api_key)
def call_claude(system: str, user: str) -> str:
response = client.messages.create(
model="claude-sonnet-4-20250514", # Fast and capable
max_tokens=4096,
system=system,
messages=[{"role": "user", "content": user}],
)
return response.content[0].text
return AIResolver(ai_call_fn=call_claude)
+631
View File
@@ -0,0 +1,631 @@
"""
Auto Merger
===========
Deterministic merge strategies that don't require AI intervention.
This module implements the merge strategies identified by ConflictDetector
as auto-mergeable. Each strategy is a pure Python algorithm that combines
changes from multiple tasks in a predictable way.
Strategies:
- COMBINE_IMPORTS: Merge import statements from multiple tasks
- HOOKS_FIRST: Add hooks at function start, then other changes
- HOOKS_THEN_WRAP: Add hooks first, then wrap return in JSX
- APPEND_FUNCTIONS: Add new functions after existing ones
- APPEND_METHODS: Add new methods to class
- COMBINE_PROPS: Merge JSX/object props
- ORDER_BY_DEPENDENCY: Analyze dependencies and order appropriately
- ORDER_BY_TIME: Apply changes in chronological order
"""
from __future__ import annotations
import logging
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
from .types import (
ChangeType,
ConflictRegion,
MergeDecision,
MergeResult,
MergeStrategy,
SemanticChange,
TaskSnapshot,
)
logger = logging.getLogger(__name__)
@dataclass
class MergeContext:
"""Context for a merge operation."""
file_path: str
baseline_content: str
task_snapshots: list[TaskSnapshot]
conflict: ConflictRegion
class AutoMerger:
"""
Performs deterministic merges without AI.
This class implements various merge strategies that can be applied
when the ConflictDetector determines changes are compatible.
Example:
merger = AutoMerger()
result = merger.merge(context, MergeStrategy.COMBINE_IMPORTS)
if result.success:
print(result.merged_content)
"""
def __init__(self):
"""Initialize the auto merger."""
self._strategy_handlers = {
MergeStrategy.COMBINE_IMPORTS: self._merge_combine_imports,
MergeStrategy.HOOKS_FIRST: self._merge_hooks_first,
MergeStrategy.HOOKS_THEN_WRAP: self._merge_hooks_then_wrap,
MergeStrategy.APPEND_FUNCTIONS: self._merge_append_functions,
MergeStrategy.APPEND_METHODS: self._merge_append_methods,
MergeStrategy.COMBINE_PROPS: self._merge_combine_props,
MergeStrategy.ORDER_BY_DEPENDENCY: self._merge_order_by_dependency,
MergeStrategy.ORDER_BY_TIME: self._merge_order_by_time,
MergeStrategy.APPEND_STATEMENTS: self._merge_append_statements,
}
def merge(
self,
context: MergeContext,
strategy: MergeStrategy,
) -> MergeResult:
"""
Perform a merge using the specified strategy.
Args:
context: The merge context with baseline and task snapshots
strategy: The merge strategy to use
Returns:
MergeResult with merged content or error
"""
handler = self._strategy_handlers.get(strategy)
if not handler:
return MergeResult(
decision=MergeDecision.FAILED,
file_path=context.file_path,
error=f"No handler for strategy: {strategy.value}",
)
try:
return handler(context)
except Exception as e:
logger.exception(f"Auto-merge failed with strategy {strategy.value}")
return MergeResult(
decision=MergeDecision.FAILED,
file_path=context.file_path,
error=f"Auto-merge failed: {str(e)}",
)
def can_handle(self, strategy: MergeStrategy) -> bool:
"""Check if this merger can handle a strategy."""
return strategy in self._strategy_handlers
# ========================================
# Strategy Implementations
# ========================================
def _merge_combine_imports(self, context: MergeContext) -> MergeResult:
"""Combine import statements from multiple tasks."""
lines = context.baseline_content.split("\n")
ext = Path(context.file_path).suffix.lower()
# Collect all imports to add
imports_to_add: list[str] = []
imports_to_remove: set[str] = set()
for snapshot in context.task_snapshots:
for change in snapshot.semantic_changes:
if change.change_type == ChangeType.ADD_IMPORT and change.content_after:
imports_to_add.append(change.content_after.strip())
elif change.change_type == ChangeType.REMOVE_IMPORT and change.content_before:
imports_to_remove.add(change.content_before.strip())
# Find where imports end in the file
import_end_line = self._find_import_section_end(lines, ext)
# Remove duplicates and already-present imports
existing_imports = set()
for i, line in enumerate(lines[:import_end_line]):
stripped = line.strip()
if self._is_import_line(stripped, ext):
existing_imports.add(stripped)
new_imports = [
imp for imp in imports_to_add
if imp not in existing_imports and imp not in imports_to_remove
]
# Remove imports that should be removed
result_lines = []
for line in lines:
if line.strip() not in imports_to_remove:
result_lines.append(line)
# Insert new imports at the import section end
if new_imports:
# Find insert position in result_lines
insert_pos = self._find_import_section_end(result_lines, ext)
for imp in reversed(new_imports):
result_lines.insert(insert_pos, imp)
merged_content = "\n".join(result_lines)
return MergeResult(
decision=MergeDecision.AUTO_MERGED,
file_path=context.file_path,
merged_content=merged_content,
conflicts_resolved=[context.conflict],
explanation=f"Combined {len(new_imports)} imports from {len(context.task_snapshots)} tasks",
)
def _merge_hooks_first(self, context: MergeContext) -> MergeResult:
"""Add hooks at function start, then apply other changes."""
content = context.baseline_content
# Collect hooks and other changes
hooks: list[str] = []
other_changes: list[SemanticChange] = []
for snapshot in context.task_snapshots:
for change in snapshot.semantic_changes:
if change.change_type == ChangeType.ADD_HOOK_CALL:
# Extract just the hook call from the change
hook_content = self._extract_hook_call(change)
if hook_content:
hooks.append(hook_content)
else:
other_changes.append(change)
# Find the function to modify
func_location = context.conflict.location
if func_location.startswith("function:"):
func_name = func_location.split(":")[1]
content = self._insert_hooks_into_function(content, func_name, hooks)
# Apply other changes (simplified - just take the latest version)
for change in other_changes:
if change.content_after:
# This is a simplification - in production we'd need smarter merging
pass
return MergeResult(
decision=MergeDecision.AUTO_MERGED,
file_path=context.file_path,
merged_content=content,
conflicts_resolved=[context.conflict],
explanation=f"Added {len(hooks)} hooks to function start",
)
def _merge_hooks_then_wrap(self, context: MergeContext) -> MergeResult:
"""Add hooks first, then wrap JSX return."""
content = context.baseline_content
hooks: list[str] = []
wraps: list[tuple[str, str]] = [] # (wrapper_component, props)
for snapshot in context.task_snapshots:
for change in snapshot.semantic_changes:
if change.change_type == ChangeType.ADD_HOOK_CALL:
hook_content = self._extract_hook_call(change)
if hook_content:
hooks.append(hook_content)
elif change.change_type == ChangeType.WRAP_JSX:
wrapper = self._extract_jsx_wrapper(change)
if wrapper:
wraps.append(wrapper)
# Get function name from conflict location
func_location = context.conflict.location
if func_location.startswith("function:"):
func_name = func_location.split(":")[1]
# First add hooks
if hooks:
content = self._insert_hooks_into_function(content, func_name, hooks)
# Then apply wraps
for wrapper_name, wrapper_props in wraps:
content = self._wrap_function_return(
content, func_name, wrapper_name, wrapper_props
)
return MergeResult(
decision=MergeDecision.AUTO_MERGED,
file_path=context.file_path,
merged_content=content,
conflicts_resolved=[context.conflict],
explanation=f"Added {len(hooks)} hooks and {len(wraps)} JSX wrappers",
)
def _merge_append_functions(self, context: MergeContext) -> MergeResult:
"""Append new functions to the file."""
content = context.baseline_content
# Collect all new functions
new_functions: list[str] = []
for snapshot in context.task_snapshots:
for change in snapshot.semantic_changes:
if change.change_type == ChangeType.ADD_FUNCTION and change.content_after:
new_functions.append(change.content_after)
# Append at the end (before any module.exports in JS)
ext = Path(context.file_path).suffix.lower()
insert_pos = self._find_function_insert_position(content, ext)
if insert_pos is not None:
lines = content.split("\n")
for func in new_functions:
lines.insert(insert_pos, "")
lines.insert(insert_pos + 1, func)
insert_pos += 2 + func.count("\n")
content = "\n".join(lines)
else:
# Just append at the end
for func in new_functions:
content += f"\n\n{func}"
return MergeResult(
decision=MergeDecision.AUTO_MERGED,
file_path=context.file_path,
merged_content=content,
conflicts_resolved=[context.conflict],
explanation=f"Appended {len(new_functions)} new functions",
)
def _merge_append_methods(self, context: MergeContext) -> MergeResult:
"""Append new methods to a class."""
content = context.baseline_content
# Collect new methods by class
new_methods: dict[str, list[str]] = {}
for snapshot in context.task_snapshots:
for change in snapshot.semantic_changes:
if change.change_type == ChangeType.ADD_METHOD and change.content_after:
# Extract class name from location
class_name = change.target.split(".")[0] if "." in change.target else None
if class_name:
if class_name not in new_methods:
new_methods[class_name] = []
new_methods[class_name].append(change.content_after)
# Insert methods into their classes
for class_name, methods in new_methods.items():
content = self._insert_methods_into_class(content, class_name, methods)
total_methods = sum(len(m) for m in new_methods.values())
return MergeResult(
decision=MergeDecision.AUTO_MERGED,
file_path=context.file_path,
merged_content=content,
conflicts_resolved=[context.conflict],
explanation=f"Added {total_methods} methods to {len(new_methods)} classes",
)
def _merge_combine_props(self, context: MergeContext) -> MergeResult:
"""Combine JSX/object props from multiple changes."""
# This is a simplified implementation
# In production, we'd parse the JSX properly
content = context.baseline_content
# Collect all prop additions
props_to_add: list[tuple[str, str]] = [] # (prop_name, prop_value)
for snapshot in context.task_snapshots:
for change in snapshot.semantic_changes:
if change.change_type == ChangeType.MODIFY_JSX_PROPS:
new_props = self._extract_new_props(change)
props_to_add.extend(new_props)
# For now, return the last version with all props
# A proper implementation would merge prop objects
if context.task_snapshots and context.task_snapshots[-1].semantic_changes:
last_change = context.task_snapshots[-1].semantic_changes[-1]
if last_change.content_after:
content = self._apply_content_change(
content, last_change.content_before, last_change.content_after
)
return MergeResult(
decision=MergeDecision.AUTO_MERGED,
file_path=context.file_path,
merged_content=content,
conflicts_resolved=[context.conflict],
explanation=f"Combined props from {len(context.task_snapshots)} tasks",
)
def _merge_order_by_dependency(self, context: MergeContext) -> MergeResult:
"""Order changes by dependency analysis."""
# Analyze dependencies between changes
ordered_changes = self._topological_sort_changes(context.task_snapshots)
content = context.baseline_content
# Apply changes in dependency order
for change in ordered_changes:
if change.content_after:
if change.change_type == ChangeType.ADD_HOOK_CALL:
func_name = change.target.split(".")[-1] if "." in change.target else change.target
hook_call = self._extract_hook_call(change)
if hook_call:
content = self._insert_hooks_into_function(content, func_name, [hook_call])
elif change.change_type == ChangeType.WRAP_JSX:
wrapper = self._extract_jsx_wrapper(change)
if wrapper:
func_name = change.target.split(".")[-1] if "." in change.target else change.target
content = self._wrap_function_return(content, func_name, wrapper[0], wrapper[1])
return MergeResult(
decision=MergeDecision.AUTO_MERGED,
file_path=context.file_path,
merged_content=content,
conflicts_resolved=[context.conflict],
explanation="Changes applied in dependency order",
)
def _merge_order_by_time(self, context: MergeContext) -> MergeResult:
"""Apply changes in chronological order."""
# Sort snapshots by start time
sorted_snapshots = sorted(context.task_snapshots, key=lambda s: s.started_at)
content = context.baseline_content
# Apply each snapshot's changes in order
for snapshot in sorted_snapshots:
for change in snapshot.semantic_changes:
if change.content_before and change.content_after:
content = self._apply_content_change(
content, change.content_before, change.content_after
)
elif change.content_after and not change.content_before:
# Addition - handled by other strategies
pass
return MergeResult(
decision=MergeDecision.AUTO_MERGED,
file_path=context.file_path,
merged_content=content,
conflicts_resolved=[context.conflict],
explanation=f"Applied {len(sorted_snapshots)} changes in chronological order",
)
def _merge_append_statements(self, context: MergeContext) -> MergeResult:
"""Append statements (variables, comments, etc.)."""
content = context.baseline_content
additions: list[str] = []
for snapshot in context.task_snapshots:
for change in snapshot.semantic_changes:
if change.is_additive and change.content_after:
additions.append(change.content_after)
# Append at appropriate location
for addition in additions:
content += f"\n{addition}"
return MergeResult(
decision=MergeDecision.AUTO_MERGED,
file_path=context.file_path,
merged_content=content,
conflicts_resolved=[context.conflict],
explanation=f"Appended {len(additions)} statements",
)
# ========================================
# Helper Methods
# ========================================
def _find_import_section_end(self, lines: list[str], ext: str) -> int:
"""Find where the import section ends."""
last_import_line = 0
for i, line in enumerate(lines):
stripped = line.strip()
if self._is_import_line(stripped, ext):
last_import_line = i + 1
elif stripped and not stripped.startswith("#") and not stripped.startswith("//"):
# Non-empty, non-comment line after imports
if last_import_line > 0:
break
return last_import_line if last_import_line > 0 else 0
def _is_import_line(self, line: str, ext: str) -> bool:
"""Check if a line is an import statement."""
if ext == ".py":
return line.startswith("import ") or line.startswith("from ")
elif ext in {".js", ".jsx", ".ts", ".tsx"}:
return line.startswith("import ") or line.startswith("export ")
return False
def _extract_hook_call(self, change: SemanticChange) -> Optional[str]:
"""Extract the hook call from a change."""
if change.content_after:
# Look for useXxx() pattern
match = re.search(r"(const\s+\{[^}]+\}\s*=\s*)?use\w+\([^)]*\);?", change.content_after)
if match:
return match.group(0)
# Also check for simple hook calls
match = re.search(r"use\w+\([^)]*\);?", change.content_after)
if match:
return match.group(0)
return None
def _extract_jsx_wrapper(self, change: SemanticChange) -> Optional[tuple[str, str]]:
"""Extract JSX wrapper component and props."""
if change.content_after:
# Look for <ComponentName ...>
match = re.search(r"<(\w+)([^>]*)>", change.content_after)
if match:
return (match.group(1), match.group(2).strip())
return None
def _insert_hooks_into_function(
self,
content: str,
func_name: str,
hooks: list[str],
) -> str:
"""Insert hooks at the start of a function."""
# Find function and insert hooks after opening brace
patterns = [
# function Component() {
rf"(function\s+{re.escape(func_name)}\s*\([^)]*\)\s*\{{)",
# const Component = () => {
rf"((?:const|let|var)\s+{re.escape(func_name)}\s*=\s*(?:async\s+)?(?:\([^)]*\)|[^=]+)\s*=>\s*\{{)",
# const Component = function() {
rf"((?:const|let|var)\s+{re.escape(func_name)}\s*=\s*function\s*\([^)]*\)\s*\{{)",
]
for pattern in patterns:
match = re.search(pattern, content)
if match:
insert_pos = match.end()
hook_text = "\n " + "\n ".join(hooks)
content = content[:insert_pos] + hook_text + content[insert_pos:]
break
return content
def _wrap_function_return(
self,
content: str,
func_name: str,
wrapper_name: str,
wrapper_props: str,
) -> str:
"""Wrap the return statement of a function in a JSX component."""
# This is simplified - a real implementation would use AST
# Find return statement with JSX
return_pattern = r"(return\s*\(\s*)(<[^>]+>)"
def replacer(match):
return_start = match.group(1)
jsx_start = match.group(2)
props = f" {wrapper_props}" if wrapper_props else ""
return f"{return_start}<{wrapper_name}{props}>\n {jsx_start}"
content = re.sub(return_pattern, replacer, content, count=1)
# Also need to close the wrapper - this is tricky without proper parsing
# For now, we'll rely on the AI resolver for complex cases
return content
def _find_function_insert_position(self, content: str, ext: str) -> Optional[int]:
"""Find the best position to insert new functions."""
lines = content.split("\n")
# Look for module.exports or export default at the end
for i in range(len(lines) - 1, -1, -1):
line = lines[i].strip()
if line.startswith("module.exports") or line.startswith("export default"):
return i
return None
def _insert_methods_into_class(
self,
content: str,
class_name: str,
methods: list[str],
) -> str:
"""Insert methods into a class body."""
# Find class closing brace
class_pattern = rf"class\s+{re.escape(class_name)}\s*(?:extends\s+\w+)?\s*\{{"
match = re.search(class_pattern, content)
if match:
# Find the matching closing brace
start = match.end()
brace_count = 1
pos = start
while pos < len(content) and brace_count > 0:
if content[pos] == "{":
brace_count += 1
elif content[pos] == "}":
brace_count -= 1
pos += 1
if brace_count == 0:
# Insert before closing brace
insert_pos = pos - 1
method_text = "\n\n " + "\n\n ".join(methods)
content = content[:insert_pos] + method_text + content[insert_pos:]
return content
def _extract_new_props(self, change: SemanticChange) -> list[tuple[str, str]]:
"""Extract newly added props from a change."""
props = []
if change.content_after and change.content_before:
# Simple diff - find props in after that aren't in before
after_props = re.findall(r"(\w+)=\{([^}]+)\}", change.content_after)
before_props = dict(re.findall(r"(\w+)=\{([^}]+)\}", change.content_before))
for name, value in after_props:
if name not in before_props:
props.append((name, value))
return props
def _apply_content_change(
self,
content: str,
old: Optional[str],
new: str,
) -> str:
"""Apply a content change by replacing old with new."""
if old and old in content:
return content.replace(old, new, 1)
return content
def _topological_sort_changes(
self,
snapshots: list[TaskSnapshot],
) -> list[SemanticChange]:
"""Sort changes by their dependencies."""
# Collect all changes
all_changes: list[SemanticChange] = []
for snapshot in snapshots:
all_changes.extend(snapshot.semantic_changes)
# Simple ordering: hooks before wraps before modifications
priority = {
ChangeType.ADD_IMPORT: 0,
ChangeType.ADD_HOOK_CALL: 1,
ChangeType.ADD_VARIABLE: 2,
ChangeType.ADD_CONSTANT: 2,
ChangeType.WRAP_JSX: 3,
ChangeType.ADD_JSX_ELEMENT: 4,
ChangeType.MODIFY_FUNCTION: 5,
ChangeType.MODIFY_JSX_PROPS: 5,
}
return sorted(
all_changes,
key=lambda c: priority.get(c.change_type, 10)
)
+642
View File
@@ -0,0 +1,642 @@
"""
Conflict Detector
=================
Detects conflicts between multiple task changes using rule-based analysis.
This module determines:
1. Which changes from different tasks overlap
2. Whether overlapping changes are compatible
3. What merge strategy can be used for compatible changes
4. Which conflicts need AI or human intervention
The goal is to resolve as many conflicts as possible without AI,
using deterministic rules based on semantic change types.
"""
from __future__ import annotations
import logging
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional
from .types import (
ChangeType,
ConflictRegion,
ConflictSeverity,
FileAnalysis,
MergeStrategy,
SemanticChange,
)
# Import debug utilities
try:
from debug import debug, debug_detailed, debug_verbose, debug_success, debug_error, debug_warning, is_debug_enabled
except ImportError:
def debug(*args, **kwargs): pass
def debug_detailed(*args, **kwargs): pass
def debug_verbose(*args, **kwargs): pass
def debug_success(*args, **kwargs): pass
def debug_error(*args, **kwargs): pass
def debug_warning(*args, **kwargs): pass
def is_debug_enabled(): return False
logger = logging.getLogger(__name__)
MODULE = "merge.conflict_detector"
@dataclass
class CompatibilityRule:
"""
A rule defining compatibility between two change types.
Attributes:
change_type_a: First change type
change_type_b: Second change type (can be same as a)
compatible: Whether these changes can be auto-merged
strategy: If compatible, which strategy to use
reason: Human-readable explanation
bidirectional: If True, rule applies both ways (a,b) and (b,a)
"""
change_type_a: ChangeType
change_type_b: ChangeType
compatible: bool
strategy: Optional[MergeStrategy] = None
reason: str = ""
bidirectional: bool = True
class ConflictDetector:
"""
Detects and classifies conflicts between task changes.
Uses a comprehensive rule base to determine compatibility
between different semantic change types, enabling maximum
auto-merge capability.
Example:
detector = ConflictDetector()
conflicts = detector.detect_conflicts({
"task-001": analysis1,
"task-002": analysis2,
})
for conflict in conflicts:
if conflict.can_auto_merge:
print(f"Can auto-merge with {conflict.merge_strategy}")
else:
print(f"Needs {conflict.severity} review")
"""
def __init__(self):
"""Initialize with default compatibility rules."""
debug(MODULE, "Initializing ConflictDetector")
self._rules = self._build_default_rules()
self._rule_index = self._index_rules()
debug_success(MODULE, "ConflictDetector initialized", rule_count=len(self._rules))
def _build_default_rules(self) -> list[CompatibilityRule]:
"""Build the default set of compatibility rules."""
rules = []
# ========================================
# IMPORT RULES - Generally compatible
# ========================================
# Multiple imports from different modules = always compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_IMPORT,
change_type_b=ChangeType.ADD_IMPORT,
compatible=True,
strategy=MergeStrategy.COMBINE_IMPORTS,
reason="Adding different imports is always compatible",
)
)
# Import addition + removal = check if same module
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_IMPORT,
change_type_b=ChangeType.REMOVE_IMPORT,
compatible=False, # Need to check if same import
strategy=MergeStrategy.AI_REQUIRED,
reason="Import add/remove may conflict if same module",
)
)
# ========================================
# FUNCTION RULES
# ========================================
# Adding different functions = compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_FUNCTION,
change_type_b=ChangeType.ADD_FUNCTION,
compatible=True,
strategy=MergeStrategy.APPEND_FUNCTIONS,
reason="Adding different functions is compatible",
)
)
# Adding function + modifying different function = compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_FUNCTION,
change_type_b=ChangeType.MODIFY_FUNCTION,
compatible=True,
strategy=MergeStrategy.APPEND_FUNCTIONS,
reason="Adding a function doesn't affect modifications to other functions",
)
)
# Modifying same function = conflict (but may be resolvable)
rules.append(
CompatibilityRule(
change_type_a=ChangeType.MODIFY_FUNCTION,
change_type_b=ChangeType.MODIFY_FUNCTION,
compatible=False,
strategy=MergeStrategy.AI_REQUIRED,
reason="Multiple modifications to same function need analysis",
)
)
# ========================================
# REACT HOOK RULES
# ========================================
# Multiple hook additions = compatible (order matters, but predictable)
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_HOOK_CALL,
change_type_b=ChangeType.ADD_HOOK_CALL,
compatible=True,
strategy=MergeStrategy.ORDER_BY_DEPENDENCY,
reason="Multiple hooks can be added with correct ordering",
)
)
# Hook addition + JSX wrap = compatible (hooks first, then wrap)
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_HOOK_CALL,
change_type_b=ChangeType.WRAP_JSX,
compatible=True,
strategy=MergeStrategy.HOOKS_THEN_WRAP,
reason="Hooks are added at function start, wrap is on return",
)
)
# Hook addition + function modification = usually compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_HOOK_CALL,
change_type_b=ChangeType.MODIFY_FUNCTION,
compatible=True,
strategy=MergeStrategy.HOOKS_FIRST,
reason="Hooks go at start, other modifications likely elsewhere",
)
)
# ========================================
# JSX RULES
# ========================================
# Multiple JSX wraps = need to determine order
rules.append(
CompatibilityRule(
change_type_a=ChangeType.WRAP_JSX,
change_type_b=ChangeType.WRAP_JSX,
compatible=True,
strategy=MergeStrategy.ORDER_BY_DEPENDENCY,
reason="Multiple wraps can be nested in correct order",
)
)
# JSX wrap + element addition = compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.WRAP_JSX,
change_type_b=ChangeType.ADD_JSX_ELEMENT,
compatible=True,
strategy=MergeStrategy.APPEND_STATEMENTS,
reason="Wrapping and adding elements are independent",
)
)
# Prop modifications = may conflict
rules.append(
CompatibilityRule(
change_type_a=ChangeType.MODIFY_JSX_PROPS,
change_type_b=ChangeType.MODIFY_JSX_PROPS,
compatible=True,
strategy=MergeStrategy.COMBINE_PROPS,
reason="Props can usually be combined if different",
)
)
# ========================================
# CLASS/METHOD RULES
# ========================================
# Adding methods to same class = compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_METHOD,
change_type_b=ChangeType.ADD_METHOD,
compatible=True,
strategy=MergeStrategy.APPEND_METHODS,
reason="Adding different methods is compatible",
)
)
# Modifying same method = conflict
rules.append(
CompatibilityRule(
change_type_a=ChangeType.MODIFY_METHOD,
change_type_b=ChangeType.MODIFY_METHOD,
compatible=False,
strategy=MergeStrategy.AI_REQUIRED,
reason="Multiple modifications to same method need analysis",
)
)
# Adding class + modifying existing class = compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_CLASS,
change_type_b=ChangeType.MODIFY_CLASS,
compatible=True,
strategy=MergeStrategy.APPEND_FUNCTIONS,
reason="New classes don't conflict with modifications",
)
)
# ========================================
# VARIABLE RULES
# ========================================
# Adding different variables = compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_VARIABLE,
change_type_b=ChangeType.ADD_VARIABLE,
compatible=True,
strategy=MergeStrategy.APPEND_STATEMENTS,
reason="Adding different variables is compatible",
)
)
# Adding constant + variable = compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_CONSTANT,
change_type_b=ChangeType.ADD_VARIABLE,
compatible=True,
strategy=MergeStrategy.APPEND_STATEMENTS,
reason="Constants and variables are independent",
)
)
# ========================================
# TYPE RULES (TypeScript)
# ========================================
# Adding different types = compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_TYPE,
change_type_b=ChangeType.ADD_TYPE,
compatible=True,
strategy=MergeStrategy.APPEND_FUNCTIONS,
reason="Adding different types is compatible",
)
)
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_INTERFACE,
change_type_b=ChangeType.ADD_INTERFACE,
compatible=True,
strategy=MergeStrategy.APPEND_FUNCTIONS,
reason="Adding different interfaces is compatible",
)
)
# Modifying same interface = conflict
rules.append(
CompatibilityRule(
change_type_a=ChangeType.MODIFY_INTERFACE,
change_type_b=ChangeType.MODIFY_INTERFACE,
compatible=False,
strategy=MergeStrategy.AI_REQUIRED,
reason="Multiple interface modifications need analysis",
)
)
# ========================================
# DECORATOR RULES (Python)
# ========================================
# Adding decorators = usually compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_DECORATOR,
change_type_b=ChangeType.ADD_DECORATOR,
compatible=True,
strategy=MergeStrategy.ORDER_BY_DEPENDENCY,
reason="Decorators can be stacked with correct order",
)
)
# ========================================
# COMMENT RULES - Low priority
# ========================================
rules.append(
CompatibilityRule(
change_type_a=ChangeType.ADD_COMMENT,
change_type_b=ChangeType.ADD_COMMENT,
compatible=True,
strategy=MergeStrategy.APPEND_STATEMENTS,
reason="Comments are independent",
)
)
# Formatting changes are always compatible
rules.append(
CompatibilityRule(
change_type_a=ChangeType.FORMATTING_ONLY,
change_type_b=ChangeType.FORMATTING_ONLY,
compatible=True,
strategy=MergeStrategy.ORDER_BY_TIME,
reason="Formatting doesn't affect semantics",
)
)
return rules
def _index_rules(self) -> dict[tuple[ChangeType, ChangeType], CompatibilityRule]:
"""Create an index for fast rule lookup."""
index = {}
for rule in self._rules:
index[(rule.change_type_a, rule.change_type_b)] = rule
if rule.bidirectional and rule.change_type_a != rule.change_type_b:
index[(rule.change_type_b, rule.change_type_a)] = rule
return index
def add_rule(self, rule: CompatibilityRule) -> None:
"""Add a custom compatibility rule."""
self._rules.append(rule)
self._rule_index[(rule.change_type_a, rule.change_type_b)] = rule
if rule.bidirectional and rule.change_type_a != rule.change_type_b:
self._rule_index[(rule.change_type_b, rule.change_type_a)] = rule
def detect_conflicts(
self,
task_analyses: dict[str, FileAnalysis],
) -> list[ConflictRegion]:
"""
Detect conflicts between multiple task changes to the same file.
Args:
task_analyses: Map of task_id -> FileAnalysis
Returns:
List of detected conflict regions
"""
task_ids = list(task_analyses.keys())
debug(MODULE, f"Detecting conflicts between {len(task_analyses)} tasks",
tasks=task_ids)
if len(task_analyses) <= 1:
debug(MODULE, "No conflicts possible with 0-1 tasks")
return [] # No conflicts possible with 0-1 tasks
conflicts: list[ConflictRegion] = []
# Group changes by location
location_changes: dict[str, list[tuple[str, SemanticChange]]] = defaultdict(list)
for task_id, analysis in task_analyses.items():
debug_detailed(MODULE, f"Processing task {task_id}",
changes_count=len(analysis.changes),
file=analysis.file_path)
for change in analysis.changes:
location_changes[change.location].append((task_id, change))
debug_detailed(MODULE, f"Grouped changes into {len(location_changes)} locations")
# Analyze each location for conflicts
for location, task_changes in location_changes.items():
if len(task_changes) <= 1:
continue # No conflict at this location
debug_verbose(MODULE, f"Checking location {location}",
task_changes_count=len(task_changes))
file_path = next(iter(task_analyses.values())).file_path
conflict = self._analyze_location_conflict(
file_path, location, task_changes
)
if conflict:
debug_detailed(MODULE, f"Conflict detected at {location}",
severity=conflict.severity.value,
can_auto_merge=conflict.can_auto_merge,
tasks=conflict.tasks_involved)
conflicts.append(conflict)
# Also check for implicit conflicts (e.g., changes to related code)
implicit_conflicts = self._detect_implicit_conflicts(task_analyses)
if implicit_conflicts:
debug_detailed(MODULE, f"Found {len(implicit_conflicts)} implicit conflicts")
conflicts.extend(implicit_conflicts)
# Summary
auto_mergeable = sum(1 for c in conflicts if c.can_auto_merge)
critical = sum(1 for c in conflicts if c.severity == ConflictSeverity.CRITICAL)
debug_success(MODULE, f"Conflict detection complete",
total_conflicts=len(conflicts),
auto_mergeable=auto_mergeable,
critical=critical)
return conflicts
def _analyze_location_conflict(
self,
file_path: str,
location: str,
task_changes: list[tuple[str, SemanticChange]],
) -> Optional[ConflictRegion]:
"""Analyze changes at a specific location for conflicts."""
tasks = [tc[0] for tc in task_changes]
changes = [tc[1] for tc in task_changes]
change_types = [c.change_type for c in changes]
# Check if all changes target the same thing
targets = {c.target for c in changes}
if len(targets) > 1:
# Different targets at same location - likely compatible
# (e.g., adding two different functions)
return None
# Check pairwise compatibility
all_compatible = True
final_strategy: Optional[MergeStrategy] = None
reasons = []
for i, (type_a, change_a) in enumerate(zip(change_types, changes)):
for type_b, change_b in zip(change_types[i + 1 :], changes[i + 1 :]):
rule = self._rule_index.get((type_a, type_b))
if rule:
if not rule.compatible:
all_compatible = False
reasons.append(rule.reason)
elif rule.strategy:
final_strategy = rule.strategy
else:
# No rule - conservative default
all_compatible = False
reasons.append(f"No rule for {type_a.value} + {type_b.value}")
# Determine severity
if all_compatible:
severity = ConflictSeverity.NONE
else:
severity = self._assess_severity(change_types, changes)
return ConflictRegion(
file_path=file_path,
location=location,
tasks_involved=tasks,
change_types=change_types,
severity=severity,
can_auto_merge=all_compatible,
merge_strategy=final_strategy if all_compatible else MergeStrategy.AI_REQUIRED,
reason=" | ".join(reasons) if reasons else "Changes are compatible",
)
def _assess_severity(
self,
change_types: list[ChangeType],
changes: list[SemanticChange],
) -> ConflictSeverity:
"""Assess the severity of a conflict."""
# Critical: Both tasks modify core logic
modify_types = {
ChangeType.MODIFY_FUNCTION,
ChangeType.MODIFY_METHOD,
ChangeType.MODIFY_CLASS,
}
modify_count = sum(1 for ct in change_types if ct in modify_types)
if modify_count >= 2:
# Check if they modify the exact same lines
line_ranges = [(c.line_start, c.line_end) for c in changes]
if self._ranges_overlap(line_ranges):
return ConflictSeverity.CRITICAL
# High: Structural changes that could break compilation
structural_types = {
ChangeType.WRAP_JSX,
ChangeType.UNWRAP_JSX,
ChangeType.REMOVE_FUNCTION,
ChangeType.REMOVE_CLASS,
}
if any(ct in structural_types for ct in change_types):
return ConflictSeverity.HIGH
# Medium: Modifications to same function/method
if modify_count >= 1:
return ConflictSeverity.MEDIUM
# Low: Likely resolvable with AI
return ConflictSeverity.LOW
def _ranges_overlap(self, ranges: list[tuple[int, int]]) -> bool:
"""Check if any line ranges overlap."""
sorted_ranges = sorted(ranges)
for i in range(len(sorted_ranges) - 1):
if sorted_ranges[i][1] >= sorted_ranges[i + 1][0]:
return True
return False
def _detect_implicit_conflicts(
self,
task_analyses: dict[str, FileAnalysis],
) -> list[ConflictRegion]:
"""Detect implicit conflicts not caught by location analysis."""
conflicts = []
# Check for function rename + function call changes
# (If task A renames a function and task B calls the old name)
# Check for import removal + usage
# (If task A removes an import and task B uses it)
# For now, these advanced checks are TODO
# The main location-based detection handles most cases
return conflicts
def get_compatible_pairs(self) -> list[tuple[ChangeType, ChangeType, MergeStrategy]]:
"""Get all compatible change type pairs and their strategies."""
pairs = []
for rule in self._rules:
if rule.compatible:
pairs.append((rule.change_type_a, rule.change_type_b, rule.strategy))
return pairs
def explain_conflict(self, conflict: ConflictRegion) -> str:
"""Generate a human-readable explanation of a conflict."""
lines = [
f"Conflict in {conflict.file_path} at {conflict.location}",
f"Tasks involved: {', '.join(conflict.tasks_involved)}",
f"Severity: {conflict.severity.value}",
"",
]
if conflict.can_auto_merge:
lines.append(f"Can be auto-merged using strategy: {conflict.merge_strategy.value}")
else:
lines.append("Cannot be auto-merged:")
lines.append(f" Reason: {conflict.reason}")
lines.append("")
lines.append("Changes:")
for ct in conflict.change_types:
lines.append(f" - {ct.value}")
return "\n".join(lines)
def analyze_compatibility(
change_a: SemanticChange,
change_b: SemanticChange,
detector: Optional[ConflictDetector] = None,
) -> tuple[bool, Optional[MergeStrategy], str]:
"""
Analyze compatibility between two specific changes.
Convenience function for quick compatibility checks.
Args:
change_a: First semantic change
change_b: Second semantic change
detector: Optional detector instance (creates one if not provided)
Returns:
Tuple of (compatible, strategy, reason)
"""
if detector is None:
detector = ConflictDetector()
rule = detector._rule_index.get((change_a.change_type, change_b.change_type))
if rule:
return (rule.compatible, rule.strategy, rule.reason)
else:
return (False, MergeStrategy.AI_REQUIRED, "No compatibility rule defined")
+687
View File
@@ -0,0 +1,687 @@
"""
File Evolution Tracker
======================
Tracks the evolution of files across multiple task modifications.
This component:
1. Captures baseline file states when worktrees are created
2. Stores file content snapshots for reference
3. Records task modifications with semantic changes
4. Persists evolution data for merge analysis
The evolution data enables the merge system to understand:
- What the file looked like before any tasks started
- What each task intended to do
- The order of modifications
- Content hashes for integrity verification
"""
from __future__ import annotations
import json
import logging
import shutil
import subprocess
from datetime import datetime
from pathlib import Path
from typing import Optional
from .types import (
FileEvolution,
SemanticChange,
TaskSnapshot,
compute_content_hash,
sanitize_path_for_storage,
)
from .semantic_analyzer import SemanticAnalyzer
# Import debug utilities
try:
from debug import debug, debug_detailed, debug_verbose, debug_success, debug_error, debug_warning, is_debug_enabled
except ImportError:
def debug(*args, **kwargs): pass
def debug_detailed(*args, **kwargs): pass
def debug_verbose(*args, **kwargs): pass
def debug_success(*args, **kwargs): pass
def debug_error(*args, **kwargs): pass
def debug_warning(*args, **kwargs): pass
def is_debug_enabled(): return False
logger = logging.getLogger(__name__)
MODULE = "merge.file_evolution"
class FileEvolutionTracker:
"""
Tracks file evolution across task modifications.
This class manages:
- Baseline capture when worktrees are created
- File content snapshots in .auto-claude/baselines/
- Task modification tracking with semantic analysis
- Persistence of evolution data
Usage:
tracker = FileEvolutionTracker(project_dir)
# When creating a worktree for a task
tracker.capture_baselines(task_id, files_to_track)
# When a task modifies a file
tracker.record_modification(task_id, file_path, old_content, new_content)
# When preparing to merge
evolution = tracker.get_file_evolution(file_path)
"""
# Default extensions to track for baselines
DEFAULT_EXTENSIONS = {
".py", ".js", ".ts", ".tsx", ".jsx",
".json", ".yaml", ".yml", ".toml",
".md", ".txt", ".html", ".css", ".scss",
".go", ".rs", ".java", ".kt", ".swift",
}
def __init__(
self,
project_dir: Path,
storage_dir: Optional[Path] = None,
semantic_analyzer: Optional[SemanticAnalyzer] = None,
):
"""
Initialize the file evolution tracker.
Args:
project_dir: Root directory of the project
storage_dir: Directory for evolution data (default: .auto-claude/)
semantic_analyzer: Optional pre-configured analyzer
"""
debug(MODULE, "Initializing FileEvolutionTracker",
project_dir=str(project_dir))
self.project_dir = Path(project_dir).resolve()
self.storage_dir = storage_dir or (self.project_dir / ".auto-claude")
self.baselines_dir = self.storage_dir / "baselines"
self.evolution_file = self.storage_dir / "file_evolution.json"
# Ensure directories exist
self.storage_dir.mkdir(parents=True, exist_ok=True)
self.baselines_dir.mkdir(parents=True, exist_ok=True)
# Semantic analyzer for extracting changes
self.analyzer = semantic_analyzer or SemanticAnalyzer()
# Load existing evolution data
self._evolutions: dict[str, FileEvolution] = {}
self._load_evolutions()
debug_success(MODULE, "FileEvolutionTracker initialized",
evolutions_loaded=len(self._evolutions))
def _load_evolutions(self) -> None:
"""Load evolution data from disk."""
if not self.evolution_file.exists():
return
try:
with open(self.evolution_file) as f:
data = json.load(f)
for file_path, evolution_data in data.items():
self._evolutions[file_path] = FileEvolution.from_dict(evolution_data)
logger.debug(f"Loaded evolution data for {len(self._evolutions)} files")
except Exception as e:
logger.error(f"Failed to load evolution data: {e}")
def _save_evolutions(self) -> None:
"""Persist evolution data to disk."""
try:
data = {
file_path: evolution.to_dict()
for file_path, evolution in self._evolutions.items()
}
with open(self.evolution_file, "w") as f:
json.dump(data, f, indent=2)
logger.debug(f"Saved evolution data for {len(self._evolutions)} files")
except Exception as e:
logger.error(f"Failed to save evolution data: {e}")
def _get_current_commit(self) -> str:
"""Get the current git commit hash."""
try:
result = subprocess.run(
["git", "rev-parse", "HEAD"],
cwd=self.project_dir,
capture_output=True,
text=True,
check=True,
)
return result.stdout.strip()
except subprocess.CalledProcessError:
return "unknown"
def _get_relative_path(self, file_path: Path | str) -> str:
"""Get path relative to project root."""
path = Path(file_path)
if path.is_absolute():
try:
return str(path.relative_to(self.project_dir))
except ValueError:
return str(path)
return str(path)
def _store_baseline_content(
self,
file_path: str,
content: str,
task_id: str,
) -> str:
"""
Store baseline content to disk.
Returns:
Path to the stored baseline file (relative to storage_dir)
"""
safe_name = sanitize_path_for_storage(file_path)
baseline_path = self.baselines_dir / task_id / f"{safe_name}.baseline"
baseline_path.parent.mkdir(parents=True, exist_ok=True)
with open(baseline_path, "w", encoding="utf-8") as f:
f.write(content)
return str(baseline_path.relative_to(self.storage_dir))
def _read_file_content(self, file_path: Path | str) -> Optional[str]:
"""Read file content, returning None if not readable."""
try:
path = Path(file_path)
if not path.is_absolute():
path = self.project_dir / path
return path.read_text(encoding="utf-8")
except Exception as e:
logger.warning(f"Could not read {file_path}: {e}")
return None
def capture_baselines(
self,
task_id: str,
files: Optional[list[Path | str]] = None,
intent: str = "",
) -> dict[str, FileEvolution]:
"""
Capture baseline state of files for a task.
Call this when creating a worktree for a new task.
Args:
task_id: Unique identifier for the task
files: List of files to capture. If None, discovers trackable files.
intent: Description of what the task intends to do
Returns:
Dictionary mapping file paths to their FileEvolution objects
"""
commit = self._get_current_commit()
captured_at = datetime.now()
captured: dict[str, FileEvolution] = {}
# Discover files if not specified
if files is None:
files = self._discover_trackable_files()
for file_path in files:
rel_path = self._get_relative_path(file_path)
content = self._read_file_content(file_path)
if content is None:
continue
# Store baseline content
baseline_path = self._store_baseline_content(rel_path, content, task_id)
content_hash = compute_content_hash(content)
# Create or update evolution
if rel_path in self._evolutions:
evolution = self._evolutions[rel_path]
# Update baseline if this is a fresh start
logger.debug(f"Updating existing evolution for {rel_path}")
else:
evolution = FileEvolution(
file_path=rel_path,
baseline_commit=commit,
baseline_captured_at=captured_at,
baseline_content_hash=content_hash,
baseline_snapshot_path=baseline_path,
)
self._evolutions[rel_path] = evolution
logger.debug(f"Created new evolution for {rel_path}")
# Create task snapshot
snapshot = TaskSnapshot(
task_id=task_id,
task_intent=intent,
started_at=captured_at,
content_hash_before=content_hash,
)
evolution.add_task_snapshot(snapshot)
captured[rel_path] = evolution
self._save_evolutions()
logger.info(f"Captured baselines for {len(captured)} files for task {task_id}")
return captured
def _discover_trackable_files(self) -> list[Path]:
"""
Discover files that should be tracked for baselines.
Uses git ls-files to get tracked files, filtering by extension.
"""
try:
result = subprocess.run(
["git", "ls-files"],
cwd=self.project_dir,
capture_output=True,
text=True,
check=True,
)
all_files = result.stdout.strip().split("\n")
trackable = []
for file_path in all_files:
if not file_path:
continue
path = Path(file_path)
if path.suffix in self.DEFAULT_EXTENSIONS:
trackable.append(self.project_dir / path)
return trackable
except subprocess.CalledProcessError:
logger.warning("Failed to list git files, returning empty list")
return []
def record_modification(
self,
task_id: str,
file_path: Path | str,
old_content: str,
new_content: str,
raw_diff: Optional[str] = None,
) -> Optional[TaskSnapshot]:
"""
Record a file modification by a task.
Call this after a task makes changes to a file.
Args:
task_id: The task that made the modification
file_path: Path to the modified file
old_content: File content before modification
new_content: File content after modification
raw_diff: Optional unified diff for reference
Returns:
Updated TaskSnapshot, or None if file not being tracked
"""
rel_path = self._get_relative_path(file_path)
# Get or create evolution
if rel_path not in self._evolutions:
logger.warning(f"File {rel_path} not being tracked, creating evolution")
self.capture_baselines(task_id, [file_path])
evolution = self._evolutions.get(rel_path)
if not evolution:
return None
# Get existing snapshot or create new one
snapshot = evolution.get_task_snapshot(task_id)
if not snapshot:
snapshot = TaskSnapshot(
task_id=task_id,
task_intent="",
started_at=datetime.now(),
content_hash_before=compute_content_hash(old_content),
)
# Analyze semantic changes
analysis = self.analyzer.analyze_diff(
rel_path, old_content, new_content
)
semantic_changes = analysis.changes
# Update snapshot
snapshot.completed_at = datetime.now()
snapshot.content_hash_after = compute_content_hash(new_content)
snapshot.semantic_changes = semantic_changes
snapshot.raw_diff = raw_diff
# Update evolution
evolution.add_task_snapshot(snapshot)
self._save_evolutions()
logger.info(
f"Recorded modification to {rel_path} by {task_id}: "
f"{len(semantic_changes)} semantic changes"
)
return snapshot
def get_file_evolution(self, file_path: Path | str) -> Optional[FileEvolution]:
"""
Get the complete evolution history for a file.
Args:
file_path: Path to the file
Returns:
FileEvolution object, or None if not tracked
"""
rel_path = self._get_relative_path(file_path)
return self._evolutions.get(rel_path)
def get_baseline_content(self, file_path: Path | str) -> Optional[str]:
"""
Get the baseline content for a file.
Args:
file_path: Path to the file
Returns:
Original baseline content, or None if not available
"""
rel_path = self._get_relative_path(file_path)
evolution = self._evolutions.get(rel_path)
if not evolution:
return None
baseline_path = self.storage_dir / evolution.baseline_snapshot_path
if baseline_path.exists():
return baseline_path.read_text(encoding="utf-8")
return None
def get_task_modifications(
self,
task_id: str,
) -> list[tuple[str, TaskSnapshot]]:
"""
Get all file modifications made by a specific task.
Args:
task_id: The task identifier
Returns:
List of (file_path, TaskSnapshot) tuples
"""
modifications = []
for file_path, evolution in self._evolutions.items():
snapshot = evolution.get_task_snapshot(task_id)
if snapshot and snapshot.semantic_changes:
modifications.append((file_path, snapshot))
return modifications
def get_files_modified_by_tasks(
self,
task_ids: list[str],
) -> dict[str, list[str]]:
"""
Get files modified by specified tasks.
Args:
task_ids: List of task identifiers
Returns:
Dictionary mapping file paths to list of task IDs that modified them
"""
file_tasks: dict[str, list[str]] = {}
for file_path, evolution in self._evolutions.items():
for snapshot in evolution.task_snapshots:
if snapshot.task_id in task_ids and snapshot.semantic_changes:
if file_path not in file_tasks:
file_tasks[file_path] = []
file_tasks[file_path].append(snapshot.task_id)
return file_tasks
def get_conflicting_files(self, task_ids: list[str]) -> list[str]:
"""
Get files modified by multiple tasks (potential conflicts).
Args:
task_ids: List of task identifiers to check
Returns:
List of file paths modified by 2+ tasks
"""
file_tasks = self.get_files_modified_by_tasks(task_ids)
return [
file_path for file_path, tasks in file_tasks.items()
if len(tasks) > 1
]
def mark_task_completed(self, task_id: str) -> None:
"""
Mark a task as completed (set completed_at on all snapshots).
Args:
task_id: The task identifier
"""
now = datetime.now()
for evolution in self._evolutions.values():
snapshot = evolution.get_task_snapshot(task_id)
if snapshot and snapshot.completed_at is None:
snapshot.completed_at = now
self._save_evolutions()
def cleanup_task(
self,
task_id: str,
remove_baselines: bool = True,
) -> None:
"""
Clean up data for a completed/cancelled task.
Args:
task_id: The task identifier
remove_baselines: Whether to remove stored baseline files
"""
# Remove task snapshots from evolutions
for evolution in self._evolutions.values():
evolution.task_snapshots = [
ts for ts in evolution.task_snapshots
if ts.task_id != task_id
]
# Remove baseline directory if requested
if remove_baselines:
baseline_dir = self.baselines_dir / task_id
if baseline_dir.exists():
shutil.rmtree(baseline_dir)
logger.debug(f"Removed baseline directory for task {task_id}")
# Clean up empty evolutions
self._evolutions = {
file_path: evolution
for file_path, evolution in self._evolutions.items()
if evolution.task_snapshots
}
self._save_evolutions()
logger.info(f"Cleaned up data for task {task_id}")
def get_active_tasks(self) -> set[str]:
"""
Get set of task IDs with active (non-completed) modifications.
Returns:
Set of task IDs
"""
active = set()
for evolution in self._evolutions.values():
for snapshot in evolution.task_snapshots:
if snapshot.completed_at is None:
active.add(snapshot.task_id)
return active
def get_evolution_summary(self) -> dict:
"""
Get a summary of tracked file evolutions.
Returns:
Dictionary with summary statistics
"""
total_files = len(self._evolutions)
all_tasks = set()
files_with_multiple_tasks = 0
total_changes = 0
for evolution in self._evolutions.values():
task_ids = [ts.task_id for ts in evolution.task_snapshots]
all_tasks.update(task_ids)
if len(task_ids) > 1:
files_with_multiple_tasks += 1
for snapshot in evolution.task_snapshots:
total_changes += len(snapshot.semantic_changes)
return {
"total_files_tracked": total_files,
"total_tasks": len(all_tasks),
"files_with_potential_conflicts": files_with_multiple_tasks,
"total_semantic_changes": total_changes,
"active_tasks": len(self.get_active_tasks()),
}
def export_for_merge(
self,
file_path: Path | str,
task_ids: Optional[list[str]] = None,
) -> Optional[dict]:
"""
Export evolution data for a file in a format suitable for merge.
This provides the data needed by the merge system to understand
what each task did and in what order.
Args:
file_path: Path to the file
task_ids: Optional list of tasks to include (default: all)
Returns:
Dictionary with merge-relevant evolution data
"""
rel_path = self._get_relative_path(file_path)
evolution = self._evolutions.get(rel_path)
if not evolution:
return None
baseline_content = self.get_baseline_content(file_path)
# Filter snapshots if task_ids specified
snapshots = evolution.task_snapshots
if task_ids:
snapshots = [
ts for ts in snapshots
if ts.task_id in task_ids
]
return {
"file_path": rel_path,
"baseline_content": baseline_content,
"baseline_commit": evolution.baseline_commit,
"baseline_hash": evolution.baseline_content_hash,
"tasks": [
{
"task_id": ts.task_id,
"intent": ts.task_intent,
"started_at": ts.started_at.isoformat(),
"completed_at": ts.completed_at.isoformat() if ts.completed_at else None,
"changes": [c.to_dict() for c in ts.semantic_changes],
"hash_before": ts.content_hash_before,
"hash_after": ts.content_hash_after,
}
for ts in snapshots
],
}
def refresh_from_git(
self,
task_id: str,
worktree_path: Path,
) -> None:
"""
Refresh task snapshots by analyzing git diff from worktree.
This is useful when we didn't capture real-time modifications
and need to retroactively analyze what a task changed.
Args:
task_id: The task identifier
worktree_path: Path to the task's worktree
"""
debug(MODULE, f"refresh_from_git() for task {task_id}",
task_id=task_id,
worktree_path=str(worktree_path))
try:
# Get list of files changed in the worktree
result = subprocess.run(
["git", "diff", "--name-only", "main...HEAD"],
cwd=worktree_path,
capture_output=True,
text=True,
check=True,
)
changed_files = [f for f in result.stdout.strip().split("\n") if f]
debug(MODULE, f"Found {len(changed_files)} changed files",
changed_files=changed_files[:10] if len(changed_files) > 10 else changed_files)
for file_path in changed_files:
# Get the diff for this file
diff_result = subprocess.run(
["git", "diff", "main...HEAD", "--", file_path],
cwd=worktree_path,
capture_output=True,
text=True,
check=True,
)
# Get content before (from main) and after (current)
try:
show_result = subprocess.run(
["git", "show", f"main:{file_path}"],
cwd=worktree_path,
capture_output=True,
text=True,
check=True,
)
old_content = show_result.stdout
except subprocess.CalledProcessError:
# File is new
old_content = ""
current_file = worktree_path / file_path
if current_file.exists():
new_content = current_file.read_text(encoding="utf-8")
else:
# File was deleted
new_content = ""
# Record the modification
self.record_modification(
task_id=task_id,
file_path=file_path,
old_content=old_content,
new_content=new_content,
raw_diff=diff_result.stdout,
)
logger.info(f"Refreshed {len(changed_files)} files from worktree for task {task_id}")
except subprocess.CalledProcessError as e:
logger.error(f"Failed to refresh from git: {e}")
File diff suppressed because it is too large Load Diff
+806
View File
@@ -0,0 +1,806 @@
"""
Semantic Analyzer
=================
Analyzes code changes at a semantic level using tree-sitter.
This module provides AST-based analysis of code changes, extracting
meaningful semantic changes like "added import", "modified function",
"wrapped JSX element" rather than line-level diffs.
When tree-sitter is not available, falls back to regex-based heuristics.
"""
from __future__ import annotations
import difflib
import logging
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional
from .types import (
ChangeType,
FileAnalysis,
SemanticChange,
compute_content_hash,
)
# Import debug utilities
try:
from debug import debug, debug_detailed, debug_verbose, debug_success, debug_error, is_debug_enabled
except ImportError:
# Fallback if debug module not available
def debug(*args, **kwargs): pass
def debug_detailed(*args, **kwargs): pass
def debug_verbose(*args, **kwargs): pass
def debug_success(*args, **kwargs): pass
def debug_error(*args, **kwargs): pass
def is_debug_enabled(): return False
logger = logging.getLogger(__name__)
MODULE = "merge.semantic_analyzer"
# Try to import tree-sitter - it's optional but recommended
TREE_SITTER_AVAILABLE = False
try:
import tree_sitter
from tree_sitter import Language, Node, Parser, Tree
TREE_SITTER_AVAILABLE = True
logger.info("tree-sitter available, using AST-based analysis")
except ImportError:
logger.warning("tree-sitter not available, using regex-based fallback")
Tree = None
Node = None
# Try to import language bindings
LANGUAGES_AVAILABLE: dict[str, Any] = {}
if TREE_SITTER_AVAILABLE:
try:
import tree_sitter_python as tspython
LANGUAGES_AVAILABLE[".py"] = tspython.language()
except ImportError:
pass
try:
import tree_sitter_javascript as tsjs
LANGUAGES_AVAILABLE[".js"] = tsjs.language()
LANGUAGES_AVAILABLE[".jsx"] = tsjs.language()
except ImportError:
pass
try:
import tree_sitter_typescript as tsts
LANGUAGES_AVAILABLE[".ts"] = tsts.language_typescript()
LANGUAGES_AVAILABLE[".tsx"] = tsts.language_tsx()
except ImportError:
pass
@dataclass
class ExtractedElement:
"""A structural element extracted from code."""
element_type: str # function, class, import, variable, etc.
name: str
start_line: int
end_line: int
content: str
parent: Optional[str] = None # For nested elements (methods in classes)
metadata: dict[str, Any] = None
def __post_init__(self):
if self.metadata is None:
self.metadata = {}
class SemanticAnalyzer:
"""
Analyzes code changes at a semantic level.
Uses tree-sitter for AST-based analysis when available,
falling back to regex-based heuristics when not.
Example:
analyzer = SemanticAnalyzer()
analysis = analyzer.analyze_diff("src/App.tsx", before_code, after_code)
for change in analysis.changes:
print(f"{change.change_type.value}: {change.target}")
"""
def __init__(self):
"""Initialize the analyzer with available parsers."""
self._parsers: dict[str, Parser] = {}
debug(MODULE, "Initializing SemanticAnalyzer", tree_sitter_available=TREE_SITTER_AVAILABLE)
if TREE_SITTER_AVAILABLE:
for ext, lang in LANGUAGES_AVAILABLE.items():
parser = Parser()
parser.language = Language(lang)
self._parsers[ext] = parser
debug_detailed(MODULE, f"Initialized parser for {ext}")
debug_success(MODULE, "SemanticAnalyzer initialized", parsers=list(self._parsers.keys()))
else:
debug(MODULE, "Using regex-based fallback (tree-sitter not available)")
def analyze_diff(
self,
file_path: str,
before: str,
after: str,
task_id: Optional[str] = None,
) -> FileAnalysis:
"""
Analyze the semantic differences between two versions of a file.
Args:
file_path: Path to the file being analyzed
before: Content before changes
after: Content after changes
task_id: Optional task ID for context
Returns:
FileAnalysis containing semantic changes
"""
ext = Path(file_path).suffix.lower()
debug(MODULE, f"Analyzing diff for {file_path}",
file_path=file_path,
extension=ext,
before_length=len(before),
after_length=len(after),
task_id=task_id)
# Use tree-sitter if available for this language
if ext in self._parsers:
debug_detailed(MODULE, f"Using tree-sitter parser for {ext}")
analysis = self._analyze_with_tree_sitter(file_path, before, after, ext)
else:
debug_detailed(MODULE, f"Using regex fallback for {ext}")
analysis = self._analyze_with_regex(file_path, before, after, ext)
debug_success(MODULE, f"Analysis complete for {file_path}",
changes_found=len(analysis.changes),
functions_modified=len(analysis.functions_modified),
functions_added=len(analysis.functions_added),
imports_added=len(analysis.imports_added),
total_lines_changed=analysis.total_lines_changed)
# Log each change at verbose level
for change in analysis.changes:
debug_verbose(MODULE, f" Change: {change.change_type.value}",
target=change.target,
location=change.location,
lines=f"{change.line_start}-{change.line_end}")
return analysis
def _analyze_with_tree_sitter(
self,
file_path: str,
before: str,
after: str,
ext: str,
) -> FileAnalysis:
"""Analyze using tree-sitter AST parsing."""
parser = self._parsers[ext]
tree_before = parser.parse(bytes(before, "utf-8"))
tree_after = parser.parse(bytes(after, "utf-8"))
# Extract structural elements from both versions
elements_before = self._extract_elements(tree_before, before, ext)
elements_after = self._extract_elements(tree_after, after, ext)
# Compare and generate semantic changes
changes = self._compare_elements(elements_before, elements_after, ext)
# Build the analysis
analysis = FileAnalysis(file_path=file_path, changes=changes)
# Populate summary fields
for change in changes:
if change.change_type in {
ChangeType.MODIFY_FUNCTION,
ChangeType.ADD_HOOK_CALL,
}:
analysis.functions_modified.add(change.target)
elif change.change_type == ChangeType.ADD_FUNCTION:
analysis.functions_added.add(change.target)
elif change.change_type == ChangeType.ADD_IMPORT:
analysis.imports_added.add(change.target)
elif change.change_type == ChangeType.REMOVE_IMPORT:
analysis.imports_removed.add(change.target)
elif change.change_type in {
ChangeType.MODIFY_CLASS,
ChangeType.ADD_METHOD,
}:
analysis.classes_modified.add(change.target.split(".")[0])
analysis.total_lines_changed += change.line_end - change.line_start + 1
return analysis
def _extract_elements(
self,
tree: Tree,
source: str,
ext: str,
) -> dict[str, ExtractedElement]:
"""Extract structural elements from a syntax tree."""
elements: dict[str, ExtractedElement] = {}
source_bytes = bytes(source, "utf-8")
source_lines = source.split("\n")
def get_text(node: Node) -> str:
return source_bytes[node.start_byte : node.end_byte].decode("utf-8")
def get_line(byte_pos: int) -> int:
# Convert byte position to line number (1-indexed)
return source[:byte_pos].count("\n") + 1
# Language-specific extraction
if ext == ".py":
self._extract_python_elements(tree.root_node, elements, get_text, get_line)
elif ext in {".js", ".jsx", ".ts", ".tsx"}:
self._extract_js_elements(tree.root_node, elements, get_text, get_line, ext)
return elements
def _extract_python_elements(
self,
node: Node,
elements: dict[str, ExtractedElement],
get_text: callable,
get_line: callable,
parent: Optional[str] = None,
):
"""Extract elements from Python AST."""
for child in node.children:
if child.type == "import_statement":
# import x, y
text = get_text(child)
# Extract module names
for name_node in child.children:
if name_node.type == "dotted_name":
name = get_text(name_node)
elements[f"import:{name}"] = ExtractedElement(
element_type="import",
name=name,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=text,
)
elif child.type == "import_from_statement":
# from x import y, z
text = get_text(child)
module = None
for sub in child.children:
if sub.type == "dotted_name":
module = get_text(sub)
break
if module:
elements[f"import_from:{module}"] = ExtractedElement(
element_type="import_from",
name=module,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=text,
)
elif child.type == "function_definition":
name_node = child.child_by_field_name("name")
if name_node:
name = get_text(name_node)
full_name = f"{parent}.{name}" if parent else name
elements[f"function:{full_name}"] = ExtractedElement(
element_type="function",
name=full_name,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=get_text(child),
parent=parent,
)
elif child.type == "class_definition":
name_node = child.child_by_field_name("name")
if name_node:
name = get_text(name_node)
elements[f"class:{name}"] = ExtractedElement(
element_type="class",
name=name,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=get_text(child),
)
# Recurse into class body for methods
body = child.child_by_field_name("body")
if body:
self._extract_python_elements(
body, elements, get_text, get_line, parent=name
)
elif child.type == "decorated_definition":
# Handle decorated functions/classes
for sub in child.children:
if sub.type in {"function_definition", "class_definition"}:
self._extract_python_elements(
child, elements, get_text, get_line, parent
)
break
# Recurse for other compound statements
elif child.type in {"if_statement", "while_statement", "for_statement", "try_statement", "with_statement"}:
self._extract_python_elements(child, elements, get_text, get_line, parent)
def _extract_js_elements(
self,
node: Node,
elements: dict[str, ExtractedElement],
get_text: callable,
get_line: callable,
ext: str,
parent: Optional[str] = None,
):
"""Extract elements from JavaScript/TypeScript AST."""
for child in node.children:
if child.type == "import_statement":
text = get_text(child)
# Try to extract the source module
source_node = child.child_by_field_name("source")
if source_node:
source = get_text(source_node).strip("'\"")
elements[f"import:{source}"] = ExtractedElement(
element_type="import",
name=source,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=text,
)
elif child.type in {"function_declaration", "function"}:
name_node = child.child_by_field_name("name")
if name_node:
name = get_text(name_node)
full_name = f"{parent}.{name}" if parent else name
elements[f"function:{full_name}"] = ExtractedElement(
element_type="function",
name=full_name,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=get_text(child),
parent=parent,
)
elif child.type == "arrow_function":
# Arrow functions are usually assigned to variables
# We'll catch these via variable declarations
pass
elif child.type in {"lexical_declaration", "variable_declaration"}:
# const/let/var declarations
for declarator in child.children:
if declarator.type == "variable_declarator":
name_node = declarator.child_by_field_name("name")
value_node = declarator.child_by_field_name("value")
if name_node:
name = get_text(name_node)
content = get_text(child)
# Check if it's a function (arrow function or function expression)
is_function = False
if value_node and value_node.type in {
"arrow_function",
"function",
}:
is_function = True
elements[f"function:{name}"] = ExtractedElement(
element_type="function",
name=name,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=content,
parent=parent,
)
else:
elements[f"variable:{name}"] = ExtractedElement(
element_type="variable",
name=name,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=content,
parent=parent,
)
elif child.type == "class_declaration":
name_node = child.child_by_field_name("name")
if name_node:
name = get_text(name_node)
elements[f"class:{name}"] = ExtractedElement(
element_type="class",
name=name,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=get_text(child),
)
# Recurse into class body
body = child.child_by_field_name("body")
if body:
self._extract_js_elements(
body, elements, get_text, get_line, ext, parent=name
)
elif child.type == "method_definition":
name_node = child.child_by_field_name("name")
if name_node:
name = get_text(name_node)
full_name = f"{parent}.{name}" if parent else name
elements[f"method:{full_name}"] = ExtractedElement(
element_type="method",
name=full_name,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=get_text(child),
parent=parent,
)
elif child.type == "export_statement":
# Recurse into exports to find the actual declaration
self._extract_js_elements(
child, elements, get_text, get_line, ext, parent
)
# TypeScript specific
elif child.type in {"interface_declaration", "type_alias_declaration"}:
name_node = child.child_by_field_name("name")
if name_node:
name = get_text(name_node)
elem_type = "interface" if "interface" in child.type else "type"
elements[f"{elem_type}:{name}"] = ExtractedElement(
element_type=elem_type,
name=name,
start_line=get_line(child.start_byte),
end_line=get_line(child.end_byte),
content=get_text(child),
)
# Recurse into statement blocks
elif child.type in {"program", "statement_block", "class_body"}:
self._extract_js_elements(
child, elements, get_text, get_line, ext, parent
)
def _compare_elements(
self,
before: dict[str, ExtractedElement],
after: dict[str, ExtractedElement],
ext: str,
) -> list[SemanticChange]:
"""Compare extracted elements to generate semantic changes."""
changes: list[SemanticChange] = []
all_keys = set(before.keys()) | set(after.keys())
for key in all_keys:
elem_before = before.get(key)
elem_after = after.get(key)
if elem_before and not elem_after:
# Element was removed
change_type = self._get_remove_change_type(elem_before.element_type)
changes.append(
SemanticChange(
change_type=change_type,
target=elem_before.name,
location=self._get_location(elem_before),
line_start=elem_before.start_line,
line_end=elem_before.end_line,
content_before=elem_before.content,
content_after=None,
)
)
elif not elem_before and elem_after:
# Element was added
change_type = self._get_add_change_type(elem_after.element_type)
changes.append(
SemanticChange(
change_type=change_type,
target=elem_after.name,
location=self._get_location(elem_after),
line_start=elem_after.start_line,
line_end=elem_after.end_line,
content_before=None,
content_after=elem_after.content,
)
)
elif elem_before and elem_after:
# Element exists in both - check if modified
if elem_before.content != elem_after.content:
change_type = self._classify_modification(
elem_before, elem_after, ext
)
changes.append(
SemanticChange(
change_type=change_type,
target=elem_after.name,
location=self._get_location(elem_after),
line_start=elem_after.start_line,
line_end=elem_after.end_line,
content_before=elem_before.content,
content_after=elem_after.content,
)
)
return changes
def _get_add_change_type(self, element_type: str) -> ChangeType:
"""Map element type to add change type."""
mapping = {
"import": ChangeType.ADD_IMPORT,
"import_from": ChangeType.ADD_IMPORT,
"function": ChangeType.ADD_FUNCTION,
"class": ChangeType.ADD_CLASS,
"method": ChangeType.ADD_METHOD,
"variable": ChangeType.ADD_VARIABLE,
"interface": ChangeType.ADD_INTERFACE,
"type": ChangeType.ADD_TYPE,
}
return mapping.get(element_type, ChangeType.UNKNOWN)
def _get_remove_change_type(self, element_type: str) -> ChangeType:
"""Map element type to remove change type."""
mapping = {
"import": ChangeType.REMOVE_IMPORT,
"import_from": ChangeType.REMOVE_IMPORT,
"function": ChangeType.REMOVE_FUNCTION,
"class": ChangeType.REMOVE_CLASS,
"method": ChangeType.REMOVE_METHOD,
"variable": ChangeType.REMOVE_VARIABLE,
}
return mapping.get(element_type, ChangeType.UNKNOWN)
def _get_location(self, element: ExtractedElement) -> str:
"""Generate a location string for an element."""
if element.parent:
return f"{element.element_type}:{element.parent}.{element.name.split('.')[-1]}"
return f"{element.element_type}:{element.name}"
def _classify_modification(
self,
before: ExtractedElement,
after: ExtractedElement,
ext: str,
) -> ChangeType:
"""Classify what kind of modification was made."""
element_type = after.element_type
if element_type == "import":
return ChangeType.MODIFY_IMPORT
if element_type in {"function", "method"}:
# Analyze the function content for specific changes
return self._classify_function_modification(before.content, after.content, ext)
if element_type == "class":
return ChangeType.MODIFY_CLASS
if element_type == "interface":
return ChangeType.MODIFY_INTERFACE
if element_type == "type":
return ChangeType.MODIFY_TYPE
if element_type == "variable":
return ChangeType.MODIFY_VARIABLE
return ChangeType.UNKNOWN
def _classify_function_modification(
self,
before: str,
after: str,
ext: str,
) -> ChangeType:
"""Classify what changed in a function."""
# Check for React hook additions
hook_pattern = r"\buse[A-Z]\w*\s*\("
hooks_before = set(re.findall(hook_pattern, before))
hooks_after = set(re.findall(hook_pattern, after))
if hooks_after - hooks_before:
return ChangeType.ADD_HOOK_CALL
if hooks_before - hooks_after:
return ChangeType.REMOVE_HOOK_CALL
# Check for JSX wrapping (more JSX elements in after)
jsx_pattern = r"<[A-Z]\w*"
jsx_before = len(re.findall(jsx_pattern, before))
jsx_after = len(re.findall(jsx_pattern, after))
if jsx_after > jsx_before:
return ChangeType.WRAP_JSX
if jsx_after < jsx_before:
return ChangeType.UNWRAP_JSX
# Check if only JSX props changed
if ext in {".jsx", ".tsx"}:
# Simplified check - if the structure is same but content differs
struct_before = re.sub(r'=\{[^}]*\}|="[^"]*"', "=...", before)
struct_after = re.sub(r'=\{[^}]*\}|="[^"]*"', "=...", after)
if struct_before == struct_after:
return ChangeType.MODIFY_JSX_PROPS
return ChangeType.MODIFY_FUNCTION
def _analyze_with_regex(
self,
file_path: str,
before: str,
after: str,
ext: str,
) -> FileAnalysis:
"""Fallback analysis using regex when tree-sitter isn't available."""
changes: list[SemanticChange] = []
# Get a unified diff
diff = list(
difflib.unified_diff(
before.splitlines(keepends=True),
after.splitlines(keepends=True),
lineterm="",
)
)
# Analyze the diff for patterns
added_lines: list[tuple[int, str]] = []
removed_lines: list[tuple[int, str]] = []
current_line = 0
for line in diff:
if line.startswith("@@"):
# Parse the line numbers
match = re.match(r"@@ -\d+(?:,\d+)? \+(\d+)", line)
if match:
current_line = int(match.group(1))
elif line.startswith("+") and not line.startswith("+++"):
added_lines.append((current_line, line[1:]))
current_line += 1
elif line.startswith("-") and not line.startswith("---"):
removed_lines.append((current_line, line[1:]))
elif not line.startswith("-"):
current_line += 1
# Detect imports
import_pattern = self._get_import_pattern(ext)
for line_num, line in added_lines:
if import_pattern and import_pattern.match(line.strip()):
changes.append(
SemanticChange(
change_type=ChangeType.ADD_IMPORT,
target=line.strip(),
location="file_top",
line_start=line_num,
line_end=line_num,
content_after=line,
)
)
for line_num, line in removed_lines:
if import_pattern and import_pattern.match(line.strip()):
changes.append(
SemanticChange(
change_type=ChangeType.REMOVE_IMPORT,
target=line.strip(),
location="file_top",
line_start=line_num,
line_end=line_num,
content_before=line,
)
)
# Detect function changes (simplified)
func_pattern = self._get_function_pattern(ext)
if func_pattern:
funcs_before = set(func_pattern.findall(before))
funcs_after = set(func_pattern.findall(after))
for func in funcs_after - funcs_before:
changes.append(
SemanticChange(
change_type=ChangeType.ADD_FUNCTION,
target=func,
location=f"function:{func}",
line_start=1,
line_end=1,
)
)
for func in funcs_before - funcs_after:
changes.append(
SemanticChange(
change_type=ChangeType.REMOVE_FUNCTION,
target=func,
location=f"function:{func}",
line_start=1,
line_end=1,
)
)
# Build analysis
analysis = FileAnalysis(file_path=file_path, changes=changes)
for change in changes:
if change.change_type == ChangeType.ADD_IMPORT:
analysis.imports_added.add(change.target)
elif change.change_type == ChangeType.REMOVE_IMPORT:
analysis.imports_removed.add(change.target)
elif change.change_type == ChangeType.ADD_FUNCTION:
analysis.functions_added.add(change.target)
elif change.change_type == ChangeType.MODIFY_FUNCTION:
analysis.functions_modified.add(change.target)
analysis.total_lines_changed = len(added_lines) + len(removed_lines)
return analysis
def _get_import_pattern(self, ext: str) -> Optional[re.Pattern]:
"""Get the import pattern for a file extension."""
patterns = {
".py": re.compile(r"^(?:from\s+\S+\s+)?import\s+"),
".js": re.compile(r"^import\s+"),
".jsx": re.compile(r"^import\s+"),
".ts": re.compile(r"^import\s+"),
".tsx": re.compile(r"^import\s+"),
}
return patterns.get(ext)
def _get_function_pattern(self, ext: str) -> Optional[re.Pattern]:
"""Get the function definition pattern for a file extension."""
patterns = {
".py": re.compile(r"def\s+(\w+)\s*\("),
".js": re.compile(r"(?:function\s+(\w+)|(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s+)?(?:function|\([^)]*\)\s*=>))"),
".jsx": re.compile(r"(?:function\s+(\w+)|(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s+)?(?:function|\([^)]*\)\s*=>))"),
".ts": re.compile(r"(?:function\s+(\w+)|(?:const|let|var)\s+(\w+)\s*(?::\s*\w+)?\s*=\s*(?:async\s+)?(?:function|\([^)]*\)\s*=>))"),
".tsx": re.compile(r"(?:function\s+(\w+)|(?:const|let|var)\s+(\w+)\s*(?::\s*\w+)?\s*=\s*(?:async\s+)?(?:function|\([^)]*\)\s*=>))"),
}
return patterns.get(ext)
def analyze_file(self, file_path: str, content: str) -> FileAnalysis:
"""
Analyze a single file's structure (not a diff).
Useful for capturing baseline state.
Args:
file_path: Path to the file
content: File content
Returns:
FileAnalysis with structural elements (no changes, just structure)
"""
# Analyze against empty string to get all elements as "additions"
return self.analyze_diff(file_path, "", content)
@property
def supported_extensions(self) -> set[str]:
"""Get the set of supported file extensions."""
if TREE_SITTER_AVAILABLE:
# Tree-sitter extensions plus regex fallbacks
return set(self._parsers.keys()) | {".py", ".js", ".jsx", ".ts", ".tsx"}
else:
# Only regex-supported extensions
return {".py", ".js", ".jsx", ".ts", ".tsx"}
def is_supported(self, file_path: str) -> bool:
"""Check if a file type is supported for semantic analysis."""
ext = Path(file_path).suffix.lower()
return ext in self.supported_extensions
+545
View File
@@ -0,0 +1,545 @@
"""
Merge System Types
==================
Core data structures for the intent-aware merge system.
These types represent the semantic understanding of code changes,
enabling intelligent conflict detection and resolution.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Any, Optional
class ChangeType(Enum):
"""
Semantic classification of code changes.
These represent WHAT changed at a semantic level, not line-level diffs.
The merge system uses these to determine compatibility between changes.
"""
# Import changes
ADD_IMPORT = "add_import"
REMOVE_IMPORT = "remove_import"
MODIFY_IMPORT = "modify_import"
# Function/method changes
ADD_FUNCTION = "add_function"
REMOVE_FUNCTION = "remove_function"
MODIFY_FUNCTION = "modify_function"
RENAME_FUNCTION = "rename_function"
# React/JSX specific
ADD_HOOK_CALL = "add_hook_call"
REMOVE_HOOK_CALL = "remove_hook_call"
WRAP_JSX = "wrap_jsx"
UNWRAP_JSX = "unwrap_jsx"
ADD_JSX_ELEMENT = "add_jsx_element"
MODIFY_JSX_PROPS = "modify_jsx_props"
# Variable/constant changes
ADD_VARIABLE = "add_variable"
REMOVE_VARIABLE = "remove_variable"
MODIFY_VARIABLE = "modify_variable"
ADD_CONSTANT = "add_constant"
# Class changes
ADD_CLASS = "add_class"
REMOVE_CLASS = "remove_class"
MODIFY_CLASS = "modify_class"
ADD_METHOD = "add_method"
REMOVE_METHOD = "remove_method"
MODIFY_METHOD = "modify_method"
ADD_PROPERTY = "add_property"
# Type changes (TypeScript)
ADD_TYPE = "add_type"
MODIFY_TYPE = "modify_type"
ADD_INTERFACE = "add_interface"
MODIFY_INTERFACE = "modify_interface"
# Python specific
ADD_DECORATOR = "add_decorator"
REMOVE_DECORATOR = "remove_decorator"
# Generic
ADD_COMMENT = "add_comment"
MODIFY_COMMENT = "modify_comment"
FORMATTING_ONLY = "formatting_only"
UNKNOWN = "unknown"
class ConflictSeverity(Enum):
"""
Severity levels for detected conflicts.
Determines how the conflict should be handled:
- NONE: No conflict, can auto-merge
- LOW: Minor overlap, likely auto-mergeable with rules
- MEDIUM: Significant overlap, may need AI assistance
- HIGH: Major conflict, likely needs human review
- CRITICAL: Incompatible changes, definitely needs human review
"""
NONE = "none"
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class MergeStrategy(Enum):
"""
Strategies for merging compatible changes.
Each strategy is implemented in AutoMerger as a deterministic algorithm.
"""
# Import strategies
COMBINE_IMPORTS = "combine_imports"
# Function body strategies
HOOKS_FIRST = "hooks_first" # Add hooks at function start, then other changes
HOOKS_THEN_WRAP = "hooks_then_wrap" # Hooks first, then JSX wrapping
APPEND_STATEMENTS = "append_statements" # Add statements in order
# Structural strategies
APPEND_FUNCTIONS = "append_functions" # Add new functions after existing
APPEND_METHODS = "append_methods" # Add new methods to class
COMBINE_PROPS = "combine_props" # Merge JSX/object props
# Ordering strategies
ORDER_BY_DEPENDENCY = "order_by_dependency" # Analyze deps and order
ORDER_BY_TIME = "order_by_time" # Apply in chronological order
# Fallback
AI_REQUIRED = "ai_required" # Cannot auto-merge, need AI
HUMAN_REQUIRED = "human_required" # Cannot auto-merge, need human
class MergeDecision(Enum):
"""
Decision outcomes from the merge system.
"""
AUTO_MERGED = "auto_merged" # Python handled it, no AI
AI_MERGED = "ai_merged" # AI resolved the conflict
NEEDS_HUMAN_REVIEW = "needs_human_review" # Flagged for human
FAILED = "failed" # Could not merge
@dataclass
class SemanticChange:
"""
A single semantic change within a file.
This represents one logical modification (e.g., "added useAuth hook")
rather than a line-level diff.
Attributes:
change_type: The semantic classification of the change
target: What was changed (function name, import path, etc.)
location: Where in the file (file_top, function:App, class:User)
line_start: Starting line number (1-indexed)
line_end: Ending line number (1-indexed)
content_before: The code before the change (for modifications)
content_after: The code after the change
metadata: Additional context (dependency info, etc.)
"""
change_type: ChangeType
target: str
location: str
line_start: int
line_end: int
content_before: Optional[str] = None
content_after: Optional[str] = None
metadata: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
"change_type": self.change_type.value,
"target": self.target,
"location": self.location,
"line_start": self.line_start,
"line_end": self.line_end,
"content_before": self.content_before,
"content_after": self.content_after,
"metadata": self.metadata,
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> SemanticChange:
"""Create from dictionary."""
return cls(
change_type=ChangeType(data["change_type"]),
target=data["target"],
location=data["location"],
line_start=data["line_start"],
line_end=data["line_end"],
content_before=data.get("content_before"),
content_after=data.get("content_after"),
metadata=data.get("metadata", {}),
)
def overlaps_with(self, other: SemanticChange) -> bool:
"""Check if this change overlaps with another in location."""
# Same location means potential conflict
if self.location == other.location:
return True
# Check line overlap
if self.line_end >= other.line_start and other.line_end >= self.line_start:
return True
return False
@property
def is_additive(self) -> bool:
"""Check if this is a purely additive change."""
additive_types = {
ChangeType.ADD_IMPORT,
ChangeType.ADD_FUNCTION,
ChangeType.ADD_HOOK_CALL,
ChangeType.ADD_VARIABLE,
ChangeType.ADD_CONSTANT,
ChangeType.ADD_CLASS,
ChangeType.ADD_METHOD,
ChangeType.ADD_PROPERTY,
ChangeType.ADD_TYPE,
ChangeType.ADD_INTERFACE,
ChangeType.ADD_DECORATOR,
ChangeType.ADD_JSX_ELEMENT,
ChangeType.ADD_COMMENT,
}
return self.change_type in additive_types
@dataclass
class FileAnalysis:
"""
Complete semantic analysis of changes to a single file.
This aggregates all semantic changes and provides summary statistics
useful for conflict detection.
Attributes:
file_path: Path to the analyzed file (relative to project root)
changes: List of semantic changes detected
functions_modified: Set of function/method names that were changed
functions_added: Set of new functions/methods
imports_added: Set of new imports
imports_removed: Set of removed imports
classes_modified: Set of modified class names
total_lines_changed: Approximate lines affected
"""
file_path: str
changes: list[SemanticChange] = field(default_factory=list)
functions_modified: set[str] = field(default_factory=set)
functions_added: set[str] = field(default_factory=set)
imports_added: set[str] = field(default_factory=set)
imports_removed: set[str] = field(default_factory=set)
classes_modified: set[str] = field(default_factory=set)
total_lines_changed: int = 0
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
"file_path": self.file_path,
"changes": [c.to_dict() for c in self.changes],
"functions_modified": list(self.functions_modified),
"functions_added": list(self.functions_added),
"imports_added": list(self.imports_added),
"imports_removed": list(self.imports_removed),
"classes_modified": list(self.classes_modified),
"total_lines_changed": self.total_lines_changed,
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> FileAnalysis:
"""Create from dictionary."""
return cls(
file_path=data["file_path"],
changes=[SemanticChange.from_dict(c) for c in data.get("changes", [])],
functions_modified=set(data.get("functions_modified", [])),
functions_added=set(data.get("functions_added", [])),
imports_added=set(data.get("imports_added", [])),
imports_removed=set(data.get("imports_removed", [])),
classes_modified=set(data.get("classes_modified", [])),
total_lines_changed=data.get("total_lines_changed", 0),
)
def get_changes_at_location(self, location: str) -> list[SemanticChange]:
"""Get all changes at a specific location."""
return [c for c in self.changes if c.location == location]
@property
def is_additive_only(self) -> bool:
"""Check if all changes are purely additive."""
return all(c.is_additive for c in self.changes)
@property
def locations_changed(self) -> set[str]:
"""Get all unique locations that were changed."""
return {c.location for c in self.changes}
@dataclass
class ConflictRegion:
"""
A detected conflict between multiple task changes.
This represents a region where two or more tasks made changes
that may not be automatically compatible.
Attributes:
file_path: The file containing the conflict
location: The specific location (e.g., "function:App")
tasks_involved: List of task IDs that modified this location
change_types: The types of changes from each task
severity: How serious the conflict is
can_auto_merge: Whether Python rules can handle this
merge_strategy: If auto-mergeable, which strategy to use
reason: Human-readable explanation of the conflict
"""
file_path: str
location: str
tasks_involved: list[str]
change_types: list[ChangeType]
severity: ConflictSeverity
can_auto_merge: bool
merge_strategy: Optional[MergeStrategy] = None
reason: str = ""
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
"file_path": self.file_path,
"location": self.location,
"tasks_involved": self.tasks_involved,
"change_types": [ct.value for ct in self.change_types],
"severity": self.severity.value,
"can_auto_merge": self.can_auto_merge,
"merge_strategy": self.merge_strategy.value if self.merge_strategy else None,
"reason": self.reason,
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> ConflictRegion:
"""Create from dictionary."""
return cls(
file_path=data["file_path"],
location=data["location"],
tasks_involved=data["tasks_involved"],
change_types=[ChangeType(ct) for ct in data["change_types"]],
severity=ConflictSeverity(data["severity"]),
can_auto_merge=data["can_auto_merge"],
merge_strategy=MergeStrategy(data["merge_strategy"]) if data.get("merge_strategy") else None,
reason=data.get("reason", ""),
)
@dataclass
class TaskSnapshot:
"""
A snapshot of a task's changes to a file.
This captures what a single task did to a file, including
the semantic understanding of its changes and intent.
Attributes:
task_id: The task identifier
task_intent: One-sentence description of what the task intended
started_at: When the task started working on this file
completed_at: When the task finished
content_hash_before: Hash of file content when task started
content_hash_after: Hash of file content when task finished
semantic_changes: List of semantic changes made
raw_diff: Optional raw unified diff for reference
"""
task_id: str
task_intent: str
started_at: datetime
completed_at: Optional[datetime] = None
content_hash_before: str = ""
content_hash_after: str = ""
semantic_changes: list[SemanticChange] = field(default_factory=list)
raw_diff: Optional[str] = None
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
"task_id": self.task_id,
"task_intent": self.task_intent,
"started_at": self.started_at.isoformat(),
"completed_at": self.completed_at.isoformat() if self.completed_at else None,
"content_hash_before": self.content_hash_before,
"content_hash_after": self.content_hash_after,
"semantic_changes": [c.to_dict() for c in self.semantic_changes],
"raw_diff": self.raw_diff,
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> TaskSnapshot:
"""Create from dictionary."""
return cls(
task_id=data["task_id"],
task_intent=data["task_intent"],
started_at=datetime.fromisoformat(data["started_at"]),
completed_at=datetime.fromisoformat(data["completed_at"]) if data.get("completed_at") else None,
content_hash_before=data.get("content_hash_before", ""),
content_hash_after=data.get("content_hash_after", ""),
semantic_changes=[SemanticChange.from_dict(c) for c in data.get("semantic_changes", [])],
raw_diff=data.get("raw_diff"),
)
@dataclass
class FileEvolution:
"""
Complete evolution history of a single file.
Tracks the baseline state and all task modifications,
enabling intelligent merge decisions with full context.
Attributes:
file_path: Path to the file (relative to project root)
baseline_commit: Git commit hash of the baseline
baseline_captured_at: When the baseline was captured
baseline_content_hash: Hash of baseline content
baseline_snapshot_path: Path to stored baseline content
task_snapshots: Ordered list of task modifications
"""
file_path: str
baseline_commit: str
baseline_captured_at: datetime
baseline_content_hash: str
baseline_snapshot_path: str
task_snapshots: list[TaskSnapshot] = field(default_factory=list)
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
"file_path": self.file_path,
"baseline_commit": self.baseline_commit,
"baseline_captured_at": self.baseline_captured_at.isoformat(),
"baseline_content_hash": self.baseline_content_hash,
"baseline_snapshot_path": self.baseline_snapshot_path,
"task_snapshots": [ts.to_dict() for ts in self.task_snapshots],
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> FileEvolution:
"""Create from dictionary."""
return cls(
file_path=data["file_path"],
baseline_commit=data["baseline_commit"],
baseline_captured_at=datetime.fromisoformat(data["baseline_captured_at"]),
baseline_content_hash=data["baseline_content_hash"],
baseline_snapshot_path=data["baseline_snapshot_path"],
task_snapshots=[TaskSnapshot.from_dict(ts) for ts in data.get("task_snapshots", [])],
)
def get_task_snapshot(self, task_id: str) -> Optional[TaskSnapshot]:
"""Get a specific task's snapshot."""
for snapshot in self.task_snapshots:
if snapshot.task_id == task_id:
return snapshot
return None
def add_task_snapshot(self, snapshot: TaskSnapshot) -> None:
"""Add or update a task snapshot."""
# Remove existing snapshot for this task if present
self.task_snapshots = [
ts for ts in self.task_snapshots
if ts.task_id != snapshot.task_id
]
self.task_snapshots.append(snapshot)
# Keep sorted by start time
self.task_snapshots.sort(key=lambda ts: ts.started_at)
@property
def tasks_involved(self) -> list[str]:
"""Get list of task IDs that modified this file."""
return [ts.task_id for ts in self.task_snapshots]
@dataclass
class MergeResult:
"""
Result of a merge operation.
Contains the outcome, merged content, and detailed information
about how the merge was performed.
Attributes:
decision: The merge decision outcome
file_path: Path to the merged file
merged_content: The final merged content (if successful)
conflicts_resolved: List of conflicts that were resolved
conflicts_remaining: List of conflicts needing human review
ai_calls_made: Number of AI calls required
tokens_used: Approximate tokens used for AI calls
explanation: Human-readable explanation of what was done
error: Error message if merge failed
"""
decision: MergeDecision
file_path: str
merged_content: Optional[str] = None
conflicts_resolved: list[ConflictRegion] = field(default_factory=list)
conflicts_remaining: list[ConflictRegion] = field(default_factory=list)
ai_calls_made: int = 0
tokens_used: int = 0
explanation: str = ""
error: Optional[str] = None
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
"decision": self.decision.value,
"file_path": self.file_path,
"merged_content": self.merged_content,
"conflicts_resolved": [c.to_dict() for c in self.conflicts_resolved],
"conflicts_remaining": [c.to_dict() for c in self.conflicts_remaining],
"ai_calls_made": self.ai_calls_made,
"tokens_used": self.tokens_used,
"explanation": self.explanation,
"error": self.error,
}
@property
def success(self) -> bool:
"""Check if merge was successful."""
return self.decision in {MergeDecision.AUTO_MERGED, MergeDecision.AI_MERGED}
@property
def needs_human_review(self) -> bool:
"""Check if human review is needed."""
return len(self.conflicts_remaining) > 0 or self.decision == MergeDecision.NEEDS_HUMAN_REVIEW
def compute_content_hash(content: str) -> str:
"""Compute a hash of file content for comparison."""
return hashlib.sha256(content.encode('utf-8')).hexdigest()[:16]
def sanitize_path_for_storage(file_path: str) -> str:
"""Convert a file path to a safe storage name."""
# Replace path separators and special chars
safe = file_path.replace("/", "_").replace("\\", "_").replace(".", "_")
return safe
+160 -3
View File
@@ -21,11 +21,13 @@ Terminology mapping (technical -> user-friendly):
- working directory -> "your project"
"""
import json
import shutil
import subprocess
import sys
from enum import Enum
from pathlib import Path
from typing import Optional
from ui import (
Icons,
@@ -44,6 +46,13 @@ from ui import (
)
from worktree import WorktreeInfo, WorktreeManager
# Import merge system
from merge import (
MergeOrchestrator,
MergeDecision,
ConflictSeverity,
)
class WorkspaceMode(Enum):
"""How auto-claude should work."""
@@ -554,17 +563,28 @@ def handle_workspace_choice(
def merge_existing_build(
project_dir: Path, spec_name: str, no_commit: bool = False
project_dir: Path,
spec_name: str,
no_commit: bool = False,
use_smart_merge: bool = True,
) -> bool:
"""
Merge an existing build into the project.
Merge an existing build into the project using intent-aware merge.
Called when user runs: python auto-claude/run.py --spec X --merge
This uses the MergeOrchestrator to:
1. Analyze semantic changes from the task
2. Detect potential conflicts with main branch
3. Auto-merge compatible changes
4. Use AI for ambiguous conflicts (if enabled)
5. Fall back to git merge for remaining changes
Args:
project_dir: The project directory
spec_name: Name of the spec
no_commit: If True, merge changes but don't commit (stage only for review in IDE)
use_smart_merge: If True, use intent-aware merge (default True)
Returns:
True if merge succeeded
@@ -594,10 +614,33 @@ def merge_existing_build(
print(box(content, width=60, style="heavy"))
manager = WorktreeManager(project_dir)
show_build_summary(manager, spec_name)
print()
# Try smart merge first if enabled
if use_smart_merge:
smart_result = _try_smart_merge(
project_dir, spec_name, worktree_path, manager
)
if smart_result is not None:
# Smart merge handled it (success or identified conflicts)
if smart_result.get("success"):
# Smart merge succeeded, now do git merge
success_result = manager.merge_worktree(
spec_name, delete_after=True, no_commit=no_commit
)
if success_result:
_print_merge_success(no_commit, smart_result.get("stats"))
return True
elif smart_result.get("conflicts"):
# Has conflicts that need resolution
_print_conflict_info(smart_result)
print()
print(muted("Attempting git merge anyway..."))
print()
# Fall back to standard git merge
success_result = manager.merge_worktree(
spec_name, delete_after=True, no_commit=no_commit
)
@@ -622,6 +665,120 @@ def merge_existing_build(
return False
def _try_smart_merge(
project_dir: Path,
spec_name: str,
worktree_path: Path,
manager: WorktreeManager,
) -> Optional[dict]:
"""
Try to use the intent-aware merge system.
Returns:
Dict with results, or None if smart merge not applicable
"""
try:
print(muted(" Analyzing changes with intent-aware merge..."))
# Initialize the orchestrator
orchestrator = MergeOrchestrator(
project_dir,
enable_ai=True, # Enable AI for ambiguous conflicts
dry_run=False,
)
# Refresh evolution data from the worktree
orchestrator.evolution_tracker.refresh_from_git(spec_name, worktree_path)
# Preview what merge would look like
preview = orchestrator.preview_merge([spec_name])
files_to_merge = len(preview.get("files_to_merge", []))
conflicts = preview.get("conflicts", [])
auto_mergeable = preview.get("summary", {}).get("auto_mergeable", 0)
print(muted(f" Found {files_to_merge} files to merge"))
if conflicts:
print(muted(f" Detected {len(conflicts)} potential conflict(s)"))
print(muted(f" Auto-mergeable: {auto_mergeable}/{len(conflicts)}"))
# Check if any conflicts need human review
needs_human = [
c for c in conflicts
if not c.get("can_auto_merge")
]
if needs_human:
return {
"success": False,
"conflicts": needs_human,
"preview": preview,
}
# All conflicts can be auto-merged or no conflicts
print(muted(" All changes compatible, proceeding with merge..."))
return {
"success": True,
"stats": {
"files_merged": files_to_merge,
"auto_resolved": auto_mergeable,
},
}
except Exception as e:
# If smart merge fails, fall back to git
print(muted(f" Smart merge unavailable: {e}"))
return None
def _print_merge_success(no_commit: bool, stats: Optional[dict] = None) -> None:
"""Print success message after merge."""
print()
if stats:
print(muted(f" Files merged: {stats.get('files_merged', 0)}"))
if stats.get('auto_resolved'):
print(muted(f" Conflicts auto-resolved: {stats.get('auto_resolved', 0)}"))
print()
if no_commit:
print_status("Changes are staged in your working directory.", "success")
print()
print("Review the changes in your IDE, then commit:")
print(highlight(" git commit -m 'your commit message'"))
print()
print("Or discard if not satisfied:")
print(muted(" git reset --hard HEAD"))
else:
print_status("Your feature has been added to your project.", "success")
def _print_conflict_info(result: dict) -> None:
"""Print information about detected conflicts."""
conflicts = result.get("conflicts", [])
print()
print_status(f"Detected {len(conflicts)} conflict(s) that need attention:", "warning")
print()
for i, conflict in enumerate(conflicts[:5], 1): # Show first 5
file_path = conflict.get("file", "unknown")
location = conflict.get("location", "")
reason = conflict.get("reason", "")
severity = conflict.get("severity", "unknown")
print(f" {i}. {highlight(file_path)}")
if location:
print(f" Location: {muted(location)}")
if reason:
print(f" Reason: {muted(reason)}")
print(f" Severity: {severity}")
print()
if len(conflicts) > 5:
print(muted(f" ... and {len(conflicts) - 5} more"))
def review_existing_build(project_dir: Path, spec_name: str) -> bool:
"""
Show what an existing build contains.
+1300
View File
File diff suppressed because it is too large Load Diff