全栈实战：10 Agent 协作完成 SaaS 项目全链路

┌─────────────────────────────────────────────────────┐
│                  TaskFlow 技术栈                      │
├─────────────────────────────────────────────────────┤
│  前端:  React 18 + TypeScript + TailwindCSS + Vite   │
│  后端:  FastAPI + Python 3.11 + SQLAlchemy 2.0       │
│  数据库: PostgreSQL 15 + Redis 7                     │
│  消息:    RabbitMQ + Celery                          │
│  部署:    Docker Compose + Nginx + Let's Encrypt     │
│  监控:    Prometheus + Grafana + Loki                │
│  测试:    Pytest + Playwright + k6                   │
└─────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│                  10 Agent 协作全景图                       │
├──────────────────────────────────────────────────────────┤
│                                                          │
│  Agent-1  📋 需求分析师 (RequirementsAgent)               │
│           ↓ PRD文档 + 用户故事                              │
│  Agent-2  🏗 架构师 (ArchitectAgent)                      │
│           ↓ 架构图 + 技术选型 + 接口契约                     │
│  Agent-3  🗄 数据库设计师 (DBAAgent)                      │
│           ↓ ER图 + 迁移脚本 + 索引策略                      │
│  Agent-4  ⚙️ 后端开发 (BackendAgent)                      │
│           ↓ FastAPI 服务 + 业务逻辑 + 数据层                 │
│  Agent-5  🎨 前端开发 (FrontendAgent)                     │
│           ↓ React 组件 + 页面 + 状态管理                    │
│  Agent-6  🔗 集成开发 (IntegrationAgent)                  │
│           ↓ API联调 + Webhook + 第三方集成                  │
│  Agent-7  🧪 测试工程师 (QAAgent)                         │
│           ↓ 单元测试 + E2E + 性能测试脚本                   │
│  Agent-8  🚀 DevOps (DevOpsAgent)                        │
│           ↓ Docker + CI/CD + Nginx + 部署脚本              │
│  Agent-9  📊 运维监控 (SREAgent)                          │
│           ↓ Prometheus + Grafana + 告警规则                │
│  Agent-10 📚 文档工程师 (DocAgent)                        │
│           ↓ API文档 + 用户手册 + 部署指南                   │
│                                                          │
└──────────────────────────────────────────────────────────┘

# agents/taskflow_agents.yaml
agents:
  requirements:
    name: "RequirementsAgent"
    role: "资深产品经理，擅长PRD撰写和敏捷用户故事拆分"
    model: "gpt-4o"
    tools: ["file_writer", "web_search", "mermaid_generator"]
    output_spec:
      format: "markdown"
      deliverables: ["prd.md", "user_stories.yaml", "acceptance_criteria.md"]

  architect:
    name: "ArchitectAgent"
    role: "10年+架构经验，精通微服务设计和领域驱动设计"
    model: "claude-sonnet-4-20250514"
    tools: ["file_writer", "diagram_generator", "code_analyzer"]
    depends_on: ["requirements"]
    output_spec:
      format: "markdown"
      deliverables: ["architecture.md", "api_contract.yaml", "tech_stack.md"]

  dba:
    name: "DBAAgent"
    role: "资深DBA，擅长PostgreSQL调优和数据库设计"
    model: "gpt-4o"
    tools: ["file_writer", "sql_runner", "schema_validator"]
    depends_on: ["architect"]
    output_spec:
      format: "sql"
      deliverables: ["schema.sql", "migrations/", "index_strategy.md"]

  backend:
    name: "BackendAgent"
    role: "资深Python工程师，精通FastAPI和异步编程"
    model: "claude-sonnet-4-20250514"
    tools: ["file_writer", "code_runner", "linter", "git_ops"]
    depends_on: ["architect", "dba"]
    output_spec:
      format: "python"
      deliverables: ["app/", "tests/", "requirements.txt"]

  frontend:
    name: "FrontendAgent"
    role: "资深前端工程师，精通React生态和UI/UX"
    model: "gpt-4o"
    tools: ["file_writer", "code_runner", "linter", "browser_preview"]
    depends_on: ["architect"]
    output_spec:
      format: "typescript"
      deliverables: ["src/", "public/", "package.json"]

  integration:
    name: "IntegrationAgent"
    role: "集成开发专家，擅长API联调和第三方服务对接"
    model: "gpt-4o"
    tools: ["file_writer", "api_tester", "webhook_simulator"]
    depends_on: ["backend", "frontend"]
    output_spec:
      format: "typescript"
      deliverables: ["integrations/", "webhooks/", "api_client.ts"]

  qa:
    name: "QAAgent"
    role: "资深QA工程师，擅长自动化测试和性能压测"
    model: "gpt-4o"
    tools: ["file_writer", "code_runner", "playwright", "k6_runner"]
    depends_on: ["backend", "frontend", "integration"]
    output_spec:
      format: "python"
      deliverables: ["tests/", "e2e/", "performance/", "test_report.md"]

  devops:
    name: "DevOpsAgent"
    role: "资深DevOps工程师，精通容器化和CI/CD"
    model: "claude-sonnet-4-20250514"
    tools: ["file_writer", "shell_executor", "docker_manager", "k8s_ops"]
    depends_on: ["backend", "frontend"]
    output_spec:
      format: "yaml"
      deliverables: ["docker-compose.yml", "Dockerfile", ".github/workflows/", "nginx/"]

  sre:
    name: "SREAgent"
    role: "资深SRE工程师，精通监控体系和故障排查"
    model: "gpt-4o"
    tools: ["file_writer", "prometheus_api", "grafana_manager", "alert_config"]
    depends_on: ["devops"]
    output_spec:
      format: "yaml"
      deliverables: ["monitoring/", "alert_rules.yml", "dashboards/", "runbooks/"]

  doc:
    name: "DocAgent"
    role: "技术文档工程师，擅长API文档和运维手册"
    model: "gpt-4o"
    tools: ["file_writer", "openapi_generator", "doc_builder"]
    depends_on: ["backend", "devops", "sre"]
    output_spec:
      format: "markdown"
      deliverables: ["docs/api/", "docs/user_guide.md", "docs/deployment.md", "docs/troubleshooting.md"]

# agents/requirements_agent.py
REQUIREMENTS_PROMPT = """
你是一个资深产品经理，负责为 TaskFlow SaaS 项目撰写产品需求文档。

## 项目概述
TaskFlow 是一个面向中小团队的任务管理 SaaS 应用，核心功能是
看板管理、任务协作和团队效率分析。

## 你的职责
1. 撰写完整的 PRD 文档（产品需求文档）
2. 拆分为敏捷用户故事（User Stories）
3. 定义验收标准（Acceptance Criteria）
4. 绘制用户流程图

## 输出要求
- 使用 Markdown 格式
- 包含功能优先级（P0/P1/P2）
- 每个用户故事包含角色、行为、价值
- 验收标准使用 Given-When-Then 格式
"""

class RequirementsAgent:
    def __init__(self, llm_client):
        self.llm = llm_client
        self.system_prompt = REQUIREMENTS_PROMPT

    async def generate_prd(self, project_brief: str) -> dict:
        """生成完整的 PRD 文档"""
        prd = await self.llm.chat(
            system=self.system_prompt,
            user=f"项目简介：{project_brief}\n\n请生成完整的 PRD 文档。"
        )
        return {
            "prd": prd,
            "user_stories": await self._extract_stories(prd),
            "flow_charts": await self._generate_mermaid_flows(prd)
        }

