Claude Code 512K行源码泄露：架构全景与七大设计启示

timeline
    title Claude Code源码泄露事件时间线
    section 3月31日
        凌晨4:00 : Anthropic推送v2.1.88到npm
        凌晨4:15 : 安全研究员Chaofan Shou(@shoucccc)发现异常
        凌晨4:30 : Shou在X平台发布发现
        上午8:00 : Anthropic确认泄露并撤包
        上午10:00 : 社区已完成初步源码分析
    section 4月1日-7日
        第1天 : 44个隐藏Feature Flag被发现
        第2天 : KAIROS守护进程架构曝光
        第3天 : BUDDY AI宠物功能曝光
        第5天 : claw-code Rust重写项目启动
        第7天 : 50+篇深度分析文章发布

// 正常的.npmignore应该排除source map
{
  "exclude": [
    "src/**",
    "*.ts",
    "*.map",      // ← 这一行的缺失导致了泄露
    "tests/**",
    ".eslintrc*"
  ]
}

graph TB
    subgraph "用户交互层"
        CLI[CLI终端界面]
        IDE[IDE集成<br/>VSCode/JetBrains]
        API[Headless API模式]
    end

    subgraph "协调层 Coordinator"
        COORD[Coordinator协调器]
        MODE[模式路由器<br/>7种执行模式]
        PERM[权限管理器<br/>5级权限模型]
    end

    subgraph "Agent核心层"
        LOOP[Agent Loop<br/>核心循环]
        CTX[上下文管理器<br/>5阶段组装]
        PROMPT[提示词层级<br/>System→User→Tool]
        MODEL[模型路由<br/>h2A动态选择]
    end

    subgraph "工具系统 Tool System"
        TOOLS[40+工具定义]
        BASH[Bash工具]
        FILE[文件工具<br/>Read/Write/Edit/Glob/Grep]
        TASK[Task工具<br/>Subagent管理]
        MCP[MCP客户端<br/>外部工具协议]
    end

    subgraph "基础设施层"
        MEM[记忆系统<br/>6层记忆架构]
        HOOK[Hook事件系统<br/>13个事件点]
        CRYPTO[加密/安全层]
        CACHE[缓存层]
    end

    CLI --> COORD
    IDE --> COORD
    API --> COORD
    COORD --> LOOP
    COORD --> MODE
    COORD --> PERM
    LOOP --> CTX
    LOOP --> PROMPT
    LOOP --> MODEL
    LOOP --> TOOLS
    TOOLS --> BASH
    TOOLS --> FILE
    TOOLS --> TASK
    TOOLS --> MCP
    LOOP --> MEM
    LOOP --> HOOK
    LOOP --> CRYPTO

// 简化版Agent Loop伪代码（基于源码逆向）
async function agentLoop(userMessage: string) {
    // 阶段1：上下文组装
    const context = await assembleContext({
        systemPrompt: buildSystemPrompt(),      // 系统提示
        claudeMd: loadClaudeMdFiles(),          // CLAUDE.md指令
        conversationHistory: getHistory(),       // 对话历史
        toolResults: getPreviousToolResults(),   // 上轮工具结果
        fileContext: detectRelevantFiles(),      // 自动检测相关文件
    });

    // 阶段2：模型调用（h2A路由）
    const response = await callModel({
        messages: context,
        tools: getToolDefinitions(),             // 40+工具定义
        model: selectModel(taskComplexity),      // 动态模型选择
    });

    // 阶段3：响应处理
    for (const block of response.content) {
        if (block.type === 'text') {
            displayToUser(block.text);
        } else if (block.type === 'tool_use') {
            // 阶段4：权限检查
            const permitted = await checkPermission(block);
            if (!permitted) {
                const userApproval = await requestUserApproval(block);
                if (!userApproval) continue;
            }
            // 阶段5：工具执行
            const result = await executeTool(block.name, block.input);
            // 阶段6：Hook触发
            await triggerHooks('tool_execution_complete', result);
            // 循环回阶段1
            return agentLoop(result);
        }
    }
}

flowchart LR
    A["阶段1<br/>系统提示<br/>~15K tokens"] --> B["阶段2<br/>CLAUDE.md<br/>项目指令"]
    B --> C["阶段3<br/>对话历史<br/>含压缩摘要"]
    C --> D["阶段4<br/>工具结果<br/>智能截断"]
    D --> E["阶段5<br/>运行时注入<br/>环境信息/文件"]

