Kanban 多 Agent 协作：Hermes Kanban 板 + 多个 Profile Agent 并行处理

┌─────────────────────────────────────────────────────────────┐
│                   人类敏捷团队 vs Agent 团队                   │
│                                                             │
│  人类团队:                                                     │
│  ┌────────┐  ┌────────┐  ┌────────┐  ┌────────┐            │
│  │ PM     │  │ 前端    │  │ 后端    │  │ 测试    │            │
│  │ 拆解任务│  │ 写UI    │  │ 写API  │  │ 写用例  │            │
│  └────────┘  └────────┘  └────────┘  └────────┘            │
│                                                             │
│  Agent 团队:                                                  │
│  ┌──────────────────────────────────────────────────────┐   │
│  │                  Hermes (Scrum Master)                │   │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌─────┐  │   │
│  │  │ Kanban板 │  │ 任务分配器│  │ 进度监控器│  │汇总器│  │   │
│  │  └──────────┘  └──────────┘  └──────────┘  └─────┘  │   │
│  └──────────────────────────────────────────────────────┘   │
│       │              │              │              │         │
│  ┌────┴────┐   ┌─────┴─────┐  ┌────┴────┐  ┌────┴────┐    │
│  │PM Agent │   │Frontend   │  │Backend  │  │Test     │    │
│  │分析需求  │   │Agent      │  │Agent    │  │Agent    │    │
│  └─────────┘   └───────────┘  └─────────┘  └─────────┘    │
└─────────────────────────────────────────────────────────────┘

# hermes/kanban/models.py
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
from datetime import datetime
import uuid

class TaskStatus(str, Enum):
    BACKLOG = "backlog"         # 待办
    TODO = "todo"              # 待处理
    IN_PROGRESS = "in_progress" # 进行中
    REVIEW = "review"          # 审查中
    DONE = "done"              # 已完成
    BLOCKED = "blocked"        # 阻塞

class TaskPriority(str, Enum):
    CRITICAL = "P0"
    HIGH = "P1"
    MEDIUM = "P2"
    LOW = "P3"

class AgentProfile(str, Enum):
    PM = "pm-analyst"
    ARCHITECT = "architect"
    FRONTEND = "frontend-dev"
    BACKEND = "backend-dev"
    QA = "qa-engineer"
    DEVOPS = "devops-engineer"
    WRITER = "tech-writer"

@dataclass
class KanbanCard:
    """看板卡片"""
    card_id: str = field(default_factory=lambda: f"CARD-{uuid.uuid4().hex[:8]}")
    title: str = ""
    description: str = ""
    status: TaskStatus = TaskStatus.BACKLOG
    priority: TaskPriority = TaskPriority.MEDIUM
    assigned_to: Optional[AgentProfile] = None
    parent_card: Optional[str] = None
    sub_cards: list[str] = field(default_factory=list)
    dependencies: list[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.utcnow)
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    output: str = ""
    output_artifacts: list[str] = field(default_factory=list)
    effort_estimate: int = 0  # 故事点
    actual_effort: int = 0
    retry_count: int = 0
    max_retries: int = 3
    tags: list[str] = field(default_factory=list)
    metadata: dict = field(default_factory=dict)

    def to_dict(self) -> dict:
        return {
            "card_id": self.card_id,
            "title": self.title,
            "description": self.description,
            "status": self.status.value,
            "priority": self.priority.value,
            "assigned_to": self.assigned_to.value if self.assigned_to else None,
            "parent_card": self.parent_card,
            "sub_cards": self.sub_cards,
            "dependencies": self.dependencies,
            "created_at": self.created_at.isoformat(),
            "started_at": self.started_at.isoformat() if self.started_at else None,
            "completed_at": self.completed_at.isoformat() if self.completed_at else None,
            "output": self.output[:500] if self.output else "",
            "output_artifacts": self.output_artifacts,
            "effort_estimate": self.effort_estimate,
            "actual_effort": self.actual_effort,
            "retry_count": self.retry_count,
            "tags": self.tags,
        }

