第五章：Git Worktree 隔离与并行智能体协作

# 主仓库，在 main 分支
/project/
  src/
  tests/

# 添加 worktree，检出到独立目录
git worktree add .auto-claude/worktrees/tasks/feature-auth auto-claude/feature-auth

# 现在你有两个工作目录：
/project/                              # main 分支
.auto-claude/worktrees/tasks/feature-auth/   # auto-claude/feature-auth 分支

class WorktreeManager:
    """
    管理每个 spec 的 Git worktree。

    约定：
    - 路径：.auto-claude/worktrees/tasks/{spec-name}/
    - 分支：auto-claude/{spec-name}
    """

    GIT_PUSH_TIMEOUT = 120   # 2 分钟（网络操作）
    CLI_TIMEOUT = 60          # 1 分钟（gh/glab 命令）
    CLI_QUERY_TIMEOUT = 30    # 30 秒（查询操作）

    def __init__(
        self,
        project_dir: Path,
        base_branch: str | None = None,
        use_local_branch: bool = False,
    ):
        self.project_dir = project_dir
        self.base_branch = base_branch or self._detect_base_branch()
        self.worktrees_dir = project_dir / ".auto-claude" / "worktrees" / "tasks"
        self._merge_lock = asyncio.Lock()  # 合并操作需要互斥

def _detect_base_branch(self) -> str:
    """
    按优先级检测基础分支。

    Priority:
    1. DEFAULT_BRANCH 环境变量（明确配置优先）
    2. 自动检测 main/master（常规约定）
    3. 回退到当前分支（带警告）
    """
    # 1. 环境变量
    env_branch = os.getenv("DEFAULT_BRANCH")
    if env_branch and self._branch_exists(env_branch):
        return env_branch

    # 2. 约定检测
    for candidate in ["main", "master"]:
        if self._branch_exists(candidate):
            return candidate

    # 3. 回退（可能不是预期行为，记录警告）
    current = self._get_current_branch()
    debug_warning(f"Could not detect main branch, using current: {current}")
    return current

async def create_worktree(self, spec_name: str) -> WorktreeInfo:
    """
    为 spec 创建独立的 git worktree。

    步骤：
    1. 确定路径和分支名
    2. 从基础分支创建新分支
    3. 在指定路径添加 worktree
    4. 返回 WorktreeInfo
    """
    # 规范化名称（去除特殊字符）
    safe_name = re.sub(r"[^a-zA-Z0-9_-]", "-", spec_name)
    worktree_path = self.worktrees_dir / safe_name
    branch_name = f"auto-claude/{safe_name}"

    # 如果 worktree 已存在，直接返回
    if worktree_path.exists():
        return self._get_worktree_info(worktree_path, branch_name, spec_name)

    # 确保目录结构存在
    self.worktrees_dir.mkdir(parents=True, exist_ok=True)

    # 创建分支并添加 worktree
    success, error = run_git(
        ["worktree", "add", "-b", branch_name, str(worktree_path), self.base_branch],
        cwd=self.project_dir,
        timeout=30,
    )

    if not success:
        raise WorktreeError(f"Failed to create worktree: {error}")

    logger.info(f"Created worktree: {worktree_path} (branch: {branch_name})")
    return self._get_worktree_info(worktree_path, branch_name, spec_name)

def _with_retry(
    operation: Callable[[], tuple[bool, T | None, str]],
    max_retries: int = 3,
    is_retryable: Callable[[str], bool] | None = None,
    on_retry: Callable[[int, str], None] | None = None,
) -> tuple[T | None, str]:
    """
    带重试的操作执行框架。

    operation 返回 (success, result, error) 三元组：
    - success=True：返回 result
    - success=False：检查是否可重试
    """
    last_error = ""

    for attempt in range(1, max_retries + 1):
        try:
            success, result, error = operation()
            if success:
                return result, ""

            last_error = error

            # 只重试可重试的错误
            if is_retryable and attempt < max_retries and is_retryable(error):
                if on_retry:
                    on_retry(attempt, error)
                backoff = 2 ** (attempt - 1)  # 1s, 2s, 4s...
                time.sleep(backoff)
                continue

            break  # 不可重试，立即放弃

        except subprocess.TimeoutExpired:
            last_error = "Operation timed out"
            if attempt < max_retries:
                backoff = 2 ** (attempt - 1)
                time.sleep(backoff)
                continue
            break

    return None, last_error