// 工具定义结构（简化）
interface ToolDefinition {
    name: string;           // 如 "Read", "Bash", "Grep"
    description: string;    // 给LLM的自然语言描述
    inputSchema: JSONSchema;// 输入参数JSON Schema
    permissions: string[];  // 所需权限级别
    execute: (input) => Promise<ToolResult>;
}

// 源码中发现的诱饵工具机制
const decoyTools = [
    { name: "UndercoverMode", description: "..." },
    { name: "SystemOverride", description: "..." },
    // 如果模型试图调用这些工具，说明它可能在"扮演"而非真正推理
];

graph TD
    L1["Level 1: 自动允许<br/>只读操作<br/>Read/Glob/Grep"] --> L2["Level 2: 首次确认<br/>文件写入<br/>Write/Edit"]
    L2 --> L3["Level 3: 每次确认<br/>命令执行<br/>Bash(破坏性)"]
    L3 --> L4["Level 4: 严格限制<br/>网络操作<br/>MCP外部调用"]
    L4 --> L5["Level 5: 完全禁止<br/>系统级操作<br/>rm -rf/特权提升"]

// 权限检查伪代码
function checkPermission(tool: Tool, input: ToolInput): PermissionDecision {
    // 1. 检查工具的基础权限级别
    const baseLevel = tool.permissions.defaultLevel;

    // 2. 根据输入参数调整（如Bash命令的具体内容）
    const adjustedLevel = analyzeInput(input);

    // 3. 检查用户配置的权限设置
    const userPolicy = getUserPolicy(tool.name);

    // 4. 检查是否已经获得过许可（会话级缓存）
    if (hasBeenApproved(tool.name, input)) return 'auto-approve';

    // 5. 综合决策
    return resolvePermission(baseLevel, adjustedLevel, userPolicy);
}

// 源码中发现的挫折检测模式（简化）
const frustrationPatterns = [
    /stop/i,
    /no\s+(that|this)\s+is\s+wrong/i,
    /you'?re?\s+not\s+listening/i,
    /forget\s+it/i,
    /this\s+is\s+useless/i,
    /I\s+said\s+/i,
    // 更多的多语言模式...
];

function detectFrustration(userMessage: string): FrustrationLevel {
    let score = 0;
    for (const pattern of frustrationPatterns) {
        if (pattern.test(userMessage)) score += pattern.weight;
    }
    if (score > FRUSTRATION_THRESHOLD) return 'high';
    if (score > LOW_THRESHOLD) return 'medium';
    return 'low';
}

flowchart TD
    INPUT[用户请求] --> ANALYZE[任务分析]
    ANALYZE --> COMPLEX{任务复杂度}
    COMPLEX -->|"高<br/>架构设计/重构"| OPUS[Opus 4.6<br/>最强推理]
    COMPLEX -->|"中<br/>功能实现/调试"| SONNET[Sonnet 4.6<br/>速度与质量平衡]
    COMPLEX -->|"低<br/>简单查找/格式化"| HAIKU[Haiku 4.5<br/>最快响应]

    OPUS --> EXEC[执行]
    SONNET --> EXEC
    HAIKU --> EXEC
    EXEC --> EVAL[结果评估]
    EVAL -->|"质量不足<br/>升级模型"| OPUS
    EVAL -->|"质量达标"| DONE[返回结果]

graph LR
    subgraph "瞬时记忆"
        W[工作记忆<br/>单次调用]
    end
    subgraph "会话记忆"
        C[对话历史<br/>单次会话]
        S[会话摘要<br/>自动压缩]
    end
    subgraph "持久记忆"
        MD[CLAUDE.md<br/>项目级]
        MEM[记忆文件<br/>全局级]
        ARCH[压缩档案<br/>历史积累]
    end

    W -->|"工具结果"| C
    C -->|"接近上限"| S
    S -->|"关键信息提取"| MEM
    MD -->|"每轮注入"| C
    MEM -->|"相关召回"| C
    ARCH -->|"搜索匹配"| C

事件点:
├── session.start          # 会话开始
├── session.end            # 会话结束
├── prompt.submit          # 用户提交提示
├── tool.call              # 工具调用前
├── tool.result            # 工具返回后
├── file.read              # 文件读取
├── file.write             # 文件写入
├── file.edit              # 文件编辑
├── bash.pre               # Bash命令执行前
├── bash.post              # Bash命令执行后
├── agent.spawn            # Subagent创建
├── agent.complete         # Subagent完成
└── context.overflow       # 上下文溢出

