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

merge logic v0.3

Browse Source
This commit is contained in:
AndyMik90
2025-12-16 12:48:36 +01:00
parent e08ab62acb
commit c5d33cd8b1
8 changed files with 2073 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
auto-claude/cli/workspace_commands.py
+79 -65
View File
@@ -132,11 +132,10 @@ def handle_cleanup_worktrees_command(project_dir: Path) -> None:
def _check_git_merge_conflicts(project_dir: Path, spec_name: str) -> dict:
"""
Check for git-level merge conflicts by attempting a dry-run merge.
Check for git-level merge conflicts WITHOUT modifying the working directory.
This detects conflicts that occur when main has diverged from the
worktree branch (e.g., other changes were merged to main since
the worktree was created).
Uses git merge-tree and git diff to detect conflicts in-memory,
which avoids triggering Vite HMR or other file watchers.
Args:
project_dir: Project root directory
@@ -152,7 +151,7 @@ def _check_git_merge_conflicts(project_dir: Path, spec_name: str) -> dict:
"""
import subprocess
debug(MODULE, "Checking for git-level merge conflicts...")
debug(MODULE, "Checking for git-level merge conflicts (non-destructive)...")
spec_branch = f"auto-claude/{spec_name}"
result = {
@@ -175,95 +174,110 @@ def _check_git_merge_conflicts(project_dir: Path, spec_name: str) -> dict:
if base_result.returncode == 0:
result["base_branch"] = base_result.stdout.strip()
# Check how many commits main is ahead of the merge-base
# Get the merge base commit
merge_base_result = subprocess.run(
["git", "merge-base", result["base_branch"], spec_branch],
cwd=project_dir,
capture_output=True,
text=True,
)
if merge_base_result.returncode == 0:
merge_base = merge_base_result.stdout.strip()
if merge_base_result.returncode != 0:
debug_warning(MODULE, "Could not find merge base")
return result
# Count commits main is ahead
ahead_result = subprocess.run(
["git", "rev-list", "--count", f"{merge_base}..{result['base_branch']}"],
cwd=project_dir,
capture_output=True,
text=True,
)
if ahead_result.returncode == 0:
commits_behind = int(ahead_result.stdout.strip())
result["commits_behind"] = commits_behind
if commits_behind > 0:
result["needs_rebase"] = True
debug(MODULE, f"Main is {commits_behind} commits ahead of worktree base")
merge_base = merge_base_result.stdout.strip()
# Try a merge --no-commit --no-ff to see if there would be conflicts
# First, save current state
stash_result = subprocess.run(
["git", "stash", "push", "-m", "merge-preview-check"],
# Count commits main is ahead
ahead_result = subprocess.run(
["git", "rev-list", "--count", f"{merge_base}..{result['base_branch']}"],
cwd=project_dir,
capture_output=True,
text=True,
)
if ahead_result.returncode == 0:
commits_behind = int(ahead_result.stdout.strip())
result["commits_behind"] = commits_behind
if commits_behind > 0:
result["needs_rebase"] = True
debug(MODULE, f"Main is {commits_behind} commits ahead of worktree base")
# Get commit hashes for merge-tree
main_commit_result = subprocess.run(
["git", "rev-parse", result["base_branch"]],
cwd=project_dir,
capture_output=True,
text=True,
)
spec_commit_result = subprocess.run(
["git", "rev-parse", spec_branch],
cwd=project_dir,
capture_output=True,
text=True,
)
try:
# Attempt the merge (dry run style)
merge_result = subprocess.run(
["git", "merge", "--no-commit", "--no-ff", spec_branch],
cwd=project_dir,
capture_output=True,
text=True,
)
if main_commit_result.returncode != 0 or spec_commit_result.returncode != 0:
debug_warning(MODULE, "Could not resolve branch commits")
return result
if merge_result.returncode != 0:
result["has_conflicts"] = True
debug(MODULE, "Git merge would have conflicts")
main_commit = main_commit_result.stdout.strip()
spec_commit = spec_commit_result.stdout.strip()
# Parse conflicting files from the output
# Look for "CONFLICT (content): Merge conflict in <file>"
import re
for line in merge_result.stdout.split("\n") + merge_result.stderr.split("\n"):
# Match patterns like "CONFLICT (content): Merge conflict in path/to/file"
match = re.search(r"CONFLICT.*:\s*(?:Merge conflict in\s+)?(.+)", line)
# Use git merge-tree to check for conflicts WITHOUT touching working directory
# This is a plumbing command that does a 3-way merge in memory
merge_tree_result = subprocess.run(
["git", "merge-tree", "--write-tree", "--no-messages", merge_base, main_commit, spec_commit],
cwd=project_dir,
capture_output=True,
text=True,
)
# merge-tree returns exit code 1 if there are conflicts
if merge_tree_result.returncode != 0:
result["has_conflicts"] = True
debug(MODULE, "Git merge-tree detected conflicts")
# Parse the output for conflicting files
# merge-tree --write-tree outputs conflict info to stderr
output = merge_tree_result.stdout + merge_tree_result.stderr
for line in output.split("\n"):
# Look for lines indicating conflicts
if "CONFLICT" in line:
# Extract file path from conflict message
import re
match = re.search(r"(?:Merge conflict in|CONFLICT.*?:)\s*(.+?)(?:\s*$|\s+\()", line)
if match:
file_path = match.group(1).strip()
if file_path and file_path not in result["conflicting_files"]:
result["conflicting_files"].append(file_path)
# Also check git status for unmerged files
status_result = subprocess.run(
["git", "diff", "--name-only", "--diff-filter=U"],
# Fallback: if we didn't parse conflicts, use diff to find files changed in both branches
if not result["conflicting_files"]:
# Files changed in main since merge-base
main_files_result = subprocess.run(
["git", "diff", "--name-only", merge_base, main_commit],
cwd=project_dir,
capture_output=True,
text=True,
)
if status_result.returncode == 0:
for line in status_result.stdout.strip().split("\n"):
if line and line not in result["conflicting_files"]:
result["conflicting_files"].append(line)
main_files = set(main_files_result.stdout.strip().split("\n")) if main_files_result.stdout.strip() else set()
debug(MODULE, f"Conflicting files: {result['conflicting_files']}")
else:
debug_success(MODULE, "Git merge would succeed without conflicts")
finally:
# Always abort the merge and restore state
subprocess.run(
["git", "merge", "--abort"],
cwd=project_dir,
capture_output=True,
text=True,
)
# Restore stash if we made one
if "No local changes" not in stash_result.stdout:
subprocess.run(
["git", "stash", "pop"],
# Files changed in spec branch since merge-base
spec_files_result = subprocess.run(
["git", "diff", "--name-only", merge_base, spec_commit],
cwd=project_dir,
capture_output=True,
text=True,
)
spec_files = set(spec_files_result.stdout.strip().split("\n")) if spec_files_result.stdout.strip() else set()
# Files modified in both = potential conflicts
conflicting = main_files & spec_files
result["conflicting_files"] = list(conflicting)
debug(MODULE, f"Found {len(conflicting)} files modified in both branches")
debug(MODULE, f"Conflicting files: {result['conflicting_files']}")
else:
debug_success(MODULE, "Git merge-tree: no conflicts detected")
except Exception as e:
debug_error(MODULE, f"Error checking git conflicts: {e}")
auto-claude/merge/__init__.py
+28
View File
@@ -41,6 +41,21 @@ from .auto_merger import AutoMerger
from .file_evolution import FileEvolutionTracker
from .ai_resolver import AIResolver, create_claude_resolver
from .orchestrator import MergeOrchestrator
from .file_timeline import (
FileTimelineTracker,
FileTimeline,
MainBranchEvent,
BranchPoint,
WorktreeState,
TaskIntent,
TaskFileView,
MergeContext,
)
from .prompts import (
build_timeline_merge_prompt,
build_simple_merge_prompt,
optimize_prompt_for_length,
)
__all__ = [
# Types
@@ -62,4 +77,17 @@ __all__ = [
"AIResolver",
"create_claude_resolver",
"MergeOrchestrator",
# File Timeline (Intent-Aware Merge System)
"FileTimelineTracker",
"FileTimeline",
"MainBranchEvent",
"BranchPoint",
"WorktreeState",
"TaskIntent",
"TaskFileView",
"MergeContext",
# Prompt Templates
"build_timeline_merge_prompt",
"build_simple_merge_prompt",
"optimize_prompt_for_length",
]
auto-claude/merge/file_timeline.py
+992
View File
@@ -0,0 +1,992 @@
"""
File Timeline Tracker
=====================
Intent-aware file evolution tracking for multi-agent merge resolution.
This module implements the File-Centric Timeline Model that tracks:
- Main branch evolution (human commits)
- Task worktree modifications (AI agent changes)
- Task branch points and intent
- Pending task awareness for forward-compatible merges
The key insight is that each file has a TIMELINE of changes from multiple sources,
and the Merge AI needs this complete context to make intelligent decisions.
Usage:
tracker = FileTimelineTracker(project_dir)
# When a task starts
tracker.on_task_start(
task_id="task-001-auth",
files_to_modify=["src/App.tsx"],
branch_point_commit="abc123",
task_intent="Add authentication via useAuth() hook"
)
# When human commits to main (via git hook)
tracker.on_main_branch_commit("def456")
# When getting merge context
context = tracker.get_merge_context("task-001-auth", "src/App.tsx")
"""
from __future__ import annotations
import json
import logging
import subprocess
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Literal
logger = logging.getLogger(__name__)
# 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
MODULE = "merge.file_timeline"
# =============================================================================
# DATA MODELS
# =============================================================================
@dataclass
class MainBranchEvent:
"""
Represents a single commit to main branch affecting a file.
These events form the "spine" of the file's timeline - the authoritative
history that all task worktrees diverge from and merge back into.
"""
# Git identification
commit_hash: str
timestamp: datetime
# Content at this point
content: str
# Source of change
source: Literal['human', 'merged_task']
merged_from_task: Optional[str] = None # If source is 'merged_task'
# Intent/reason for change
commit_message: str = ""
# For richer context (optional)
author: Optional[str] = None
diff_summary: Optional[str] = None # e.g., "+15 -3 lines"
def to_dict(self) -> dict:
return {
"commit_hash": self.commit_hash,
"timestamp": self.timestamp.isoformat(),
"content": self.content,
"source": self.source,
"merged_from_task": self.merged_from_task,
"commit_message": self.commit_message,
"author": self.author,
"diff_summary": self.diff_summary,
}
@classmethod
def from_dict(cls, data: dict) -> MainBranchEvent:
return cls(
commit_hash=data["commit_hash"],
timestamp=datetime.fromisoformat(data["timestamp"]),
content=data["content"],
source=data["source"],
merged_from_task=data.get("merged_from_task"),
commit_message=data.get("commit_message", ""),
author=data.get("author"),
diff_summary=data.get("diff_summary"),
)
@dataclass
class BranchPoint:
"""The exact point a task branched from main."""
commit_hash: str
content: str
timestamp: datetime
def to_dict(self) -> dict:
return {
"commit_hash": self.commit_hash,
"content": self.content,
"timestamp": self.timestamp.isoformat(),
}
@classmethod
def from_dict(cls, data: dict) -> BranchPoint:
return cls(
commit_hash=data["commit_hash"],
content=data["content"],
timestamp=datetime.fromisoformat(data["timestamp"]),
)
@dataclass
class WorktreeState:
"""Current state of a file in a task's worktree."""
content: str
last_modified: datetime
def to_dict(self) -> dict:
return {
"content": self.content,
"last_modified": self.last_modified.isoformat(),
}
@classmethod
def from_dict(cls, data: dict) -> WorktreeState:
return cls(
content=data["content"],
last_modified=datetime.fromisoformat(data["last_modified"]),
)
@dataclass
class TaskIntent:
"""What the task intends to do with this file."""
title: str
description: str
from_plan: bool = False # True if extracted from implementation_plan.json
def to_dict(self) -> dict:
return {
"title": self.title,
"description": self.description,
"from_plan": self.from_plan,
}
@classmethod
def from_dict(cls, data: dict) -> TaskIntent:
return cls(
title=data["title"],
description=data["description"],
from_plan=data.get("from_plan", False),
)
@dataclass
class TaskFileView:
"""
A single task's relationship with a specific file.
This captures everything we need to know about how one task
sees and modifies one file.
"""
task_id: str
# The exact point this task branched from main
branch_point: BranchPoint
# Current state in the task's worktree (None if not modified yet)
worktree_state: Optional[WorktreeState] = None
# What the task intends to do
task_intent: TaskIntent = field(default_factory=lambda: TaskIntent("", ""))
# Drift tracking - how many commits happened in main since branch
commits_behind_main: int = 0
# Lifecycle status
status: Literal['active', 'merged', 'abandoned'] = 'active'
merged_at: Optional[datetime] = None
def to_dict(self) -> dict:
return {
"task_id": self.task_id,
"branch_point": self.branch_point.to_dict(),
"worktree_state": self.worktree_state.to_dict() if self.worktree_state else None,
"task_intent": self.task_intent.to_dict(),
"commits_behind_main": self.commits_behind_main,
"status": self.status,
"merged_at": self.merged_at.isoformat() if self.merged_at else None,
}
@classmethod
def from_dict(cls, data: dict) -> TaskFileView:
return cls(
task_id=data["task_id"],
branch_point=BranchPoint.from_dict(data["branch_point"]),
worktree_state=WorktreeState.from_dict(data["worktree_state"]) if data.get("worktree_state") else None,
task_intent=TaskIntent.from_dict(data["task_intent"]) if data.get("task_intent") else TaskIntent("", ""),
commits_behind_main=data.get("commits_behind_main", 0),
status=data.get("status", "active"),
merged_at=datetime.fromisoformat(data["merged_at"]) if data.get("merged_at") else None,
)
@dataclass
class FileTimeline:
"""
The core data structure tracking a single file's complete history.
This is the "file-centric" view - instead of asking "what did Task X change?",
we ask "what happened to File Y over time, from ALL sources?"
"""
file_path: str
# Main branch evolution - the authoritative history
main_branch_history: List[MainBranchEvent] = field(default_factory=list)
# Each task's isolated view of this file
task_views: Dict[str, TaskFileView] = field(default_factory=dict)
# Metadata
created_at: datetime = field(default_factory=datetime.now)
last_updated: datetime = field(default_factory=datetime.now)
def add_main_event(self, event: MainBranchEvent) -> None:
"""Add a main branch event and increment drift for all active tasks."""
self.main_branch_history.append(event)
self.last_updated = datetime.now()
# Update commits_behind_main for all active tasks
for task_view in self.task_views.values():
if task_view.status == 'active':
task_view.commits_behind_main += 1
def add_task_view(self, task_view: TaskFileView) -> None:
"""Add or update a task's view of this file."""
self.task_views[task_view.task_id] = task_view
self.last_updated = datetime.now()
def get_task_view(self, task_id: str) -> Optional[TaskFileView]:
"""Get a task's view of this file."""
return self.task_views.get(task_id)
def get_active_tasks(self) -> List[TaskFileView]:
"""Get all tasks that are still active (not merged/abandoned)."""
return [tv for tv in self.task_views.values() if tv.status == 'active']
def get_events_since_commit(self, commit_hash: str) -> List[MainBranchEvent]:
"""Get all main branch events since a given commit."""
events = []
found_commit = False
for event in self.main_branch_history:
if found_commit:
events.append(event)
if event.commit_hash == commit_hash:
found_commit = True
return events
def get_current_main_state(self) -> Optional[MainBranchEvent]:
"""Get the most recent main branch event."""
if self.main_branch_history:
return self.main_branch_history[-1]
return None
def to_dict(self) -> dict:
return {
"file_path": self.file_path,
"main_branch_history": [e.to_dict() for e in self.main_branch_history],
"task_views": {k: v.to_dict() for k, v in self.task_views.items()},
"created_at": self.created_at.isoformat(),
"last_updated": self.last_updated.isoformat(),
}
@classmethod
def from_dict(cls, data: dict) -> FileTimeline:
timeline = cls(
file_path=data["file_path"],
created_at=datetime.fromisoformat(data["created_at"]),
last_updated=datetime.fromisoformat(data["last_updated"]),
)
timeline.main_branch_history = [
MainBranchEvent.from_dict(e) for e in data.get("main_branch_history", [])
]
timeline.task_views = {
k: TaskFileView.from_dict(v) for k, v in data.get("task_views", {}).items()
}
return timeline
@dataclass
class MergeContext:
"""
The complete context package provided to the Merge AI.
This is the "situational awareness" the AI needs to make intelligent
merge decisions.
"""
file_path: str
# The task being merged
task_id: str
task_intent: TaskIntent
# Task's starting point
task_branch_point: BranchPoint
# What happened in main since task branched (ordered from oldest to newest)
main_evolution: List[MainBranchEvent]
# Task's changes
task_worktree_content: str
# Current main state
current_main_content: str
current_main_commit: str
# Other tasks that also touch this file (for forward-compatibility)
other_pending_tasks: List[Dict] # [{task_id, intent, branch_point, commits_behind}]
# Metrics
total_commits_behind: int
total_pending_tasks: int
def to_dict(self) -> dict:
return {
"file_path": self.file_path,
"task_id": self.task_id,
"task_intent": self.task_intent.to_dict(),
"task_branch_point": self.task_branch_point.to_dict(),
"main_evolution": [e.to_dict() for e in self.main_evolution],
"task_worktree_content": self.task_worktree_content,
"current_main_content": self.current_main_content,
"current_main_commit": self.current_main_commit,
"other_pending_tasks": self.other_pending_tasks,
"total_commits_behind": self.total_commits_behind,
"total_pending_tasks": self.total_pending_tasks,
}
# =============================================================================
# FILE TIMELINE TRACKER SERVICE
# =============================================================================
class FileTimelineTracker:
"""
Central service managing all file timelines.
This service is the "brain" of the intent-aware merge system. It:
- Creates and manages FileTimeline objects
- Handles events from git hooks and task lifecycle
- Provides merge context to the AI resolver
- Persists timelines to JSON storage
"""
def __init__(self, project_path: Path, storage_path: Optional[Path] = None):
"""
Initialize the file timeline tracker.
Args:
project_path: Root directory of the project
storage_path: Directory for timeline storage (default: .auto-claude/)
"""
debug(MODULE, "Initializing FileTimelineTracker",
project_path=str(project_path))
self.project_path = Path(project_path).resolve()
self.storage_path = storage_path or (self.project_path / ".auto-claude")
self.timelines_dir = self.storage_path / "file-timelines"
# Ensure storage directory exists
self.timelines_dir.mkdir(parents=True, exist_ok=True)
# In-memory cache of timelines
self._timelines: Dict[str, FileTimeline] = {}
# Load existing timelines
self._load_from_storage()
debug_success(MODULE, "FileTimelineTracker initialized",
timelines_loaded=len(self._timelines))
# =========================================================================
# EVENT HANDLERS
# =========================================================================
def on_task_start(
self,
task_id: str,
files_to_modify: List[str],
files_to_create: Optional[List[str]] = None,
branch_point_commit: Optional[str] = None,
task_intent: str = "",
task_title: str = "",
) -> None:
"""
Called when a task creates its worktree and starts work.
This captures the task's "branch point" - what the file looked like
when the task started, which is crucial for understanding what the
task actually changed vs what was already there.
"""
debug(MODULE, f"on_task_start: {task_id}",
files_to_modify=files_to_modify,
branch_point=branch_point_commit)
# Get actual branch point commit if not provided
if not branch_point_commit:
branch_point_commit = self._get_current_main_commit()
timestamp = datetime.now()
for file_path in files_to_modify:
# Get or create timeline for this file
timeline = self._get_or_create_timeline(file_path)
# Get file content at branch point
content = self._get_file_content_at_commit(file_path, branch_point_commit)
if content is None:
# File doesn't exist at this commit - might be created by task
content = ""
# Create task file view
task_view = TaskFileView(
task_id=task_id,
branch_point=BranchPoint(
commit_hash=branch_point_commit,
content=content,
timestamp=timestamp,
),
task_intent=TaskIntent(
title=task_title or task_id,
description=task_intent,
from_plan=bool(task_intent),
),
commits_behind_main=0,
status='active',
)
timeline.add_task_view(task_view)
self._persist_timeline(file_path)
debug_success(MODULE, f"Task {task_id} registered with {len(files_to_modify)} files")
def on_main_branch_commit(self, commit_hash: str) -> None:
"""
Called via git post-commit hook when human commits to main.
This tracks the "drift" - how many commits have happened in main
since each task branched.
"""
debug(MODULE, f"on_main_branch_commit: {commit_hash}")
# Get list of files changed in this commit
changed_files = self._get_files_changed_in_commit(commit_hash)
for file_path in changed_files:
# Only update existing timelines (we don't create new ones for random files)
if file_path not in self._timelines:
continue
timeline = self._timelines[file_path]
# Get file content at this commit
content = self._get_file_content_at_commit(file_path, commit_hash)
if content is None:
continue
# Get commit metadata
commit_info = self._get_commit_info(commit_hash)
# Create main branch event
event = MainBranchEvent(
commit_hash=commit_hash,
timestamp=datetime.now(),
content=content,
source='human',
commit_message=commit_info.get('message', ''),
author=commit_info.get('author'),
diff_summary=commit_info.get('diff_summary'),
)
timeline.add_main_event(event)
self._persist_timeline(file_path)
debug_success(MODULE, f"Processed main commit {commit_hash[:8]}",
files_updated=len(changed_files))
def on_task_worktree_change(
self,
task_id: str,
file_path: str,
new_content: str,
) -> None:
"""
Called when AI agent modifies a file in its worktree.
This updates the task's "worktree state" - what the file currently
looks like in that task's isolated workspace.
"""
debug(MODULE, f"on_task_worktree_change: {task_id} -> {file_path}")
timeline = self._timelines.get(file_path)
if not timeline:
# Create timeline if it doesn't exist
timeline = self._get_or_create_timeline(file_path)
task_view = timeline.get_task_view(task_id)
if not task_view:
debug_warning(MODULE, f"Task {task_id} not registered for {file_path}")
return
# Update worktree state
task_view.worktree_state = WorktreeState(
content=new_content,
last_modified=datetime.now(),
)
self._persist_timeline(file_path)
def on_task_merged(self, task_id: str, merge_commit: str) -> None:
"""
Called after a task is successfully merged to main.
This updates the timeline to show:
1. The task is now merged
2. Main branch has a new commit (from this merge)
"""
debug(MODULE, f"on_task_merged: {task_id}")
# Get list of files this task modified
task_files = self.get_files_for_task(task_id)
for file_path in task_files:
timeline = self._timelines.get(file_path)
if not timeline:
continue
task_view = timeline.get_task_view(task_id)
if not task_view:
continue
# Mark task as merged
task_view.status = 'merged'
task_view.merged_at = datetime.now()
# Add main branch event for the merge
content = self._get_file_content_at_commit(file_path, merge_commit)
if content:
event = MainBranchEvent(
commit_hash=merge_commit,
timestamp=datetime.now(),
content=content,
source='merged_task',
merged_from_task=task_id,
commit_message=f"Merged from {task_id}",
)
timeline.add_main_event(event)
self._persist_timeline(file_path)
debug_success(MODULE, f"Task {task_id} marked as merged")
def on_task_abandoned(self, task_id: str) -> None:
"""
Called if a task is cancelled/abandoned.
"""
debug(MODULE, f"on_task_abandoned: {task_id}")
task_files = self.get_files_for_task(task_id)
for file_path in task_files:
timeline = self._timelines.get(file_path)
if not timeline:
continue
task_view = timeline.get_task_view(task_id)
if task_view:
task_view.status = 'abandoned'
self._persist_timeline(file_path)
# =========================================================================
# QUERY METHODS
# =========================================================================
def get_merge_context(self, task_id: str, file_path: str) -> Optional[MergeContext]:
"""
Build complete merge context for AI resolver.
This is the key method that produces the "situational awareness"
the Merge AI needs.
"""
debug(MODULE, f"get_merge_context: {task_id} -> {file_path}")
timeline = self._timelines.get(file_path)
if not timeline:
debug_warning(MODULE, f"No timeline found for {file_path}")
return None
task_view = timeline.get_task_view(task_id)
if not task_view:
debug_warning(MODULE, f"Task {task_id} not found in timeline for {file_path}")
return None
# Get main evolution since task branched
main_evolution = timeline.get_events_since_commit(task_view.branch_point.commit_hash)
# Get current main state
current_main = timeline.get_current_main_state()
current_main_content = current_main.content if current_main else task_view.branch_point.content
current_main_commit = current_main.commit_hash if current_main else task_view.branch_point.commit_hash
# Get task's worktree content
worktree_content = ""
if task_view.worktree_state:
worktree_content = task_view.worktree_state.content
else:
# Try to get from worktree path
worktree_content = self._get_worktree_file_content(task_id, file_path)
# Get other pending tasks
other_tasks = []
for tv in timeline.get_active_tasks():
if tv.task_id != task_id:
other_tasks.append({
"task_id": tv.task_id,
"intent": tv.task_intent.description,
"branch_point": tv.branch_point.commit_hash,
"commits_behind": tv.commits_behind_main,
})
context = MergeContext(
file_path=file_path,
task_id=task_id,
task_intent=task_view.task_intent,
task_branch_point=task_view.branch_point,
main_evolution=main_evolution,
task_worktree_content=worktree_content,
current_main_content=current_main_content,
current_main_commit=current_main_commit,
other_pending_tasks=other_tasks,
total_commits_behind=task_view.commits_behind_main,
total_pending_tasks=len(other_tasks),
)
debug_success(MODULE, f"Built merge context",
commits_behind=task_view.commits_behind_main,
main_events=len(main_evolution),
other_tasks=len(other_tasks))
return context
def get_files_for_task(self, task_id: str) -> List[str]:
"""Return all files this task is tracking."""
files = []
for file_path, timeline in self._timelines.items():
if task_id in timeline.task_views:
files.append(file_path)
return files
def get_pending_tasks_for_file(self, file_path: str) -> List[TaskFileView]:
"""Return all active tasks that modify this file."""
timeline = self._timelines.get(file_path)
if not timeline:
return []
return timeline.get_active_tasks()
def get_task_drift(self, task_id: str) -> Dict[str, int]:
"""Return commits-behind-main for each file in task."""
drift = {}
for file_path, timeline in self._timelines.items():
task_view = timeline.get_task_view(task_id)
if task_view and task_view.status == 'active':
drift[file_path] = task_view.commits_behind_main
return drift
def has_timeline(self, file_path: str) -> bool:
"""Check if a file has an active timeline."""
return file_path in self._timelines
def get_timeline(self, file_path: str) -> Optional[FileTimeline]:
"""Get the timeline for a file."""
return self._timelines.get(file_path)
# =========================================================================
# PERSISTENCE
# =========================================================================
def _load_from_storage(self) -> None:
"""Load timelines from disk on startup."""
index_path = self.timelines_dir / "index.json"
if not index_path.exists():
return
try:
with open(index_path) as f:
index = json.load(f)
for file_path in index.get("files", []):
timeline_file = self._get_timeline_file_path(file_path)
if timeline_file.exists():
with open(timeline_file) as f:
data = json.load(f)
self._timelines[file_path] = FileTimeline.from_dict(data)
debug(MODULE, f"Loaded {len(self._timelines)} timelines from storage")
except Exception as e:
logger.error(f"Failed to load timelines: {e}")
def _persist_timeline(self, file_path: str) -> None:
"""Save a single timeline to disk."""
timeline = self._timelines.get(file_path)
if not timeline:
return
try:
# Save timeline file
timeline_file = self._get_timeline_file_path(file_path)
timeline_file.parent.mkdir(parents=True, exist_ok=True)
with open(timeline_file, "w") as f:
json.dump(timeline.to_dict(), f, indent=2)
# Update index
self._update_index()
except Exception as e:
logger.error(f"Failed to persist timeline for {file_path}: {e}")
def _update_index(self) -> None:
"""Update the index file with all tracked files."""
index_path = self.timelines_dir / "index.json"
index = {
"files": list(self._timelines.keys()),
"last_updated": datetime.now().isoformat(),
}
with open(index_path, "w") as f:
json.dump(index, f, indent=2)
def _get_timeline_file_path(self, file_path: str) -> Path:
"""Get the storage path for a file's timeline."""
# Encode path: src/App.tsx -> src_App.tsx.json
safe_name = file_path.replace("/", "_").replace("\\", "_")
return self.timelines_dir / f"{safe_name}.json"
def _get_or_create_timeline(self, file_path: str) -> FileTimeline:
"""Get existing timeline or create new one."""
if file_path not in self._timelines:
self._timelines[file_path] = FileTimeline(file_path=file_path)
return self._timelines[file_path]
# =========================================================================
# GIT HELPERS
# =========================================================================
def _get_current_main_commit(self) -> str:
"""Get the current HEAD commit on main branch."""
try:
result = subprocess.run(
["git", "rev-parse", "HEAD"],
cwd=self.project_path,
capture_output=True,
text=True,
check=True,
)
return result.stdout.strip()
except subprocess.CalledProcessError:
return "unknown"
def _get_file_content_at_commit(self, file_path: str, commit_hash: str) -> Optional[str]:
"""Get file content at a specific commit."""
try:
result = subprocess.run(
["git", "show", f"{commit_hash}:{file_path}"],
cwd=self.project_path,
capture_output=True,
text=True,
)
if result.returncode == 0:
return result.stdout
return None
except Exception:
return None
def _get_files_changed_in_commit(self, commit_hash: str) -> List[str]:
"""Get list of files changed in a commit."""
try:
result = subprocess.run(
["git", "diff-tree", "--no-commit-id", "--name-only", "-r", commit_hash],
cwd=self.project_path,
capture_output=True,
text=True,
check=True,
)
return [f for f in result.stdout.strip().split("\n") if f]
except subprocess.CalledProcessError:
return []
def _get_commit_info(self, commit_hash: str) -> dict:
"""Get commit metadata."""
info = {}
try:
# Get commit message
result = subprocess.run(
["git", "log", "-1", "--format=%s", commit_hash],
cwd=self.project_path,
capture_output=True,
text=True,
)
if result.returncode == 0:
info["message"] = result.stdout.strip()
# Get author
result = subprocess.run(
["git", "log", "-1", "--format=%an", commit_hash],
cwd=self.project_path,
capture_output=True,
text=True,
)
if result.returncode == 0:
info["author"] = result.stdout.strip()
# Get diff stat
result = subprocess.run(
["git", "diff-tree", "--stat", "--no-commit-id", commit_hash],
cwd=self.project_path,
capture_output=True,
text=True,
)
if result.returncode == 0:
info["diff_summary"] = result.stdout.strip().split("\n")[-1] if result.stdout.strip() else None
except Exception:
pass
return info
def _get_worktree_file_content(self, task_id: str, file_path: str) -> str:
"""Get file content from a task's worktree."""
# Extract spec name from task_id (remove 'task-' prefix if present)
spec_name = task_id.replace("task-", "") if task_id.startswith("task-") else task_id
worktree_path = self.project_path / ".worktrees" / spec_name / file_path
if worktree_path.exists():
return worktree_path.read_text(encoding="utf-8")
return ""
# =========================================================================
# CAPTURE METHODS (for integration with existing code)
# =========================================================================
def capture_worktree_state(self, task_id: str, worktree_path: Path) -> None:
"""
Capture the current state of all modified files in a worktree.
Called before merge to ensure we have the latest worktree content.
"""
debug(MODULE, f"capture_worktree_state: {task_id}")
try:
# Get all changed files in worktree vs main
result = subprocess.run(
["git", "diff", "--name-only", "main...HEAD"],
cwd=worktree_path,
capture_output=True,
text=True,
)
if result.returncode != 0:
return
changed_files = [f for f in result.stdout.strip().split("\n") if f]
for file_path in changed_files:
full_path = worktree_path / file_path
if full_path.exists():
content = full_path.read_text(encoding="utf-8")
self.on_task_worktree_change(task_id, file_path, content)
debug_success(MODULE, f"Captured {len(changed_files)} files from worktree")
except Exception as e:
logger.error(f"Failed to capture worktree state: {e}")
def initialize_from_worktree(
self,
task_id: str,
worktree_path: Path,
task_intent: str = "",
task_title: str = "",
) -> None:
"""
Initialize timeline tracking from an existing worktree.
Used for retroactive registration of tasks that were created
before the timeline system was in place.
"""
debug(MODULE, f"initialize_from_worktree: {task_id}")
try:
# Get the branch point (merge-base with main)
result = subprocess.run(
["git", "merge-base", "main", "HEAD"],
cwd=worktree_path,
capture_output=True,
text=True,
)
if result.returncode != 0:
debug_warning(MODULE, "Could not determine branch point")
return
branch_point = result.stdout.strip()
# Get changed files
result = subprocess.run(
["git", "diff", "--name-only", f"{branch_point}...HEAD"],
cwd=worktree_path,
capture_output=True,
text=True,
)
if result.returncode != 0:
return
changed_files = [f for f in result.stdout.strip().split("\n") if f]
# Register task for these files
self.on_task_start(
task_id=task_id,
files_to_modify=changed_files,
branch_point_commit=branch_point,
task_intent=task_intent,
task_title=task_title,
)
# Capture current worktree state
self.capture_worktree_state(task_id, worktree_path)
# Calculate drift (commits behind main)
result = subprocess.run(
["git", "rev-list", "--count", f"{branch_point}..main"],
cwd=self.project_path,
capture_output=True,
text=True,
)
if result.returncode == 0:
drift = int(result.stdout.strip())
for file_path in changed_files:
timeline = self._timelines.get(file_path)
if timeline:
task_view = timeline.get_task_view(task_id)
if task_view:
task_view.commits_behind_main = drift
self._persist_timeline(file_path)
debug_success(MODULE, f"Initialized from worktree",
files=len(changed_files),
branch_point=branch_point[:8])
except Exception as e:
logger.error(f"Failed to initialize from worktree: {e}")
auto-claude/merge/hooks/post-commit
Executable
+43
View File
@@ -0,0 +1,43 @@
#!/bin/bash
#
# Git post-commit hook for FileTimelineTracker
# =============================================
#
# This hook notifies the FileTimelineTracker when human commits
# are made to the main branch, enabling drift tracking.
#
# Installation:
# Copy to .git/hooks/post-commit and make executable
# Or use: python -m auto_claude.merge.install_hook
#
COMMIT_HASH=$(git rev-parse HEAD)
BRANCH=$(git rev-parse --abbrev-ref HEAD)
# Only track commits to main/master branch
# Skip if we're in a worktree (auto-claude branches)
if [[ "$BRANCH" == "main" ]] || [[ "$BRANCH" == "master" ]]; then
# Check if this is the main working directory (not a worktree)
# Worktrees have a .git file pointing to the main repo, not a .git directory
if [[ -d ".git" ]]; then
# Find python executable
if command -v python3 &> /dev/null; then
PYTHON=python3
elif command -v python &> /dev/null; then
PYTHON=python
else
# Python not found, skip silently
exit 0
fi
# Try to notify the tracker
# Run in background to avoid slowing down commits
($PYTHON -m auto_claude.merge.tracker_cli notify-commit "$COMMIT_HASH" 2>/dev/null &) &
# Don't let hook failures block commits
exit 0
fi
fi
# Not main branch or in worktree, do nothing
exit 0
auto-claude/merge/install_hook.py
+185
View File
@@ -0,0 +1,185 @@
"""
Git Hook Installer for FileTimelineTracker
==========================================
Installs the post-commit hook for tracking main branch commits.
Usage:
python -m auto_claude.merge.install_hook [--project-path /path/to/project]
"""
import argparse
import shutil
import stat
import sys
from pathlib import Path
HOOK_SCRIPT = '''#!/bin/bash
#
# Git post-commit hook for FileTimelineTracker
# =============================================
#
# This hook notifies the FileTimelineTracker when human commits
# are made to the main branch, enabling drift tracking.
#
COMMIT_HASH=$(git rev-parse HEAD)
BRANCH=$(git rev-parse --abbrev-ref HEAD)
# Only track commits to main/master branch
# Skip if we're in a worktree (auto-claude branches)
if [[ "$BRANCH" == "main" ]] || [[ "$BRANCH" == "master" ]]; then
# Check if this is the main working directory (not a worktree)
# Worktrees have a .git file pointing to the main repo, not a .git directory
if [[ -d ".git" ]]; then
# Find python executable
if command -v python3 &> /dev/null; then
PYTHON=python3
elif command -v python &> /dev/null; then
PYTHON=python
else
# Python not found, skip silently
exit 0
fi
# Try to notify the tracker
# Run in background to avoid slowing down commits
($PYTHON -m auto_claude.merge.tracker_cli notify-commit "$COMMIT_HASH" 2>/dev/null &) &
# Don't let hook failures block commits
exit 0
fi
fi
# Not main branch or in worktree, do nothing
exit 0
'''
def find_project_root() -> Path:
"""Find the project root by looking for .git directory."""
current = Path.cwd()
while current != current.parent:
if (current / ".git").exists():
return current
current = current.parent
return Path.cwd()
def install_hook(project_path: Path) -> bool:
"""Install the post-commit hook to a project."""
git_dir = project_path / ".git"
# Handle worktrees (where .git is a file, not directory)
if git_dir.is_file():
# Read the gitdir from the file
content = git_dir.read_text().strip()
if content.startswith("gitdir:"):
git_dir = Path(content.split(":", 1)[1].strip())
else:
print(f"Error: Cannot parse .git file at {git_dir}")
return False
if not git_dir.is_dir():
print(f"Error: No .git directory found at {project_path}")
return False
hooks_dir = git_dir / "hooks"
hooks_dir.mkdir(exist_ok=True)
hook_path = hooks_dir / "post-commit"
# Check if hook already exists
if hook_path.exists():
existing = hook_path.read_text()
if "FileTimelineTracker" in existing:
print(f"Hook already installed at {hook_path}")
return True
# Backup existing hook
backup_path = hooks_dir / "post-commit.backup"
shutil.copy(hook_path, backup_path)
print(f"Backed up existing hook to {backup_path}")
# Append our hook to existing
with open(hook_path, "a") as f:
f.write("\n\n# FileTimelineTracker integration\n")
f.write(HOOK_SCRIPT.split("#!/bin/bash", 1)[1]) # Skip shebang
print(f"Appended FileTimelineTracker hook to {hook_path}")
else:
# Write new hook
hook_path.write_text(HOOK_SCRIPT)
print(f"Created new hook at {hook_path}")
# Make executable
hook_path.chmod(hook_path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
print("Hook is now executable")
return True
def uninstall_hook(project_path: Path) -> bool:
"""Remove the post-commit hook from a project."""
git_dir = project_path / ".git"
if git_dir.is_file():
content = git_dir.read_text().strip()
if content.startswith("gitdir:"):
git_dir = Path(content.split(":", 1)[1].strip())
hook_path = git_dir / "hooks" / "post-commit"
if not hook_path.exists():
print("No hook to uninstall")
return True
content = hook_path.read_text()
if "FileTimelineTracker" not in content:
print("Hook does not contain FileTimelineTracker integration")
return True
# Check if we can restore from backup
backup_path = git_dir / "hooks" / "post-commit.backup"
if backup_path.exists():
shutil.move(backup_path, hook_path)
print(f"Restored original hook from backup")
else:
# Remove the hook entirely
hook_path.unlink()
print(f"Removed hook at {hook_path}")
return True
def main():
parser = argparse.ArgumentParser(
description="Install/uninstall FileTimelineTracker git hook"
)
parser.add_argument(
"--project-path",
type=Path,
help="Path to project (default: current directory)",
)
parser.add_argument(
"--uninstall",
action="store_true",
help="Uninstall the hook",
)
args = parser.parse_args()
project_path = args.project_path or find_project_root()
if args.uninstall:
success = uninstall_hook(project_path)
else:
success = install_hook(project_path)
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()
auto-claude/merge/prompts.py
+291
View File
@@ -0,0 +1,291 @@
"""
AI Merge Prompt Templates
=========================
Templates for providing rich context to the AI merge resolver,
using the FileTimelineTracker's complete file evolution data.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .file_timeline import MergeContext, MainBranchEvent
def build_timeline_merge_prompt(context: "MergeContext") -> str:
"""
Build a complete merge prompt using FileTimelineTracker context.
This provides the AI with full situational awareness:
- Task's starting point (branch point)
- Complete main branch evolution since branch
- Task's intent and changes
- Other pending tasks that will merge later
Args:
context: MergeContext from FileTimelineTracker.get_merge_context()
Returns:
Formatted prompt string for AI merge resolution
"""
# Build main evolution section
main_evolution_section = _build_main_evolution_section(context)
# Build pending tasks section
pending_tasks_section = _build_pending_tasks_section(context)
prompt = f'''MERGING: {context.file_path}
TASK: {context.task_id} ({context.task_intent.title})
{"=" * 79}
TASK'S STARTING POINT
Branched from commit: {context.task_branch_point.commit_hash[:12]}
Branched at: {context.task_branch_point.timestamp}
{"─" * 79}
```
{context.task_branch_point.content}
```
{"=" * 79}
{main_evolution_section}
CURRENT MAIN CONTENT (commit {context.current_main_commit[:12]}):
{"─" * 79}
```
{context.current_main_content}
```
{"=" * 79}
TASK'S CHANGES
Intent: "{context.task_intent.description or context.task_intent.title}"
{"─" * 79}
```
{context.task_worktree_content}
```
{"=" * 79}
{pending_tasks_section}
YOUR TASK:
1. Merge {context.task_id}'s changes into the current main version
2. PRESERVE all changes from main branch commits listed above
- Every human commit since the task branched must be retained
- Every previously merged task's changes must be retained
3. APPLY {context.task_id}'s changes
- Intent: {context.task_intent.description or context.task_intent.title}
- The task's changes should achieve its stated intent
4. ENSURE COMPATIBILITY with pending tasks
{_build_compatibility_instructions(context)}
5. OUTPUT only the complete merged file content
{"=" * 79}
'''
return prompt
def _build_main_evolution_section(context: "MergeContext") -> str:
"""Build the main branch evolution section of the prompt."""
if not context.main_evolution:
return f"""MAIN BRANCH EVOLUTION (0 commits since task branched)
{"─" * 79}
No changes have been made to main branch since this task started.
"""
lines = [f"MAIN BRANCH EVOLUTION ({len(context.main_evolution)} commits since task branched)"]
lines.append("─" * 79)
lines.append("")
for event in context.main_evolution:
source_label = event.source.upper()
if event.source == 'merged_task' and event.merged_from_task:
source_label = f"MERGED FROM {event.merged_from_task}"
lines.append(f'COMMIT {event.commit_hash[:12]} [{source_label}]: "{event.commit_message}"')
lines.append(f"Timestamp: {event.timestamp}")
if event.diff_summary:
lines.append(f"Changes: {event.diff_summary}")
else:
lines.append("Changes: See content evolution below")
lines.append("")
return "\n".join(lines)
def _build_pending_tasks_section(context: "MergeContext") -> str:
"""Build the other pending tasks section."""
separator = "─" * 79
if not context.other_pending_tasks:
return f"""OTHER TASKS MODIFYING THIS FILE
{separator}
No other tasks are pending for this file.
"""
lines = ["OTHER TASKS ALSO MODIFYING THIS FILE (not yet merged)"]
lines.append("─" * 79)
lines.append("")
for task in context.other_pending_tasks:
task_id = task.get("task_id", "unknown")
intent = task.get("intent", "No intent specified")
branch_point = task.get("branch_point", "unknown")[:12]
commits_behind = task.get("commits_behind", 0)
lines.append(f"• {task_id} (branched at {branch_point}, {commits_behind} commits behind)")
lines.append(f' Intent: "{intent}"')
lines.append("")
return "\n".join(lines)
def _build_compatibility_instructions(context: "MergeContext") -> str:
"""Build compatibility instructions based on pending tasks."""
if not context.other_pending_tasks:
return "- No other tasks pending for this file"
lines = [f"- {len(context.other_pending_tasks)} other task(s) will merge after this"]
lines.append(" - Structure your merge to accommodate their upcoming changes:")
for task in context.other_pending_tasks:
task_id = task.get("task_id", "unknown")
intent = task.get("intent", "")
if intent:
lines.append(f" - {task_id}: {intent[:80]}...")
else:
lines.append(f" - {task_id}")
return "\n".join(lines)
def build_simple_merge_prompt(
file_path: str,
main_content: str,
worktree_content: str,
base_content: str | None,
spec_name: str,
language: str,
task_intent: dict | None = None,
) -> str:
"""
Build a simple three-way merge prompt (fallback when timeline not available).
This is the traditional merge prompt without full timeline context.
"""
intent_section = ""
if task_intent:
intent_section = f"""
=== FEATURE BRANCH INTENT ({spec_name}) ===
Task: {task_intent.get('title', spec_name)}
Description: {task_intent.get('description', 'No description')}
"""
if task_intent.get('spec_summary'):
intent_section += f"Summary: {task_intent['spec_summary']}\n"
base_section = base_content if base_content else "(File did not exist in common ancestor)"
prompt = f'''You are a code merge expert. Merge the following conflicting versions of a file.
FILE: {file_path}
The file was modified in both the main branch and in the "{spec_name}" feature branch.
Your task is to produce a merged version that incorporates ALL changes from both branches.
{intent_section}
=== COMMON ANCESTOR (base) ===
{base_section}
=== MAIN BRANCH VERSION ===
{main_content}
=== FEATURE BRANCH VERSION ({spec_name}) ===
{worktree_content}
MERGE RULES:
1. Keep ALL imports from both versions
2. Keep ALL new functions/components from both versions
3. If the same function was modified differently, combine the changes logically
4. Preserve the intent of BOTH branches - main's changes are important too
5. If there's a genuine semantic conflict (same thing done differently), prefer the feature branch version but include main's additions
6. The merged code MUST be syntactically valid {language}
Output ONLY the merged code, wrapped in triple backticks:
```{language}
merged code here
```
'''
return prompt
def optimize_prompt_for_length(
context: "MergeContext",
max_content_chars: int = 50000,
max_evolution_events: int = 10,
) -> "MergeContext":
"""
Optimize a MergeContext for prompt length by trimming large content.
For very long files or many commits, this summarizes the middle
parts to keep the prompt within reasonable bounds.
Args:
context: Original MergeContext
max_content_chars: Maximum characters for file content
max_evolution_events: Maximum main branch events to include
Returns:
Modified MergeContext with trimmed content
"""
# Trim main evolution to first N and last N events if too long
if len(context.main_evolution) > max_evolution_events:
half = max_evolution_events // 2
first_events = context.main_evolution[:half]
last_events = context.main_evolution[-half:]
# Create a placeholder event for the middle
from datetime import datetime
from .file_timeline import MainBranchEvent
omitted_count = len(context.main_evolution) - max_evolution_events
placeholder = MainBranchEvent(
commit_hash="...",
timestamp=datetime.now(),
content="[Content omitted for brevity]",
source="human",
commit_message=f"({omitted_count} commits omitted for brevity)",
)
context.main_evolution = first_events + [placeholder] + last_events
# Trim content if too long
def _trim_content(content: str, label: str) -> str:
if len(content) > max_content_chars:
half = max_content_chars // 2
return (
content[:half]
+ f"\n\n... [{label}: {len(content) - max_content_chars} chars omitted] ...\n\n"
+ content[-half:]
)
return content
context.task_branch_point.content = _trim_content(
context.task_branch_point.content, "branch point"
)
context.task_worktree_content = _trim_content(
context.task_worktree_content, "worktree"
)
context.current_main_content = _trim_content(
context.current_main_content, "main"
)
return context
auto-claude/merge/tracker_cli.py
+233
View File
@@ -0,0 +1,233 @@
"""
FileTimelineTracker CLI
=======================
CLI interface for the FileTimelineTracker service.
Used by git hooks and manual operations.
Usage:
python -m auto_claude.merge.tracker_cli notify-commit <hash>
python -m auto_claude.merge.tracker_cli show-timeline <file_path>
python -m auto_claude.merge.tracker_cli show-drift <task_id>
"""
import argparse
import json
import sys
from pathlib import Path
from .file_timeline import FileTimelineTracker
def find_project_root() -> Path:
"""Find the project root by looking for .auto-claude or .git directory."""
current = Path.cwd()
# Walk up until we find .auto-claude or .git
while current != current.parent:
if (current / ".auto-claude").exists() or (current / ".git").exists():
return current
current = current.parent
# Default to cwd
return Path.cwd()
def get_tracker() -> FileTimelineTracker:
"""Get the FileTimelineTracker instance for this project."""
project_path = find_project_root()
return FileTimelineTracker(project_path)
def cmd_notify_commit(args):
"""Handle the notify-commit command from git post-commit hook."""
tracker = get_tracker()
commit_hash = args.commit_hash
print(f"[FileTimelineTracker] Processing commit: {commit_hash[:8]}")
tracker.on_main_branch_commit(commit_hash)
print(f"[FileTimelineTracker] Commit processed successfully")
def cmd_show_timeline(args):
"""Show the timeline for a file."""
tracker = get_tracker()
file_path = args.file_path
timeline = tracker.get_timeline(file_path)
if not timeline:
print(f"No timeline found for: {file_path}")
return
print(f"\n=== Timeline for: {file_path} ===\n")
print(f"Created: {timeline.created_at}")
print(f"Last Updated: {timeline.last_updated}")
print(f"\n--- Main Branch History ({len(timeline.main_branch_history)} events) ---")
for i, event in enumerate(timeline.main_branch_history):
print(f" [{i+1}] {event.commit_hash[:8]} ({event.source}): {event.commit_message[:50]}...")
print(f"\n--- Task Views ({len(timeline.task_views)} tasks) ---")
for task_id, view in timeline.task_views.items():
status = f"[{view.status.upper()}]"
behind = f"{view.commits_behind_main} commits behind"
print(f" {task_id} {status} - {behind}")
print(f" Branch point: {view.branch_point.commit_hash[:8]}")
print(f" Intent: {view.task_intent.title}")
def cmd_show_drift(args):
"""Show commits-behind-main for a task."""
tracker = get_tracker()
task_id = args.task_id
drift = tracker.get_task_drift(task_id)
if not drift:
print(f"No files found for task: {task_id}")
return
print(f"\n=== Drift Report for: {task_id} ===\n")
total_drift = 0
for file_path, commits_behind in sorted(drift.items()):
print(f" {file_path}: {commits_behind} commits behind")
total_drift = max(total_drift, commits_behind)
print(f"\n Max drift: {total_drift} commits")
def cmd_show_context(args):
"""Show merge context for a task and file."""
tracker = get_tracker()
task_id = args.task_id
file_path = args.file_path
context = tracker.get_merge_context(task_id, file_path)
if not context:
print(f"No merge context available for {task_id} -> {file_path}")
return
print(f"\n=== Merge Context for: {task_id} -> {file_path} ===\n")
print(f"Task Intent: {context.task_intent.title}")
print(f" {context.task_intent.description}")
print(f"\nBranch Point: {context.task_branch_point.commit_hash[:8]}")
print(f"Current Main: {context.current_main_commit[:8]}")
print(f"Commits Behind: {context.total_commits_behind}")
print(f"Other Pending Tasks: {context.total_pending_tasks}")
if context.other_pending_tasks:
print("\n--- Other Pending Tasks ---")
for task in context.other_pending_tasks:
print(f" {task['task_id']}: {task['intent'][:50]}...")
print(f"\n--- Main Evolution ({len(context.main_evolution)} events) ---")
for event in context.main_evolution:
print(f" {event.commit_hash[:8]} ({event.source}): {event.commit_message[:50]}...")
def cmd_list_files(args):
"""List all tracked files."""
tracker = get_tracker()
print("\n=== Tracked Files ===\n")
# Access internal _timelines
if not tracker._timelines:
print("No files currently tracked.")
return
for file_path in sorted(tracker._timelines.keys()):
timeline = tracker._timelines[file_path]
active_tasks = len([tv for tv in timeline.task_views.values() if tv.status == 'active'])
main_events = len(timeline.main_branch_history)
print(f" {file_path}: {active_tasks} active tasks, {main_events} main events")
def cmd_init_from_worktree(args):
"""Initialize tracking from an existing worktree."""
tracker = get_tracker()
task_id = args.task_id
worktree_path = Path(args.worktree_path).resolve()
if not worktree_path.exists():
print(f"Worktree path does not exist: {worktree_path}")
sys.exit(1)
print(f"Initializing tracking for {task_id} from {worktree_path}")
tracker.initialize_from_worktree(
task_id=task_id,
worktree_path=worktree_path,
task_intent=args.intent or "",
task_title=args.title or task_id,
)
print("Done.")
def main():
parser = argparse.ArgumentParser(
description="FileTimelineTracker CLI",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
subparsers = parser.add_subparsers(dest="command", help="Available commands")
# notify-commit
notify_parser = subparsers.add_parser(
"notify-commit",
help="Notify tracker of a new commit (called by git post-commit hook)"
)
notify_parser.add_argument("commit_hash", help="The commit hash")
notify_parser.set_defaults(func=cmd_notify_commit)
# show-timeline
timeline_parser = subparsers.add_parser(
"show-timeline",
help="Show the timeline for a file"
)
timeline_parser.add_argument("file_path", help="The file path (relative to project)")
timeline_parser.set_defaults(func=cmd_show_timeline)
# show-drift
drift_parser = subparsers.add_parser(
"show-drift",
help="Show commits-behind-main for a task"
)
drift_parser.add_argument("task_id", help="The task ID")
drift_parser.set_defaults(func=cmd_show_drift)
# show-context
context_parser = subparsers.add_parser(
"show-context",
help="Show merge context for a task and file"
)
context_parser.add_argument("task_id", help="The task ID")
context_parser.add_argument("file_path", help="The file path")
context_parser.set_defaults(func=cmd_show_context)
# list-files
list_parser = subparsers.add_parser(
"list-files",
help="List all tracked files"
)
list_parser.set_defaults(func=cmd_list_files)
# init-from-worktree
init_parser = subparsers.add_parser(
"init-from-worktree",
help="Initialize tracking from an existing worktree"
)
init_parser.add_argument("task_id", help="The task ID")
init_parser.add_argument("worktree_path", help="Path to the worktree")
init_parser.add_argument("--intent", help="Task intent description")
init_parser.add_argument("--title", help="Task title")
init_parser.set_defaults(func=cmd_init_from_worktree)
args = parser.parse_args()
if not args.command:
parser.print_help()
sys.exit(1)
args.func(args)
if __name__ == "__main__":
main()
auto-claude/workspace.py
+222 -77
View File
@@ -64,8 +64,12 @@ from merge import (
MergeDecision,
ConflictSeverity,
FileEvolutionTracker,
FileTimelineTracker,
)
# Track if we've already tried to install the git hook this session
_git_hook_check_done = False
MODULE = "workspace"
@@ -291,6 +295,9 @@ def setup_workspace(
print()
print_status("Setting up separate workspace...", "progress")
# Ensure timeline tracking hook is installed (once per session)
_ensure_timeline_hook_installed(project_dir)
manager = WorktreeManager(project_dir)
manager.setup()
@@ -308,9 +315,132 @@ def setup_workspace(
print_status(f"Workspace ready: {worktree_info.path.name}", "success")
print()
# Initialize FileTimelineTracker for this task
_initialize_timeline_tracking(
project_dir=project_dir,
spec_name=spec_name,
worktree_path=worktree_info.path,
source_spec_dir=localized_spec_dir or source_spec_dir,
)
return worktree_info.path, manager, localized_spec_dir
def _ensure_timeline_hook_installed(project_dir: Path) -> None:
"""
Ensure the FileTimelineTracker git post-commit hook is installed.
This enables tracking human commits to main branch for drift detection.
Called once per session during first workspace setup.
"""
global _git_hook_check_done
if _git_hook_check_done:
return
_git_hook_check_done = True
try:
git_dir = project_dir / ".git"
if not git_dir.exists():
return # Not a git repo
# Handle worktrees (where .git is a file, not directory)
if git_dir.is_file():
content = git_dir.read_text().strip()
if content.startswith("gitdir:"):
git_dir = Path(content.split(":", 1)[1].strip())
else:
return
hook_path = git_dir / "hooks" / "post-commit"
# Check if hook already installed
if hook_path.exists():
if "FileTimelineTracker" in hook_path.read_text():
debug(MODULE, "FileTimelineTracker hook already installed")
return
# Auto-install the hook (silent, non-intrusive)
from merge.install_hook import install_hook
install_hook(project_dir)
debug(MODULE, "Auto-installed FileTimelineTracker git hook")
except Exception as e:
# Non-fatal - hook installation is optional
debug_warning(MODULE, f"Could not auto-install timeline hook: {e}")
def _initialize_timeline_tracking(
project_dir: Path,
spec_name: str,
worktree_path: Path,
source_spec_dir: Path | None = None,
) -> None:
"""
Initialize FileTimelineTracker for a new task.
This registers the task's branch point and the files it intends to modify,
enabling intent-aware merge conflict resolution later.
"""
try:
tracker = FileTimelineTracker(project_dir)
# Get task intent from implementation plan
task_intent = ""
task_title = spec_name
files_to_modify = []
if source_spec_dir:
plan_path = source_spec_dir / "implementation_plan.json"
if plan_path.exists():
import json
with open(plan_path) as f:
plan = json.load(f)
task_title = plan.get("title", spec_name)
task_intent = plan.get("description", "")
# Extract files from phases/subtasks
for phase in plan.get("phases", []):
for subtask in phase.get("subtasks", []):
files_to_modify.extend(subtask.get("files", []))
# Get the current branch point commit
import subprocess
result = subprocess.run(
["git", "rev-parse", "HEAD"],
cwd=project_dir,
capture_output=True,
text=True,
)
branch_point = result.stdout.strip() if result.returncode == 0 else None
if files_to_modify and branch_point:
# Register the task with known files
tracker.on_task_start(
task_id=spec_name,
files_to_modify=list(set(files_to_modify)), # Dedupe
branch_point_commit=branch_point,
task_intent=task_intent,
task_title=task_title,
)
debug(MODULE, f"Timeline tracking initialized for {spec_name}",
files_tracked=len(files_to_modify),
branch_point=branch_point[:8] if branch_point else None)
else:
# Initialize retroactively from worktree if no plan
tracker.initialize_from_worktree(
task_id=spec_name,
worktree_path=worktree_path,
task_intent=task_intent,
task_title=task_title,
)
except Exception as e:
# Non-fatal - timeline tracking is supplementary
debug_warning(MODULE, f"Could not initialize timeline tracking: {e}")
print(muted(f" Note: Timeline tracking could not be initialized: {e}"))
def show_build_summary(manager: WorktreeManager, spec_name: str) -> None:
"""Show a summary of what was built."""
summary = manager.get_change_summary(spec_name)
@@ -766,6 +896,14 @@ def _try_smart_merge_inner(
try:
print(muted(" Analyzing changes with intent-aware merge..."))
# Capture worktree state in FileTimelineTracker before merge
try:
timeline_tracker = FileTimelineTracker(project_dir)
timeline_tracker.capture_worktree_state(spec_name, worktree_path)
debug(MODULE, "Captured worktree state for timeline tracking")
except Exception as e:
debug_warning(MODULE, f"Could not capture worktree state: {e}")
# Initialize the orchestrator
debug(MODULE, "Initializing MergeOrchestrator",
project_dir=str(project_dir),
@@ -1286,9 +1424,10 @@ def _record_merge_completion(
spec_name: str,
merged_files: list[str],
task_intent: str = "",
merge_commit: str = "",
) -> None:
"""
Record completed merge in the Evolution Tracker.
Record completed merge in both Evolution Tracker and FileTimelineTracker.
This enables future AI merges to understand the history of file changes,
creating a knowledge chain for intelligent conflict resolution.
@@ -1298,16 +1437,18 @@ def _record_merge_completion(
spec_name: The task/spec that was merged
merged_files: List of file paths that were merged
task_intent: Description of what the task accomplished
merge_commit: The commit hash of the merge (for timeline tracking)
"""
# Get intent from implementation plan if not provided
if not task_intent:
intent_data = _get_task_intent(project_dir, spec_name)
if intent_data:
task_intent = intent_data.get("description", "") or intent_data.get("title", spec_name)
# Track in FileEvolutionTracker (legacy system)
try:
tracker = FileEvolutionTracker(project_dir)
# Get intent from implementation plan if not provided
if not task_intent:
intent_data = _get_task_intent(project_dir, spec_name)
if intent_data:
task_intent = intent_data.get("description", "") or intent_data.get("title", spec_name)
# Mark the task as completed for all its tracked files
tracker.mark_task_completed(spec_name)
@@ -1324,11 +1465,38 @@ def _record_merge_completion(
# Save updates
tracker._save_evolutions()
debug(MODULE, f"Recorded merge in FileEvolutionTracker",
spec_name=spec_name, files=len(merged_files))
except Exception as e:
debug_warning(MODULE, f"Could not record in FileEvolutionTracker: {e}")
# Track in FileTimelineTracker (new intent-aware system)
try:
timeline_tracker = FileTimelineTracker(project_dir)
# Get merge commit if not provided
if not merge_commit:
import subprocess
result = subprocess.run(
["git", "rev-parse", "HEAD"],
cwd=project_dir,
capture_output=True,
text=True,
)
merge_commit = result.stdout.strip() if result.returncode == 0 else "unknown"
# Mark task as merged in timeline tracker
timeline_tracker.on_task_merged(spec_name, merge_commit)
debug(MODULE, f"Recorded merge in FileTimelineTracker",
spec_name=spec_name, merge_commit=merge_commit[:8])
print(muted(f" Recorded merge completion for {len(merged_files)} files"))
except Exception as e:
# Non-fatal - this is supplementary tracking
print(muted(f" Could not record merge completion: {e}"))
debug_warning(MODULE, f"Could not record in FileTimelineTracker: {e}")
print(muted(f" Note: Could not record merge completion: {e}"))
def _get_task_intent(project_dir: Path, spec_name: str) -> Optional[dict]:
@@ -1553,6 +1721,7 @@ def _merge_file_with_ai(
Use AI to merge a conflicting file.
This enhanced version includes:
- FileTimelineTracker context for full situational awareness
- Task intent from implementation_plan.json
- Binary file detection
- File size limits
@@ -1561,6 +1730,7 @@ def _merge_file_with_ai(
Returns merged content, or None if AI couldn't resolve.
"""
from merge import create_claude_resolver
from merge.prompts import build_timeline_merge_prompt, build_simple_merge_prompt
debug(MODULE, f"AI merge starting for: {file_path}",
spec_name=spec_name,
@@ -1599,85 +1769,60 @@ def _merge_file_with_ai(
print(muted(f" AI not available, trying heuristic merge..."))
return _heuristic_merge(main_content, worktree_content, base_content)
# Try to get timeline context for richer merge prompt
timeline_context = None
if project_dir:
try:
tracker = FileTimelineTracker(project_dir)
timeline_context = tracker.get_merge_context(spec_name, file_path)
if timeline_context:
debug(MODULE, "Using FileTimelineTracker context",
commits_behind=timeline_context.total_commits_behind,
pending_tasks=timeline_context.total_pending_tasks,
main_events=len(timeline_context.main_evolution))
except Exception as e:
debug_warning(MODULE, f"Could not get timeline context: {e}")
# Determine language
language = resolver._infer_language(file_path)
debug(MODULE, "Detected language", language=language)
# Quick Win 1: Get task intent for richer context
task_intent = None
if project_dir:
task_intent = _get_task_intent(project_dir, spec_name)
if task_intent:
debug(MODULE, "Loaded task intent",
title=task_intent.get('title'),
num_subtasks=len(task_intent.get('subtasks', [])))
# Build prompt - use timeline context if available, fallback to simple prompt
if timeline_context and timeline_context.total_commits_behind > 0:
# Use the rich timeline-based prompt with full situational awareness
debug(MODULE, "Building timeline-based merge prompt",
commits_behind=timeline_context.total_commits_behind,
main_events=len(timeline_context.main_evolution),
pending_tasks=timeline_context.total_pending_tasks)
print(muted(f" Using timeline context ({timeline_context.total_commits_behind} commits behind, {timeline_context.total_pending_tasks} pending tasks)"))
prompt = build_timeline_merge_prompt(timeline_context)
else:
# Fallback to simple three-way merge prompt
debug(MODULE, "Building simple merge prompt (no timeline context)")
# Build intent context string
intent_context = ""
if task_intent:
intent_context = f"""
=== FEATURE BRANCH INTENT ({spec_name}) ===
Task: {task_intent.get('title', spec_name)}
Description: {task_intent.get('description', 'No description')}
"""
if task_intent.get('spec_summary'):
intent_context += f"Summary: {task_intent['spec_summary']}\n"
# Get task intent for basic context
task_intent = None
if project_dir:
task_intent = _get_task_intent(project_dir, spec_name)
if task_intent:
debug(MODULE, "Loaded task intent",
title=task_intent.get('title'),
num_subtasks=len(task_intent.get('subtasks', [])))
# Add relevant subtasks
relevant_subtasks = [
st for st in task_intent.get('subtasks', [])
if st.get('status') in ('completed', 'in_progress')
]
if relevant_subtasks:
intent_context += "Completed work:\n"
for st in relevant_subtasks[:5]: # Limit to 5
intent_context += f" - {st.get('title', 'Unknown')}\n"
# Get context about recent merges to this file (for v2)
recent_merges_context = ""
if project_dir:
recent_merges = _get_recent_merges_context(project_dir, file_path)
if recent_merges:
recent_merges_context = "\n=== RECENT CHANGES TO THIS FILE ===\n"
for merge in recent_merges:
recent_merges_context += f"- Task '{merge['task_id']}': {merge['intent']}\n"
# Build a focused prompt for three-way merge with intent
prompt = f'''You are a code merge expert. Merge the following conflicting versions of a file.
FILE: {file_path}
The file was modified in both the main branch and in the "{spec_name}" feature branch.
Your task is to produce a merged version that incorporates ALL changes from both branches.
{intent_context}{recent_merges_context}
=== COMMON ANCESTOR (base) ===
{base_content if base_content else "(File did not exist in common ancestor)"}
=== MAIN BRANCH VERSION ===
{main_content}
=== FEATURE BRANCH VERSION ({spec_name}) ===
{worktree_content}
MERGE RULES:
1. Keep ALL imports from both versions
2. Keep ALL new functions/components from both versions
3. If the same function was modified differently, combine the changes logically
4. Preserve the intent of BOTH branches - main's changes are important too
5. If there's a genuine semantic conflict (same thing done differently), prefer the feature branch version but include main's additions
6. The merged code MUST be syntactically valid {language}
Output ONLY the merged code, wrapped in triple backticks:
```{language}
merged code here
```
'''
prompt = build_simple_merge_prompt(
file_path=file_path,
main_content=main_content,
worktree_content=worktree_content,
base_content=base_content,
spec_name=spec_name,
language=language,
task_intent=task_intent,
)
try:
debug(MODULE, "Calling AI for merge",
file_path=file_path,
has_intent_context=bool(intent_context),
has_recent_merges=bool(recent_merges_context))
has_timeline_context=timeline_context is not None)
response = resolver.ai_call_fn(
"You are an expert code merge assistant. Output only the merged code. The code MUST be syntactically valid.",