# 使用示例
result, error = _with_retry(
    operation=lambda: push_branch(branch_name),
    max_retries=3,
    is_retryable=_is_retryable_network_error,
    on_retry=lambda attempt, err: logger.warning(
        f"Push attempt {attempt} failed: {err}, retrying..."
    ),
)

# merge/types.py（概念化）

class ChangeType(Enum):
    # 结构变更
    FUNCTION_ADDED = "function_added"
    FUNCTION_MODIFIED = "function_modified"
    FUNCTION_DELETED = "function_deleted"
    CLASS_ADDED = "class_added"
    CLASS_MODIFIED = "class_modified"

    # 接口变更
    PARAMETER_ADDED = "parameter_added"
    PARAMETER_REMOVED = "parameter_removed"
    RETURN_TYPE_CHANGED = "return_type_changed"

    # 内容变更
    IMPORT_ADDED = "import_added"
    IMPORT_REMOVED = "import_removed"
    CONSTANT_CHANGED = "constant_changed"
    BODY_MODIFIED = "body_modified"

class MergeDecision(Enum):
    AUTO_MERGED = "auto_merged"           # 确定性规则可以解决
    AI_MERGED = "ai_merged"               # AI 辅助解决了歧义
    DIRECT_COPY = "direct_copy"           # 只有一个任务修改了此文件
    NEEDS_HUMAN_REVIEW = "needs_human_review"  # 需要人工介入
    FAILED = "failed"                     # 合并失败

class MergeOrchestrator:
    """
    合并流水线的统一协调者。

    最大化自动化，最小化 AI Token 使用。
    """

    def merge_task(
        self,
        task_id: str,
        worktree_path: Path | None = None,
        target_branch: str = "main",
        progress_callback: MergeProgressCallback | None = None,
    ) -> MergeReport:

# 阶段进度报告（前端实时更新用）
# 0%   → ANALYZING:          开始语义分析
# 25%  → DETECTING_CONFLICTS: 冲突检测
# 50%  → RESOLVING:           逐文件合并
# 75%  → VALIDATING:          结果验证
# 100% → COMPLETE:            完成

def merge_task(self, task_id, worktree_path=None, ...):
    report = MergeReport(started_at=datetime.now())

    try:
        # 1. 加载文件演化数据
        self.evolution_tracker.refresh_from_git(task_id, worktree_path)

        # 2. 获取此任务修改的文件列表
        modifications = self.evolution_tracker.get_task_modifications(task_id)

        # 3. 逐文件合并
        for file_path, snapshot in modifications:
            result = self._merge_file(file_path, [snapshot], target_branch)

            # 特殊处理：DIRECT_COPY 需要从 worktree 读取实际文件
            if result.decision == MergeDecision.DIRECT_COPY:
                content, success = self._read_worktree_file(file_path, worktree_path)
                if success:
                    result.merged_content = content
                else:
                    result.decision = MergeDecision.FAILED

            report.file_results[file_path] = result
            self._update_stats(report.stats, result)

    except Exception as e:
        report.success = False
        report.error = str(e)

    report.completed_at = datetime.now()
    return report

def merge_tasks(
    self,
    requests: list[TaskMergeRequest],
    target_branch: str = "main",
) -> MergeReport:
    # 1. 按优先级排序（优先级高的任务在冲突时优先）
    requests = sorted(requests, key=lambda r: -r.priority)

    # 2. 为所有任务刷新演化数据
    for request in requests:
        self.evolution_tracker.refresh_from_git(
            request.task_id, request.worktree_path
        )

    # 3. 找出被任意任务修改的所有文件
    task_ids = [r.task_id for r in requests]
    file_tasks = self.evolution_tracker.get_files_modified_by_tasks(task_ids)
    # file_tasks: {"src/auth.py": ["task-001", "task-002"], "src/db.py": ["task-002"], ...}

    # 4. 逐文件合并（考虑多任务冲突）
    for file_path, modifying_tasks in file_tasks.items():
        # 获取所有修改此文件的任务的语义快照
        snapshots = [
            evolution.get_task_snapshot(tid)
            for tid in modifying_tasks
        ]
        result = self._merge_file(file_path, snapshots, target_branch)
        report.file_results[file_path] = result