# TaskFlow PRD v1.0

## 1. 产品概述
TaskFlow 帮助中小团队（5-50人）以可视化的方式管理工作任务，
替代传统的 Excel/邮件管理方式。

## 2. 用户角色
| 角色 | 描述 | 权限 |
|------|------|------|
| Owner | 团队创建者 | 全部权限 |
| Admin | 管理员 | 项目管理、成员管理 |
| Member | 普通成员 | 任务操作、评论 |
| Viewer | 只读成员 | 仅查看 |

## 3. P0 功能（MVP必须）
### 3.1 用户注册登录
- 邮箱注册 + 密码登录
- 第三方 OAuth（Google/GitHub）
- JWT Token 认证

### 3.2 看板管理
- 创建/编辑/删除看板
- 看板列：待办 → 进行中 → 已完成
- 拖拽卡片改变状态

## 4. 验收标准示例
**US-01: 用户注册**
- Given 用户访问注册页面
- When 填写有效邮箱和密码并提交
- Then 系统发送验证邮件
- And 用户点击验证链接后账户激活

# agents/architect_agent.py
ARCHITECT_PROMPT = """
你是一个拥有10年+经验的系统架构师。
基于 PRD 文档，设计 TaskFlow 的系统架构。

## 你的职责
1. 设计整体架构（服务划分、数据流）
2. 定义 API 接口契约
3. 制定技术选型和编码规范
4. 设计安全方案（认证/授权/数据加密）
"""

class ArchitectAgent:
    async def design_architecture(self, prd: str) -> dict:
        architecture = await self.llm.chat(
            system=ARCHITECT_PROMPT,
            user=f"PRD文档：\n{prd}\n\n请设计系统架构。"
        )

        return {
            "architecture_md": architecture,
            "api_contract": await self._generate_api_contract(architecture),
            "diagrams": await self._generate_architecture_diagrams(architecture),
            "coding_standards": await self._generate_standards(architecture)
        }

# TaskFlow 系统架构 v1.0

## 1. 整体架构

                ┌─────────────┐
                │   Cloudflare │
                │   CDN/WAF   │
                └──────┬──────┘
                       │
                ┌──────▼──────┐
                │   Nginx     │
                │  反向代理    │
                └──────┬──────┘
                       │
        ┌──────────────┼──────────────┐
        ▼              ▼              ▼
 ┌──────────┐  ┌──────────┐  ┌──────────┐
 │  Frontend │  │  Backend  │  │  WebSocket│
 │  (Vite)   │  │ (FastAPI) │  │  Server  │
 │  :3000    │  │  :8000    │  │  :8001   │
 └──────────┘  └─────┬────┘  └──────────┘
                     │
        ┌────────────┼────────────┐
        ▼            ▼            ▼
 ┌──────────┐  ┌──────────┐  ┌──────────┐
 │PostgreSQL│  │  Redis   │  │ RabbitMQ │
 │  :5432   │  │  :6379   │  │  :5672   │
 └──────────┘  └──────────┘  └─────┬────┘
                                   │
                             ┌─────▼─────┐
                             │  Celery   │
                             │ Workers   │
                             └───────────┘

## 2. API 接口契约（OpenAPI 3.0）

```yaml
openapi: 3.0.3
info:
  title: TaskFlow API
  version: 1.0.0
paths:
  /api/v1/auth/register:
    post:
      summary: 用户注册
      requestBody:
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/RegisterRequest'
      responses:
        201:
          description: 注册成功
        409:
          description: 邮箱已存在

Tenant (租户)
  ├── User (用户)
  ├── Project (项目)
  │   ├── Board (看板)
  │   │   ├── Column (列)
  │   │   └── Card (卡片/任务)
  │   ├── Label (标签)
  │   └── Comment (评论)
  └── Notification (通知)

## 五、Agent-3：DBA → 数据库设计与迁移

### 5.1 数据库设计

```python
# agents/dba_agent.py
DBA_PROMPT = """
你是资深DBA，负责设计TaskFlow的数据库方案。

## 职责
1. 基于架构设计生成完整DDL
2. 设计索引策略（覆盖高频查询）
3. 编写Alembic迁移脚本
4. 提供SQL优化建议
"""

class DBAAgent:
    async def design_database(self, api_contract: str) -> dict:
        schema_sql = await self.llm.chat(
            system=DBA_PROMPT,
            user=f"接口契约：\n{api_contract}\n\n请设计数据库Schema。"
        )

        return {
            "schema": schema_sql,
            "migrations": await self._generate_migrations(schema_sql),
            "seed_data": await self._generate_seed_data(),
            "index_strategy": await self._design_indexes(schema_sql)
        }

# app/models/base.py
from datetime import datetime
from sqlalchemy import Column, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped

class BaseModel(DeclarativeBase):
    __abstract__ = True

    id: Mapped[int] = Column(Integer, primary_key=True, autoincrement=True)
    created_at: Mapped[datetime] = Column(
        DateTime, server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = Column(
        DateTime, server_default=func.now(), onupdate=func.now()
    )
    is_deleted: Mapped[bool] = Column(
        Boolean, default=False, index=True
    )

# app/models/tenant.py
class Tenant(BaseModel):
    __tablename__ = "tenants"

    name: Mapped[str] = Column(String(100), nullable=False)
    slug: Mapped[str] = Column(String(50), unique=True, nullable=False, index=True)
    plan: Mapped[str] = Column(String(20), default="free")  # free/pro/enterprise
    max_users: Mapped[int] = Column(Integer, default=10)
    settings: Mapped[dict] = Column(JSON, default={})

# app/models/user.py
class User(BaseModel):
    __tablename__ = "users"

    email: Mapped[str] = Column(String(255), unique=True, nullable=False, index=True)
    password_hash: Mapped[str] = Column(String(255), nullable=False)
    display_name: Mapped[str] = Column(String(100))
    avatar_url: Mapped[str] = Column(String(500))
    tenant_id: Mapped[int] = Column(Integer, ForeignKey("tenants.id"), index=True)
    role: Mapped[str] = Column(String(20), default="member")  # owner/admin/member/viewer

    tenant = relationship("Tenant", back_populates="users")

# app/models/board.py
class Board(BaseModel):
    __tablename__ = "boards"

    name: Mapped[str] = Column(String(200), nullable=False)
    description: Mapped[str] = Column(Text)
    project_id: Mapped[int] = Column(Integer, ForeignKey("projects.id"), index=True)
    position: Mapped[int] = Column(Integer, default=0)

    project = relationship("Project", back_populates="boards")
    columns = relationship("BoardColumn", back_populates="board",
                          cascade="all, delete-orphan")

# app/models/card.py
class Card(BaseModel):
    __tablename__ = "cards"

    title: Mapped[str] = Column(String(500), nullable=False)
    description: Mapped[str] = Column(Text)
    board_id: Mapped[int] = Column(Integer, ForeignKey("boards.id"), index=True)
    column_id: Mapped[int] = Column(Integer, ForeignKey("board_columns.id"), index=True)
    assignee_id: Mapped[int] = Column(Integer, ForeignKey("users.id"))
    priority: Mapped[str] = Column(String(20), default="medium")  # low/medium/high/urgent
    due_date: Mapped[datetime] = Column(DateTime, nullable=True)
    position: Mapped[int] = Column(Integer, default=0)
    labels: Mapped[list] = Column(ARRAY(String), default=[])

    # 索引策略 - 高频查询覆盖
    __table_args__ = (
        Index("ix_cards_board_position", "board_id", "position"),
        Index("ix_cards_assignee_status", "assignee_id", "column_id"),
    )

# migrations/versions/001_initial_schema.py
"""initial schema

Revision ID: 001
Revises:
Create Date: 2025-01-15
"""
from alembic import op
import sqlalchemy as sa

def upgrade():
    # 创建租户表
    op.create_table('tenants',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('name', sa.String(100), nullable=False),
        sa.Column('slug', sa.String(50), unique=True, nullable=False),
        sa.Column('plan', sa.String(20), server_default='free'),
        sa.Column('max_users', sa.Integer, server_default='10'),
        sa.Column('settings', sa.JSON, server_default='{}'),
        sa.Column('created_at', sa.DateTime, server_default=sa.func.now()),
        sa.Column('updated_at', sa.DateTime, server_default=sa.func.now(),
                  onupdate=sa.func.now()),
        sa.Column('is_deleted', sa.Boolean, server_default='false'),
    )
    op.create_index('ix_tenants_slug', 'tenants', ['slug'])

    # 创建用户表
    op.create_table('users',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('email', sa.String(255), unique=True, nullable=False),
        sa.Column('password_hash', sa.String(255), nullable=False),
        sa.Column('display_name', sa.String(100)),
        sa.Column('avatar_url', sa.String(500)),
        sa.Column('tenant_id', sa.Integer, sa.ForeignKey('tenants.id')),
        sa.Column('role', sa.String(20), server_default='member'),
        sa.Column('created_at', sa.DateTime, server_default=sa.func.now()),
        sa.Column('updated_at', sa.DateTime, server_default=sa.func.now(),
                  onupdate=sa.func.now()),
        sa.Column('is_deleted', sa.Boolean, server_default='false'),
    )
    op.create_index('ix_users_email', 'users', ['email'])
    op.create_index('ix_users_tenant_id', 'users', ['tenant_id'])

    # ... 其他表（boards, columns, cards, comments等）

def downgrade():
    op.drop_table('cards')
    op.drop_table('board_columns')
    op.drop_table('boards')
    op.drop_table('users')
    op.drop_table('tenants')

# agents/backend_agent.py
BACKEND_PROMPT = """
你是资深Python工程师，精通FastAPI和异步编程。

## 职责
1. 实现全部API端点
2. 编写业务逻辑层
3. 实现认证/授权中间件
4. 编写异步任务和消息处理
5. 遵守PEP8和团队编码规范
"""

class BackendAgent:
    async def implement_backend(
        self,
        api_contract: str,
        schema: str,
        coding_standards: str
    ) -> dict:
        # 分模块并行开发
        tasks = [
            self._develop_auth(api_contract, schema),
            self._develop_boards(api_contract, schema),
            self._develop_cards(api_contract, schema),
            self._develop_users(api_contract, schema),
        ]
        modules = await asyncio.gather(*tasks)

        return {
            "app_structure": self._generate_project_structure(),
            "modules": modules,
            "middleware": await self._develop_middleware(coding_standards),
            "celery_tasks": await self._develop_celery_tasks()
        }

taskflow-backend/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI 应用入口
│   ├── config.py               # 配置管理
│   ├── database.py             # 数据库连接
│   ├── dependencies.py         # 依赖注入
│   ├── middleware/
│   │   ├── auth.py             # JWT 认证中间件
│   │   ├── tenant.py           # 多租户中间件
│   │   └── rate_limit.py       # 速率限制
│   ├── models/                 # SQLAlchemy 模型
│   │   ├── base.py
│   │   ├── user.py
│   │   ├── tenant.py
│   │   ├── board.py
│   │   └── card.py
│   ├── schemas/                # Pydantic 请求/响应模型
│   │   ├── auth.py
│   │   ├── user.py
│   │   ├── board.py
│   │   └── card.py
│   ├── routers/                # API 路由
│   │   ├── auth.py
│   │   ├── users.py
│   │   ├── tenants.py
│   │   ├── boards.py
│   │   └── cards.py
│   ├── services/               # 业务逻辑层
│   │   ├── auth_service.py
│   │   ├── user_service.py
│   │   └── board_service.py
│   ├── tasks/                  # Celery 异步任务
│   │   ├── email.py
│   │   └── notifications.py
│   └── utils/
│       ├── security.py         # 密码加密、JWT
│       └── validators.py       # 业务校验
├── tests/                      # 测试目录
├── alembic/                    # 数据库迁移
├── docker-compose.yml
├── Dockerfile
└── requirements.txt

# app/main.py
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from contextlib import asynccontextmanager

from app.config import settings
from app.database import engine, Base
from app.middleware.auth import AuthMiddleware
from app.middleware.tenant import TenantMiddleware
from app.routers import auth, users, tenants, boards, cards

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    # 启动：创建数据库表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # 关闭：清理连接池
    await engine.dispose()

app = FastAPI(
    title="TaskFlow API",
    version="1.0.0",
    lifespan=lifespan
)

# 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(TrustedHostMiddleware, allowed_hosts=settings.ALLOWED_HOSTS)
app.add_middleware(AuthMiddleware)
app.add_middleware(TenantMiddleware)

# 路由注册
app.include_router(auth.router, prefix="/api/v1/auth", tags=["认证"])
app.include_router(users.router, prefix="/api/v1/users", tags=["用户"])
app.include_router(tenants.router, prefix="/api/v1/tenants", tags=["租户"])
app.include_router(boards.router, prefix="/api/v1/boards", tags=["看板"])
app.include_router(cards.router, prefix="/api/v1/cards", tags=["卡片"])

@app.get("/health")
async def health_check():
    return {"status": "ok", "version": "1.0.0"}

# app/middleware/auth.py
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from jose import JWTError, jwt
from app.config import settings
from app.database import get_async_session
from app.models.user import User
from sqlalchemy import select

class AuthMiddleware(BaseHTTPMiddleware):
    """JWT 认证中间件"""

    EXEMPT_PATHS = {
        "/health",
        "/api/v1/auth/register",
        "/api/v1/auth/login",
        "/docs", "/openapi.json",
    }

    async def dispatch(self, request: Request, call_next):
        # 白名单路径直接放行
        if request.url.path in self.EXEMPT_PATHS:
            return await call_next(request)

        token = self._extract_token(request)
        if not token:
            return JSONResponse(
                status_code=401,
                content={"detail": "未提供认证令牌"}
            )

        try:
            payload = jwt.decode(
                token, settings.JWT_SECRET, algorithms=["HS256"]
            )
            user_id = payload.get("sub")
            tenant_id = payload.get("tenant_id")

            if not user_id:
                raise ValueError("无效的token")

            # 将用户信息注入请求
            request.state.user_id = int(user_id)
            request.state.tenant_id = int(tenant_id)

        except (JWTError, ValueError):
            return JSONResponse(
                status_code=401,
                content={"detail": "认证令牌无效或已过期"}
            )

        return await call_next(request)

    def _extract_token(self, request: Request) -> str | None:
        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            return None
        return auth_header.split(" ", 1)[1]

# app/routers/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.auth import RegisterRequest, LoginRequest, TokenResponse
from app.services.auth_service import AuthService

router = APIRouter()

@router.post("/register", status_code=status.HTTP_201_CREATED)
async def register(
    request: RegisterRequest,
    db: AsyncSession = Depends(get_db)
):
    """用户注册"""
    auth_service = AuthService(db)
    user = await auth_service.register(
        email=request.email,
        password=request.password,
        display_name=request.display_name,
    )
    return {
        "message": "注册成功，请验证邮箱",
        "user_id": user.id
    }

@router.post("/login", response_model=TokenResponse)
async def login(
    request: LoginRequest,
    db: AsyncSession = Depends(get_db)
):
    """用户登录"""
    auth_service = AuthService(db)
    token = await auth_service.authenticate(
        email=request.email,
        password=request.password
    )
    return token

# app/services/auth_service.py
from datetime import datetime, timedelta
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from passlib.context import CryptContext
from jose import jwt
from app.config import settings
from app.models.user import User
from app.models.tenant import Tenant

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class AuthService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def register(
        self,
        email: str,
        password: str,
        display_name: str = None
    ) -> User:
        """用户注册"""
        # 检查邮箱是否已存在
        existing = await self.db.execute(
            select(User).where(User.email == email)
        )
        if existing.scalar_one_or_none():
            raise HTTPException(
                status_code=409,
                detail="邮箱已被注册"
            )

        # 创建默认租户
        tenant = Tenant(
            name=f"{display_name or email} 的团队",
            slug=f"{email.split('@')[0]}-{id(email)}"
        )
        self.db.add(tenant)
        await self.db.flush()

        # 创建用户
        user = User(
            email=email,
            password_hash=pwd_context.hash(password),
            display_name=display_name or email,
            tenant_id=tenant.id,
            role="owner"
        )
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)

        return user

    async def authenticate(self, email: str, password: str) -> dict:
        """用户登录，返回JWT token"""
        result = await self.db.execute(
            select(User).where(User.email == email)
        )
        user = result.scalar_one_or_none()

        if not user or not pwd_context.verify(password, user.password_hash):
            raise HTTPException(
                status_code=401,
                detail="邮箱或密码错误"
            )

        # 生成JWT
        expire = datetime.utcnow() + timedelta(
            minutes=settings.JWT_EXPIRE_MINUTES
        )
        token = jwt.encode(
            {
                "sub": str(user.id),
                "tenant_id": str(user.tenant_id),
                "role": user.role,
                "exp": expire,
            },
            settings.JWT_SECRET,
            algorithm="HS256"
        )

        return {
            "access_token": token,
            "token_type": "bearer",
            "expires_in": settings.JWT_EXPIRE_MINUTES * 60
        }