@dataclass
class KanbanBoard:
    """看板"""
    board_id: str = field(default_factory=lambda: f"BOARD-{uuid.uuid4().hex[:6]}")
    name: str = ""
    description: str = ""
    cards: dict[str, KanbanCard] = field(default_factory=dict)
    wip_limits: dict[TaskStatus, int] = field(default_factory=lambda: {
        TaskStatus.IN_PROGRESS: 5,
        TaskStatus.REVIEW: 3,
    })
    created_at: datetime = field(default_factory=datetime.utcnow)

    @property
    def columns(self) -> dict[TaskStatus, list[KanbanCard]]:
        """按状态分列"""
        cols = {status: [] for status in TaskStatus}
        for card in self.cards.values():
            cols[card.status].append(card)
        # 每列按优先级排序
        for status in cols:
            cols[status].sort(key=lambda c: TaskPriority[c.priority])
        return cols

    @property
    def is_complete(self) -> bool:
        """所有任务是否完成"""
        return all(
            c.status == TaskStatus.DONE
            for c in self.cards.values()
        )

    @property
    def progress(self) -> float:
        """完成进度"""
        if not self.cards:
            return 0.0
        done = sum(1 for c in self.cards.values() if c.status == TaskStatus.DONE)
        return done / len(self.cards) * 100

# hermes/kanban/manager.py
import asyncio
import json
from typing import Optional, Callable