@dataclass
class MergeStats:
    files_processed: int = 0
    files_auto_merged: int = 0        # 规则自动解决
    files_ai_merged: int = 0          # AI 辅助解决
    files_need_review: int = 0        # 需要人工
    files_failed: int = 0             # 失败

    conflicts_detected: int = 0
    conflicts_auto_resolved: int = 0
    conflicts_ai_resolved: int = 0

    ai_calls_made: int = 0            # AI 调用次数（成本指标）
    estimated_tokens_used: int = 0    # Token 用量（成本指标）
    duration_seconds: float = 0.0

    @property
    def auto_merge_rate(self) -> float:
        """无 AI 自动解决率——越高越省钱"""
        if self.conflicts_detected == 0:
            return 1.0
        return self.conflicts_auto_resolved / self.conflicts_detected

@dataclass
class WorktreeInfo:
    path: Path
    branch: str
    spec_name: str
    base_branch: str
    is_active: bool = True
    commit_count: int = 0         # 此 worktree 的提交数
    files_changed: int = 0        # 修改的文件数
    additions: int = 0            # 新增行数
    deletions: int = 0            # 删除行数
    last_commit_date: datetime | None = None
    days_since_last_commit: int | None = None

# lab_05_worktree.py

import os
import subprocess
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

@dataclass
class TaskWorkspace:
    """代表一个写作任务的独立工作空间。"""
    task_id: str
    path: Path
    branch: str
    base_branch: str
    files_created: list[str] = field(default_factory=list)
    files_modified: list[str] = field(default_factory=list)

class WorkspaceManager:
    """管理写作任务的独立工作空间。"""

    def __init__(self, project_dir: Path):
        self.project_dir = project_dir
        self.workspaces_dir = project_dir / ".workspaces"
        self.workspaces_dir.mkdir(exist_ok=True)

    def create_workspace(self, task_id: str) -> TaskWorkspace:
        """
        TODO Part A: 为任务创建独立工作空间

        要求：
        1. 在 self.workspaces_dir / task_id 创建目录
        2. 将 project_dir 的内容"快照"到工作空间（简化版：复制关键文件）
        3. 记录基础状态（用于后续对比）
        4. 返回 TaskWorkspace

        提示：在真实系统中这是 git worktree add，
        在本实验中用目录复制模拟隔离效果。
        """
        workspace_path = self.workspaces_dir / task_id
        workspace_path.mkdir(exist_ok=True)

        # TODO: 复制项目文件到工作空间
        # TODO: 记录基础状态快照
        # TODO: 返回 TaskWorkspace
        pass

    def get_modifications(self, task_id: str) -> dict[str, str]:
        """
        TODO Part A: 检测工作空间相对于基础状态的修改

        返回 {文件路径: "created" | "modified" | "deleted"}

        提示：对比工作空间文件与基础状态快照
        """
        pass

    def list_workspaces(self) -> list[TaskWorkspace]:
        """返回所有活跃的工作空间。"""
        workspaces = []
        for task_dir in self.workspaces_dir.iterdir():
            if task_dir.is_dir() and not task_dir.name.startswith("_"):
                # TODO: 读取并返回工作空间信息
                pass
        return workspaces

from enum import Enum

class MergeResult(Enum):
    AUTO_MERGED = "auto_merged"
    CONFLICT = "conflict"
    DIRECT_COPY = "direct_copy"

@dataclass
class FileMergeResult:
    file_path: str
    decision: MergeResult
    content: Optional[str] = None
    conflict_info: Optional[str] = None

