v3.3.0: PM agent + task chain + workspace + file output

pipeline_service/agent_loop.py

@@ -1,9 +1,17 @@
-"""Agent loop — 角色agent循环认领执行任务（v3.2.0）。
+"""Agent loop — 角色agent循环认领执行任务（v3.3.0）。
+
+v3.3.0 新增：
+- PM（项目经理）角色：审核交付件，决定通过/驳回
+- 任务链：requirement → design → develop → test → deploy
+  前序角色完成后 → PM审核通过 → 自动创建下一角色任务
+- 产物文件输出：交付件写入项目工作目录 deliverables/{role}/
+- 工作目录自动创建：确保 workspace_dir 存在
 
 架构：
 - 主agent（cockpit_chat）理解客户意图，开发类意图写入 pipeline_tasks（带 role）。
 - 各角色agent（需求/设计/开发/测试/运维）调 role_agent_run/role_agent_loop
   循环认领自己角色的任务并执行。
+- PM agent 调 pm_review_loop 循环审核处于 review 状态的任务。
 - 认领是原子的：UPDATE ... WHERE state='submitted' + claimed_by 令牌校验，
   多agent并发不会重复认领。
 - 角色agent缺信息时返回 need_info → 写入 pipeline_agent_questions，
@@ -13,6 +21,7 @@
 
 import asyncio
 import json
+import os
 import logging
 
 logger = logging.getLogger("pipeline.agent_loop")
@@ -25,8 +34,23 @@ ROLE_ALIASES = {
     'testing': 'test', 'qa': 'test', 'tester': 'test',
     'deployment': 'deploy', 'release': 'deploy',
     'operation': 'ops', 'operations': 'ops', 'maintenance': 'ops', '运维': 'ops',
+    'pm': 'pm', 'project_manager': 'pm', '项目经理': 'pm', 'manager': 'pm',
 }
 
+# 任务链：前序角色完成后，PM审核通过 → 自动创建下一角色任务
+# None 表示链末端
+ROLE_CHAIN = {
+    'requirement': 'design',
+    'design': 'develop',
+    'develop': 'test',
+    'test': 'deploy',
+    'deploy': None,  # 部署完成 = 链结束
+}
+
+# PM 审核后任务状态
+TASK_REVIEW = 'review'    # 角色agent完成，待PM审核
+TASK_APPROVED = 'approved'  # PM审核通过
+
 
 def _normalize_role(role):
     r = (role or '').strip().lower()
@@ -39,12 +63,59 @@ ROLE_PROMPT = """你是软件开发产线项目中的「__ROLE__」角色agent
 标题：__TITLE__
 参数：__PARAMS__
 __QNA__
+
+## 项目信息
+- 工作目录：__WORKSPACE__
+- 你的产出物将保存到：__DELIVERABLE_DIR__
+
 请站在你的角色立场完成这项工作。
 
+## 产出规范
+1. 产出内容必须完整、可交付、可直接使用
+2. 产出格式：根据角色不同，产出对应格式：
+   - requirement（需求分析）：需求规格文档（Markdown格式）
+   - design（设计）：设计文档 / 架构图描述（Markdown格式）
+   - develop（开发）：可运行的代码 + 代码说明
+   - test（测试）：测试用例 + 测试报告（Markdown格式）
+   - deploy（部署）：部署文档 / 配置 / 脚本
+
 输出要求——返回纯 JSON（不要 markdown 包裹）：
-- 能完成时：{"status":"done","deliverable_type":"...","result":"完整交付内容"}
+- 能完成时：{"status":"done","deliverable_type":"文档类型(requirement_doc/design_doc/code/test_report/deploy_doc)","result":"完整交付内容","summary":"一段话概述产出的关键结论"}
 - 缺少关键信息无法继续时：{"status":"need_info","question":"要向主agent/客户提出的具体问题","partial":"已完成的部分"}"""
 
