# tools/adaptive_executor.py
from typing import Callable, Any, Dict
import subprocess
import json

class AdaptiveExecutor:
    def __init__(self):
        self.tool_history = {}
        self.failure_patterns = {}

    async def execute_with_adaptation(
        self,
        primary_tool: Callable,
        fallback_tools: list[Callable],
        context: Dict[str, Any]
    ) -> tuple[bool, Any]:
        """
        执行工具，失败时自动适应

        返回: (success, result)
        """
        # 尝试主工具
        try:
            result = await primary_tool(**context)
            self._record_success(primary_tool.__name__, context)
            return True, result
        except Exception as e:
            error_type = type(e).__name__
            error_msg = str(e)

            # 分析失败模式
            strategy = self._analyze_failure(error_type, error_msg)

            # 根据策略选择适应方案
            if strategy == "retry_with_modified_params":
                modified_context = self._modify_params(context, error_msg)
                return await self.execute_with_adaptation(
                    primary_tool, fallback_tools, modified_context
                )

            elif strategy == "switch_to_fallback":
                for fallback in fallback_tools:
                    try:
                        result = await fallback(**context)
                        self._record_success(fallback.__name__, context)
                        return True, result
                    except Exception:
                        continue

            # 所有策略都失败
            return False, error_msg

    def _analyze_failure(self, error_type: str, error_msg: str) -> str:
        """分析失败原因，返回适应策略"""
        if "timeout" in error_msg.lower():
            return "increase_timeout"
        elif "permission" in error_msg.lower():
            return "escalate_permissions"
        elif "not found" in error_msg.lower():
            return "switch_to_fallback"
        else:
            return "retry_with_modified_params"

    def _modify_params(self, context: Dict, error_msg: str) -> Dict:
        """根据错误信息修改参数"""
        modified = context.copy()

        if "timeout" in error_msg:
            modified["timeout"] = modified.get("timeout", 30) * 2

        return modified

    def _record_success(self, tool_name: str, context: Dict):
        """记录成功模式"""
        key = f"{tool_name}:{json.dumps(context, sort_keys=True)}"
        self.tool_history[key] = self.tool_history.get(key, 0) + 1

# 使用A1框架
executor = AdaptiveExecutor()

async def git_push(branch: str):
    subprocess.run(["git", "push", "origin", branch], check=True)

async def git_push_with_force(branch: str):
    subprocess.run(["git", "push", "origin", branch, "--force-with-lease"], check=True)

async def git_push_via_ssh(branch: str):
    subprocess.run(["GIT_SSH=ssh", "git", "push", "origin", branch], check=True)

# 执行（自动适应）
success, result = await executor.execute_with_adaptation(
    primary_tool=git_push,
    fallback_tools=[git_push_with_force, git_push_via_ssh],
    context={"branch": "main"}
)

if success:
    print("✅ Push成功")
else:
    print(f"❌ Push失败: {result}")

# agents/quality_feedback_agent.py
from typing import List, Tuple
import subprocess