# agents/frontend_agent.py
FRONTEND_PROMPT = """
你是资深前端工程师，精通React生态。

## 职责
1. 搭建Vite + React + TypeScript项目
2. 实现全部页面和组件
3. 实现状态管理（Zustand）
4. 实现路由守卫和API拦截器
5. 响应式设计 + 暗色模式
"""

class FrontendAgent:
    async def develop_frontend(self, api_contract: str) -> dict:
        # 并行开发各模块
        tasks = [
            self._setup_project(),
            self._develop_auth_pages(),
            self._develop_board_pages(),
            self._develop_card_components(),
            self._develop_layout(),
        ]
        modules = await asyncio.gather(*tasks)

        return {
            "project": modules,
            "tailwind_config": await self._generate_tailwind_config(),
        }

taskflow-frontend/
├── src/
│   ├── main.tsx
│   ├── App.tsx
│   ├── vite-env.d.ts
│   ├── config/
│   │   └── api.ts                    # API 配置
│   ├── api/
│   │   ├── client.ts                 # Axios 实例 + 拦截器
│   │   ├── auth.ts                   # 认证 API
│   │   ├── boards.ts                 # 看板 API
│   │   └── cards.ts                  # 卡片 API
│   ├── store/
│   │   ├── auth.ts                   # 认证状态
│   │   ├── boards.ts                 # 看板状态
│   │   └── ui.ts                     # UI 状态（主题等）
│   ├── components/
│   │   ├── ui/                       # 基础UI组件
│   │   │   ├── Button.tsx
│   │   │   ├── Modal.tsx
│   │   │   └── Input.tsx
│   │   ├── board/                    # 看板组件
│   │   │   ├── BoardView.tsx
│   │   │   ├── BoardColumn.tsx
│   │   │   └── BoardCard.tsx
│   │   └── layout/                   # 布局组件
│   │       ├── Header.tsx
│   │       ├── Sidebar.tsx
│   │       └── MainLayout.tsx
│   ├── pages/
│   │   ├── Login.tsx
│   │   ├── Register.tsx
│   │   ├── Dashboard.tsx
│   │   └── BoardPage.tsx
│   ├── hooks/
│   │   ├── useAuth.ts
│   │   └── useWebSocket.ts
│   └── utils/
│       └── helpers.ts
├── public/
├── index.html
├── tailwind.config.js
├── vite.config.ts
└── package.json

// src/api/client.ts
import axios, { AxiosError, InternalAxiosRequestConfig } from "axios";
import { useAuthStore } from "@/store/auth";

const api = axios.create({
  baseURL: import.meta.env.VITE_API_URL || "/api/v1",
  timeout: 10000,
  headers: {
    "Content-Type": "application/json",
  },
});

// 请求拦截器 - 自动附加 Token
api.interceptors.request.use(
  (config: InternalAxiosRequestConfig) => {
    const token = useAuthStore.getState().token;
    if (token) {
      config.headers.Authorization = `Bearer ${token}`;
    }
    return config;
  },
  (error) => Promise.reject(error)
);

// 响应拦截器 - 统一错误处理
api.interceptors.response.use(
  (response) => response,
  (error: AxiosError) => {
    if (error.response?.status === 401) {
      useAuthStore.getState().logout();
      window.location.href = "/login";
    }
    return Promise.reject(error);
  }
);

export default api;

// src/components/board/BoardView.tsx
import { useState } from "react";
import { DragDropContext, Droppable, Draggable } from "@hello-pangea/dnd";
import BoardColumn from "./BoardColumn";
import { useBoardStore, BoardColumn as ColumnType } from "@/store/boards";

interface BoardViewProps {
  boardId: string;
}

export default function BoardView({ boardId }: BoardViewProps) {
  const { columns, moveCard } = useBoardStore();

  const handleDragEnd = (result: any) => {
    const { source, destination, draggableId } = result;

    if (!destination) return;
    if (
      source.droppableId === destination.droppableId &&
      source.index === destination.index
    ) return;

    moveCard({
      cardId: draggableId,
      fromColumn: source.droppableId,
      toColumn: destination.droppableId,
      toIndex: destination.index,
    });
  };

  return (
    <DragDropContext onDragEnd={handleDragEnd}>
      <div className="flex gap-4 p-4 overflow-x-auto h-full bg-gray-100">
        {columns.map((column: ColumnType) => (
          <Droppable key={column.id} droppableId={column.id}>
            {(provided) => (
              <div ref={provided.innerRef} {...provided.droppableProps}>
                <BoardColumn column={column} />
                {provided.placeholder}
              </div>
            )}
          </Droppable>
        ))}
      </div>
    </DragDropContext>
  );
}

// src/store/auth.ts
import { create } from "zustand";
import { persist } from "zustand/middleware";
import { login as loginApi, register as registerApi } from "@/api/auth";

interface User {
  id: number;
  email: string;
  display_name: string;
  role: string;
}