+PM_PROMPT = """你是软件开发产线项目中的「项目经理（PM）」角色agent。
+
+你的职责是审核其他角色agent产出的交付件，确保质量和完整性，并推动项目前进。
+
+## 任务链
+项目按照以下阶段推进：
+1. requirement（需求分析）→ 2. design（设计）→ 3. develop（开发）→ 4. test（测试）→ 5. deploy（部署）
+
+当前待审核的任务：
+标题：__TITLE__
+角色：__ROLE__
+交付件内容：
+__DELIVERABLE__
+
+## 你的决策
+请基于交付件内容做出判断，输出纯JSON（不要markdown包裹）：
+
+1. 审核通过，启动下一阶段：
+   {"status":"approved","comment":"审核意见","next_task_title":"下一阶段任务标题","next_task_description":"下一阶段任务详细描述"}
+
+2. 审核不通过，需要修改：
+   {"status":"rejected","comment":"驳回原因和具体修改要求","questions":"需要角色agent澄清的具体问题"}
+
+3. 项目已完成（deploy阶段通过后）：
+   {"status":"completed","comment":"项目完成总结"}
+
+## 审核标准
+- requirement：需求是否清晰、完整、可量化
+- design：方案是否合理、技术选型是否恰当、是否覆盖需求
+- develop：代码是否可运行、是否遵循规范、是否实现设计
+- test：测试覆盖是否充分、是否发现关键问题
+- deploy：部署文档是否完整、配置是否正确"""
+
 
 def _get_db():
     from sqlor.dbpools import DBPools
@@ -57,22 +128,40 @@ def _get_db():
     return db
 
 
-async def _claim_task(sor, tenant_id: str, role: str):
-    """原子认领一条符合角色的 submitted 任务。返回 task record 或 None。
+def _resolve_workspace(workspace_dir):
+    """解析工作目录，如果不可写则回退到 /tmp。确保目录存在。"""
+    if not workspace_dir:
+        workspace_dir = '/tmp/pipeline_workspaces/default'
+    # 检测是否可写
+    try:
+        parent = os.path.dirname(workspace_dir)
+        if parent and not os.access(parent, os.W_OK):
+            # 回退到 /tmp
+            proj_name = os.path.basename(workspace_dir)
+            workspace_dir = f'/tmp/pipeline_workspaces/{proj_name}'
+    except Exception:
+        workspace_dir = f'/tmp/pipeline_workspaces/{os.path.basename(workspace_dir) or "default"}'
+    # 确保目录存在
+    os.makedirs(workspace_dir, exist_ok=True)
+    return workspace_dir
+
+
+async def _claim_task(sor, tenant_id: str, role: str, state: str = 'submitted'):
+    """原子认领一条符合角色的任务。返回 task record 或 None。
 
     认领策略：
-    1. 选出最早一条 state='submitted' 且 role 匹配（或 role 为空）的任务，
+    1. 选出最早一条符合状态和角色的任务，
        且排除有步骤记录的 DAG 任务（那些归 executor）。
