feat: 新增 QC 角色（质量门禁）+ 部署拆分为 deploy_test/deploy_prod + requirement 明确部署环境需求

- SDL_ROLES 6 环新链：requirement→design→develop→deploy_test→test→deploy_prod，新增 agent.qc
- 状态机新增 qc_review：角色交付→QC 合规/质量检查→PM 验收→下一环
- qc_review_run + qc poller；不合规退回重做
- requirement 职责增加部署环境需求明确

pipeline_service/agent_loop.py

@@ -26,7 +26,9 @@ ROLE_ALIASES = {
     'requirement': 'agent.requirement', 'requirements': 'agent.requirement', 'requirement_analysis': 'agent.requirement',
     'designer': 'agent.design', 'ui': 'agent.design', 'ux': 'agent.design',
     'testing': 'agent.test', 'qa': 'agent.test', 'tester': 'agent.test',
-    'deployment': 'agent.deploy', 'release': 'agent.deploy',
+    'deployment': 'agent.deploy_prod', 'release': 'agent.deploy_prod', 'production': 'agent.deploy_prod',
+    'staging': 'agent.deploy_test', 'deploy_staging': 'agent.deploy_test',
+    'qc': 'agent.qc', 'quality': 'agent.qc', 'quality_control': 'agent.qc', '质量控制': 'agent.qc',
     'operation': 'agent.ops', 'operations': 'agent.ops', 'maintenance': 'agent.ops', '运维': 'agent.ops',
     'pm': 'agent.pm', 'project_manager': 'agent.pm', '项目经理': 'agent.pm', 'manager': 'agent.pm',
 }