interface AuthState {
  user: User | null;
  token: string | null;
  isAuthenticated: boolean;
  isLoading: boolean;
  login: (email: string, password: string) => Promise<void>;
  register: (email: string, password: string, name: string) => Promise<void>;
  logout: () => void;
}

export const useAuthStore = create<AuthState>()(
  persist(
    (set) => ({
      user: null,
      token: null,
      isAuthenticated: false,
      isLoading: false,

      login: async (email, password) => {
        set({ isLoading: true });
        try {
          const { access_token } = await loginApi(email, password);
          // 获取用户信息...
          set({
            token: access_token,
            isAuthenticated: true,
            isLoading: false,
          });
        } catch (error) {
          set({ isLoading: false });
          throw error;
        }
      },

      register: async (email, password, name) => {
        set({ isLoading: true });
        try {
          await registerApi(email, password, name);
          set({ isLoading: false });
        } catch (error) {
          set({ isLoading: false });
          throw error;
        }
      },

      logout: () => {
        set({ user: null, token: null, isAuthenticated: false });
      },
    }),
    { name: "taskflow-auth" }
  )
);

# agents/integration_agent.py
INTEGRATION_PROMPT = """
你是集成开发专家。

## 职责
1. 前后端 API 联调，修复接口不一致
2. 实现 Webhook 系统（事件通知）
3. 对接第三方服务（GitHub/Slack等）
4. 编写 API 客户端 SDK
"""

class IntegrationAgent:
    async def integrate(self, backend_output: str, frontend_output: str) -> dict:
        return {
            "api_client": await self._generate_api_client(backend_output),
            "webhooks": await self._implement_webhooks(),
            "third_party": await self._integrate_third_party(),
            "integration_tests": await self._run_integration_tests()
        }

# app/webhooks/dispatcher.py
import httpx
import hashlib
import hmac
from celery import shared_task
from app.config import settings

@shared_task(bind=True, max_retries=3)
def dispatch_webhook(self, webhook_url: str, event: dict):
    """异步发送 Webhook 通知"""
    payload = {
        "event": event["type"],
        "timestamp": event.get("timestamp"),
        "data": event["data"],
    }

    # 计算签名
    signature = hmac.new(
        settings.WEBHOOK_SECRET.encode(),
        str(payload).encode(),
        hashlib.sha256
    ).hexdigest()

    try:
        response = httpx.post(
            webhook_url,
            json=payload,
            headers={
                "Content-Type": "application/json",
                "X-Webhook-Signature": f"sha256={signature}",
            },
            timeout=10.0
        )
        response.raise_for_status()
    except Exception as exc:
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

// src/integrations/api-client.ts
import api from "@/api/client";

export class TaskFlowClient {
  async getBoards() {
    const { data } = await api.get("/boards");
    return data;
  }

  async createBoard(board: { name: string; projectId: number }) {
    const { data } = await api.post("/boards", board);
    return data;
  }

  async moveCard(cardId: string, columnId: string, position: number) {
    const { data } = await api.patch(`/cards/${cardId}/move`, {
      column_id: columnId,
      position,
    });
    return data;
  }
}

# agents/qa_agent.py
QA_PROMPT = """
你是资深QA工程师。

## 职责
1. 编写单元测试（覆盖率 > 80%）
2. 编写E2E测试（Playwright）
3. 编写性能测试（k6）
4. 生成测试报告
"""

class QAAgent:
    async def test_all(self, backend_output: str, frontend_output: str) -> dict:
        # 并行执行三类测试
        unit, e2e, performance = await asyncio.gather(
            self._run_unit_tests(),
            self._run_e2e_tests(),
            self._run_performance_tests()
        )

        return {
            "unit_report": unit,
            "e2e_report": e2e,
            "performance_report": performance,
            "coverage": await self._calculate_coverage(),
            "summary": await self._generate_test_summary(unit, e2e, performance)
        }

# tests/test_auth_service.py
import pytest
from unittest.mock import AsyncMock, patch
from app.services.auth_service import AuthService
from app.models.user import User
from app.models.tenant import Tenant
from fastapi import HTTPException

@pytest.fixture
def mock_db():
    session = AsyncMock()
    return session

@pytest.mark.asyncio
async def test_register_success(mock_db):
    """测试用户注册成功"""
    mock_db.execute = AsyncMock(return_value=AsyncMock(
        scalar_one_or_none=AsyncMock(return_value=None)
    ))

    service = AuthService(mock_db)
    user = await service.register(
        email="test@example.com",
        password="SecurePass123!",
        display_name="Test User"
    )

    assert user.email == "test@example.com"
    assert user.role == "owner"
    assert mock_db.add.call_count == 2  # tenant + user

@pytest.mark.asyncio
async def test_register_duplicate_email(mock_db):
    """测试邮箱重复注册"""
    existing_user = User(email="test@example.com")
    mock_db.execute = AsyncMock(return_value=AsyncMock(
        scalar_one_or_none=AsyncMock(return_value=existing_user)
    ))

    service = AuthService(mock_db)

    with pytest.raises(HTTPException) as exc:
        await service.register(
            email="test@example.com",
            password="SecurePass123!"
        )

    assert exc.value.status_code == 409
    assert "邮箱已被注册" in str(exc.value.detail)

@pytest.mark.asyncio
async def test_login_success(mock_db):
    """测试登录成功"""
    from passlib.context import CryptContext
    pwd_context = CryptContext(schemes=["bcrypt"])

    user = User(
        id=1,
        email="test@example.com",
        password_hash=pwd_context.hash("SecurePass123!"),
        tenant_id=1,
        role="owner"
    )
    mock_db.execute = AsyncMock(return_value=AsyncMock(
        scalar_one_or_none=AsyncMock(return_value=user)
    ))

    service = AuthService(mock_db)
    result = await service.authenticate("test@example.com", "SecurePass123!")

    assert "access_token" in result
    assert result["token_type"] == "bearer"

@pytest.mark.asyncio
async def test_login_wrong_password(mock_db):
    """测试密码错误"""
    user = User(
        id=1,
        email="test@example.com",
        password_hash="$2b$12$wronghash",
        tenant_id=1,
        role="owner"
    )
    mock_db.execute = AsyncMock(return_value=AsyncMock(
        scalar_one_or_none=AsyncMock(return_value=user)
    ))

    service = AuthService(mock_db)

    with pytest.raises(HTTPException) as exc:
        await service.authenticate("test@example.com", "WrongPassword")

    assert exc.value.status_code == 401

// e2e/tests/auth.spec.ts
import { test, expect } from "@playwright/test";

test.describe("认证流程", () => {
  test("用户注册并登录", async ({ page }) => {
    // 访问注册页面
    await page.goto("/register");

    // 填写注册表单
    await page.fill('input[name="email"]', "e2e-test@example.com");
    await page.fill('input[name="password"]', "TestPass123!");
    await page.fill('input[name="displayName"]', "E2E Tester");
    await page.click('button[type="submit"]');

    // 验证跳转到登录页
    await expect(page).toHaveURL("/login");

    // 登录
    await page.fill('input[name="email"]', "e2e-test@example.com");
    await page.fill('input[name="password"]', "TestPass123!");
    await page.click('button[type="submit"]');

    // 验证跳转到仪表盘
    await expect(page).toHaveURL("/dashboard");
    await expect(page.locator("text=欢迎")).toBeVisible();
  });

  test("看板操作", async ({ page }) => {
    // 登录后访问看板
    await page.goto("/boards/1");

    // 创建新卡片
    await page.click('button:has-text("添加卡片")');
    await page.fill('input[placeholder="卡片标题"]', "新测试任务");
    await page.click('button:has-text("创建")');

    // 验证卡片出现
    await expect(page.locator('text="新测试任务"')).toBeVisible();

    // 拖拽卡片
    const card = page.locator('text="新测试任务"');
    const targetColumn = page.locator('[data-column="done"]');
    await card.dragTo(targetColumn);

    // 验证卡片移动成功
    await expect(targetColumn.locator('text="新测试任务"')).toBeVisible();
  });
});