class KanbanManager:
    """Hermes Kanban 看板管理器"""

    def __init__(self, name: str = "Agent Sprint"):
        self.board = KanbanBoard(name=name)
        self.agent_handlers: dict[AgentProfile, Callable] = {}
        self.event_log: list[dict] = []
        self._subscribers: list[Callable] = []

    def register_agent(self, profile: AgentProfile, handler: Callable):
        """注册 Agent 处理器"""
        self.agent_handlers[profile] = handler
        self._log_event("agent_registered", {
            "profile": profile.value,
            "handler": handler.__name__,
        })

    def add_card(self, card: KanbanCard) -> str:
        """添加卡片到看板"""
        self.board.cards[card.card_id] = card
        self._log_event("card_added", card.to_dict())
        return card.card_id

    def move_card(self, card_id: str, new_status: TaskStatus):
        """移动卡片到新状态"""
        if card_id not in self.board.cards:
            raise ValueError(f"卡片 {card_id} 不存在")

        card = self.board.cards[card_id]

        # 检查 WIP 限制
        if new_status in self.board.wip_limits:
            current_count = sum(
                1 for c in self.board.cards.values()
                if c.status == new_status
            )
            if current_count >= self.board.wip_limits[new_status]:
                self._log_event("wip_limit_reached", {
                    "status": new_status.value,
                    "limit": self.board.wip_limits[new_status],
                })
                return False

        old_status = card.status
        card.status = new_status

        if new_status == TaskStatus.IN_PROGRESS:
            card.started_at = datetime.utcnow()
        elif new_status == TaskStatus.DONE:
            card.completed_at = datetime.utcnow()

        self._log_event("card_moved", {
            "card_id": card_id,
            "from": old_status.value,
            "to": new_status.value,
        })

        # 通知订阅者
        self._notify_subscribers("card_moved", card)
        return True

    def get_ready_tasks(self, profile: AgentProfile) -> list[KanbanCard]:
        """获取某个 Agent 可以处理的任务"""
        ready = []
        for card in self.board.cards.values():
            # 状态为 TODO 且分配给该 Agent
            if card.status == TaskStatus.TODO and card.assigned_to == profile:
                # 检查依赖
                deps_met = all(
                    self.board.cards[dep_id].status == TaskStatus.DONE
                    for dep_id in card.dependencies
                    if dep_id in self.board.cards
                )
                if deps_met:
                    ready.append(card)
        return ready

    async def execute_card(self, card: KanbanCard) -> KanbanCard:
        """执行卡片任务"""
        if card.assigned_to not in self.agent_handlers:
            raise ValueError(f"Agent {card.assigned_to} 未注册")

        handler = self.agent_handlers[card.assigned_to]
        self.move_card(card.card_id, TaskStatus.IN_PROGRESS)

        try:
            # 调用 Agent 处理器
            result = await handler(card)

            card.output = result.get("output", "")
            card.output_artifacts = result.get("artifacts", [])
            card.actual_effort = result.get("effort", 0)

            # 移动到审查
            self.move_card(card.card_id, TaskStatus.REVIEW)
            self._log_event("card_executed", card.to_dict())

            return card

        except Exception as e:
            card.retry_count += 1
            if card.retry_count >= card.max_retries:
                self.move_card(card.card_id, TaskStatus.BLOCKED)
                self._log_event("card_blocked", {
                    "card_id": card.card_id,
                    "error": str(e),
                    "retries": card.retry_count,
                })
            else:
                self.move_card(card.card_id, TaskStatus.TODO)
            raise

    def display_board(self) -> str:
        """生成看板文本视图"""
        lines = [
            f"╔══════════════════════════════════════════════════════╗",
            f"║  {self.board.name:^52}  ║",
            f"║  进度: {self.board.progress:.0f}%  "
            f"卡片: {len(self.board.cards)}                       ║",
            f"╠══════════════════════════════════════════════════════╣",
        ]

        status_names = {
            TaskStatus.BACKLOG: "📋 待办",
            TaskStatus.TODO: "📝 待处理",
            TaskStatus.IN_PROGRESS: "🔄 进行中",
            TaskStatus.REVIEW: "🔍 审查中",
            TaskStatus.DONE: "✅ 已完成",
            TaskStatus.BLOCKED: "🚫 阻塞",
        }

        columns = self.board.columns
        max_cards = max(len(cards) for cards in columns.values()) or 1

        for row in range(max_cards + 2):
            line = "║"
            for status in TaskStatus:
                cards = columns[status]
                width = 14
                if row == 0:
                    line += f" {status_names[status]:^{width}} │"
                elif row == 1:
                    limit = self.board.wip_limits.get(status, "∞")
                    line += f" {'─' * width} │"
                elif row - 2 < len(cards):
                    card = cards[row - 2]
                    priority_icon = {"P0": "🔴", "P1": "🟠", "P2": "🟡", "P3": "🟢"}
                    icon = priority_icon.get(card.priority, "⚪")
                    label = f"{icon} {card.title[:10]}"
                    line += f" {label:<{width}} │"
                else:
                    line += f" {' ' * width} │"
            lines.append(line)

        lines.append("╚══════════════════════════════════════════════════════╝")
        return "\n".join(lines)

    def _log_event(self, event_type: str, details: dict):
        """记录事件"""
        self.event_log.append({
            "timestamp": datetime.utcnow().isoformat(),
            "event": event_type,
            "details": details,
        })

    def _notify_subscribers(self, event: str, card: KanbanCard):
        """通知订阅者"""
        for sub in self._subscribers:
            try:
                sub(event, card)
            except Exception:
                pass

    def subscribe(self, callback: Callable):
        """订阅看板事件"""
        self._subscribers.append(callback)

    def export_board(self) -> dict:
        """导出看板数据"""
        return {
            "board_id": self.board.board_id,
            "name": self.board.name,
            "created_at": self.board.created_at.isoformat(),
            "progress": self.board.progress,
            "cards": {cid: c.to_dict() for cid, c in self.board.cards.items()},
            "event_log": self.event_log[-50:],  # 最近 50 条
        }

# hermes/kanban/profiles/base.py
from abc import ABC, abstractmethod
from typing import Any

class AgentProfile(ABC):
    """Agent Profile 基类"""

    def __init__(self, name: str, tools: list[str] = None):
        self.name = name
        self.tools = tools or []
        self.system_prompt = self._build_system_prompt()

    @abstractmethod
    def _build_system_prompt(self) -> str:
        """构建系统提示词"""
        pass

    @abstractmethod
    async def handle_task(self, card: KanbanCard) -> dict:
        """处理看板任务"""
        pass


