1. 感知（Perceive）→ 理解项目结构、代码上下文、用户需求
2. 规划（Plan）    → 拆解任务、制定实施步骤
3. 执行（Act）     → 生成/修改代码、运行测试、执行命令
4. 反馈（Reflect） → 观察执行结果、自我修正、迭代优化

# 自主编程 Agent 核心循环架构
from dataclasses import dataclass, field
from typing import List, Optional, Callable
import json

@dataclass
class CodingAgentConfig:
    model: str = "claude-opus-4"
    max_iterations: int = 25
    auto_run_tests: bool = True
    auto_commit: bool = False
    allowed_tools: List[str] = field(default_factory=lambda: [
        "read_file", "write_file", "patch", "terminal", "search_files"
    ])

@dataclass
class AgentState:
    task: str
    project_context: dict = field(default_factory=dict)
    conversation_history: List[dict] = field(default_factory=list)
    iteration: int = 0
    completed: bool = False
    last_error: Optional[str] = None

class AutonomousCodingAgent:
    """自主编程 Agent 的核心引擎"""
    
    def __init__(self, config: CodingAgentConfig):
        self.config = config
        self.state = None
    
    def perceive(self, project_path: str) -> dict:
        """感知阶段：理解项目上下文"""
        context = {}
        # 读取关键项目文件
        context["structure"] = self._get_project_structure(project_path)
        context["technologies"] = self._detect_technologies(project_path)
        context["entry_points"] = self._find_entry_points(project_path)
        context["configs"] = self._parse_config_files(project_path)
        return context
    
    def plan(self, task: str, context: dict) -> List[dict]:
        """规划阶段：制定实施步骤"""
        # LLM 驱动的任务分解
        plan_steps = self._llm_task_decomposition(task, context)
        return plan_steps
    
    def act(self, step: dict) -> dict:
        """执行阶段：执行单个步骤"""
        result = self._execute_step(step)
        return result
    
    def reflect(self, result: dict) -> str:
        """反馈阶段：分析执行结果"""
        if result.get("error"):
            return self._autocorrect(result["error"])
        if self._should_verify(result):
            return self._verify_implementation(result)
        return "continue"
    
    def run(self, task: str, project_path: str) -> dict:
        """完整执行循环"""
        self.state = AgentState(task=task)
        
        # Phase 1: Perceive
        context = self.perceive(project_path)
        self.state.project_context = context
        
        # Phase 2: Plan
        plan = self.plan(task, context)
        
        # Phase 3: Act-Reflect Loop
        for step in plan:
            while self.state.iteration < self.config.max_iterations:
                self.state.iteration += 1
                
                result = self.act(step)
                feedback = self.reflect(result)
                
                if feedback == "continue":
                    break
                elif feedback == "done":
                    self.state.completed = True
                    return {"status": "success", "plan": plan}
        
        return {"status": "completed", "summary": self._generate_summary()}
    
    def _get_project_structure(self, path: str) -> dict:
        """获取项目结构"""
        import os
        structure = {}
        for root, dirs, files in os.walk(path):
            rel_path = os.path.relpath(root, path)
            if rel_path == ".":
                structure["files"] = files[:30]  # 限制数量
            else:
                structure[rel_path] = files[:10]
        return structure
    
    def _detect_technologies(self, path: str) -> List[str]:
        """检测项目使用的技术栈"""
        import os
        techs = []
        for item in ["package.json", "pyproject.toml", "Cargo.toml", 
                     "go.mod", "Gemfile", "requirements.txt"]:
            if os.path.exists(os.path.join(path, item)):
                techs.append(item)
        return techs
    
    def _find_entry_points(self, path: str) -> List[str]:
        """寻找项目入口文件"""
        import glob
        patterns = ["main.py", "app.py", "index.js", "src/main.*", "src/index.*"]
        entries = []
        for pattern in patterns:
            entries.extend(glob.glob(os.path.join(path, pattern)))
        return entries
    
    def _parse_config_files(self, path: str) -> dict:
        """解析配置文件"""
        import os, json
        configs = {}
        for cfg_file in ["package.json", "tsconfig.json", ".env.example", 
                         "pyproject.toml", "Makefile"]:
            filepath = os.path.join(path, cfg_file)
            if os.path.exists(filepath):
                try:
                    with open(filepath) as f:
                        content = f.read()
                    if cfg_file.endswith(".json"):
                        configs[cfg_file] = json.loads(content)
                    else:
                        configs[cfg_file] = content[:200]
                except:
                    configs[cfg_file] = "parse_error"
        return configs
    
    def _llm_task_decomposition(self, task: str, context: dict) -> List[dict]:
        """LLM 驱动的任务分解 - 实际调用 LLM API"""
        # 生产环境中调用 LLM API
        prompt = f"""给定任务：{task}
项目上下文：{json.dumps(context, ensure_ascii=False)[:500]}
请将任务分解为具体的实施步骤，每个步骤包含：
- step: 步骤名称
- action: 操作类型 (read/write/patch/run)
- description: 详细描述
- verification: 验证方法"""
        # 解析 LLM 响应
        return [
            {"step": "分析现有代码", "action": "read", "description": "阅读相关文件了解当前实现"},
            {"step": "实现功能", "action": "write", "description": "生成新代码或修改现有代码"},
            {"step": "运行测试", "action": "run", "description": "执行测试验证功能正确性"},
        ]
    
    def _execute_step(self, step: dict) -> dict:
        """执行单个步骤 - 实际调用工具"""
        return {"status": "ok", "output": f"Executed: {step['step']}"}
    
    def _autocorrect(self, error: str) -> str:
        """自动修正错误"""
        return "retry" if "timeout" in error else "continue"
    
    def _should_verify(self, result: dict) -> bool:
        """判断是否需要验证"""
        return result.get("status") == "ok"
    
    def _verify_implementation(self, result: dict) -> str:
        """验证实现正确性"""
        return "continue"
    
    def _generate_summary(self) -> str:
        """生成执行摘要"""
        return f"完成 {self.state.iteration} 次迭代的任务执行"