class QualityFeedbackAgent:
    def __init__(self, quality_threshold: float = 0.8):
        self.threshold = quality_threshold
        self.feedback_history = []

    async def generate_with_feedback(
        self,
        generator: Callable,
        prompt: str,
        max_iterations: int = 3
    ) -> Tuple[bool, str]:
        """
        生成代码，根据质量反馈迭代优化

        返回: (success, final_output)
        """
        for iteration in range(max_iterations):
            # 生成输出
            output = await generator(prompt)

            # 评估质量
            quality = await self._evaluate_quality(output)

            print(f"Iteration {iteration + 1}: Quality = {quality:.2f}")

            # 如果达到阈值，返回
            if quality >= self.threshold:
                self._record_success(prompt, output)
                return True, output

            # 否则，生成反馈并重新生成
            feedback = await self._generate_feedback(output, quality)
            prompt = self._refine_prompt(prompt, feedback)

        # 达到最大迭代次数
        return False, output

    async def _evaluate_quality(self, code: str) -> float:
        """评估代码质量（0-1）"""
        score = 0.0
        max_score = 0.0

        # 1. Linting检查
        try:
            result = subprocess.run(
                ["ruff", "check", "--output-format=json", "-"],
                input=code,
                capture_output=True,
                text=True
            )
            if result.returncode == 0:
                score += 0.3
            max_score += 0.3
        except Exception:
            pass

        # 2. 类型检查
        try:
            result = subprocess.run(
                ["mypy", "-", "--no-error-summary"],
                input=code,
                capture_output=True,
                text=True
            )
            if result.returncode == 0:
                score += 0.3
            max_score += 0.3
        except Exception:
            pass

        # 3. 测试通过率
        try:
            # 假设测试已经设置
            result = subprocess.run(
                ["pytest", "-q"],
                capture_output=True,
                text=True,
                timeout=30
            )
            if result.returncode == 0:
                score += 0.4
            max_score += 0.4
        except Exception:
            pass

        return score / max_score if max_score > 0 else 0.0

    async def _generate_feedback(self, code: str, quality: float) -> str:
        """生成改进反馈"""
        feedback = []

        # 运行各种检查工具
        checks = {
            "lint": ["ruff", "check", "-"],
            "type": ["mypy", "-"],
        }

        for check_name, cmd in checks.items():
            try:
                result = subprocess.run(
                    cmd,
                    input=code,
                    capture_output=True,
                    text=True
                )
                if result.returncode != 0:
                    feedback.append(f"### {check_name.upper()} 失败:\n{result.stdout}")
            except Exception:
                pass

        return "\n\n".join(feedback)

    def _refine_prompt(self, original_prompt: str, feedback: str) -> str:
        """根据反馈优化提示"""
        return f"""{original_prompt}

上次尝试的反馈:
{feedback}

请根据这些反馈修复问题。"""

    def _record_success(self, prompt: str, output: str):
        """记录成功模式"""
        self.feedback_history.append({
            "prompt": prompt,
            "output": output,
            "success": True
        })

# 使用A2框架
agent = QualityFeedbackAgent(quality_threshold=0.8)

async def code_generator(prompt: str) -> str:
    # 这里调用Claude API
    response = await claude.messages.create(
        model="claude-opus-4-5",
        max_tokens=4000,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

# 生成并迭代优化
success, code = await agent.generate_with_feedback(
    generator=code_generator,
    prompt="写一个Python函数来计算斐波那契数列",
    max_iterations=3
)

if success:
    print("✅ 代码质量达标")
    print(code)
else:
    print("⚠️ 未达到质量阈值，但这是最佳结果")
    print(code)

## 260106 工作日志

### 今日任务
- [x] 任务1: XXX (2小时)
- [x] 任务2: XXX (1.5小时)
- [ ] 任务3: XXX (未完成)

### AI使用统计
- 代码生成: 15次
- 调试: 3次
- 代码审查: 5次
- 文档生成: 2次

### 生产力指标
- PR完成: 2个
- 代码行数: 500 (手动: 50, AI: 450)
- Bug修复: 3个
- 学习时间: 1小时

### 关键发现
1. 发现XX任务用AI可节省80%时间
2. 提示模板Y效果最好
3. 工具Z在XX场景下不适用

### 失败教训
- 案例: XXX
- 原因: 提示不够清晰
- 改进: 添加具体示例

### 明日计划
1. 优化XX提示模板
2. 学习YY新技术
3. 完成ZZ任务

#!/bin/bash
# verify-ai-setup.sh - 验证AI工具配置

echo "🔍 验证AI辅助编程环境..."

# 1. Claude Code
if command -v claude &> /dev/null; then
    echo "✅ Claude Code: $(claude --version)"
else
    echo "❌ Claude Code未安装"
fi

# 2. Git
if command -v git &> /dev/null; then
    echo "✅ Git: $(git --version)"
else
    echo "❌ Git未安装"
fi

# 3. Python工具（如果适用）
if command -v python3 &> /dev/null; then
    echo "✅ Python: $(python3 --version)"

    if command -v ruff &> /dev/null; then
        echo "✅ Ruff: $(ruff --version)"
    fi

    if command -v mypy &> /dev/null; then
        echo "✅ MyPy: $(mypy --version)"
    fi
fi

# 4. Node工具（如果适用）
if command -v node &> /dev/null; then
    echo "✅ Node: $(node --version)"

    if command -v eslint &> /dev/null; then
        echo "✅ ESLint: $(eslint --version)"
    fi
fi

# 5. MCP配置
if [ -f ~/.claude.json ]; then
    echo "✅ MCP配置文件存在"
    # 可以添加更多检查
else
    echo "⚠️ ~/.claude.json不存在"
fi

echo "✨ 验证完成！"