class PMAgent(AgentProfile):
    """产品经理 Agent"""

    def _build_system_prompt(self) -> str:
        return """你是一位资深产品经理，擅长需求分析和任务拆解。
你的职责：
1. 理解用户需求，明确功能边界
2. 将大需求拆解为可执行的子任务
3. 评估每个任务的优先级和复杂度
4. 定义验收标准

输出格式：JSON，包含 tasks 列表，每个任务有 title、description、priority、effort、assignee。"""

    async def handle_task(self, card: KanbanCard) -> dict:
        # 模拟需求分析
        tasks = self._breakdown_requirement(card.description)
        return {
            "output": json.dumps(tasks, ensure_ascii=False, indent=2),
            "artifacts": ["requirements.md", "task_breakdown.json"],
            "effort": len(tasks),
            "sub_tasks": tasks,
        }

    def _breakdown_requirement(self, requirement: str) -> list[dict]:
        """拆解需求为子任务"""
        # 实际实现中会调用 LLM
        return [
            {"title": "需求文档", "desc": "编写详细需求文档",
             "priority": "P0", "effort": 3, "assignee": "pm-analyst"},
            {"title": "技术方案", "desc": "设计系统架构和接口",
             "priority": "P0", "effort": 5, "assignee": "architect"},
            {"title": "数据库设计", "desc": "设计数据库 Schema",
             "priority": "P1", "effort": 3, "assignee": "backend-dev"},
            {"title": "前端页面", "desc": "实现用户界面",
             "priority": "P1", "effort": 8, "assignee": "frontend-dev"},
            {"title": "后端 API", "desc": "实现业务逻辑接口",
             "priority": "P1", "effort": 8, "assignee": "backend-dev"},
            {"title": "测试用例", "desc": "编写测试用例并执行",
             "priority": "P2", "effort": 5, "assignee": "qa-engineer"},
            {"title": "部署配置", "desc": "配置 CI/CD 和部署环境",
             "priority": "P2", "effort": 3, "assignee": "devops-engineer"},
            {"title": "文档编写", "desc": "编写 API 文档和用户手册",
             "priority": "P3", "effort": 3, "assignee": "tech-writer"},
        ]


class ArchitectAgent(AgentProfile):
    """架构师 Agent"""

    def _build_system_prompt(self) -> str:
        return """你是一位资深系统架构师，擅长技术方案设计。
你的职责：
1. 根据需求设计系统架构
2. 定义 API 接口和数据结构
3. 选择合适的技术栈
4. 输出架构图和技术文档"""

    async def handle_task(self, card: KanbanCard) -> dict:
        return {
            "output": "架构设计完成：微服务架构，3 个核心服务",
            "artifacts": ["architecture.md", "api_spec.yaml", "db_schema.sql"],
            "effort": 5,
        }


class FrontendAgent(AgentProfile):
    """前端开发 Agent"""

    def _build_system_prompt(self) -> str:
        return """你是一位资深前端开发工程师，擅长 React/Vue 开发。
你的职责：
1. 根据设计稿实现页面
2. 编写可复用的组件
3. 确保响应式设计和可访问性
4. 编写单元测试"""

    async def handle_task(self, card: KanbanCard) -> dict:
        return {
            "output": "前端页面开发完成：登录页、仪表盘、设置页",
            "artifacts": ["src/App.tsx", "src/components/", "src/pages/"],
            "effort": 8,
        }


class BackendAgent(AgentProfile):
    """后端开发 Agent"""

    def _build_system_prompt(self) -> str:
        return """你是一位资深后端开发工程师，擅长 Python/FastAPI 开发。
你的职责：
1. 根据 API 规范实现接口
2. 编写数据库模型和迁移
3. 实现业务逻辑和权限控制
4. 编写集成测试"""

    async def handle_task(self, card: KanbanCard) -> dict:
        return {
            "output": "后端 API 开发完成：12 个端点，数据库迁移",
            "artifacts": ["app/api/", "app/models/", "migrations/"],
            "effort": 8,
        }