-    2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state='submitted'
+    2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state=原状态
     3. 用令牌回查校验——别的agent抢先时校验失败，返回 None。
     """
     role = _normalize_role(role)
     recs = await sor.sqlExe(
         "SELECT id, title, params, pipeline_id, tenant_id, role FROM pipeline_tasks "
-        "WHERE tenant_id=${tid}$ AND state='submitted' AND (role=${role}$ OR role='') "
+        "WHERE tenant_id=${tid}$ AND state=${state}$ AND (role=${role}$ OR role='') "
         "AND NOT EXISTS (SELECT 1 FROM pipeline_task_steps s WHERE s.task_id=pipeline_tasks.id) "
         "ORDER BY created_at ASC LIMIT 1",
-        {"tid": tenant_id, "role": role})
+        {"tid": tenant_id, "state": state, "role": role})
     if not recs:
         return None
 
@@ -83,8 +172,8 @@ async def _claim_task(sor, tenant_id: str, role: str):
     claim_token = getID()
     await sor.sqlExe(
         "UPDATE pipeline_tasks SET state='running', claimed_by=${cb}$ "
-        "WHERE id=${tid}$ AND state='submitted'",
-        {"cb": claim_token, "tid": task_id})
+        "WHERE id=${tid}$ AND state=${state}$",
+        {"cb": claim_token, "tid": task_id, "state": state})
 
     # 令牌校验：确认是本agent抢到的
     check = await sor.sqlExe(
@@ -96,6 +185,15 @@ async def _claim_task(sor, tenant_id: str, role: str):
     return task
 
 
+async def _get_workspace_dir(sor, project_id):
+    """获取项目的工作目录，确保可写并存在。"""
+    recs = await sor.sqlExe(
+        "SELECT workspace_dir FROM sd_projects WHERE id=${pid}$",
+        {"pid": project_id})
+    ws = getattr(recs[0], 'workspace_dir', '') if recs else ''
+    return _resolve_workspace(ws)
+
+
 async def _build_qna_section(sor, task_id: str) -> str:
     """把该任务已回答的问答历史注入prompt，让角色agent带着答案继续。"""
     from .questions import get_task_qna
@@ -125,16 +223,90 @@ def _parse_result(raw: str) -> dict:
     return {"status": "done", "result": raw}
 
 
+async def _write_deliverable_file(workspace_dir, role, task_id, deliverable_type, content):
+    """将交付件写入项目工作目录的文件中。"""
+    role_dir = os.path.join(workspace_dir, 'deliverables', role)
+    os.makedirs(role_dir, exist_ok=True)
+    # 文件名：任务ID_类型.md
+    safe_type = (deliverable_type or 'deliverable').replace('/', '_')
+    filename = f"{task_id}_{safe_type}.md"
+    file_path = os.path.join(role_dir, filename)
+    try:
+        with open(file_path, 'w', encoding='utf-8') as f:
+            f.write(content or '')
+        logger.info(f"deliverable written: {file_path}")
+        return file_path
+    except Exception as e:
+        logger.error(f"failed to write deliverable file: {file_path} err={e}")
+        return ''
+
+
+async def _get_deliverable_content(sor, task_id):
+    """获取任务的最新交付件内容。"""
+    recs = await sor.sqlExe(
+        "SELECT content, deliverable_type FROM pipeline_deliverables "
+        "WHERE task_id=${tid}$ ORDER BY created_at DESC LIMIT 1",
+        {"tid": task_id})
+    if recs:
+        return getattr(recs[0], 'content', '') or '', getattr(recs[0], 'deliverable_type', '') or ''
+    return '', ''
+
+
+async def _get_next_role(current_role):
+    """获取任务链中的下一个角色。"""
+    return ROLE_CHAIN.get(_normalize_role(current_role))
+
+
+async def _create_next_task(sor, project_id, task, next_role, pm_comment=''):
+    """为下一个角色创建任务。"""
+    from appPublic.uniqueID import getID
+    title = getattr(task, 'title', '') or ''
+    params_str = getattr(task, 'params', '{}') or '{}'
+    try:
+        params = json.loads(params_str) if isinstance(params_str, str) else params_str
+    except (json.JSONDecodeError, TypeError):
+        params = {}
+
+    new_title = f"{title}（{next_role}阶段）"
+    new_params = {
+        **params,
+        'previous_role': _normalize_role(getattr(task, 'role', '')),
+        'previous_task_id': getattr(task, 'id', ''),
+        'pm_comment': pm_comment,
+    }
+
+    new_task_id = getID()
+    await sor.C('pipeline_tasks', {
+        'id': new_task_id,
+        'tenant_id': project_id,
+        'pipeline_id': 'role_task',
+        'title': new_title,
+        'params': json.dumps(new_params, ensure_ascii=False),
+        'role': next_role,
+        'state': 'submitted',
+        'claimed_by': None,
+    })
+    logger.info(f"_create_next_task: {next_role} task={new_task_id} from {getattr(task, 'id', '')}")
+    return new_task_id, new_title
+
+
 async def role_agent_run(project_id: str, role: str, agent_id: str = None, model_name: str = None) -> dict:
-    """角色agent单次迭代：认领一条任务 → 执行 → 交付或提问。
+    """角色agent单次迭代：认领一条任务 → 执行 → 交付。
+
+    非PM角色完成 → 任务置 review（等PM审核）
+    PM角色 → 由 pm_review_run 处理
 
     Returns:
         {"status": "idle"} 没有待办任务
-        {"status": "completed", "task_id", "deliverable_id"} 完成交付
+        {"status": "completed", "task_id", "deliverable_id"} 完成交付（已进入review）
         {"status": "need_info", "task_id", "question_id", "question"} 缺信息已提问
         {"status": "failed", "task_id", "error"} 执行失败
     """
     role = _normalize_role(role)