// performance/load-test.js
import http from "k6/http";
import { check, sleep } from "k6";
import { rate, trend } from "k6/metrics";

// 自定义指标
const apiErrorRate = new rate("api_errors");
const apiLatency = new trend("api_latency");

export const options = {
  stages: [
    { duration: "30s", target: 50 },    // 30秒内升到50并发
    { duration: "2m", target: 50 },     // 保持50并发2分钟
    { duration: "30s", target: 200 },   // 30秒内升到200并发（压力测试）
    { duration: "1m", target: 200 },    // 保持200并发1分钟
    { duration: "30s", target: 0 },     // 30秒内降到0
  ],
  thresholds: {
    http_req_duration: ["p(95)<500"],  // 95%请求<500ms
    api_errors: ["rate<0.01"],          // 错误率<1%
  },
};

const BASE_URL = __ENV.BASE_URL || "http://localhost:8000";
let authToken = "";

export function setup() {
  // 登录获取token
  const loginRes = http.post(`${BASE_URL}/api/v1/auth/login`, JSON.stringify({
    email: "test@example.com",
    password: "TestPass123!",
  }), {
    headers: { "Content-Type": "application/json" },
  });

  check(loginRes, { "login success": (r) => r.status === 200 });
  return { token: loginRes.json("access_token") };
}

export default function (data) {
  const headers = {
    "Content-Type": "application/json",
    "Authorization": `Bearer ${data.token}`,
  };

  // 获取看板列表
  const boardsRes = http.get(`${BASE_URL}/api/v1/boards`, { headers });
  apiErrorRate.add(boardsRes.status >= 400);
  apiLatency.add(boardsRes.timings.duration);

  check(boardsRes, {
    "boards status 200": (r) => r.status === 200,
    "boards response time < 200ms": (r) => r.timings.duration < 200,
  });

  sleep(1);
}

# agents/devops_agent.py
DEVOPS_PROMPT = """
你是资深DevOps工程师。

## 职责
1. 编写Dockerfile（多阶段构建）
2. 编写docker-compose配置
3. 配置GitHub Actions CI/CD流水线
4. 编写Nginx反向代理配置
"""

class DevOpsAgent:
    async def setup_infrastructure(self) -> dict:
        return {
            "dockerfiles": await self._generate_dockerfiles(),
            "docker_compose": await self._generate_compose(),
            "github_actions": await self._generate_ci_cd(),
            "nginx": await self._generate_nginx_config()
        }

# backend/Dockerfile
# 构建阶段
FROM python:3.11-slim AS builder

WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# 运行阶段
FROM python:3.11-slim

WORKDIR /app

# 从构建阶段复制依赖
COPY --from=builder /install /usr/local

# 非root用户运行
RUN groupadd -r appuser && useradd -r -g appuser appuser
COPY --chown=appuser:appuser . .

USER appuser

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# frontend/Dockerfile
FROM node:20-alpine AS builder

WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build

# Nginx 运行阶段
FROM nginx:alpine

COPY --from=builder /app/dist /usr/share/nginx/html
COPY nginx.conf /etc/nginx/conf.d/default.conf

EXPOSE 80

CMD ["nginx", "-g", "daemon off;"]

# docker-compose.yml
version: "3.9"

services:
  # 反向代理
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
      - ./nginx/ssl:/etc/nginx/ssl:ro
    depends_on:
      - frontend
      - backend
    restart: unless-stopped
    networks:
      - taskflow-network

  # 前端
  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile
    expose:
      - "80"
    restart: unless-stopped
    networks:
      - taskflow-network

  # 后端
  backend:
    build:
      context: ./backend
      dockerfile: Dockerfile
    expose:
      - "8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://taskflow:${DB_PASSWORD}@db:5432/taskflow
      - REDIS_URL=redis://:${REDIS_PASSWORD}@redis:6379/0
      - JWT_SECRET=${JWT_SECRET}
      - WEBHOOK_SECRET=${WEBHOOK_SECRET}
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_healthy
    restart: unless-stopped
    networks:
      - taskflow-network

  # Celery Worker
  celery-worker:
    build:
      context: ./backend
      dockerfile: Dockerfile
    command: ["celery", "-A", "app.tasks", "worker", "--loglevel=info", "--concurrency=4"]
    environment:
      - DATABASE_URL=postgresql+asyncpg://taskflow:${DB_PASSWORD}@db:5432/taskflow
      - REDIS_URL=redis://:${REDIS_PASSWORD}@redis:6379/0
      - CELERY_BROKER_URL=amqp://guest:${RABBITMQ_PASSWORD}@rabbitmq:5672//
    depends_on:
      - rabbitmq
      - redis
    restart: unless-stopped
    networks:
      - taskflow-network

  # Celery Beat (定时任务)
  celery-beat:
    build:
      context: ./backend
      dockerfile: Dockerfile
    command: ["celery", "-A", "app.tasks", "beat", "--loglevel=info"]
    environment:
      - CELERY_BROKER_URL=amqp://guest:${RABBITMQ_PASSWORD}@rabbitmq:5672//
    depends_on:
      - rabbitmq
    restart: unless-stopped
    networks:
      - taskflow-network

  # 数据库
  db:
    image: postgres:15-alpine
    environment:
      - POSTGRES_USER=taskflow
      - POSTGRES_PASSWORD=${DB_PASSWORD}
      - POSTGRES_DB=taskflow
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./backend/alembic/migrations:/docker-entrypoint-initdb.d
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U taskflow"]
      interval: 5s
      timeout: 5s
      retries: 5
    restart: unless-stopped
    networks:
      - taskflow-network

  # Redis
  redis:
    image: redis:7-alpine
    command: redis-server --requirepass ${REDIS_PASSWORD}
    volumes:
      - redisdata:/data
    healthcheck:
      test: ["CMD", "redis-cli", "-a", "${REDIS_PASSWORD}", "ping"]
      interval: 5s
      timeout: 5s
      retries: 5
    restart: unless-stopped
    networks:
      - taskflow-network

  # RabbitMQ
  rabbitmq:
    image: rabbitmq:3-management-alpine
    environment:
      - RABBITMQ_DEFAULT_PASS=${RABBITMQ_PASSWORD}
    volumes:
      - rabbitmqdata:/var/lib/rabbitmq
    restart: unless-stopped
    networks:
      - taskflow-network

  # 监控 - Prometheus
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'
    ports:
      - "9090:9090"
    restart: unless-stopped
    networks:
      - taskflow-network

  # 监控 - Grafana
  grafana:
    image: grafana/grafana:latest
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
    volumes:
      - grafana_data:/var/lib/grafana
      - ./monitoring/dashboards:/etc/grafana/provisioning/dashboards:ro
    ports:
      - "3001:3000"
    depends_on:
      - prometheus
    restart: unless-stopped
    networks:
      - taskflow-network