// Subagent类型定义（简化）
type SubagentType =
    | 'Explore'      // 快速搜索Agent
    | 'Plan'         // 架构规划Agent
    | 'general-purpose' // 通用Agent
    | string;        // 自定义Agent

interface TaskConfig {
    subagent_type: SubagentType;
    prompt: string;
    model?: 'opus' | 'sonnet' | 'haiku';
    isolation?: 'worktree';  // Git隔离
    run_in_background?: boolean;
    max_turns?: number;
}

flowchart TD
    MAIN[主Agent<br/>Opus 4.6] --> T1[Task: Explore<br/>Haiku<br/>只读工具]
    MAIN --> T2[Task: Plan<br/>Sonnet<br/>只读工具]
    MAIN --> T3[Task: General<br/>Sonnet<br/>全部工具]
    MAIN --> T4[Task: Worktree<br/>Sonnet<br/>全部工具+Git隔离]

    T1 -->|"搜索结果"| MAIN
    T2 -->|"实施方案"| MAIN
    T3 -->|"代码修改"| MAIN
    T4 -->|"分支+修改"| MAIN

    MAIN --> SYNTHESIS[结果综合<br/>质量评估]
    SYNTHESIS --> OUTPUT[最终输出]

KAIROS特性（从Feature Flag推断）：
- 24/7后台运行，监控代码库变化
- 自动响应CI/CD失败
- 持续集成测试和代码质量检查
- 类似于"AI队友"而非"工具"

BUDDY特性：
- 拥有"情绪"状态
- 根据用户交互模式变化
- 提供更拟人化的交互体验
- 可能用于提升用户粘性和使用频率

Undercover Mode功能：
- 剥离AI生成痕迹
- 修改代码风格以匹配项目既有风格
- 添加"人性化"的代码注释风格
- 在git历史中隐藏AI参与的证据

ULTRAPLAN特性：
- 自动将大型任务分解为多阶段计划
- 生成详细的实施步骤
- 跨多轮对话保持计划一致性
- 支持计划的动态调整和回滚

flowchart LR
    LEAK["3月31日<br/>源码泄露"] --> ANALYSIS["4月1-3日<br/>架构分析"]
    ANALYSIS --> DESIGN["4月4日<br/>Rust设计"]
    DESIGN --> IMPL["4月5-15日<br/>~20K行Rust"]
    IMPL --> ALPHA["4月中旬<br/>Alpha发布"]
    ALPHA --> PRODUCTION["4月底<br/>生产可用"]

// claw-code核心循环（简化）
async fn agent_loop(ctx: &mut Context) -> Result<Output> {
    loop {
        let context = ctx.assemble_context().await?;
        let response = ctx.call_model(context).await?;

        match response {
            Response::Text(text) => {
                ctx.display(&text);
                return Ok(Output::Text(text));
            }
            Response::ToolCall(call) => {
                if !ctx.check_permission(&call)? {
                    ctx.request_approval(&call).await?;
                }
                let result = ctx.execute_tool(call).await?;
                ctx.add_tool_result(result);
                // 继续循环
            }
            Response::Stop => return Ok(Output::Complete),
        }
    }
}

// 最小Agent Harness骨架
class AgentHarness {
    private tools: Map<string, Tool>;
    private context: ContextManager;
    private permissions: PermissionManager;

    async run(userMessage: string): Promise<string> {
        while (true) {
            // 1. 组装上下文
            const messages = this.context.build(userMessage);

            // 2. 调用LLM
            const response = await this.callLLM(messages, this.toolSchemas());

            // 3. 处理响应
            if (response.type === 'text') return response.content;
            if (response.type === 'tool_use') {
                // 4. 权限检查
                await this.permissions.check(response);
                // 5. 执行工具
                const result = await this.tools.get(response.name).execute(response.input);
                // 6. 将结果加入上下文
                this.context.addToolResult(result);
            }
        }
    }
}

# 推荐的CI/CD安全检查
security-check:
  script:
    - npm pack --dry-run
    - | 
      if find . -name "*.map" | grep -q .; then
        echo "ERROR: Source maps found in package!"
        exit 1
      fi
    - |
      if tar -tzf *.tgz | grep -E "\.(ts|map)$"; then
        echo "ERROR: Source files in package!"
        exit 1
      fi