class QAAgent(AgentProfile):
    """测试工程师 Agent"""

    def _build_system_prompt(self) -> str:
        return """你是一位资深 QA 工程师，擅长自动化测试。
你的职责：
1. 编写测试用例
2. 执行单元测试、集成测试、E2E 测试
3. 生成测试报告
4. 发现并记录 Bug"""

    async def handle_task(self, card: KanbanCard) -> dict:
        return {
            "output": "测试完成：通过 45/48 用例，3 个 Bug 已修复",
            "artifacts": ["tests/", "coverage_report.html"],
            "effort": 5,
        }


class DevOpsAgent(AgentProfile):
    """运维工程师 Agent"""

    def _build_system_prompt(self) -> str:
        return """你是一位资深 DevOps 工程师，擅长容器化和 CI/CD。
你的职责：
1. 编写 Dockerfile 和 docker-compose
2. 配置 CI/CD 流水线
3. 设置监控和告警
4. 编写部署文档"""

    async def handle_task(self, card: KanbanCard) -> dict:
        return {
            "output": "部署配置完成：Docker + K8s + CI/CD",
            "artifacts": ["Dockerfile", "docker-compose.yml", ".github/workflows/"],
            "effort": 3,
        }


class TechWriterAgent(AgentProfile):
    """技术文档工程师 Agent"""

    def _build_system_prompt(self) -> str:
        return """你是一位资深技术文档工程师。
你的职责：
1. 编写 API 文档
2. 编写用户使用手册
3. 编写部署指南
4. 维护 Changelog"""

    async def handle_task(self, card: KanbanCard) -> dict:
        return {
            "output": "文档编写完成：API 文档、用户手册、部署指南",
            "artifacts": ["docs/api.md", "docs/user-guide.md", "CHANGELOG.md"],
            "effort": 3,
        }

# hermes/kanban/engine.py
import asyncio
from datetime import datetime

class KanbanEngine:
    """Kanban 协作引擎"""

    def __init__(self, manager: KanbanManager):
        self.manager = manager
        self.agents: dict[AgentProfile, AgentProfile] = {}

    def register_agent(self, profile: AgentProfile, handler):
        """注册 Agent"""
        self.agents[profile] = handler
        self.manager.register_agent(profile, handler)

    async def run_sprint(self) -> dict:
        """运行一个 Sprint"""
        print(f"\n{'='*60}")
        print(f"🚀 Sprint 开始: {self.manager.board.name}")
        print(f"{'='*60}\n")

        iteration = 0
        max_iterations = 50  # 防止无限循环

        while not self.manager.board.is_complete and iteration < max_iterations:
            iteration += 1
            print(f"\n--- 迭代 {iteration} ---")
            print(self.manager.display_board())

            # 获取所有可执行的任务
            tasks_to_run = []
            for profile in AgentProfile:
                if profile in self.agents:
                    ready = self.manager.get_ready_tasks(profile)
                    tasks_to_run.extend(ready)

            if not tasks_to_run:
                # 检查是否全部完成
                if self.manager.board.is_complete:
                    break
                # 检查是否有阻塞
                blocked = [
                    c for c in self.manager.board.cards.values()
                    if c.status == TaskStatus.BLOCKED
                ]
                if blocked:
                    print(f"\n⚠️  阻塞任务: {[c.card_id for c in blocked]}")
                    # 尝试解除阻塞
                    for card in blocked:
                        self.manager.move_card(card.card_id, TaskStatus.TODO)
                        card.retry_count = 0
                    continue
                # 还有未完成但不在 TODO 状态的任务
                pending = [
                    c for c in self.manager.board.cards.values()
                    if c.status not in (TaskStatus.DONE, TaskStatus.TODO)
                ]
                if pending:
                    await asyncio.sleep(0.5)  # 等待进行中的任务完成
                    continue
                break

            # 并行执行任务
            coroutines = []
            for card in tasks_to_run:
                handler = self.agents[card.assigned_to]
                coroutines.append(self.manager.execute_card(card))

            if coroutines:
                results = await asyncio.gather(*coroutines, return_exceptions=True)

                # 处理审查 -> 自动通过
                for card, result in zip(tasks_to_run, results):
                    if isinstance(result, KanbanCard):
                        self.manager.move_card(card.card_id, TaskStatus.DONE)
                        print(f"  ✅ {card.title} 已完成")
                    elif isinstance(result, Exception):
                        print(f"  ❌ {card.title} 执行失败: {result}")

        # 生成 Sprint 报告
        report = self._generate_sprint_report()
        print(f"\n{'='*60}")
        print("📊 Sprint 完成!")
        print(f"{'='*60}")
        print(report)

        return self.manager.export_board()

    def _generate_sprint_report(self) -> str:
        """生成 Sprint 报告"""
        board = self.manager.board
        lines = [
            f"📋 Sprint 报告: {board.name}",
            f"总任务数: {len(board.cards)}",
            f"完成进度: {board.progress:.0f}%",
            f"",
            f"按状态分布:",
        ]

        for status, cards in board.columns.items():
            if cards:
                lines.append(f"  {status.value}: {len(cards)} 个任务")

        total_effort = sum(c.actual_effort for c in board.cards.values())
        lines.append(f"\n总工作量: {total_effort} 故事点")
        lines.append(f"\n产出物:")
        for card in board.cards.values():
            if card.output_artifacts:
                lines.append(f"  {card.title}: {', '.join(card.output_artifacts)}")

        return "\n".join(lines)

