Upload folder using huggingface_hub

README.md

@@ -1,13 +1,40 @@
 ---
-title: CodeAgent MCP
-emoji: 🏃
-colorFrom: green
-colorTo: indigo
+title: CodeAgent-MCP
+emoji: "\U0001F916"
+colorFrom: blue
+colorTo: purple
 sdk: gradio
-sdk_version: 6.14.0
-python_version: '3.13'
+sdk_version: "4.44.1"
 app_file: app.py
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# CodeAgent-MCP
+
+Multi-Agent Code Generation System with MCP Protocol.
+
+**Planner** (task decomposition) -> **Coder** (code generation) -> **Reviewer** (code review) feedback loop.
+
+## How to use
+
+1. Enter your DeepSeek / OpenAI API key
+2. Choose a provider (default = DeepSeek)
+3. Describe the code you want to generate
+4. Click "Start" and watch the multi-agent system work
+
+## Architecture
+
+- **Planner Agent**: Decomposes complex requirements into 2-4 subtasks
+- **Coder Agent**: Generates code with optional MCP tool integration
+- **Reviewer Agent**: Scores code quality (1-10) and provides improvement suggestions
+- **Orchestrator**: Manages Coder-Reviewer feedback loop until quality threshold is met
+
+## Project Series
+
+1. [small-llms-tool-use](https://github.com/XIECHENG6/small-llms-tool-use) - Function calling fine-tuning (86-89% exact match)
+2. [agenttune](https://github.com/XIECHENG6/agenttune) - Multi-step ReAct reasoning (100% task success rate)
+3. [smallrag](https://github.com/XIECHENG6/smallrag) - RAG optimization (chunk_size=512 + MMR + top-k=5)
+4. **CodeAgent-MCP** (this project) - Multi-Agent system integration
+
+[GitHub](https://github.com/XIECHENG6/CodeAgent-MCP)

app.py

@@ -0,0 +1,143 @@
+"""
+CodeAgent-MCP — HuggingFace Spaces Demo (Gradio).
+
+Multi-Agent code generation with Planner → Coder → Reviewer loop.
+"""
+
+import asyncio
+import json
+import os
+import time
+
+import gradio as gr
+
+from src.core.config import load_settings, load_agents_config
+from src.core.llm_client import LLMClient
+from src.core.orchestrator import Orchestrator
+from src.agents import PlannerAgent, CoderAgent, ReviewerAgent
+
+
+EXAMPLE_TASKS = [
+    "实现一个 LRU Cache，支持 get 和 put 操作，要求 O(1) 时间复杂度",
+    "实现一个简单的 Stack 数据结构，支持 push, pop, peek, is_empty 方法",
+    "编写一个配置管理器，支持从 YAML/JSON 加载，支持点号路径访问如 config.get('db.host')",
+    "实现一个令牌桶限流器 TokenBucketRateLimiter，支持 acquire() 和装饰器用法",
+]
+
+
+async def run_agents(requirement: str, api_key: str, provider: str, progress=gr.Progress()):
+    if not api_key.strip():
+        return "请输入 API Key", "", ""
+
+    os.environ["OPENAI_API_KEY"] = api_key.strip()
+
+    settings = load_settings()
+    agents_config = load_agents_config()
+
+    llm = LLMClient.from_settings(provider, settings)
+    planner = PlannerAgent(agents_config["planner"], llm)
+    coder = CoderAgent(agents_config["coder"], llm, mcp_manager=None)
+    reviewer = ReviewerAgent(agents_config["reviewer"], llm)
+    orchestrator = Orchestrator(
+        planner=planner, coder=coder, reviewer=reviewer,
+        config=settings["orchestrator"],
+    )
+
+    progress(0.1, desc="Planner 正在拆解任务...")
+    start = time.time()
+    result = await orchestrator.run(requirement)
+    elapsed = time.time() - start
+
+    code_blocks = []
+    log_lines = []
+
+    log_lines.append(f"**任务拆分**: {len(result.plan)} 个子任务")
+    for i, task in enumerate(result.plan):
+        log_lines.append(f"  T{i+1}: {task['description'][:80]}")
+    log_lines.append("")
+
+    for i, r in enumerate(result.results):
+        score = r["review"]["score"] if r.get("review") else "N/A"
+        status_icon = "✅" if r["status"] == "completed" else "⚠️"
+        log_lines.append(
+            f"{status_icon} **Task {i+1}**: score={score}/10, "
+            f"attempts={r['attempts']}, status={r['status']}"
+        )
+        if r.get("code"):
+            code_blocks.append(r["code"])
+
+    log_lines.append("")
+    log_lines.append(f"**总 Token**: {result.total_tokens:,}")
+    log_lines.append(f"**耗时**: {elapsed:.1f}s")
+
+    completed = sum(1 for r in result.results if r["status"] == "completed")
+    scores = [r["review"]["score"] for r in result.results if r.get("review")]
+    avg_score = sum(scores) / len(scores) if scores else 0
+
+    stats = json.dumps({
+        "completion_rate": f"{completed}/{len(result.results)}",
+        "avg_score": round(avg_score, 1),
+        "total_tokens": result.total_tokens,
+        "elapsed_seconds": round(elapsed, 1),
+    }, indent=2, ensure_ascii=False)
+
+    return "\n\n".join(code_blocks), "\n".join(log_lines), stats
+
+
+def create_demo():
+    with gr.Blocks(title="CodeAgent-MCP", theme=gr.themes.Soft()) as demo:
+        gr.Markdown(
+            "# 🤖 CodeAgent-MCP\n"
+            "**Multi-Agent Code Generation System** — "
+            "Planner (任务拆解) → Coder (代码生成) → Reviewer (代码审查) 反馈循环\n\n"
+            "基于 MCP 协议的多 Agent 协作代码开发系统。"
+        )
+
+        with gr.Row():
+            with gr.Column(scale=1):
+                api_key = gr.Textbox(
+                    label="API Key (DeepSeek / OpenAI)",
+                    type="password",
+                    placeholder="sk-...",
+                )
+                provider = gr.Dropdown(
+                    choices=["default", "siliconflow", "openai"],
+                    value="default",
+                    label="LLM Provider",
+                )
+                requirement = gr.Textbox(
+                    label="开发需求",
+                    placeholder="请描述你想实现的功能...",
+                    lines=3,
+                )
+                examples = gr.Examples(
+                    examples=[[e] for e in EXAMPLE_TASKS],
+                    inputs=[requirement],
+                )
+                run_btn = gr.Button("🚀 开始生成", variant="primary")
+
+            with gr.Column(scale=2):
+                code_output = gr.Markdown(label="生成的代码")
+                with gr.Row():
+                    log_output = gr.Markdown(label="执行日志")
+                    stats_output = gr.Code(label="统计数据", language="json")
+
+        run_btn.click(
+            fn=run_agents,
+            inputs=[requirement, api_key, provider],
+            outputs=[code_output, log_output, stats_output],
+        )
+
+        gr.Markdown(
+            "---\n"
+            "**架构**: 自研 300 行编排器，不依赖 LangChain | "
+            "**项目系列**: small-llms-tool-use → agenttune → smallrag → CodeAgent-MCP\n\n"
+            "[GitHub](https://github.com/XIECHENG6/CodeAgent-MCP)"
+        )
+
+    return demo
+
+
+if __name__ == "__main__":
+    demo = create_demo()
+    demo.launch()

config/agents.yaml

@@ -0,0 +1,73 @@
+planner:
+  name: "Planner Agent"
+  provider: "default"
+  temperature: 0.3
+  system_prompt: |
+    你是一个经验丰富的技术负责人。你的职责是：
+    1. 分析用户的开发需求
+    2. 将需求拆解为具体的、可执行的子任务列表
+    3. 为每个子任务标注依赖关系和执行顺序
+
+    输出格式要求(严格JSON):
+    {
+      "tasks": [
+        {
+          "task_id": "T1",
+          "description": "具体要做什么",
+          "dependencies": []
+        }
+      ]
+    }
+
+    注意：
+    - 每个子任务应该足够具体，能被一个开发者独立完成
+    - 合理安排依赖关系，无依赖的任务排在前面
+    - 不要拆分过细，2-4个子任务为宜
+    - 同一个类/模块的方法应合并为一个任务，不要逐个方法拆分
+    - 典型拆分：T1=核心实现(所有类和方法), T2=测试, T3=集成验证(可选)
+
+coder:
+  name: "Coder Agent"
+  provider: "default"
+  temperature: 0.5
+  max_tool_rounds: 10
+  system_prompt: |
+    你是一个高级Python开发工程师。根据任务描述编写高质量的 Python 代码，包含类型注解，遵循 PEP 8。
+
+    当你有工具可用时，严格按以下顺序操作：
+    第1步：如果 workspace 有已有文件，file_read 需要参考的文件
+    第2步：用 file_write 写入主源码文件（一次调用写完整个文件）
+    第3步：用 file_write 写入测试文件
+    第4步（可选）：用 shell_exec 运行 pytest 验证
+
+    禁止事项：
+    - 不要在空目录调用 file_list 或 file_search
+    - 不要写完文件后再 file_read 自己刚写的文件
+    - 不要多次小片段追加写入，一次 file_write 写完整个文件
+
+    完成后简要说明做了什么。
+
+reviewer:
+  name: "Reviewer Agent"
+  provider: "default"
+  temperature: 0.2
+  system_prompt: |
+    你是一个资深代码审查专家。你的职责是：
+    1. 审查 Coder 生成的代码
+    2. 从以下维度评分(0-10):
+       - 正确性：逻辑是否正确，边界条件是否处理
+       - 可读性：命名、结构是否清晰
+       - 健壮性：异常处理、输入校验是否完善
+       - 测试覆盖：是否有足够的测试
+    3. 给出综合评分和具体改进建议
+
+    输出格式(严格JSON):
+    {
+      "score": 8.5,
+      "passed": true,
+      "issues": ["issue1", "issue2"],
+      "suggestions": ["suggestion1"],
+      "summary": "总体评价"
+    }
+
+    评分标准：>= 7.0 为通过，< 7.0 需要返回给 Coder 修改。

config/mcp_servers.yaml

@@ -0,0 +1,24 @@
+servers:
+  file-server:
+    command: "python"
+    args: ["-m", "src.mcp.servers.file_server"]
+    description: "文件读写与搜索"
+    enabled: true
+
+  shell-server:
+    command: "python"
+    args: ["-m", "src.mcp.servers.shell_server"]
+    description: "Shell 命令执行 (受限安全沙箱)"
+    enabled: true
+
+  git-server:
+    command: "python"
+    args: ["-m", "src.mcp.servers.git_server"]
+    description: "Git 操作 (status/diff/commit/log)"
+    enabled: true
+
+  rag-server:
+    command: "python"
+    args: ["-m", "src.mcp.servers.rag_server"]
+    description: "代码知识库 RAG 检索 (需要 sentence-transformers + faiss)"
+    enabled: false

config/settings.yaml

@@ -0,0 +1,28 @@
+providers:
+  default:
+    base_url: "https://api.deepseek.com"
+    model: "deepseek-chat"
+    temperature: 0.5
+    max_tokens: 4096
+
+  siliconflow:
+    base_url: "https://api.siliconflow.cn/v1"
+    model: "Qwen/Qwen2.5-72B-Instruct"
+    temperature: 0.5
+    max_tokens: 4096
+
+  openai:
+    base_url: "https://api.openai.com/v1"
+    model: "gpt-4o-mini"
+    temperature: 0.5
+    max_tokens: 4096
+
+orchestrator:
+  max_review_rounds: 3
+  review_threshold: 7.0
+  skip_review_for_simple: false
+
+logging:
+  level: "INFO"
+  show_tool_calls: true
+  show_token_usage: true

requirements.txt

@@ -0,0 +1,6 @@
+openai>=1.0
+pydantic>=2.0
+pyyaml>=6.0
+rich>=13.0
+gradio>=4.0
+nest_asyncio

src/agents/__init__.py

@@ -0,0 +1,3 @@
+from .planner import PlannerAgent
+from .coder import CoderAgent
+from .reviewer import ReviewerAgent

src/agents/coder.py

@@ -0,0 +1,31 @@
+from ..core.agent_base import AgentBase
+from ..core.llm_client import LLMClient
+
+
+class CoderAgent(AgentBase):
+    def __init__(self, config: dict, llm_client: LLMClient, mcp_manager=None):
+        super().__init__(config, llm_client, mcp_manager)
+        self.workspace_files: list[str] = []
+
+    def set_workspace_files(self, files: list[str]):
+        self.workspace_files = files
+
+    def format_input(self, task) -> str:
+        if isinstance(task, str):
+            prompt = task
+        elif isinstance(task, dict):
+            desc = task.get("description", str(task))
+            deps = task.get("dependencies", [])
+            prompt = f"请完成以下任务:\n{desc}"
+            if deps:
+                prompt += f"\n\n依赖的前置任务: {', '.join(deps)}"
+        else:
+            prompt = str(task)
+
+        if self.mcp and self.workspace_files:
+            prompt += f"\n\nWorkspace 已有文件: {', '.join(self.workspace_files)}"
+            prompt += "\n可以用 file_read 读取已有文件作为参考，然后直接 file_write 写入新文件。"
+        elif self.mcp:
+            prompt += "\n\nWorkspace 为空，请直接用 file_write 写入代码文件，不需要先 file_list。"
+
+        return prompt

src/agents/planner.py

@@ -0,0 +1,12 @@
+from ..core.agent_base import AgentBase
+from ..core.llm_client import LLMClient
+
+
+class PlannerAgent(AgentBase):
+    def __init__(self, config: dict, llm_client: LLMClient):
+        super().__init__(config, llm_client, mcp_manager=None)
+
+    def format_input(self, task) -> str:
+        if isinstance(task, str):
+            return task
+        return f"请分析以下开发需求并拆解为子任务：\n\n{task}"

src/agents/reviewer.py

@@ -0,0 +1,19 @@
+from ..core.agent_base import AgentBase
+from ..core.llm_client import LLMClient
+
+
+class ReviewerAgent(AgentBase):
+    def __init__(self, config: dict, llm_client: LLMClient):
+        super().__init__(config, llm_client, mcp_manager=None)
+
+    def format_input(self, task) -> str:
+        if isinstance(task, dict) and "task" in task and "code" in task:
+            task_info = task["task"]
+            code = task["code"]
+            desc = task_info.get("description", str(task_info)) if isinstance(task_info, dict) else str(task_info)
+            return (
+                f"任务需求: {desc}\n\n"
+                f"Coder 的输出:\n{code}\n\n"
+                f"请审查代码质量并给出评分。"
+            )
+        return f"请审查以下代码:\n\n{task}"

src/core/__init__.py

@@ -0,0 +1,3 @@
+from .message import Role, AgentType, Message, TaskItem, ReviewResult
+from .llm_client import LLMClient
+from .orchestrator import Orchestrator

src/core/agent_base.py

@@ -0,0 +1,88 @@
+import json
+import logging
+from abc import ABC, abstractmethod
+
+from .llm_client import LLMClient
+
+logger = logging.getLogger(__name__)
+
+
+class AgentBase(ABC):
+    def __init__(self, config: dict, llm_client: LLMClient, mcp_manager=None):
+        self.name = config["name"]
+        self.system_prompt = config["system_prompt"]
+        self.llm = llm_client
+        self.mcp = mcp_manager
+        self.max_tool_rounds = config.get("max_tool_rounds", 5)
+        self.max_tool_result_chars = config.get("max_tool_result_chars", 8000)
+        self.conversation: list[dict] = []
+        self.total_tokens_used = 0
+
+    async def run(self, user_input: str) -> str:
+        self.conversation = [
+            {"role": "system", "content": self.system_prompt},
+            {"role": "user", "content": user_input},
+        ]
+
+        tools = self.mcp.get_openai_tools() if self.mcp else None
+
+        for round_idx in range(self.max_tool_rounds):
+            response = await self.llm.chat(
+                messages=self.conversation,
+                tools=tools,
+            )
+            self.total_tokens_used += (
+                response["usage"]["prompt_tokens"] + response["usage"]["completion_tokens"]
+            )
+
+            if not response["tool_calls"]:
+                return response["content"]
+
+            assistant_msg = {"role": "assistant", "content": response["content"]}
+            assistant_msg["tool_calls"] = [
+                {
+                    "id": tc["id"],
+                    "type": "function",
+                    "function": {
+                        "name": tc["function"],
+                        "arguments": json.dumps(tc["arguments"], ensure_ascii=False),
+                    },
+                }
+                for tc in response["tool_calls"]
+            ]
+            self.conversation.append(assistant_msg)
+
+            for tc in response["tool_calls"]:
+                tool_result = await self._execute_tool(tc["function"], tc["arguments"])
+                self.conversation.append({
+                    "role": "tool",
+                    "tool_call_id": tc["id"],
+                    "content": tool_result,
+                })
+
+            logger.info(f"[{self.name}] Round {round_idx + 1}: "
+                        f"called {[tc['function'] for tc in response['tool_calls']]}")
+
+        self.conversation.append({
+            "role": "user",
+            "content": "你已达到最大工具调用轮次，请基于当前进度给出最终回答。",
+        })
+        response = await self.llm.chat(messages=self.conversation)
+        return response["content"]
+
+    async def _execute_tool(self, tool_name: str, arguments: dict) -> str:
+        if not self.mcp:
+            return f"Error: No MCP manager available to execute tool '{tool_name}'"
+        try:
+            result = await self.mcp.call_tool(tool_name, arguments)
+            text = str(result)
+            if len(text) > self.max_tool_result_chars:
+                text = text[:self.max_tool_result_chars] + f"\n... (truncated, {len(str(result))} chars total)"
+            return text
+        except Exception as e:
+            logger.error(f"[{self.name}] Tool '{tool_name}' failed: {e}")
+            return f"工具调用失败: {type(e).__name__}: {str(e)}"
+
+    @abstractmethod
+    def format_input(self, task) -> str:
+        pass

src/core/config.py

@@ -0,0 +1,27 @@
+import os
+from pathlib import Path
+import yaml
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+CONFIG_DIR = PROJECT_ROOT / "config"
+
+
+def load_yaml(path: Path) -> dict:
+    with open(path, "r", encoding="utf-8") as f:
+        return yaml.safe_load(f)
+
+
+def load_settings() -> dict:
+    return load_yaml(CONFIG_DIR / "settings.yaml")
+
+
+def load_agents_config() -> dict:
+    return load_yaml(CONFIG_DIR / "agents.yaml")
+
+
+def load_mcp_config() -> dict:
+    return load_yaml(CONFIG_DIR / "mcp_servers.yaml")
+
+
+def get_project_root() -> Path:
+    return PROJECT_ROOT

src/core/llm_client.py

@@ -0,0 +1,83 @@
+import os
+import json
+import logging
+from openai import AsyncOpenAI
+
+logger = logging.getLogger(__name__)
+
+
+class LLMClient:
+    def __init__(self, config: dict):
+        self.client = AsyncOpenAI(
+            api_key=config.get("api_key") or os.getenv("OPENAI_API_KEY"),
+            base_url=config.get("base_url", "https://api.deepseek.com"),
+        )
+        self.model = config.get("model", "deepseek-chat")
+        self.temperature = config.get("temperature", 0.5)
+        self.max_tokens = config.get("max_tokens", 4096)
+
+    @classmethod
+    def from_settings(cls, provider: str = "default", settings: dict | None = None) -> "LLMClient":
+        if settings is None:
+            from .config import load_settings
+            settings = load_settings()
+        provider_config = settings["providers"][provider]
+        return cls(provider_config)
+
+    async def chat(
+        self,
+        messages: list[dict],
+        tools: list[dict] | None = None,
+        tool_choice: str = "auto",
+        temperature: float | None = None,
+    ) -> dict:
+        kwargs = {
+            "model": self.model,
+            "messages": messages,
+            "temperature": temperature or self.temperature,
+            "max_tokens": self.max_tokens,
+        }
+        if tools:
+            kwargs["tools"] = tools
+            kwargs["tool_choice"] = tool_choice
+
+        response = await self.client.chat.completions.create(**kwargs)
+        choice = response.choices[0]
+
+        result = {
+            "content": choice.message.content or "",
+            "tool_calls": [],
+            "usage": {
+                "prompt_tokens": response.usage.prompt_tokens if response.usage else 0,
+                "completion_tokens": response.usage.completion_tokens if response.usage else 0,
+            },
+        }
+
+        if choice.message.tool_calls:
+            for tc in choice.message.tool_calls:
+                arguments = self._safe_parse_arguments(tc.function.arguments)
+                result["tool_calls"].append({
+                    "id": tc.id,
+                    "function": tc.function.name,
+                    "arguments": arguments,
+                })
+
+        logger.debug(
+            f"LLM call: model={self.model}, "
+            f"tokens={result['usage']['prompt_tokens']}+{result['usage']['completion_tokens']}"
+        )
+        return result
+
+    def _safe_parse_arguments(self, raw: str) -> dict:
+        try:
+            return json.loads(raw)
+        except json.JSONDecodeError:
+            import re
+            match = re.search(r'\{.*\}', raw, re.DOTALL)
+            if match:
+                try:
+                    return json.loads(match.group())
+                except json.JSONDecodeError:
+                    pass
+            logger.warning(f"Failed to parse tool arguments: {raw[:200]}")
+            return {"_raw": raw}

src/core/message.py

@@ -0,0 +1,42 @@
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Optional
+
+
+class Role(Enum):
+    SYSTEM = "system"
+    USER = "user"
+    ASSISTANT = "assistant"
+    TOOL = "tool"
+
+
+class AgentType(Enum):
+    PLANNER = "planner"
+    CODER = "coder"
+    REVIEWER = "reviewer"
+
+
+@dataclass
+class Message:
+    role: Role
+    content: str
+    tool_call_id: Optional[str] = None
+    name: Optional[str] = None
+
+
+@dataclass
+class TaskItem:
+    task_id: str
+    description: str
+    status: str = "pending"
+    dependencies: list[str] = field(default_factory=list)
+    result: Optional[str] = None
+
+
+@dataclass
+class ReviewResult:
+    score: float
+    passed: bool
+    issues: list[str] = field(default_factory=list)
+    suggestions: list[str] = field(default_factory=list)
+    summary: str = ""

src/core/orchestrator.py

@@ -0,0 +1,193 @@
+import json
+import logging
+import re
+from dataclasses import dataclass, field
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ExecutionResult:
+    plan: list[dict]
+    results: list[dict]
+    execution_log: list[dict]
+    total_tokens: int = 0
+
+
+class Orchestrator:
+    def __init__(self, planner, coder, reviewer, config: dict):
+        self.planner = planner
+        self.coder = coder
+        self.reviewer = reviewer
+        self.max_review_rounds = config.get("max_review_rounds", 3)
+        self.review_threshold = config.get("review_threshold", 7.0)
+        self.execution_log: list[dict] = []
+
+    async def run(self, user_requirement: str) -> ExecutionResult:
+        self.execution_log = []
+
+        plan_output = await self.planner.run(user_requirement)
+        tasks = self._parse_plan(plan_output)
+        self._log("plan", {"raw_output": plan_output, "parsed_tasks": tasks})
+
+        logger.info(f"[Orchestrator] Plan: {len(tasks)} tasks")
+
+        results = []
+        for i, task in enumerate(tasks):
+            logger.info(f"[Orchestrator] Executing task {i+1}/{len(tasks)}: {task['description'][:60]}")
+            task_result = await self._execute_task(task)
+            results.append(task_result)
+
+        total_tokens = (
+            self.planner.total_tokens_used
+            + self.coder.total_tokens_used
+            + self.reviewer.total_tokens_used
+        )
+
+        return ExecutionResult(
+            plan=tasks,
+            results=results,
+            execution_log=self.execution_log,
+            total_tokens=total_tokens,
+        )
+
+    async def _execute_task(self, task: dict) -> dict:
+        code_output = None
+        review = None
+
+        for attempt in range(self.max_review_rounds):
+            await self._sync_workspace_files()
+
+            if attempt == 0:
+                coder_input = self.coder.format_input(task)
+            else:
+                coder_input = self.coder.format_input(task)
+                coder_input += (
+                    f"\n\n--- Reviewer 反馈 (得分: {review['score']}/10) ---\n"
+                    f"问题: {json.dumps(review['issues'], ensure_ascii=False)}\n"
+                    f"建议: {json.dumps(review['suggestions'], ensure_ascii=False)}\n"
+                    f"请根据反馈修改代码。如果 workspace 已有文件，用 file_read 读取后修改再 file_write 写回。"
+                )
+
+            code_output = await self.coder.run(coder_input)
+            self._log("coder", {"task_id": task.get("task_id"), "attempt": attempt, "output_preview": code_output[:500]})
+
+            reviewer_input = self.reviewer.format_input({
+                "task": task,
+                "code": code_output,
+            })
+            review_raw = await self.reviewer.run(reviewer_input)
+            review = self._parse_review(review_raw)
+            self._log("reviewer", {"task_id": task.get("task_id"), "attempt": attempt, "review": review})
+
+            logger.info(f"  [Review] Attempt {attempt+1}: score={review['score']}, passed={review['passed']}")
+
+            if review["passed"]:
+                return {
+                    "task": task,
+                    "code": code_output,
+                    "review": review,
+                    "attempts": attempt + 1,
+                    "status": "completed",
+                }
+
+        return {
+            "task": task,
+            "code": code_output,
+            "review": review,
+            "attempts": self.max_review_rounds,
+            "status": "max_attempts_reached",
+        }
+
+    def _parse_plan(self, plan_text: str) -> list[dict]:
+        parsed = self._extract_json(plan_text)
+        if parsed and "tasks" in parsed:
+            return parsed["tasks"]
+
+        tasks = []
+        lines = plan_text.strip().split("\n")
+        for i, line in enumerate(lines):
+            line = line.strip()
+            if re.match(r'^[\d]+[.)\-]', line):
+                desc = re.sub(r'^[\d]+[.)\-]\s*', '', line)
+                tasks.append({"task_id": f"T{i+1}", "description": desc, "dependencies": []})
+
+        if not tasks:
+            tasks = [{"task_id": "T1", "description": plan_text, "dependencies": []}]
+
+        return tasks
+
+    def _parse_review(self, review_text: str) -> dict:
+        parsed = self._extract_json(review_text)
+        if parsed and "score" in parsed:
+            parsed.setdefault("passed", parsed["score"] >= self.review_threshold)
+            parsed.setdefault("issues", [])
+            parsed.setdefault("suggestions", [])
+            parsed.setdefault("summary", "")
+            return parsed
+
+        score_match = re.search(r'(\d+\.?\d*)\s*/\s*10', review_text)
+        score = float(score_match.group(1)) if score_match else 5.0
+
+        return {
+            "score": score,
+            "passed": score >= self.review_threshold,
+            "issues": [],
+            "suggestions": [],
+            "summary": review_text[:200],
+        }
+
+    def _extract_json(self, text: str) -> dict | None:
+        if "```json" in text:
+            match = re.search(r'```json\s*(.*?)```', text, re.DOTALL)
+            if match:
+                try:
+                    return json.loads(match.group(1).strip())
+                except json.JSONDecodeError:
+                    pass
+        if "```" in text:
+            match = re.search(r'```\s*(.*?)```', text, re.DOTALL)
+            if match:
+                try:
+                    return json.loads(match.group(1).strip())
+                except json.JSONDecodeError:
+                    pass
+        try:
+            return json.loads(text)
+        except json.JSONDecodeError:
+            pass
+        match = re.search(r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}', text, re.DOTALL)
+        if match:
+            try:
+                return json.loads(match.group())
+            except json.JSONDecodeError:
+                pass
+        return None
+
+    async def _sync_workspace_files(self):
+        if not hasattr(self.coder, 'set_workspace_files'):
+            return
+        if not self.coder.mcp:
+            return
+        try:
+            result = await self.coder.mcp.call_tool("file_list", {"directory": "."})
+            text = str(result)
+            if text == "(empty directory)" or text.startswith("Error"):
+                self.coder.set_workspace_files([])
+                return
+            files = []
+            for line in text.split('\n'):
+                line = line.strip()
+                if not line or line.endswith('/'):
+                    continue
+                name = re.sub(r'\s*\(\d+B\)\s*$', '', line)
+                if name:
+                    files.append(name)
+            self.coder.set_workspace_files(files)
+            logger.info(f"[Orchestrator] Workspace files: {files}")
+        except Exception as e:
+            logger.debug(f"[Orchestrator] Could not list workspace: {e}")
+            self.coder.set_workspace_files([])
+
+    def _log(self, stage: str, data: dict):
+        self.execution_log.append({"stage": stage, **data})

src/utils/logger.py

@@ -0,0 +1,18 @@
+import logging
+import sys
+
+from rich.console import Console
+from rich.logging import RichHandler
+
+console = Console()
+
+
+def setup_logging(level: str = "INFO"):
+    logging.basicConfig(
+        level=getattr(logging, level.upper(), logging.INFO),
+        format="%(message)s",
+        datefmt="[%X]",
+        handlers=[RichHandler(console=console, rich_tracebacks=True, show_path=False)],
+    )
+    logging.getLogger("httpx").setLevel(logging.WARNING)
+    logging.getLogger("openai").setLevel(logging.WARNING)