graph TD
    LEAK[源码泄露] --> A[架构知识民主化]
    LEAK --> B[claw-code等开源项目]
    LEAK --> C[安全意识提升]

    A --> D[更多高质量Agent出现]
    B --> E[多Provider Agent工具]
    C --> F[更严格的发布流程]

    D --> G[行业标准加速形成]
    E --> G
    F --> G

    G --> H[AI Agent工程化成熟]

"""mini_agent.py - 基于Claude Code架构原则的最小Agent"""

import json
from typing import Any
from anthropic import Anthropic

class Tool:
    """工具基类 - 模仿Claude Code的工具定义"""
    def __init__(self, name: str, description: str, input_schema: dict,
                 permission_level: int = 1):
        self.name = name
        self.description = description
        self.input_schema = input_schema
        self.permission_level = permission_level  # 1=自动, 2=确认, 3=禁止

    async def execute(self, **kwargs) -> str:
        raise NotImplementedError

class ReadFileTool(Tool):
    def __init__(self):
        super().__init__(
            name="read_file",
            description="Read a file from the filesystem",
            input_schema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path"}
                },
                "required": ["path"]
            },
            permission_level=1  # 只读，自动允许
        )

    async def execute(self, path: str, **kwargs) -> str:
        try:
            with open(path, 'r') as f:
                return f.read()[:10000]  # 安全截断
        except FileNotFoundError:
            return f"Error: File not found: {path}"

class ContextManager:
    """上下文管理器 - 5阶段组装"""
    def __init__(self, system_prompt: str, max_tokens: int = 180000):
        self.system_prompt = system_prompt
        self.messages = []
        self.max_tokens = max_tokens

    def add_user(self, text: str):
        self.messages.append({"role": "user", "content": text})

    def add_assistant(self, text: str):
        self.messages.append({"role": "assistant", "content": text})

    def add_tool_result(self, tool_use_id: str, result: str):
        self.messages.append({
            "role": "user",
            "content": [{
                "type": "tool_result",
                "tool_use_id": tool_use_id,
                "content": result
            }]
        })

    def should_compress(self) -> bool:
        """简单的token估算"""
        total = len(self.system_prompt)
        for msg in self.messages:
            total += len(str(msg))
        return total > self.max_tokens * 0.8  # 80%阈值触发压缩

class MiniAgent:
    """最小Agent - 展示Claude Code核心架构"""
    def __init__(self, api_key: str):
        self.client = Anthropic(api_key=api_key)
        self.tools: list[Tool] = []
        self.ctx = ContextManager(
            system_prompt="You are a helpful coding assistant."
        )
        self.auto_approved = set()  # 已批准的工具+输入组合

    def register_tool(self, tool: Tool):
        self.tools.append(tool)

    def _tool_schemas(self) -> list[dict]:
        return [{
            "name": t.name,
            "description": t.description,
            "input_schema": t.input_schema
        } for t in self.tools]

    async def run(self, user_message: str) -> str:
        self.ctx.add_user(user_message)

        while True:  # Agent Loop
            response = self.client.messages.create(
                model="claude-sonnet-4-6-20250514",
                max_tokens=4096,
                system=self.ctx.system_prompt,
                messages=self.ctx.messages,
                tools=self._tool_schemas()
            )

            for block in response.content:
                if block.type == "text":
                    self.ctx.add_assistant(block.text)
                    return block.text

                elif block.type == "tool_use":
                    tool = next(t for t in self.tools if t.name == block.name)

                    # 权限检查（简化版Claude Code的5级模型）
                    if tool.permission_level <= 1:
                        pass  # 自动允许
                    elif tool.permission_level == 2:
                        key = f"{tool.name}:{json.dumps(block.input)}"
                        if key not in self.auto_approved:
                            approve = input(f"Allow {tool.name}? [y/N] ")
                            if approve.lower() != 'y':
                                self.ctx.add_tool_result(block.id, "Denied by user")
                                continue
                            self.auto_approved.add(key)

                    result = await tool.execute(**block.input)
                    self.ctx.add_tool_result(block.id, result)

# 使用示例
if __name__ == "__main__":
    import asyncio

    async def main():
        agent = MiniAgent(api_key="your-api-key")
        agent.register_tool(ReadFileTool())
        result = await agent.run("Read the file README.md and summarize it")
        print(result)

    asyncio.run(main())