# hermes/examples/kanban_sprint_demo.py

async def run_sprint_demo():
    """运行完整的 Sprint 演示"""

    # 创建看板
    manager = KanbanManager(name="SaaS 平台开发 Sprint #1")

    # 注册 Agent
    profiles = {
        AgentProfile.PM: PMAgent("pm-analyst"),
        AgentProfile.ARCHITECT: ArchitectAgent("architect"),
        AgentProfile.FRONTEND: FrontendAgent("frontend-dev"),
        AgentProfile.BACKEND: BackendAgent("backend-dev"),
        AgentProfile.QA: QAAgent("qa-engineer"),
        AgentProfile.DEVOPS: DevOpsAgent("devops-engineer"),
        AgentProfile.WRITER: TechWriterAgent("tech-writer"),
    }

    # PM 拆解需求
    pm_card = KanbanCard(
        title="需求分析：用户管理系统",
        description="开发一个包含注册、登录、权限管理的用户管理系统",
        assigned_to=AgentProfile.PM,
        priority=TaskPriority.CRITICAL,
        effort_estimate=5,
    )
    manager.add_card(pm_card)

    # 引擎
    engine = KanbanEngine(manager)
    for profile, agent in profiles.items():
        engine.register_agent(profile, agent.handle_task)

    # 运行 Sprint
    result = await engine.run_sprint()

    # 最终看板
    print("\n📋 最终看板状态:")
    print(manager.display_board())

    return result


# 运行
asyncio.run(run_sprint_demo())

╔══════════════════════════════════════════════════════╗
║          SaaS 平台开发 Sprint #1                      ║
║  进度: 87%  卡片: 8                                  ║
╠══════════════════════════════════════════════════════╣
║  📋 待办      │ 📝 待处理   │ 🔄 进行中  │ 🔍 审查中  │
║  ──────────────┼──────────────┼─────────────┼───────────┤
║  🔴 数据库设  │              │ 🟡 前端页  │ 🔴 需求文档│
║  🔴 后端 API  │              │             │            │
║  🟡 测试用例  │              │             │            │
║              │              │             │            │
║  🟡 部署配置  │              │             │            │
║  🟢 文档编写  │              │             │            │
║              │              │             │            │
║  🟢 技术方    │              │             │            │
╚══════════════════════════════════════════════════════╝