volumes:
  pgdata:
  redisdata:
  rabbitmqdata:
  prometheus_data:
  grafana_data:

networks:
  taskflow-network:
    driver: bridge

# .github/workflows/ci-cd.yml
name: TaskFlow CI/CD

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

env:
  REGISTRY: ghcr.io
  IMAGE_BACKEND: taskflow-backend
  IMAGE_FRONTEND: taskflow-frontend

jobs:
  # 代码质量检查
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: 后端 Lint
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install ruff mypy
      - run: ruff check app/ tests/
      - run: ruff format --check app/ tests/
      - run: mypy app/

      - name: 前端 Lint
        uses: actions/setup-node@v4
        with:
          node-version: "20"
      - run: cd frontend && npm ci
      - run: cd frontend && npm run lint

  # 单元测试
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: testpass
          POSTGRES_DB: taskflow_test
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v4

      - name: 运行单元测试
        run: |
          pip install -r requirements.txt
          pytest tests/ -v --cov=app --cov-report=xml

      - name: 上传覆盖率
        uses: codecov/codecov-action@v4
        with:
          file: ./coverage.xml

  # E2E 测试
  e2e:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: 启动应用
        run: docker compose up -d

      - name: 等待服务就绪
        run: sleep 30

      - name: 运行E2E测试
        run: |
          cd frontend
          npx playwright install --with-deps
          npx playwright test

      - name: 上传测试结果
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: playwright-report
          path: frontend/playwright-report/

  # 构建镜像
  build:
    needs: [lint, test]
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    outputs:
      backend-tag: ${{ steps.meta-backend.outputs.tags }}
      frontend-tag: ${{ steps.meta-frontend.outputs.tags }}

    steps:
      - uses: actions/checkout@v4

      - name: 登录 Container Registry
        uses: docker/login-action@v3
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: 构建后端镜像
        uses: docker/build-push-action@v5
        with:
          context: ./backend
          push: true
          tags: ${{ env.REGISTRY }}/${{ github.repository }}/${{ env.IMAGE_BACKEND }}:${{ github.sha }}

      - name: 构建前端镜像
        uses: docker/build-push-action@v5
        with:
          context: ./frontend
          push: true
          tags: ${{ env.REGISTRY }}/${{ github.repository }}/${{ env.IMAGE_FRONTEND }}:${{ github.sha }}

  # 部署到生产环境
  deploy:
    needs: [build, e2e]
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    environment: production

    steps:
      - uses: actions/checkout@v4

      - name: 部署到服务器
        uses: appleboy/ssh-action@v1
        with:
          host: ${{ secrets.PROD_HOST }}
          username: deploy
          key: ${{ secrets.SSH_PRIVATE_KEY }}
          script: |
            cd /opt/taskflow
            docker compose pull
            docker compose up -d --remove-orphans
            docker system prune -f

# agents/sre_agent.py
SRE_PROMPT = """
你是资深SRE工程师。

## 职责
1. 配置Prometheus监控指标采集
2. 创建Grafana仪表盘
3. 设置告警规则（AlertManager）
4. 编写Runbook运维手册
"""

class SREAgent:
    async def setup_monitoring(self) -> dict:
        return {
            "prometheus_config": await self._generate_prometheus_config(),
            "grafana_dashboards": await self._generate_dashboards(),
            "alert_rules": await self._generate_alert_rules(),
            "runbooks": await self._generate_runbooks()
        }

# monitoring/prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: "taskflow-backend"
    metrics_path: "/metrics"
    static_configs:
      - targets: ["backend:8000"]

  - job_name: "nginx"
    static_configs:
      - targets: ["nginx:9113"]

  - job_name: "postgres"
    static_configs:
      - targets: ["postgres-exporter:9187"]

  - job_name: "redis"
    static_configs:
      - targets: ["redis-exporter:9121"]

  - job_name: "celery"
    static_configs:
      - targets: ["celery-exporter:9808"]

# monitoring/alert_rules.yml
groups:
  - name: taskflow_critical
    rules:
      # 服务宕机
      - alert: ServiceDown
        expr: up{job=~"taskflow.*"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "服务 {{ $labels.job }} 宕机"
          description: "实例 {{ $labels.instance }} 已停止响应超过1分钟"

      # 高错误率
      - alert: HighErrorRate
        expr: |
          sum(rate(http_requests_total{status=~"5.."}[5m]))
          / sum(rate(http_requests_total[5m])) > 0.05
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "API 错误率超过 5%"
          description: "当前错误率: {{ $value | humanizePercentage }}"

      # 高延迟
      - alert: HighLatency
        expr: histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m])) > 1
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "P95 延迟超过 1 秒"
          description: "当前P95延迟: {{ $value }}s"

  - name: database_health
    rules:
      # 数据库连接池耗尽
      - alert: DBConnectionPoolExhausted
        expr: pg_stat_activity_count / pg_settings_max_connections > 0.8
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "数据库连接池使用率超过 80%"

      # Redis 内存使用过高
      - alert: RedisHighMemory
        expr: redis_memory_used_bytes / redis_memory_max_bytes > 0.85
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Redis 内存使用率超过 85%"

# Runbook: 数据库连接池耗尽

## 告警名称
DBConnectionPoolExhausted

## 严重程度
CRITICAL

## 影响
新请求无法获取数据库连接，服务出现 503 错误

## 排查步骤

### 1. 检查当前连接数
```bash
docker compose exec db psql -U taskflow -c \
  "SELECT count(*) FROM pg_stat_activity WHERE datname='taskflow';"

docker compose exec db psql -U taskflow -c \
  "SELECT query, state, duration FROM pg_stat_activity
   WHERE state != 'idle' ORDER BY duration DESC LIMIT 10;"

docker compose restart backend celery-worker

## 十二、Agent-10：文档工程师 → 全套文档

### 12.1 文档生成

```python
# agents/doc_agent.py
DOC_PROMPT = """
你是技术文档工程师。

## 职责
1. 基于代码生成 OpenAPI/Swagger 文档
2. 编写用户手册
3. 编写部署指南
4. 编写故障排查手册
"""

class DocAgent:
    async def generate_docs(
        self,
        backend_output: str,
        devops_output: str,
        sre_output: str
    ) -> dict:
        return {
            "api_docs": await self._generate_openapi_docs(backend_output),
            "user_guide": await self._generate_user_guide(),
            "deployment_guide": await self._generate_deployment_guide(devops_output),
            "troubleshooting": await self._generate_troubleshooting(sre_output)
        }

# docs/api/openapi.yaml
openapi: 3.0.3
info:
  title: TaskFlow API
  description: TaskFlow 任务管理 SaaS 平台的 RESTful API
  version: 1.0.0
  contact:
    name: TaskFlow Team
    email: api@taskflow.dev

servers:
  - url: https://api.taskflow.dev/api/v1
    description: 生产环境
  - url: http://localhost:8000/api/v1
    description: 本地开发

security:
  - BearerAuth: []