class SimpleMerger:
    """
    简化的语义合并器。

    合并规则：
    - 只有一个任务修改 → DIRECT_COPY
    - 多个任务修改，无重叠行 → AUTO_MERGED（将所有修改合并）
    - 多个任务修改同一行 → CONFLICT（需要人工）
    """

    def merge_file(
        self,
        file_path: str,
        baseline_content: str,
        task_modifications: dict[str, str],  # {task_id: modified_content}
    ) -> FileMergeResult:
        """
        TODO Part B: 实现文件合并逻辑

        参数：
        - baseline_content: 合并前的原始文件内容
        - task_modifications: 每个任务修改后的文件内容

        要求：
        1. 如果只有一个任务修改 → DIRECT_COPY
        2. 如果多个任务修改：
           a. 计算每个任务相对于 baseline 的行级差异
           b. 检查是否有行级冲突（不同任务修改了同一行）
           c. 无冲突 → 合并所有差异 → AUTO_MERGED
           d. 有冲突 → 返回 CONFLICT + 冲突说明

        提示：使用 difflib 库进行差异计算
        """
        if len(task_modifications) == 0:
            return FileMergeResult(
                file_path=file_path,
                decision=MergeResult.DIRECT_COPY,
                content=baseline_content,
            )

        if len(task_modifications) == 1:
            task_id, content = next(iter(task_modifications.items()))
            return FileMergeResult(
                file_path=file_path,
                decision=MergeResult.DIRECT_COPY,
                content=content,
            )

        # TODO: 实现多任务合并逻辑
        pass

    def _compute_line_changes(
        self,
        baseline: str,
        modified: str,
    ) -> dict[int, str]:
        """
        TODO Part B (辅助函数): 计算修改的行号映射

        返回 {行号: 新内容}，表示相对于 baseline 改变了哪些行。

        提示：使用 difflib.ndiff() 或 unified_diff()
        """
        pass

def test_parallel_writing():
    """
    完整测试：模拟两个写作智能体并行工作，然后合并。
    """
    import tempfile

    with tempfile.TemporaryDirectory() as project_dir:
        project_dir = Path(project_dir)

        # 创建初始文档
        doc_path = project_dir / "document.md"
        doc_path.write_text("""# 文档标题

## 第一节
这是第一节的内容。

## 第二节
这是第二节的内容。
""")

        manager = WorkspaceManager(project_dir)
        merger = SimpleMerger()

        # 模拟两个写作任务
        ws1 = manager.create_workspace("task-writer-a")
        ws2 = manager.create_workspace("task-writer-b")

        # Writer A 修改第一节
        (ws1.path / "document.md").write_text("""# 文档标题

## 第一节
这是第一节的内容，Writer A 添加了更多细节。
Writer A 新增的额外说明。

## 第二节
这是第二节的内容。
""")

        # Writer B 修改第二节
        (ws2.path / "document.md").write_text("""# 文档标题

## 第一节
这是第一节的内容。

## 第二节
这是第二节的内容，Writer B 进行了扩充。
Writer B 添加了实例和说明。
""")

        # 合并
        baseline = doc_path.read_text()
        modifications = {
            "task-writer-a": (ws1.path / "document.md").read_text(),
            "task-writer-b": (ws2.path / "document.md").read_text(),
        }

        result = merger.merge_file("document.md", baseline, modifications)

        print(f"合并决策: {result.decision.value}")
        print(f"合并结果:\n{result.content}")

        # 验证
        assert result.decision == MergeResult.AUTO_MERGED, \
            f"Expected AUTO_MERGED, got {result.decision}"
        assert "Writer A 新增的额外说明" in result.content, \
            "A 的修改应该在合并结果中"
        assert "Writer B 添加了实例和说明" in result.content, \
            "B 的修改应该在合并结果中"

        print("✓ 并行写作合并测试通过！")


def test_conflict_detection():
    """测试：两个任务修改同一行时应报告冲突。"""
    merger = SimpleMerger()

    baseline = "第一行\n第二行\n第三行\n"
    modifications = {
        "task-a": "第一行\n任务A修改了第二行\n第三行\n",
        "task-b": "第一行\n任务B也修改了第二行\n第三行\n",
    }

    result = merger.merge_file("test.txt", baseline, modifications)
    assert result.decision == MergeResult.CONFLICT, \
        "同行修改应报告冲突"
    assert result.conflict_info is not None, \
        "应提供冲突说明"

    print(f"✓ 冲突检测正确: {result.conflict_info}")


if __name__ == "__main__":
    test_parallel_writing()
    test_conflict_detection()
    print("\n✓ All tests passed!")