[18:30:00] Sprint 开始
[18:30:01] 任务入队: CARD-a1b2c3d4 (需求分析：用户管理系统)
[18:30:02] pm-analyst 开始处理 CARD-a1b2c3d4
[18:30:05] CARD-a1b2c3d4 完成，拆解出 8 个子任务
[18:30:05] 子任务加入看板
[18:30:06] architect 开始处理 CARD-e5f6g7h8 (技术方案)
[18:30:06] backend-dev 开始处理 CARD-i9j0k1l2 (数据库设计)
[18:30:08] CARD-e5f6g7h8 完成 → 审查
[18:30:09] CARD-e5f6g7h8 通过审查 → 已完成
[18:30:10] CARD-i9j0k1l2 完成 → 审查 → 已完成
[18:30:11] frontend-dev 开始处理 CARD-m3n4o5p6 (前端页面)
[18:30:11] backend-dev 开始处理 CARD-q7r8s9t0 (后端 API)
...
[18:35:00] Sprint 完成! 8/8 任务已完成

def create_dependency_chain(manager: KanbanManager) -> list[str]:
    """创建任务依赖链"""
    # 需求分析 → 技术方案 → 前后端开发 → 测试 → 部署
    cards = []

    pm = KanbanCard(
        title="需求分析",
        assigned_to=AgentProfile.PM,
        priority=TaskPriority.CRITICAL,
        effort_estimate=3,
    )
    pm.status = TaskStatus.TODO
    manager.add_card(pm)
    cards.append(pm.card_id)

    arch = KanbanCard(
        title="技术方案",
        assigned_to=AgentProfile.ARCHITECT,
        priority=TaskPriority.CRITICAL,
        effort_estimate=5,
        dependencies=[pm.card_id],
    )
    manager.add_card(arch)
    cards.append(arch.card_id)

    backend = KanbanCard(
        title="后端开发",
        assigned_to=AgentProfile.BACKEND,
        priority=TaskPriority.HIGH,
        effort_estimate=8,
        dependencies=[arch.card_id],
    )
    manager.add_card(backend)
    cards.append(backend.card_id)

    frontend = KanbanCard(
        title="前端开发",
        assigned_to=AgentProfile.FRONTEND,
        priority=TaskPriority.HIGH,
        effort_estimate=8,
        dependencies=[arch.card_id],
    )
    manager.add_card(frontend)
    cards.append(frontend.card_id)

    # 前后端完成后才能测试
    qa = KanbanCard(
        title="测试",
        assigned_to=AgentProfile.QA,
        priority=TaskPriority.MEDIUM,
        effort_estimate=5,
        dependencies=[backend.card_id, frontend.card_id],
    )
    manager.add_card(qa)

    return cards

class WIPController:
    """WIP 限制控制器"""

    def __init__(self, limits: dict[TaskStatus, int]):
        self.limits = limits

    def can_move(self, board: KanbanBoard, card_id: str, to_status: TaskStatus) -> bool:
        """检查是否可以移动"""
        if to_status not in self.limits:
            return True

        current = sum(
            1 for c in board.cards.values()
            if c.status == to_status and c.card_id != card_id
        )
        return current < self.limits[to_status]

    def suggest_next(self, board: KanbanBoard) -> Optional[KanbanCard]:
        """建议下一个应该处理的任务"""
        # 优先处理阻塞的任务
        blocked = [c for c in board.cards.values() if c.status == TaskStatus.BLOCKED]
        if blocked:
            return min(blocked, key=lambda c: TaskPriority[c.priority])

        # 其次处理审查中的任务（减少 WIP）
        review = board.columns[TaskStatus.REVIEW]
        if review:
            return review[0]

        # 然后是进行中的
        in_progress = board.columns[TaskStatus.IN_PROGRESS]
        if in_progress:
            return in_progress[0]

        # 最后是新任务
        todo = board.columns[TaskStatus.TODO]
        if todo:
            return todo[0]

        return None