第四章：Claude Agent SDK 深度解剖

# apps/backend/core/client.py

def create_client(
    phase: str,
    task_id: Optional[str] = None,
    worktree_path: Optional[str] = None,
    project_path: Optional[str] = None,
    subtask_index: Optional[int] = None,
    is_parallel: bool = False,
    spec_name: Optional[str] = None,
    spec_path: Optional[str] = None,
) -> ClaudeSDKClient:
    """
    工厂函数：构建一个完全配置的 SDK 客户端。
    所有 AI 交互必须经过此函数，绝不直接实例化 SDK。
    """

# apps/backend/core/client.py（简化）

_project_cache: dict[str, Any] = {}
_cache_lock = threading.Lock()

def _get_cached_project_data(project_path: str) -> dict[str, Any]:
    """
    双重检查锁定：在不牺牲并发性能的前提下保证线程安全。
    """
    # 第一次检查：无锁快速路径
    if project_path in _project_cache:
        return _project_cache[project_path]

    # 获取锁
    with _cache_lock:
        # 第二次检查：防止另一个线程在我们等待锁时已经写入
        if project_path in _project_cache:
            return _project_cache[project_path]

        # 只有我们是第一个到达的线程，才执行昂贵的 I/O
        data = _load_project_data(project_path)
        _project_cache[project_path] = data
        return data

# 路径解析逻辑（简化）

def _resolve_working_directory(
    worktree_path: Optional[str],
    project_path: Optional[str],
) -> str:
    """
    优先级：worktree_path > project_path > 当前目录
    """
    if worktree_path and os.path.exists(worktree_path):
        return worktree_path
    if project_path and os.path.exists(project_path):
        return project_path
    return os.getcwd()

# apps/backend/core/phase_config.py（概念化展示）

PHASE_CONFIGS = {
    "spec_critic": {
        "model": "claude-opus-4-6",       # 最强推理
        "thinking_budget": 31999,          # 最大思考空间
        "extended_thinking": True,
    },
    "planner": {
        "model": "claude-sonnet-4-6",     # 平衡性能
        "thinking_budget": 10000,
        "extended_thinking": True,
    },
    "coder": {
        "model": "claude-sonnet-4-6",     # 最佳编码模型
        "thinking_budget": 5000,
        "extended_thinking": False,        # 编码不需要深度推理
    },
    "qa_reviewer": {
        "model": "claude-sonnet-4-6",
        "thinking_budget": 8000,
        "extended_thinking": True,         # 评审需要仔细推理
    },
}

# 不同阶段的工具许可（简化）

PHASE_TOOL_ALLOWLISTS = {
    "planner": [
        "Read", "Write", "Glob", "Grep",
        "WebSearch", "WebFetch",
        # Planner 不需要 Bash——它只规划，不执行
    ],
    "coder": [
        "Read", "Write", "Edit", "Glob", "Grep",
        "Bash",           # Coder 需要运行命令
        "WebSearch",
        "TodoWrite",      # 进度跟踪
    ],
    "qa_reviewer": [
        "Read", "Glob", "Grep",
        "Bash",           # 需要运行测试
        "WebFetch",       # 可能需要查阅文档
        # QA 不能写入文件——只读评估
    ],
    "qa_fixer": [
        "Read", "Write", "Edit", "Glob", "Grep",
        "Bash",
    ],
}

def _build_filesystem_permissions(
    worktree_path: str,
    project_path: str,
    spec_path: Optional[str],
) -> list[dict]:
    """
    构建文件系统访问许可列表。
    """
    allowed_paths = [
        worktree_path,           # 主工作目录
        "/tmp",                  # 临时文件
    ]

    # spec 目录：智能体可以读取需求规格
    if spec_path and os.path.exists(spec_path):
        allowed_paths.append(spec_path)

    # 项目根目录只读：智能体可以查阅项目配置，但不能修改
    if project_path != worktree_path:
        allowed_paths.append(f"{project_path}:read-only")

    return [{"type": "filesystem", "paths": allowed_paths}]

def _expand_worktree_permissions(
    base_permissions: list,
    is_parallel: bool,
    project_path: str,
) -> list:
    """
    并行模式下，允许读取其他 worktree（只读）。
    """
    if not is_parallel:
        return base_permissions

    # 查找所有活跃的 worktree
    worktrees = _list_active_worktrees(project_path)
    for wt in worktrees:
        # 以只读方式添加其他 worktree
        base_permissions.append({
            "path": wt.path,
            "access": "read-only"
        })

    return base_permissions