┌─────────────────────────────────────────────────────┐
│                    ACP 协议栈                         │
├─────────────────────────────────────────────────────┤
│  Agent 层: 任务委派 / 子任务管理 / 状态追踪 / 结果回传  │
├─────────────────────────────────────────────────────┤
│  Tool 层: tools/list / tools/call / resources/read   │
├─────────────────────────────────────────────────────┤
│  Transport 层: stdio / SSE / WebSocket               │
├─────────────────────────────────────────────────────┤
│  Message 层: JSON-RPC 2.0 / 请求-响应 / 通知          │
└─────────────────────────────────────────────────────┘

# ACP 协议核心消息格式
@dataclass
class ACPMessage:
    jsonrpc: str = "2.0"
    method: str = ""
    params: dict = field(default_factory=dict)
    id: Optional[str] = None

# Agent 委派请求
DELEGATE_TASK_REQUEST = {
    "jsonrpc": "2.0",
    "method": "tasks/delegate",
    "params": {
        "goal": "实现用户登录功能",
        "context": {
            "project_path": "/app",
            "tech_stack": ["Python", "FastAPI", "SQLAlchemy"],
            "constraints": ["使用 JWT 认证", "支持 OAuth2 登录"],
        },
        "toolsets": ["file", "terminal", "search"],
        "max_iterations": 50
    },
    "id": "req-001"
}

# 状态同步响应
TASK_STATUS_RESPONSE = {
    "jsonrpc": "2.0",
    "result": {
        "status": "in_progress",
        "current_step": "编写用户模型",
        "progress": 0.45,
        "estimated_remaining": "2m 30s",
        "intermediate_result": {
            "files_created": ["models/user.py", "routes/auth.py"],
            "tests_passing": 3
        }
    },
    "id": "req-001"
}

# 安装与启动
npm install -g @anthropic/claude-code
claude

# 关键命令
claude -p "解释这段代码的逻辑"            # 快速查询
claude -p "重构 main.py 中的数据库查询"    # 直接修改
claude --resume                            # 恢复上次会话
claude --model claude-opus-4-20260501      # 指定模型

# 开发工作流
claude -p "实现用户注册 API，包含邮箱验证和密码加密"
# Claude Code 会自动:
# 1. 读取项目结构和现有代码
# 2. 规划实施步骤
# 3. 生成 routes/auth.py, models/user.py, services/email.py
# 4. 运行测试验证
# 5. 提交 git commit