+    if role == 'pm':
+        # PM不执行角色任务，走pm_review_run
+        return {"status": "idle", "message": "PM agent请使用 pm_review_run"}
+
     db = _get_db()
     async with db.sqlorContext("pipeline") as sor:
         task = await _claim_task(sor, project_id, role)
@@ -145,12 +317,19 @@ async def role_agent_run(project_id: str, role: str, agent_id: str = None, model
         title = getattr(task, "title", "") or ""
         params_str = getattr(task, "params", "{}") or "{}"
 
+        # 获取工作目录
+        workspace_dir = await _get_workspace_dir(sor, project_id)
+        deliverable_dir = os.path.join(workspace_dir, 'deliverables', role)
+        os.makedirs(deliverable_dir, exist_ok=True)
+
         qna_section = await _build_qna_section(sor, task_id)
         prompt = (ROLE_PROMPT
                   .replace('__ROLE__', role)
                   .replace('__TITLE__', title)
                   .replace('__PARAMS__', params_str)
-                  .replace('__QNA__', qna_section))
+                  .replace('__QNA__', qna_section)
+                  .replace('__WORKSPACE__', workspace_dir)
+                  .replace('__DELIVERABLE_DIR__', deliverable_dir))
 
         from .llm_bridge import llm_call
         try:
@@ -172,22 +351,34 @@ async def role_agent_run(project_id: str, role: str, agent_id: str = None, model
             return {"status": "need_info", "task_id": task_id,
                     "question_id": qid, "question": parsed["question"]}
 
-        # 完成 → 写交付件
+        # 完成 → 写交付件到文件 + DB
         from appPublic.uniqueID import getID
         did = getID()
         result_text = parsed.get("result") or raw
+        deliverable_type = parsed.get("deliverable_type") or role
+        summary = parsed.get("summary", "")
+
+        # 写文件
+        file_path = await _write_deliverable_file(
+            workspace_dir, role, task_id, deliverable_type, result_text)
+
+        # 写DB
         await sor.C("pipeline_deliverables", {
             "id": did, "project_id": project_id, "task_id": task_id,
-            "deliverable_type": parsed.get("deliverable_type") or role,
+            "deliverable_type": deliverable_type,
             "title": title, "content": result_text,
+            "file_path": file_path,
             "quality_score": 80, "review_status": "pending",
             "created_by": agent_id or role,
         })
+
+        # 置为 review 状态，等待PM审核
         await sor.sqlExe(
-            "UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$",
-            {"tid": task_id})
-        logger.info(f"role_agent_run completed: task={task_id} role={role} deliverable={did}")
-        return {"status": "completed", "task_id": task_id, "deliverable_id": did}
+            "UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$",
+            {"st": TASK_REVIEW, "tid": task_id})
+        logger.info(f"role_agent_run completed(review): task={task_id} role={role} deliverable={did} file={file_path}")
+        return {"status": "completed", "task_id": task_id, "deliverable_id": did,
+                "next_state": TASK_REVIEW, "file_path": file_path}
 
 
 async def role_agent_loop(project_id: str, role: str, agent_id: str = None,
@@ -203,14 +394,157 @@ async def role_agent_loop(project_id: str, role: str, agent_id: str = None,
     return results
 
 
+# ── PM 审核 ──
+
+async def pm_review_run(project_id: str, agent_id: str = None, model_name: str = None) -> dict:
+    """PM审核单次迭代：认领一条 review 状态的任务 → 审核交付件 → 批准/驳回。
+
+    Returns:
+        {"status": "idle"} 没有待审核任务
+        {"status": "approved", "task_id", "next_task_id", "next_role"} 审核通过，已创建下一阶段任务
+        {"status": "rejected", "task_id"} 驳回，任务回到 submitted
+        {"status": "completed", "task_id"} 项目完成（链末端）
+    """
+    db = _get_db()
+    async with db.sqlorContext("pipeline") as sor:
+        # 认领 review 状态的任务（不限角色，PM审核所有角色产出）
+        task = await _claim_task(sor, project_id, 'pm', state=TASK_REVIEW)
+        if not task:
+            return {"status": "idle", "message": "没有待审核任务"}
+
+        task_id = task.id
+        title = getattr(task, "title", "") or ""
+        task_role = _normalize_role(getattr(task, "role", "") or "")
+
+        # 获取交付件内容
+        deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id)
+        if not deliverable_content:
+            # 没有交付件 → 驳回
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$",
+                {"tid": task_id})
+            return {"status": "rejected", "task_id": task_id,
+                    "reason": "没有找到交付件内容，任务已退回"}
+
+        # 截断过长内容（LLM有token上限）
+        content_preview = deliverable_content[:6000]
+        if len(deliverable_content) > 6000:
+            content_preview += "\n\n...（内容过长已截断）"
+
+        prompt = (PM_PROMPT
+                  .replace('__TITLE__', title)
+                  .replace('__ROLE__', task_role)
+                  .replace('__DELIVERABLE__', content_preview))
+
+        from .llm_bridge import llm_call
+        try:
+            raw = await llm_call(prompt, model=model_name, temperature=0.3)
+        except Exception as e:
+            logger.error(f"pm_review_run llm failed: task={task_id} err={e}")
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$",
+                {"tid": task_id})
+            return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}
+
+        parsed = _parse_result(raw)
+        decision = parsed.get("status", "rejected")
+        comment = parsed.get("comment", "")
+
+        if decision == "approved":
+            # 获取工作目录
+            workspace_dir = await _get_workspace_dir(sor, project_id)
+
+            # 写PM审核记录到交付件
+            from appPublic.uniqueID import getID
+            pm_did = getID()
+            await sor.C("pipeline_deliverables", {
+                "id": pm_did, "project_id": project_id, "task_id": task_id,
+                "deliverable_type": "pm_review",
+                "title": f"PM审核：{title}",
+                "content": json.dumps(parsed, ensure_ascii=False),
+                "file_path": os.path.join(workspace_dir, 'deliverables', 'pm',
+                                         f"{task_id}_review.md"),
+                "quality_score": 100, "review_status": "approved",
+                "created_by": agent_id or "pm",
+            })
+
+            # 更新原交付件审核状态
+            await sor.sqlExe(
+                "UPDATE pipeline_deliverables SET review_status='approved', "
+                "review_comment=${cm}$ WHERE task_id=${tid}$",
+                {"cm": comment, "tid": task_id})
+
+            # 任务置为 approved
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$",
+                {"st": TASK_APPROVED, "tid": task_id})
+
+            # 检查是否有下一阶段
+            next_role = await _get_next_role(task_role)
+            if next_role:
+                next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment)
+                logger.info(f"pm_review: approved task={task_id}, created next={next_tid} role={next_role}")
+                return {"status": "approved", "task_id": task_id,
+                        "next_task_id": next_tid, "next_role": next_role,
+                        "next_title": next_title, "comment": comment}
+            else:
+                # 链末端（deploy审核通过 = 项目完成）
+                logger.info(f"pm_review: project completed, task={task_id} role={task_role}")
+                return {"status": "completed", "task_id": task_id,
+                        "comment": comment or "项目所有阶段已完成"}
+
+        elif decision == "rejected":
+            # 驳回：任务回到 submitted，附带PM驳回原因作为question
+            from .questions import agent_ask
+            rejection_question = parsed.get("questions") or comment or "交付件不满足要求，请修改后重新提交"
+            await agent_ask(project_id, task_id, "pm", rejection_question,
+                           context={"pm_comment": comment, "deliverable_type": deliverable_type})
+            # 任务回到 submitted（通过回答问题的resume机制）
+            # 但agent_ask已经把任务置waiting，需要让主agent/客户确认后才能回到submitted
+            # 简化为：直接回 submitted
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$",
+                {"tid": task_id})
+            logger.info(f"pm_review: rejected task={task_id}")
+            return {"status": "rejected", "task_id": task_id,
+                    "comment": comment, "question": rejection_question}
+
+        else:
+            # completed（项目完成）
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$",
+                {"tid": task_id})
+            return {"status": "completed", "task_id": task_id,
+                    "comment": comment or "项目完成"}
+
+
+async def pm_review_loop(project_id: str, agent_id: str = None,
+                         model_name: str = None, max_iterations: int = 20) -> list:
+    """PM审核循环：连续审核处于 review 状态的任务。"""
+    results = []
+    for _ in range(max_iterations):
+        r = await pm_review_run(project_id, agent_id, model_name)
+        results.append(r)
+        if r["status"] in ("idle", "failed"):
+            break
+        await asyncio.sleep(1)
+    return results
+
+
 # ── 向后兼容别名（旧前端按 run_agent_loop/agent_loop 调用）──
 
 async def run_agent_loop(project_id, role_name, model_name=None):
     """旧接口：单次迭代。role_name 为空时按 'dev' 处理。"""