# MCP 服务器配置（简化）

def _build_mcp_servers(
    phase: str,
    project_path: str,
    settings: dict,
) -> list[MCPServerConfig]:
    """
    根据阶段和项目配置，注册合适的 MCP 服务器。
    """
    servers = []

    # Context7：文档和库参考（所有阶段都可用）
    servers.append(MCPServerConfig(
        name="context7",
        command="npx",
        args=["-y", "@upstash/context7-mcp"],
    ))

    # Graphiti：记忆知识图谱（仅在启用时）
    if settings.get("graphiti_enabled"):
        servers.append(MCPServerConfig(
            name="graphiti",
            command="python",
            args=["-m", "graphiti_mcp"],
            env={"NEO4J_URI": settings["neo4j_uri"]},
        ))

    # Electron MCP：GUI 测试（仅 QA 阶段）
    if phase in ("qa_reviewer", "qa_fixer"):
        if settings.get("electron_mcp_enabled"):
            servers.append(MCPServerConfig(
                name="electron",
                command="node",
                args=["electron-mcp-server.js"],
            ))

    # Puppeteer：Web 测试（仅 QA 阶段）
    if phase in ("qa_reviewer", "qa_fixer"):
        servers.append(MCPServerConfig(
            name="puppeteer",
            command="npx",
            args=["-y", "@anthropic-ai/mcp-server-puppeteer"],
        ))

    # Linear：项目管理集成（仅在配置时）
    if settings.get("linear_enabled"):
        servers.append(MCPServerConfig(
            name="linear",
            command="npx",
            args=["-y", "linear-mcp-server"],
            env={"LINEAR_API_KEY": settings["linear_api_key"]},
        ))

    return servers

# apps/backend/core/client.py

def _patch_sdk_message_parser():
    """
    补丁：处理 SDK 未知消息类型，避免解析崩溃。

    Claude 有时会返回实验性的消息类型，旧版 SDK 会抛出
    ValueError: Unknown message type。这个补丁让解析器
    优雅降级，而不是崩溃。
    """
    from claude_agent_sdk import _internal

    original_parse = _internal.MessageParser.parse

    def patched_parse(self, data: dict) -> Any:
        try:
            return original_parse(self, data)
        except ValueError as e:
            if "Unknown message type" in str(e):
                # 返回通用消息对象，而不是崩溃
                logger.warning(f"SDK parse warning: {e}. Skipping unknown message.")
                return None
            raise  # 其他 ValueError 正常抛出

    _internal.MessageParser.parse = patched_parse

# apps/backend/core/client.py

WINDOWS_CMD_MAX_LENGTH = 32768  # Windows CreateProcess 限制

def _sanitize_for_windows(options: ClaudeAgentOptions) -> ClaudeAgentOptions:
    """
    Windows 专用：确保命令行参数不超过系统限制。

    MCP 服务器配置、系统提示、工具描述都通过命令行传递。
    超过 32,768 字符会导致进程启动失败。
    """
    if not sys.platform.startswith("win"):
        return options  # 非 Windows 不需要处理

    # 计算总命令行长度
    total_length = _estimate_cmdline_length(options)

    if total_length > WINDOWS_CMD_MAX_LENGTH:
        logger.warning(
            f"Command line length {total_length} exceeds Windows limit "
            f"{WINDOWS_CMD_MAX_LENGTH}. Truncating system prompt."
        )
        # 策略：优先保留工具配置，截断系统提示末尾
        options = _truncate_system_prompt(options, WINDOWS_CMD_MAX_LENGTH - 1000)

    return options

# apps/backend/core/auth.py（概念化）

class AuthResolver:
    """
    统一认证接口：调用者不需要知道是 OAuth 还是 API Key。
    """

    def resolve(self, profile_name: Optional[str] = None) -> AuthCredentials:
        if profile_name:
            return self._resolve_profile(profile_name)
        else:
            return self._resolve_oauth()

    def _resolve_oauth(self) -> AuthCredentials:
        """从 OS 密钥链读取 OAuth token，必要时自动刷新。"""
        token = keychain.get("claude-code-oauth-token")
        if self._is_expired(token):
            token = self._refresh_token(token)
        return AuthCredentials(type="oauth", token=token)

    def _resolve_profile(self, name: str) -> AuthCredentials:
        """从加密存储读取 API profile。"""
        profile = encrypted_store.get(f"profile:{name}")
        return AuthCredentials(
            type="api_key",
            api_key=profile.api_key,
            base_url=profile.endpoint,  # 支持 z.ai 等兼容端点
        )