# Claude Code 上下文管理策略
class ClaudeCodeContextManager:
    """模仿 Claude Code 的上下文管理策略"""
    
    def __init__(self, max_tokens: int = 200_000):
        self.max_tokens = max_tokens
        self.budget = {
            "project_map": 0.05,    # 5% 用于项目结构图
            "file_contents": 0.40,  # 40% 用于文件内容
            "conversation": 0.40,   # 40% 用于对话历史
            "tool_results": 0.10,   # 10% 用于工具执行结果
            "system": 0.05,         # 5% 用于系统指令
        }
    
    def build_context(self, project_path: str, task: str, 
                      conversation: list, relevant_files: List[str]) -> dict:
        """构建优化的上下文包"""
        context = {}
        
        # 1. 项目地图（5%）
        context["project_map"] = self._summarize_project(project_path)
        
        # 2. 相关文件内容（40%）
        context["files"] = self._select_and_summarize_files(
            relevant_files, 
            int(self.max_tokens * self.budget["file_contents"])
        )
        
        # 3. 对话摘要（40%）
        context["conversation"] = self._summarize_conversation(
            conversation,
            int(self.max_tokens * self.budget["conversation"])
        )
        
        # 4. 工具指令集（10%）
        context["tools"] = self._get_tool_definitions(
            int(self.max_tokens * self.budget["tool_results"])
        )
        
        return context
    
    def _summarize_project(self, path: str) -> str:
        """生成项目结构概要"""
        return f"Project: {path}\nDependencies: ...\nStructure: ..."
    
    def _select_and_summarize_files(self, files: List[str], budget: int) -> dict:
        """基于预算选择并摘要文件内容"""
        selected = {}
        for f in files[:10]:  # 最多选10个文件
            content = self._read_and_summarize(f, budget // 10)
            selected[f] = content
        return selected
    
    def _summarize_conversation(self, conversation: List[dict], budget: int) -> str:
        """压缩对话历史"""
        if len(conversation) <= 4:
            return str(conversation)
        # 保留最近的2轮完整对话
        recent = conversation[-2:]
        older = conversation[:-2]
        summary = f"[前{len(older)}轮对话摘要]"
        return f"{summary}\n{recent}"
    
    def _get_tool_definitions(self, budget: int) -> List[dict]:
        """获取工具定义"""
        return [
            {"name": "read_file", "description": "读取文件内容"},
            {"name": "write_file", "description": "写入文件"},
            {"name": "patch", "description": "修改文件片段"},
            {"name": "terminal", "description": "执行 Shell 命令"},
        ]
    
    def _read_and_summarize(self, path: str, budget: int) -> str:
        """读取并摘要文件内容"""
        return f"[{path} 摘要: {budget} tokens]"

用户输入: "实现一个 RESTful API 用户管理系统"
         │
         ▼
┌─────────────────────┐
│  Planner (o4-mini)   │ ← 轻量级规划，快速分解任务
│  "1. 设计数据模型    │
│   2. 实现 CRUD 路由  │
│   3. 添加认证中间件" │
└─────────┬───────────┘
          │ ACP task/delegate
          ▼
┌─────────────────────┐
│  Coder (o3/o4-mini)  │ ← 专注于代码生成与修改
│  - 文件读写能力      │
│  - 终端执行能力      │
│  - 自动修正循环      │
└─────────────────────┘
          │
          ▼
    代码产出 + 测试验证

# Codex CLI 的 Planner-Coder 架构
@dataclass
class CodexConfig:
    planner_model: str = "o4-mini"     # 规划模型
    coder_model: str = "o3"             # 执行模型  
    max_plan_steps: int = 10
    auto_test: bool = True

class CodexPlanner:
    """Codex CLI 规划器 - 负责任务分解"""
    
    def __init__(self, config: CodexConfig):
        self.config = config
        self.model = config.planner_model
    
    def decompose(self, task: str, context: dict) -> List[dict]:
        """将复杂任务分解为可执行的子任务"""
        # 分析任务复杂度
        complexity = self._estimate_complexity(task)
        
        if complexity > 0.7:
            # 高复杂度任务 - 分层分解
            return self._hierarchical_decompose(task, context)
        else:
            # 低复杂度任务 - 线性分解
            return self._linear_decompose(task, context)
    
    def _estimate_complexity(self, task: str) -> float:
        """估计任务复杂度 (0-1)"""
        keywords = {
            "高": ["认证", "数据库", "支付", "部署", "并发", "分布式"],
            "中": ["API", "CRUD", "缓存", "队列", "WebSocket"],
            "低": ["格式化", "重构", "修复", "测试"],
        }
        score = 0.3
        for word in task:
            if word in keywords["高"]:
                score = max(score, 0.8)
            elif word in keywords["中"]:
                score = max(score, 0.5)
        return score
    
    def _hierarchical_decompose(self, task: str, context: dict) -> List[dict]:
        """分层任务分解"""
        return [
            {"level": 0, "step": "项目分析", "action": "analyze"},
            {"level": 1, "step": "数据层实现", "action": "implement"},
            {"level": 2, "step": "业务逻辑层", "action": "implement"},
            {"level": 3, "step": "接口层实现", "action": "implement"},
            {"level": 4, "step": "集成测试", "action": "test"},
        ]
    
    def _linear_decompose(self, task: str, context: dict) -> List[dict]:
        """线性任务分解"""
        return [
            {"step": "代码分析", "action": "read"},
            {"step": "实现修改", "action": "patch"},
            {"step": "运行测试", "action": "run"},
        ]

class CodexCoder:
    """Codex CLI 执行器 - 专注于代码生成"""
    
    def __init__(self, config: CodexConfig):
        self.config = config
        self.model = config.coder_model
        self.max_retries = 3
    
    def execute(self, step: dict) -> dict:
        """执行编码步骤，带自动修正"""
        for attempt in range(self.max_retries):
            try:
                if step["action"] == "implement":
                    result = self._generate_code(step)
                elif step["action"] == "patch":
                    result = self._modify_code(step)
                elif step["action"] == "test":
                    result = self._run_and_fix(step)
                else:
                    result = self._analyze_code(step)
                
                if self._verify(result):
                    return result
            except Exception as e:
                if attempt == self.max_retries - 1:
                    return {"error": str(e)}
        return {"error": "max retries exceeded"}
    
    def _generate_code(self, step: dict) -> dict:
        """生成新代码"""
        return {"files": ["models/user.py"], "tests": ["test_user.py"]}
    
    def _modify_code(self, step: dict) -> dict:
        """修改现有代码"""
        return {"patched": ["routes/auth.py"]}
    
    def _run_and_fix(self, step: dict) -> dict:
        """运行测试并自动修复"""
        return {"tests_passed": 5, "tests_failed": 0}
    
    def _analyze_code(self, step: dict) -> dict:
        """分析现有代码"""
        return {"analysis": "项目使用 FastAPI + SQLAlchemy"}
    
    def _verify(self, result: dict) -> bool:
        """验证执行结果"""
        return "error" not in result

# 完整的 Codex CLI Pipeline
class CodexPipeline:
    """Codex CLI 完整执行流水线"""
    
    def __init__(self, config: CodexConfig = None):
        self.config = config or CodexConfig()
        self.planner = CodexPlanner(self.config)
        self.coder = CodexCoder(self.config)
    
    def run(self, task: str, project_path: str) -> dict:
        """执行完整开发任务"""
        # 1. 项目感知
        project_context = self._perceive_project(project_path)
        
        # 2. 任务规划
        plan = self.planner.decompose(task, project_context)
        print(f"📋 任务分解为 {len(plan)} 个步骤")
        
        # 3. 逐步执行
        results = []
        for step in plan:
            result = self.coder.execute(step)
            results.append(result)
            status = "✅" if "error" not in result else "❌"
            print(f"{status} {step['step']}")
        
        # 4. 生成报告
        return {
            "task": task,
            "steps_completed": len([r for r in results if "error" not in r]),
            "steps_total": len(plan),
            "status": "success" if all("error" not in r for r in results) else "partial"
        }
    
    def _perceive_project(self, path: str) -> dict:
        """感知项目"""
        return {"path": path, "type": "web_app"}

# 多 Agent 编码团队的协作架构
@dataclass
class CodeReviewResult:
    file: str
    issues: List[dict]
    score: float
    suggestions: List[str]

class CodeReviewAgent:
    """代码审查 Agent"""
    
    def review(self, file_path: str, content: str) -> CodeReviewResult:
        issues = self._find_issues(content)
        score = self._calculate_score(issues)
        suggestions = self._generate_suggestions(issues)
        return CodeReviewResult(
            file=file_path, 
            issues=issues, 
            score=score,
            suggestions=suggestions
        )
    
    def _find_issues(self, content: str) -> List[dict]:
        """发现代码问题"""
        issues = []
        # 安全检查
        if "eval(" in content or "exec(" in content:
            issues.append({
                "type": "security",
                "severity": "critical",
                "message": "禁止使用 eval/exec"
            })
        # 性能检查
        if "for" in content and "range(len(" in content:
            issues.append({
                "type": "performance",
                "severity": "warning",
                "message": "建议使用 enumerate 替代 range(len())"
            })
        # 风格检查
        if len(content.split('\n')) > 500:
            issues.append({
                "type": "style",
                "severity": "info",
                "message": "建议将长文件拆分为模块"
            })
        return issues
    
    def _calculate_score(self, issues: List[dict]) -> float:
        """计算代码质量评分 (0-100)"""
        penalties = {"critical": 20, "error": 10, "warning": 5, "info": 1}
        total_penalty = sum(penalties.get(i["severity"], 0) for i in issues)
        return max(0, 100 - total_penalty)
    
    def _generate_suggestions(self, issues: List[dict]) -> List[str]:
        """生成改进建议"""
        return [f"[{i['severity'].upper()}] {i['message']}" for i in issues]

class TestGenerationAgent:
    """测试生成 Agent"""
    
    def generate_tests(self, source_code: str, source_path: str) -> str:
        """根据源码自动生成测试代码"""
        framework = self._detect_test_framework(source_path)
        test_content = self._llm_generate_tests(source_code, framework)
        return test_content
    
    def _detect_test_framework(self, path: str) -> str:
        """检测项目使用的测试框架"""
        import os
        if os.path.exists("pytest.ini") or os.path.exists("pyproject.toml"):
            return "pytest"
        return "unittest"
    
    def _llm_generate_tests(self, source: str, framework: str) -> str:
        """LLM 驱动的测试生成"""
        prompt = f"""基于以下代码生成 {framework} 测试用例：
{source[:1000]}...
要求：
1. 覆盖正常路径和异常路径
2. 使用 Mock 隔离外部依赖
3. 测试命名遵循 AAA 模式"""
        return f"# Generated {framework} tests\n# {len(source)} lines analyzed"

class ArchitectureAgent:
    """架构设计 Agent"""
    
    def review_design(self, changes: List[dict], project_structure: dict) -> dict:
        """审查变更是否符含架构规范"""
        violations = []
        
        # 检查分层违规
        for change in changes:
            if self._check_layer_violation(change, project_structure):
                violations.append({
                    "type": "layer_violation",
                    "file": change.get("file"),
                    "message": "业务逻辑不可以直接访问数据层"
                })
        
        # 检查循环依赖
        deps = self._analyze_dependencies(changes)
        circular = self._find_circular_dependencies(deps)
        
        return {
            "approved": len(violations) == 0 and len(circular) == 0,
            "violations": violations,
            "circular_deps": circular,
            "suggestions": self._generate_improvements(violations, circular)
        }
    
    def _check_layer_violation(self, change: dict, structure: dict) -> bool:
        """检查分层违规"""
        return False
    
    def _analyze_dependencies(self, changes: List[dict]) -> dict:
        """分析依赖关系"""
        return {}
    
    def _find_circular_dependencies(self, deps: dict) -> List[str]:
        """检测循环依赖"""
        return []
    
    def _generate_improvements(self, violations: List[dict], 
                                 circular: List[str]) -> List[str]:
        """生成架构改进建议"""
        suggestions = []
        for v in violations:
            suggestions.append(f"重构 {v['file']}: {v['message']}")
        return suggestions

# 多 Agent 编排器
class MultiAgentCodingOrchestrator:
    """多 Agent 编码协作编排器"""
    
    def __init__(self):
        self.planner = CodexPlanner(CodexConfig())
        self.coder = CodexCoder(CodexConfig())
        self.reviewer = CodeReviewAgent()
        self.tester = TestGenerationAgent()
        self.architect = ArchitectureAgent()
    
    def develop_feature(self, task: str, project_path: str) -> dict:
        """完整的功能开发流程"""
        print(f"🚀 开始开发: {task}")
        
        # Phase 1: 架构审查
        print("📐 架构审查中...")
        arch_review = self.architect.review_design(
            [{"file": project_path}], {}
        )
        if not arch_review["approved"]:
            return {"status": "blocked", "reason": "架构设计不通过"}
        
        # Phase 2: 任务规划
        print("📋 规划实施步骤...")
        context = {"path": project_path}
        plan = self.planner.decompose(task, context)
        
        # Phase 3: 编码执行 + 审查循环
        results = []
        for step in plan:
            # 编码
            result = self.coder.execute(step)
            
            # 审查
            if result.get("files"):
                for f in result["files"]:
                    review = self.reviewer.review(f, "")
                    if review.score < 60:
                        print(f"  ⚠️ {f} 得分 {review.score}，需要改进")
            
            # 测试
            if result.get("files"):
                for f in result["files"]:
                    tests = self.tester.generate_tests("", f)
                    
            results.append(result)
        
        print("✅ 功能开发完成")
        return {"status": "success", "steps": len(results)}