-    return await role_agent_run(project_id, role_name or "dev", model_name=model_name)
+    role = _normalize_role(role_name or "develop")
+    if role == 'pm':
+        return await pm_review_run(project_id, model_name=model_name)
+    return await role_agent_run(project_id, role_name or "develop", model_name=model_name)
 
 
 async def agent_loop(project_id, role_name, model_name=None, max_iterations=10):
     """旧接口：循环执行。role_name 为空时按 'dev' 处理。"""
-    return await role_agent_loop(project_id, role_name or "dev",
+    role = _normalize_role(role_name or "develop")
+    if role == 'pm':
+        return await pm_review_loop(project_id, model_name=model_name, max_iterations=max_iterations)
+    return await role_agent_loop(project_id, role_name or "develop",
                                  model_name=model_name, max_iterations=max_iterations)

pipeline_service/init.py

@@ -16,6 +16,7 @@ from appPublic.log import debug
 
 from .state import (
     TASK_SUBMITTED, TASK_RUNNING, TASK_COMPLETED, TASK_FAILED, TASK_PAUSED, TASK_CANCELLED, TASK_WAITING,
+    TASK_REVIEW, TASK_APPROVED,
     build_step_graph, get_cascade_rerun_steps, get_rerun_from_next,
 )
 from .storage import (
@@ -37,9 +38,10 @@ from .questions import (
     get_question, list_questions, get_task_qna,
 )
 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
 
 MODULE_NAME = "pipeline_service"
-MODULE_VERSION = "3.2.0"
+MODULE_VERSION = "3.3.0"
 
 
 async def pipeline_submit(tenant_id, pipeline_id, owner_id, title, params=None):
@@ -508,6 +510,10 @@ def load_pipeline_service():
     env.agent_loop = agent_loop          # 向后兼容
     env.run_agent_loop = run_agent_loop  # 向后兼容
 
+    # PM agent (v3.3.0: 项目经理审核 + 任务链)
+    env.pm_review_run = pm_review_run
+    env.pm_review_loop = pm_review_loop
+
     # Question loop (角色agent ↔ 主agent ↔ 客户)
     env.agent_ask = agent_ask
     env.question_answer = answer_question
@@ -554,5 +560,35 @@ def load_pipeline_service():
     from ahserver.configuredServer import add_startup
     add_startup(_role_poller)
 
+    # PM review poller (v3.3.0): auto-dispatch review-state tasks to PM agent
+    async def _pm_poller(app):
+        from sqlor.dbpools import DBPools as _DBP
+        pm_db = _DBP()
+
+        async def _pm_poll_loop():
+            while True:
+                try:
+                    async with pm_db.sqlorContext("pipeline") as sor:
+                        recs = await sor.sqlExe(
+                            "SELECT id, tenant_id, role FROM pipeline_tasks "
+                            "WHERE state='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:
+                                debug(f"pm_poller dispatching review task={tid}")
+                                asyncio.ensure_future(
+                                    pm_review_run(pid, agent_id="pm-poller"))
+                except Exception as e:
+                    debug(f"pm_poller error: {e}")
+                await asyncio.sleep(15)
+
+        asyncio.create_task(_pm_poll_loop())
+        debug("pm poller started")
+
+    add_startup(_pm_poller)
+
     debug(f"[{MODULE_NAME}] v{MODULE_VERSION} loaded — pipeline engine with role-agent + question loop")
     return True

pipeline_service/state.py

@@ -20,6 +20,8 @@ TASK_FAILED = "failed"
 TASK_PAUSED = "paused"
 TASK_CANCELLED = "cancelled"
 TASK_WAITING = "waiting"        # 等待人工步骤完成
+TASK_REVIEW = "review"         # 角色agent完成，待PM审核
+TASK_APPROVED = "approved"     # PM审核通过
 
 # Human task statuses
 HUMAN_PENDING = "pending"