@@ -34,9 +36,10 @@ ROLE_ALIASES = {
 ROLE_CHAIN = {
     'agent.requirement': 'agent.design',
     'agent.design': 'agent.develop',
-    'agent.develop': 'agent.test',
-    'agent.test': 'agent.deploy',
-    'agent.deploy': None,
+    'agent.develop': 'agent.deploy_test',
+    'agent.deploy_test': 'agent.test',
+    'agent.test': 'agent.deploy_prod',
+    'agent.deploy_prod': None,
 }
 
 TASK_REVIEW = 'review'
@@ -461,11 +464,42 @@ __REPO_STATE__
 完成：{"action":"review_complete","comment":"总结"}
 
 审核标准：
-- requirement：需求是否清晰完整可量化
+- requirement：需求是否清晰完整可量化（含部署环境需求是否明确）
 - design：方案合理、覆盖需求、技术可行  
 - develop：必须用 git_status/read_file 检查代码是否实际写入仓库
+- deploy_test：测试环境部署配置完整、服务可访问
 - test：测试覆盖充分、发现问题记录完整
-- deploy：部署配置完整、可一键部署"""
+- deploy_prod：生产部署配置完整、可一键部署、有回滚方案"""
+
+
+QC_SYSTEM_PROMPT = """你是质量控制工程师（QC）。对交付件做合规检查和质量检查，不合规直接退回重做。
+
+## 检查维度
+1. 项目规范检查：产出是否按 SDLC 仓库标准路径/命名/格式产出（docs/ 目录结构、apps/modules 描述文件、文件命名规范）
+2. 项目过程规范：是否遵循各阶段流程规范（develop 是否实际 git 提交、test 是否覆盖充分、deploy 配置是否完整、requirement 是否明确部署环境需求）
+3. 产出质量：内容是否完整、可量化、可验收、无重大缺陷、无空泛套话
+
+## 待检查
+标题：__TITLE__
+角色：__ROLE__
+工作目录：__WORKSPACE__
+
+## 工具
+- read_file(path) — 读工作空间文件
+- list_files(path) — 列目录
+- git_status() — 查看git状态
+- run_shell(command) — 执行命令（编译/测试验证）
+
+## 输出格式（每次一个JSON）
+查看文件：{"action":"tool_call","tool":"read_file","params":{"path":"相对路径"}}
+检查通过：{"action":"review_approve","comment":"合规，质量达标"}
+不合规退回：{"action":"review_reject","comment":"退回原因","questions":"具体问题清单，逐条列出"}
+
+## 检查流程
+1. 读交付件内容，检查是否按 SDLC 规范产出（路径/命名/格式）
+2. 检查过程合规（git 提交、目录结构、必填项）
+3. 判断产出质量是否达标（完整、可量化、可验收）
+4. 合规则 review_approve；不合规则 review_reject 并逐条列出问题清单"""
 
 
 # ── Agent 核心逻辑 ──
@@ -1043,7 +1077,7 @@ async def _pm_list_tasks(sor, project_id, params):
     await sor.sqlExe("COMMIT", {})
     if not recs:
         return "暂无任务"
-    icons = {"submitted": "⏳", "running": "🔄", "review": "👀",
+    icons = {"submitted": "⏳", "running": "🔄", "review": "👀", "qc_review": "🔍",
              "approved": "✅", "completed": "✔️", "failed": "❌", "waiting": "⏸️"}
     lines = ["| 状态 | 任务 | 角色 |", "|------|------|------|"]
     for r in recs:
@@ -1218,6 +1252,102 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
             return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}
 
 
+async def qc_review_run(project_id, agent_id=None, model_name=None):
+    """QC 质量门禁：认领 qc_review 状态任务，做合规+质量检查，通过转 review，不合规退回重做。"""
+    from .task_capability import S_QC_REVIEW
+    db = _get_db()
+    async with db.sqlorContext("pipeline") as sor:
+        model_name, org_id = await _resolve_llm_context(sor, project_id, 'qc', model_name)
+        task = await _claim_task(sor, project_id, '', state=S_QC_REVIEW, match_role=False, set_state='qc_review')
+        if not task:
+            return {"status": "idle", "message": "没有待 QC 检查的任务"}
+
+        task_id = task.id
+        title = getattr(task, "title", "") or ""
+        task_role = _normalize_role(getattr(task, "role", "") or "")
+        workspace_dir = await _get_workspace_dir(sor, project_id)
+
+        try:
+            await sor.sqlExe("COMMIT", {})
+        except Exception:
+            pass
+
+        deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id)
+        if not deliverable_content:
+            from .task_capability import qc_reject_task
+            await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment="没有交付件")
+            return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"}
+
+        content_preview = deliverable_content[:6000]
+
+        qc_system = QC_SYSTEM_PROMPT.replace('__TITLE__', title)\
+            .replace('__ROLE__', task_role)\
+            .replace('__WORKSPACE__', workspace_dir)
+
+        msgs = [{"role": "system", "content": qc_system}]
+        msgs.append({"role": "user", "content": f"请检查以下交付件（类型:{deliverable_type}）：\n\n{content_preview}"})
+
+        from .llm_bridge import llm_call_msgs
+
+        decision = None
+        for turn in range(4):
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET updated_at=NOW() WHERE id=${tid}$ AND state='qc_review'",
+                {"tid": task_id})
+            await sor.sqlExe("COMMIT", {})
+            if turn >= 2:
+                msgs.append({"role": "user", "content":
+                    "已检查足够信息。现在必须立即输出 review_approve 或 review_reject，禁止再调用其它工具。"})
+            try:
+                raw = await llm_call_msgs(msgs, model=model_name, temperature=0.2, org_id=org_id)
+            except Exception as e:
+                err_msg = f"{type(e).__name__}: {str(e)[:400]}"
+                from .task_capability import mark_failed
+                await mark_failed(task_id, project_id, who="agent.qc", agent_id=agent_id, error=err_msg)
+                return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}
+
+            act = _parse_agent_action(raw)
+
+            if act.get('action') == 'review_approve':
+                decision = {'status': 'approved', 'comment': act.get('comment', '')}
+                break
+            elif act.get('action') == 'review_reject':
+                decision = {'status': 'rejected', 'comment': act.get('comment', ''), 'questions': act.get('questions', '')}
+                break
+            elif act.get('action') == 'tool_call':
+                tool = act.get('tool', '')
+                params = act.get('params', {})
+                result = await _exec_agent_tool(tool, params, workspace_dir)
+                msgs.append({"role": "assistant", "content": raw})
+                msgs.append({"role": "user", "content": f"工具 {tool} 结果：\n{result}"})
+            else:
+                # 无明确决策 → 默认通过（避免 QC 卡死阻塞流程）
+                decision = {'status': 'approved', 'comment': raw[:200]}
+                break
+
+        if not decision:
+            decision = {'status': 'approved', 'comment': 'QC 检查超时，默认通过'}
+
+        if decision['status'] == 'approved':
+            from .task_capability import qc_approve_task
+            await qc_approve_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment=decision.get('comment', ''))
+            logger.info(f"qc_review_run approve: task={task_id} -> review")
+            return {"status": "approved", "task_id": task_id, "comment": decision.get('comment', '')}
+        else:
+            comment = decision.get('comment', '')
+            from .communication import raise_problem
+            rejection_q = decision.get("questions") or comment or "交付件不合规"
+            await raise_problem("qc_reject", rejection_q, "agent.qc",
+                                tenant_id=project_id, task_id=task_id,
+                                first_handler_role=task_role,
+                                context={"qc_comment": comment, "deliverable_type": deliverable_type},
+                                suspend_task=False)
+            from .task_capability import qc_reject_task
+            await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment=comment)
+            logger.info(f"qc_review_run reject: task={task_id} -> submitted")
+            return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q}
+
+
 # ── 失败任务处理（第二层：PM/cockpit 判定重跑，最多3次，超限报故障给用户）──
 
 _RETRY_HINTS = (

pipeline_service/init.py

@@ -48,7 +48,7 @@ from .task_capability import (
     revive_task, set_task_state,
 )
 from .agent_loop import role_agent_run, role_agent_loop, agent_loop, run_agent_loop
-from .agent_loop import pm_review_run, pm_review_loop, handle_failed_task
+from .agent_loop import pm_review_run, pm_review_loop, qc_review_run, handle_failed_task
 
 MODULE_NAME = "pipeline_service"
 MODULE_VERSION = "3.4.0"
@@ -540,6 +540,7 @@ def load_pipeline_service():
     # PM agent (v3.3.0: 项目经理审核 + 任务链)
     env.pm_review_run = pm_review_run
     env.pm_review_loop = pm_review_loop
+    env.qc_review_run = qc_review_run
 
     # Question loop (角色agent ↔ 主agent ↔ 客户)
     env.agent_ask = agent_ask
@@ -693,6 +694,56 @@ def load_pipeline_service():
 
     add_startup(_pm_poller)
 
+    # QC review poller: auto-dispatch qc_review-state tasks to QC agent（合规/质量门禁）
+    async def _qc_poller(app):
+        from sqlor.dbpools import DBPools as _DBP2
+        qc_db = _DBP2()
+        _qc_dispatched = set()
+
+        async def _qc_poll_once(qc_db):
+            """执行一轮 QC poll。可能因连接获取/SQL 锁等待卡住，由外层 wait_for 超时保护。"""
+            async with qc_db.sqlorContext("pipeline") as sor:
+                # 回收僵尸 qc_review 任务：QC 检查进程崩溃后 claimed_by 残留，重新放回检查队列。
+                await sor.sqlExe(
+                    "UPDATE pipeline_tasks SET claimed_by=NULL, updated_at=NOW() "
+                    "WHERE state='qc_review' AND claimed_by IS NOT NULL "
+                    "AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {})
+                recs = await sor.sqlExe(
+                    "SELECT id, tenant_id, role FROM pipeline_tasks "
+                    "WHERE state='qc_review' AND claimed_by IS NULL "
+                    "ORDER BY created_at ASC LIMIT 5",
+                    {})
+                for rec in (recs or []):
+                    tid = getattr(rec, 'id', '')
+                    pid = getattr(rec, 'tenant_id', '')
+                    if tid and pid and tid not in _qc_dispatched:
+                        _qc_dispatched.add(tid)
+                        debug(f"qc_poller dispatching qc_review task={tid}")
+
+                        async def _qc_dispatch(tid, pid):
+                            try:
+                                await qc_review_run(pid, agent_id="qc-poller")
+                            except Exception as e:
+                                debug(f"qc_poller dispatch error: task={tid} err={e}")
+                            finally:
+                                _qc_dispatched.discard(tid)
+
+                        asyncio.ensure_future(_qc_dispatch(tid, pid))
+
+        async def _qc_poll_loop():
+            while True:
+                try:
+                    # watchdog：每轮 QC poll 最多 60 秒，超时则跳过本轮，防止 poller 永久停摆
+                    await asyncio.wait_for(_qc_poll_once(qc_db), timeout=60)
+                except Exception as e:
+                    debug(f"qc_poller error/timeout: {e}")
+                await asyncio.sleep(15)
+
+        asyncio.create_task(_qc_poll_loop())
+        debug("qc poller started")
+
+    add_startup(_qc_poller)
+
     # Failed task poller: 失败任务自动重跑(最多3次)，超限报故障给用户
     async def _failed_poller(app):
         from sqlor.dbpools import DBPools as _DBP3

pipeline_service/sdlc_ability.py

@@ -200,10 +200,14 @@ SDL_TOOLS = [
 
 # ── prompt 片段（追加到 core 通用心智之后） ──
 
-SDL_PROMPT = """你是「开发产线」的驾驶舱 agent，负责软件项目的全流程管理（需求 → 设计 → 开发 → 测试 → 部署）。
+SDL_PROMPT = """你是「开发产线」的驾驶舱 agent，负责软件项目的全流程管理（需求 → 设计 → 开发 → 部署测试环境 → 测试 → 部署生产环境）。
+
+## 部署环境需求（项目开始阶段必须明确）
+- 除功能需求外，需求阶段还要明确部署环境需求：测试环境 + 生产环境的资源规格、依赖服务、端口、环境变量、高可用/备份/域名证书/监控告警等。
+- 用户未提供时，在会话中逐步追问明确（用 ask_user / list_questions），不要跳过；明确后再创建 requirement 任务。
 
 ## 典型场景（帮助判断何时用哪个工具，非强制）
-- 用户提出新的开发需求/功能（"实现XX""XX系统要做XX"）→ 用 create_task 创建任务（可按 requirement→design→develop 拆分），再 start_agents 启动。
+- 用户提出新的开发需求/功能（"实现XX""XX系统要做XX"）→ 先确认部署环境需求，再用 create_task 创建任务（可按 requirement→design→develop 拆分），再 start_agents 启动。
 - 用户问进展/状态 → list_tasks / diagnose_project / check_progress。
 - 用户问待回答问题 → list_questions / answer_question。
 - 用户报告异常/故障 → 先用 diagnose_project 定位根因，再用 task_detail 查详情。
@@ -234,7 +238,11 @@ SDL_ROLES = [
   - apps/<应用名>.md: 应用描述、部署环境、端口
   - modules/<模块名>.md: 模块功能、仓库URL(先填写规划地址)、技术栈、依赖
   - 每个应用至少关联一个模块
-内容: 项目概述、用户角色及权限、功能列表(每个功能:输入/处理/输出/验收标准)、非功能需求、业务流程。""",
+内容: 项目概述、用户角色及权限、功能列表(每个功能:输入/处理/输出/验收标准)、非功能需求、业务流程。
+部署环境需求（项目开始阶段必须明确，含测试环境与生产环境）:
+  - 测试环境: 资源规格(CPU/内存/磁盘)、依赖服务(DB/缓存/中间件)、端口、环境变量
+  - 生产环境: 资源规格、高可用要求、备份策略、域名/证书、监控告警
+  - 若需求未明确部署环境信息，在需求文档中标注「待明确」，并列出需向用户确认的问题清单（后续会话逐步明确）。""",
         next_role="agent.design",
     ),
     RoleSpec(
@@ -269,32 +277,67 @@ SDL_ROLES = [
   8. write_file 写 docs/02-develop/dev-notes.md 记录开发内容
   9. deliver 交付，files 列出所有产出文件路径
 PM审核: git_status 检查模块仓库有提交记录。""",
+        next_role="agent.deploy_test",
+    ),
+    RoleSpec(
+        name="agent.deploy_test",
+        description="测试环境部署工程师",
+        aliases=["deploy_staging", "staging"],
+        system_prompt="""你是测试环境部署工程师。把开发完成的代码部署到测试环境，供测试工程师验证。
+
+产出路径:
+  - config/test/ 下: 测试环境 Dockerfile、docker-compose.yml、nginx 配置、部署脚本
+  - docs/04-deploy/deploy-test-env.md: 测试环境部署步骤、环境配置、访问地址
+流程:
+  1. read_file 读 docs/00-requirement/ 的部署环境需求 + docs/01-design/ 的设计
+  2. 准备测试环境部署配置（容器/依赖服务/端口/环境变量）
+  3. 部署到测试环境，run_shell 验证服务可访问
+  4. 产出部署文档，deliver 交付，files 列出配置文件路径
+PM审核: 检查测试环境部署配置完整、服务可访问。""",
         next_role="agent.test",
     ),
     RoleSpec(
         name="agent.test",
         description="测试工程师",
         aliases=["testing", "qa", "tester"],
-        system_prompt="""你是测试工程师。按 SDLC 仓库标准产出文档。
+        system_prompt="""你是测试工程师。在测试环境上执行测试，按 SDLC 仓库标准产出文档。
 
 产出路径: docs/03-test/ 目录下:
   - test-plan.md: 测试策略(单元/集成/端到端)
   - test-cases.md: 用例清单(编号/前置条件/步骤/预期结果)
   - test-report.md: 执行结果、Bug清单、覆盖率
 测试脚本放到 files。""",
-        next_role="agent.deploy",
+        next_role="agent.deploy_prod",
     ),
     RoleSpec(
-        name="agent.deploy",
-        description="部署运维工程师",
-        aliases=["deployment", "release"],
-        system_prompt="""你是部署运维工程师。按 SDLC 仓库标准产出文档和配置。
+        name="agent.deploy_prod",
+        description="生产环境部署工程师",
+        aliases=["deploy_production", "production", "release"],
+        system_prompt="""你是生产环境部署工程师。测试通过后，把代码部署到生产环境。
 
 产出路径:
-  - docs/04-deploy/deploy-guide.md: 环境要求、部署步骤、回滚方案
+  - config/prod/ 下: 生产环境 Dockerfile、docker-compose.yml、nginx 配置
+  - docs/04-deploy/deploy-prod-env.md: 生产部署步骤、回滚方案
   - docs/04-deploy/release-notes.md: 发布说明
-  - config/ 下: Dockerfile、nginx配置、docker-compose.yml 等
-产出后用 result 输出部署说明，files 列出所有配置文件路径。""",
+流程:
+  1. 确认测试验证通过（read_file 读 docs/03-test/test-report.md）
+  2. 准备生产配置（高可用、备份、监控告警、域名证书）
+  3. 部署到生产环境，验证服务可用
+  4. 产出部署文档 + 发布说明，deliver 交付，files 列出配置文件
+PM审核: 检查生产部署配置完整、可一键部署、有回滚方案。""",
+        next_role="",
+    ),
+    RoleSpec(
+        name="agent.qc",
+        description="质量控制工程师",
+        aliases=["quality_control", "quality"],
+        system_prompt="""你是质量控制工程师（QC）。对每项产出做合规检查和质量检查，不合规直接退回重做。
+
+检查维度:
+  1. 项目规范检查: 产出是否按 SDLC 仓库标准路径/命名/格式产出
+  2. 项目过程规范: 是否遵循各阶段流程规范（develop 是否实际 git 提交、test 是否覆盖充分、deploy 配置是否完整）
+  3. 产出质量: 内容是否完整、可量化、可验收、无重大缺陷
+不合规处理: review_reject 退回重做，questions 列出具体问题清单。""",
         next_role="",
     ),
 ]
@@ -368,7 +411,7 @@ async def _h_list_tasks(sor, p, ctx):
     icons = {
         "submitted": "⏳", "running": "🔄", "review": "👀",
         "approved": "✅", "completed": "✔️", "failed": "❌",
-        "waiting": "⏸️",
+        "waiting": "⏸️", "qc_review": "🔍",
     }
     lines = ["| 状态 | 任务 | 角色 | ID |", "|------|------|------|----|"]
     for r in recs:

pipeline_service/task_capability.py

@@ -19,7 +19,8 @@ logger = logging.getLogger("pipeline.task_capability")
 # 任务状态（SDLC 默认，状态机语义见 task skill）
 S_SUBMITTED = "submitted"   # 待认领
 S_RUNNING = "running"       # 角色 agent 执行中
-S_REVIEW = "review"         # 已提交待审核
+S_QC_REVIEW = "qc_review"   # 已交付，待 QC 合规/质量检查
+S_REVIEW = "review"         # QC 通过，待 PM 审核验收
 S_APPROVED = "approved"     # 审核通过
 S_COMPLETED = "completed"   # 全流程完成
 S_WAITING = "waiting"       # 挂起等回答
@@ -105,11 +106,23 @@ async def claim_task(tenant_id, role, agent_id, from_state=S_SUBMITTED,
 
 
 async def submit_task(task_id, tenant_id, who=None, agent_id=None):
-    """提交产出：running → review（清 claimed_by）。"""
-    return await _transition(task_id, tenant_id, S_RUNNING, S_REVIEW, 'submit',
+    """提交产出：running → qc_review（清 claimed_by，先过 QC 合规/质量门禁）。"""
+    return await _transition(task_id, tenant_id, S_RUNNING, S_QC_REVIEW, 'submit',
                              who=who, agent_id=agent_id)
 
 
+async def qc_approve_task(task_id, tenant_id, who=None, agent_id=None, comment=None):
+    """QC 通过：qc_review → review（进入 PM 审核验收）。"""
+    return await _transition(task_id, tenant_id, S_QC_REVIEW, S_REVIEW, 'qc_approve',
+                             who=who, agent_id=agent_id, detail=comment)
+
+
+async def qc_reject_task(task_id, tenant_id, who=None, agent_id=None, comment=None):
+    """QC 退回：qc_review → submitted（清 claimed_by，被退角色重新认领重做）。"""
+    return await _transition(task_id, tenant_id, S_QC_REVIEW, S_SUBMITTED, 'qc_reject',
+                             who=who, agent_id=agent_id, detail=comment)
+
+
 async def approve_task(task_id, tenant_id, who=None, agent_id=None, comment=None):
     """审核通过：review → approved。"""
     return await _transition(task_id, tenant_id, S_REVIEW, S_APPROVED, 'approve',