# 多账户管理（简化）

class ProfileScorer:
    """根据使用量和可用性为账户评分，选择最佳账户。"""

    def select_best_profile(self) -> ProfileCredentials:
        profiles = self._load_all_profiles()
        scored = [(self._score(p), p) for p in profiles]
        scored.sort(reverse=True)  # 分数高 = 使用量低 = 优先选择

        best_score, best_profile = scored[0]
        if best_score < MINIMUM_VIABLE_SCORE:
            raise RateLimitError("All profiles exhausted. Please wait.")

        return best_profile

    def _score(self, profile: ProfileCredentials) -> float:
        usage_penalty = profile.recent_usage_tokens / profile.daily_limit
        availability_bonus = 1.0 if not profile.is_rate_limited else 0.0
        return availability_bonus - usage_penalty

# 正确用法
async def run_coder_agent(task: Task, worktree_path: str):
    client = create_client(
        phase="coder",
        task_id=task.id,
        worktree_path=worktree_path,
    )

    async with client as session:
        result = await session.run(
            prompt=build_coder_prompt(task),
            system=load_prompt("coder.md"),
        )

    # 退出 async with 时，SDK 自动：
    # - 关闭 MCP 服务器进程
    # - 释放网络连接
    # - 清理临时资源
    return result

# apps/backend/core/phase_event.py

import json
import sys

class ExecutionPhase(Enum):
    PLANNING = "planning"
    CODING = "coding"
    QA_REVIEW = "qa_review"
    QA_FIXING = "qa_fixing"
    COMPLETE = "complete"
    FAILED = "failed"
    RATE_LIMIT_PAUSED = "rate_limit_paused"
    AUTH_FAILURE_PAUSED = "auth_failure_paused"

def emit_phase(
    phase: ExecutionPhase,
    task_id: str,
    metadata: Optional[dict] = None,
):
    """
    向 stdout 写入结构化阶段事件。
    Electron 主进程监听此输出并更新 UI。
    """
    event = {
        "type": "__EXEC_PHASE__",
        "phase": phase.value,
        "task_id": task_id,
        "timestamp": time.time(),
        "metadata": metadata or {},
    }
    # 使用特殊前缀确保可解析
    print(f"__EXEC_PHASE__:{json.dumps(event)}", flush=True)

// apps/frontend/src/main/agent/agent-process.ts（概念化）

const agentProcess = spawn("python", ["run.py", "--spec", specId]);

agentProcess.stdout.on("data", (data: Buffer) => {
    const lines = data.toString().split("\n");
    for (const line of lines) {
        if (line.startsWith("__EXEC_PHASE__:")) {
            const event = JSON.parse(line.slice("__EXEC_PHASE__:".length));
            handlePhaseEvent(event);  // 更新 Kanban 状态
        } else {
            // 普通日志输出显示在终端
            terminalStore.appendOutput(line);
        }
    }
});

# 错误分类体系

class AgentError(Exception):
    pass

class RateLimitError(AgentError):
    """触发：暂停 + 切换账户 + 重试"""
    recovery = "switch_account_and_retry"

class AuthFailureError(AgentError):
    """触发：暂停 + 通知用户 + 等待干预"""
    recovery = "pause_and_notify"

class ContextWindowError(AgentError):
    """触发：压缩上下文 + 重试"""
    recovery = "compact_context_and_retry"

class ToolError(AgentError):
    """触发：记录错误 + 继续（让智能体自我纠正）"""
    recovery = "log_and_continue"

class FatalError(AgentError):
    """触发：标记任务失败 + 人工干预"""
    recovery = "escalate_to_human"

# lab_04_client_factory.py

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import os

class AgentRole(Enum):
    RESEARCHER = "researcher"
    WRITER = "writer"
    CRITIC = "critic"

@dataclass
class ToolPermissions:
    allowed_tools: list[str]
    allowed_write_paths: list[str]  # 空列表 = 禁止写入
    allowed_read_paths: list[str]

@dataclass
class AgentConfig:
    role: AgentRole
    model: str
    permissions: ToolPermissions
    thinking_budget: int
    working_dir: str

# TODO Part A: 实现工具许可配置
# 为每个角色定义合适的工具集和路径许可
# 要求：
# - RESEARCHER：只能读文件，不能写
# - WRITER：能读写，但只能写入 output_dir
# - CRITIC：只能读，可以向 reviews/ 子目录写评审意见