paths:
  /boards:
    get:
      summary: 获取看板列表
      tags: [看板]
      parameters:
        - name: project_id
          in: query
          schema:
            type: integer
          description: 项目ID筛选
        - name: page
          in: query
          schema:
            type: integer
            default: 1
        - name: page_size
          in: query
          schema:
            type: integer
            default: 20
      responses:
        200:
          description: 看板列表
          content:
            application/json:
              schema:
                type: object
                properties:
                  items:
                    type: array
                    items:
                      $ref: '#/components/schemas/Board'
                  total:
                    type: integer
                  page:
                    type: integer
                  page_size:
                    type: integer

    post:
      summary: 创建看板
      tags: [看板]
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/CreateBoardRequest'
      responses:
        201:
          description: 创建成功

  /cards/{card_id}/move:
    patch:
      summary: 移动卡片
      tags: [卡片]
      parameters:
        - name: card_id
          in: path
          required: true
          schema:
            type: string
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [column_id, position]
              properties:
                column_id:
                  type: string
                  description: 目标列ID
                position:
                  type: integer
                  description: 目标位置索引
      responses:
        200:
          description: 移动成功

components:
  securitySchemes:
    BearerAuth:
      type: http
      scheme: bearer
      bearerFormat: JWT

  schemas:
    Board:
      type: object
      properties:
        id:
          type: integer
        name:
          type: string
        description:
          type: string
        project_id:
          type: integer
        position:
          type: integer
        created_at:
          type: string
          format: date-time

    CreateBoardRequest:
      type: object
      required: [name]
      properties:
        name:
          type: string
          maxLength: 200
        description:
          type: string
        project_id:
          type: integer

# TaskFlow 用户手册

## 快速开始

### 1. 注册账户
1. 访问 https://app.taskflow.dev/register
2. 输入邮箱和密码
3. 点击"注册"按钮
4. 查收验证邮件并点击验证链接

### 2. 创建第一个项目
1. 登录后进入仪表盘
2. 点击"新建项目"
3. 输入项目名称和描述
4. 选择项目模板（看板/列表）

### 3. 使用看板管理任务
1. 进入项目后点击看板视图
2. 默认包含三列：待办 → 进行中 → 已完成
3. 点击"添加卡片"创建任务
4. 拖拽卡片到不同列改变状态
5. 点击卡片查看详情、添加评论、设置截止日期

### 4. 团队协作
1. 在项目设置中点击"邀请成员"
2. 输入成员邮箱并选择角色
3. 被邀请者收到邮件后加入项目
4. 使用 @提及 功能通知特定成员

# orchestration/saas_pipeline.py
import asyncio
from hermes import Workflow, Task, Agent

class SaaSPipeline:
    """10 Agent 协作 SaaS 全链路编排"""

    def __init__(self):
        self.agents = {
            "requirements": Agent("RequirementsAgent"),
            "architect": Agent("ArchitectAgent"),
            "dba": Agent("DBAAgent"),
            "backend": Agent("BackendAgent"),
            "frontend": Agent("FrontendAgent"),
            "integration": Agent("IntegrationAgent"),
            "qa": Agent("QAAgent"),
            "devops": Agent("DevOpsAgent"),
            "sre": Agent("SREAgent"),
            "doc": Agent("DocAgent"),
        }

    def build_workflow(self) -> Workflow:
        workflow = Workflow(name="TaskFlow-SaaS-Build")

        # Phase 1: 需求 & 架构（串行）
        workflow.add_task(
            Task("requirements", self.agents["requirements"]),
            depends_on=[]
        )
        workflow.add_task(
            Task("architect", self.agents["architect"]),
            depends_on=["requirements"]
        )
        workflow.add_task(
            Task("dba", self.agents["dba"]),
            depends_on=["architect"]
        )

        # Phase 2: 开发（并行）
        workflow.add_task(
            Task("backend", self.agents["backend"]),
            depends_on=["architect", "dba"]
        )
        workflow.add_task(
            Task("frontend", self.agents["frontend"]),
            depends_on=["architect"]
        )

        # Phase 3: 集成 & 测试
        workflow.add_task(
            Task("integration", self.agents["integration"]),
            depends_on=["backend", "frontend"]
        )
        workflow.add_task(
            Task("qa", self.agents["qa"]),
            depends_on=["backend", "frontend", "integration"]
        )

        # Phase 4: 部署（并行）
        workflow.add_task(
            Task("devops", self.agents["devops"]),
            depends_on=["backend", "frontend"]
        )
        workflow.add_task(
            Task("sre", self.agents["sre"]),
            depends_on=["devops"]
        )

        # Phase 5: 文档
        workflow.add_task(
            Task("doc", self.agents["doc"]),
            depends_on=["backend", "devops", "sre"]
        )

        return workflow

    async def run(self):
        """执行完整流水线"""
        workflow = self.build_workflow()

        print("🚀 启动 TaskFlow SaaS 全链路构建...")
        result = await workflow.execute()

        print(f"✅ 构建完成!")
        print(f"   总耗时: {result.duration:.1f}s")
        print(f"   成功任务: {result.succeeded}/{result.total}")
        print(f"   产出文件: {result.output_files}")

        return result

时间轴 ────────────────────────────────────────────────────────────►

Phase 1: 需求&架构        [===Req===][===Arch===][==DBA==]
                                        │
Phase 2: 开发              ┌──[===Backend===]──┐
                          [===Arch===]          │
                           │       └──[=Frontend=]──┘
Phase 3: 集成&测试                        [==Integration==]
                                               │
                                       [========QA=========]

Phase 4: 部署              ┌──[===DevOps===]──┐
                          │                   │
                          └──────────[==SRE==]

Phase 5: 文档                              [===Doc===]

总预计时间: ~25-30 分钟（10 Agent 并行协作）
传统团队: ~2-3 周

taskflow-project/
├── docs/                           # Agent-1,10
│   ├── prd.md                      # 产品需求文档
│   ├── user_stories.yaml           # 用户故事
│   ├── api/                        # API 文档
│   │   └── openapi.yaml
│   ├── user_guide.md               # 用户手册
│   ├── deployment.md               # 部署指南
│   └── troubleshooting.md          # 故障排查
├── architecture/                   # Agent-2
│   ├── architecture.md             # 架构设计文档
│   ├── tech_stack.md               # 技术选型
│   └── diagrams/                   # 架构图
├── database/                       # Agent-3
│   ├── schema.sql                  # 数据库DDL
│   ├── migrations/                 # 迁移脚本
│   ├── seed_data.sql               # 种子数据
│   └── index_strategy.md           # 索引策略
├── backend/                        # Agent-4
│   ├── app/                        # FastAPI 应用
│   ├── tests/                      # 单元测试
│   ├── requirements.txt
│   └── Dockerfile
├── frontend/                       # Agent-5
│   ├── src/                        # React 应用
│   ├── package.json
│   └── Dockerfile
├── integrations/                   # Agent-6
│   ├── api_client.ts               # API 客户端
│   ├── webhooks/                   # Webhook 系统
│   └── third_party/                # 第三方集成
├── tests/                          # Agent-7
│   ├── e2e/                        # E2E 测试
│   ├── performance/                # 性能测试
│   └── test_report.md              # 测试报告
├── infrastructure/                 # Agent-8
│   ├── docker-compose.yml          # Docker Compose
│   ├── nginx/                      # Nginx 配置
│   └── .github/workflows/          # CI/CD
├── monitoring/                     # Agent-9
│   ├── prometheus.yml              # Prometheus 配置
│   ├── alert_rules.yml             # 告警规则
│   ├── dashboards/                 # Grafana 仪表盘
│   └── runbooks/                   # 运维手册
└── README.md                       # 项目总览