ROLE_PERMISSIONS: dict[AgentRole, callable] = {
    # TODO: 填写每个角色的许可配置
    # 提示：使用工厂函数，接受 working_dir 参数
    # 示例：
    # AgentRole.RESEARCHER: lambda working_dir: ToolPermissions(
    #     allowed_tools=[...],
    #     allowed_write_paths=[],
    #     allowed_read_paths=[working_dir, "/tmp"],
    # ),
}


def create_agent_config(
    role: AgentRole,
    working_dir: str,
    output_dir: Optional[str] = None,
) -> AgentConfig:
    """
    工厂函数：根据角色创建完整配置。
    """
    if not os.path.exists(working_dir):
        raise ValueError(f"Working directory does not exist: {working_dir}")

    # TODO Part A: 完成工厂函数
    # 1. 从 ROLE_PERMISSIONS 获取该角色的许可
    # 2. 如果提供了 output_dir，动态添加到 WRITER 的写入许可中
    # 3. 根据角色选择合适的模型（researcher 和 critic 用 haiku，writer 用 sonnet）
    # 4. 根据角色设置思考预算（critic 需要最多的思考空间）
    pass

import threading
import time

class ConfigCache:
    """
    线程安全的配置缓存，使用双重检查锁定。
    """

    def __init__(self):
        self._cache: dict[str, AgentConfig] = {}
        self._lock = threading.Lock()
        self._load_count = 0  # 追踪实际加载次数（用于验证）

    def get_or_load(self, cache_key: str, loader: callable) -> AgentConfig:
        """
        TODO Part B: 实现双重检查锁定

        要求：
        1. 先在无锁状态下检查缓存（快速路径）
        2. 如果缓存未命中，获取锁
        3. 在持有锁时再次检查（防止竞争条件）
        4. 如果确实需要加载，调用 loader() 并存入缓存
        5. 递增 self._load_count

        注意：loader() 是一个耗时操作（模拟文件读取），
        在并发场景下应该只被调用一次。
        """
        pass

    def get_load_count(self) -> int:
        return self._load_count


def test_concurrent_caching():
    """验证：10 个并发线程只触发 1 次实际加载。"""
    cache = ConfigCache()

    def slow_loader():
        """模拟耗时的配置加载"""
        time.sleep(0.1)  # 模拟 I/O
        return {"model": "claude-sonnet-4-6", "budget": 5000}

    results = []
    errors = []

    def worker():
        try:
            config = cache.get_or_load("test-project", slow_loader)
            results.append(config)
        except Exception as e:
            errors.append(e)

    threads = [threading.Thread(target=worker) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert len(errors) == 0, f"Errors: {errors}"
    assert len(results) == 10, "All threads should get a result"
    assert cache.get_load_count() == 1, (
        f"Expected 1 load, got {cache.get_load_count()}. "
        "Race condition detected!"
    )
    print(f"✓ 10 concurrent threads, {cache.get_load_count()} actual load. "
          f"Cache is thread-safe.")

def verify_permission_isolation():
    """
    验证：不同角色的权限确实被隔离。
    """
    import tempfile

    with tempfile.TemporaryDirectory() as tmp_dir:
        output_dir = os.path.join(tmp_dir, "output")
        os.makedirs(output_dir)

        researcher_config = create_agent_config(
            role=AgentRole.RESEARCHER,
            working_dir=tmp_dir,
        )

        writer_config = create_agent_config(
            role=AgentRole.WRITER,
            working_dir=tmp_dir,
            output_dir=output_dir,
        )

        critic_config = create_agent_config(
            role=AgentRole.CRITIC,
            working_dir=tmp_dir,
        )

        # 验证 Researcher 没有写入权限
        assert len(researcher_config.permissions.allowed_write_paths) == 0, \
            "Researcher should have NO write permissions"

        # 验证 Writer 的写入范围被限制
        assert output_dir in writer_config.permissions.allowed_write_paths, \
            "Writer should be able to write to output_dir"
        assert tmp_dir not in writer_config.permissions.allowed_write_paths or \
               output_dir in writer_config.permissions.allowed_write_paths, \
            "Writer's write scope should be limited"

        # 验证 Critic 使用最高思考预算
        assert critic_config.thinking_budget >= writer_config.thinking_budget, \
            "Critic should have more thinking budget than Writer"

        print("✓ Permission isolation verified")
        print(f"  Researcher write paths: {researcher_config.permissions.allowed_write_paths}")
        print(f"  Writer write paths: {writer_config.permissions.allowed_write_paths}")
        print(f"  Critic thinking budget: {critic_config.thinking_budget}")


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