refactor(agent_loop): 状态机迁移走 task 能力 + 角色规范化 agent.{role}

- 角色规范: agent 角色 agent.{role}(_normalize_role 补前缀+ROLE_ALIASES/ROLE_CHAIN/SDL_ROLES 改规范名)，
  人角色 {orgtype}.{role}(默认 owner.superuser)
- role_agent_run/pm_review_run/handle_failed_task 硬编码状态 SQL 改 task_capability 工具
  (submit/approve/reject/complete/mark_failed/retry，CAS+审计)
- communication/questions/sdlc_ability 默认 handler 角色规范化(main_agent→agent.main_agent, customer→owner.superuser)
- 通用助手路径(AgentExecutor)不动，保持 hermes 式自由对话

pipeline_service/agent_loop.py

@@ -20,21 +20,21 @@ import logging
 logger = logging.getLogger("pipeline.agent_loop")
 
 ROLE_ALIASES = {
-    'developer': 'develop', 'dev': 'develop', 'development': 'develop', 'coding': 'develop',
+    'developer': 'agent.develop', 'dev': 'agent.develop', 'development': 'agent.develop', 'coding': 'agent.develop',
-    'requirement': 'requirement', 'requirements': 'requirement', 'requirement_analysis': 'requirement',
+    'requirement': 'agent.requirement', 'requirements': 'agent.requirement', 'requirement_analysis': 'agent.requirement',
-    'designer': 'design', 'ui': 'design', 'ux': 'design',
+    'designer': 'agent.design', 'ui': 'agent.design', 'ux': 'agent.design',
-    'testing': 'test', 'qa': 'test', 'tester': 'test',
+    'testing': 'agent.test', 'qa': 'agent.test', 'tester': 'agent.test',
-    'deployment': 'deploy', 'release': 'deploy',
+    'deployment': 'agent.deploy', 'release': 'agent.deploy',
-    'operation': 'ops', 'operations': 'ops', 'maintenance': 'ops', '运维': 'ops',
+    'operation': 'agent.ops', 'operations': 'agent.ops', 'maintenance': 'agent.ops', '运维': 'agent.ops',
-    'pm': 'pm', 'project_manager': 'pm', '项目经理': 'pm', 'manager': 'pm',
+    'pm': 'agent.pm', 'project_manager': 'agent.pm', '项目经理': 'agent.pm', 'manager': 'agent.pm',
 }
 
 ROLE_CHAIN = {
-    'requirement': 'design',
+    'agent.requirement': 'agent.design',
-    'design': 'develop',
+    'agent.design': 'agent.develop',
-    'develop': 'test',
+    'agent.develop': 'agent.test',
-    'test': 'deploy',
+    'agent.test': 'agent.deploy',
-    'deploy': None,
+    'agent.deploy': None,
 }
 
 TASK_REVIEW = 'review'
@@ -50,8 +50,12 @@ _ALLOWED_WORKDIRS = [
 
 
 def _normalize_role(role):
+    """角色规范化：别名映射 + 补 agent. 前缀（人角色 {orgtype}.{role} 保留原样）。"""
     r = (role or '').strip().lower()
-    return ROLE_ALIASES.get(r, r)
+    r = ROLE_ALIASES.get(r, r)
+    if r and '.' not in r:
+        r = f"agent.{r}"
+    return r
 
 
 async def _resolve_pipeline_id(project_id):
@@ -455,7 +459,7 @@ async def _build_qna_section(sor, task_id, role, agent_id=None):
     if qna:
         lines.append("历史问答：")
         for item in qna:
-            src = "客户" if item.get('answer_source') == "customer" else "主agent"
+            src = "业主" if item.get('answer_source') == "owner.superuser" else "主agent"
             lines.append(f"问：{item.get('question','')}")
             lines.append(f"答({src})：{item.get('answer','')}")
     return "\n".join(lines)
@@ -756,7 +760,8 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
                 resp = await llm_call_msgs_native(msgs, tools=tools_schema, model=model_name, temperature=0.4)
             except Exception as e:
                 err_msg = f"{type(e).__name__}: {str(e)[:400]}"
-                await sor.sqlExe("UPDATE pipeline_tasks SET state='failed', last_error=${e}$ WHERE id=${tid}$", {"e": err_msg, "tid": task_id})
+                from .task_capability import mark_failed
+                await mark_failed(task_id, project_id, who=role, agent_id=agent_id, error=err_msg)
                 logger.error(f"role_agent_run llm failed: task={task_id} err={e}")
                 return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}
 
@@ -813,7 +818,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
             qid = await raise_problem("need_info", ask_question, role,
                                       from_agentid=agent_id,
                                       tenant_id=project_id, task_id=task_id,
-                                      first_handler_role="main_agent",
+                                      first_handler_role="agent.main_agent",
                                       context={"title": title})
             return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": ask_question}
 
@@ -879,11 +884,10 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
                 logger.info(f"git_commit_push {repo_name}: rc={r.get('rc', -1)} {r.get('message', '')[:100]}")
         git_result = {"rc": 0, "message": "; ".join(git_results) if git_results else "无仓库可提交"}
 
-        await sor.sqlExe(
+        from .task_capability import submit_task
-            "UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$",
+        ok, _ = await submit_task(task_id, project_id, who=role, agent_id=agent_id)
-            {"st": TASK_REVIEW, "tid": task_id})
         logger.info(f"role_agent_run completed: task={task_id} role={role} "
-                     f"deliverable={did} files={len(files_written)} git={git_result.get('rc',-1)}")
+                     f"deliverable={did} files={len(files_written)} git={git_result.get('rc',-1)} submit={ok}")
         return {"status": "completed", "task_id": task_id, "deliverable_id": did,
                 "files_written": len(files_written), "git_result": git_result}
 
@@ -913,7 +917,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
 
         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})
+            from .task_capability import reject_task
+            await reject_task(task_id, project_id, who="agent.pm", agent_id=agent_id, comment="没有交付件")
             return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"}
 
         repo_state = await _get_repo_state(workspace_dir)
@@ -949,7 +954,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
                 raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3)
             except Exception as e:
                 err_msg = f"{type(e).__name__}: {str(e)[:400]}"
-                await sor.sqlExe("UPDATE pipeline_tasks SET state='failed', last_error=${e}$ WHERE id=${tid}$", {"e": err_msg, "tid": task_id})
+                from .task_capability import mark_failed
+                await mark_failed(task_id, project_id, who="agent.pm", agent_id=agent_id, error=err_msg)
                 return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}
 
             act = _parse_agent_action(raw)
@@ -999,7 +1005,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
                 "AND (problem_type='review_reject' "
                 "     OR ((problem_type IS NULL OR problem_type='') AND from_role IN ('pm','cockpit')))",
                 {"a": "角色已响应，PM审核通过", "role": task_role or "role_agent", "tid": task_id})
-            await sor.sqlExe("UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$", {"st": TASK_APPROVED, "tid": task_id})
+            from .task_capability import approve_task
+            await approve_task(task_id, project_id, who="agent.pm", agent_id=agent_id, comment=comment)
             next_role = await _get_next_role(task_role, project_id)
             if next_role:
                 next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment)
@@ -1012,12 +1019,13 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
             rejection_q = decision.get("questions") or comment or "交付件不满足要求"
             # 审核退回：首处理方=被退角色（该角色任意 agent 重新认领响应）。
             # suspend_task=False —— 任务回 submitted（重新认领）而非 waiting（挂起等回答）。
-            await raise_problem("review_reject", rejection_q, "pm",
+            await raise_problem("review_reject", rejection_q, "agent.pm",
                                 tenant_id=project_id, task_id=task_id,
                                 first_handler_role=task_role,
                                 context={"pm_comment": comment, "deliverable_type": deliverable_type},
                                 suspend_task=False)
-            await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL WHERE id=${tid}$", {"tid": task_id})
+            from .task_capability import reject_task
+            await reject_task(task_id, project_id, who="agent.pm", agent_id=agent_id, comment=comment)
             return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q}
 
         else:
@@ -1027,7 +1035,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
                 "UPDATE pipeline_deliverables SET review_status='approved', review_comment=${cm}$ "
                 "WHERE task_id=${tid}$ AND review_status='pending'",
                 {"cm": comment, "tid": task_id})
-            await sor.sqlExe("UPDATE pipeline_tasks SET state='completed', claimed_by=NULL WHERE id=${tid}$", {"tid": task_id})
+            from .task_capability import complete_task
+            await complete_task(task_id, project_id, who="agent.pm", agent_id=agent_id)
             return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}
 
 
@@ -1089,22 +1098,19 @@ async def handle_failed_task(task_id: str, project_id: str) -> dict:
             decision = _classify_failure(last_error)
 
         if decision == "retry":
-            await sor.sqlExe(
+            from .task_capability import retry_task
-                "UPDATE pipeline_tasks SET retry_count=retry_count+1, state='submitted', "
+            ok, _ = await retry_task(task_id, project_id, who="agent.pm")
-                "claimed_by=NULL, last_error=NULL, updated_at=NOW() "
+            logger.info("failed task retry: task=%s role=%s attempt=%d ok=%s", task_id, role, retry_count + 1, ok)
-                "WHERE id=${tid}$ AND state='failed'",
-                {"tid": task_id})
-            logger.info("failed task retry: task=%s role=%s attempt=%d", task_id, role, retry_count + 1)
             return {"status": "retry", "task_id": task_id, "attempt": retry_count + 1}
 
         # 报故障：建问题通知用户，任务置 waiting 等待人工介入
         from .communication import raise_problem
-        reporter = "cockpit" if role == "pm" else "pm"
+        reporter = "agent.main_agent" if role == "agent.pm" else "agent.pm"
         qid = await raise_problem(
             "fault_report",
             f"任务「{title}」已失败 {retry_count} 次，自动重试无法解决，请人工介入。失败原因：{last_error or '未知'}",
             reporter, tenant_id=project_id, task_id=task_id,
-            first_handler_role="main_agent",
+            first_handler_role="agent.main_agent",
             context={"fault": True, "last_error": last_error, "retry_count": retry_count})
         logger.info("failed task fault: task=%s role=%s qid=%s", task_id, role, qid)
         return {"status": "fault", "task_id": task_id, "question_id": qid}

pipeline_service/communication.py

@@ -56,7 +56,7 @@ async def raise_problem(problem_type, question, from_role, from_agentid="",
     """
     db, dbname = _get_db()
     async with db.sqlorContext(dbname) as sor:
-        hr = first_handler_role or "main_agent"
+        hr = first_handler_role or "agent.main_agent"
         ha = first_handler_agentid or ""
         qid = getID()
         ctx_json = json.dumps(context, ensure_ascii=False, default=str) if context else None
@@ -83,7 +83,7 @@ async def raise_problem(problem_type, question, from_role, from_agentid="",
         return qid
 
 
-async def resolve_problem(question_id, answer, answered_by="", answer_source="main_agent",
+async def resolve_problem(question_id, answer, answered_by="", answer_source="agent.main_agent",
                           resume_task=True):
     """当前处理方解决了问题 → 停止冒泡(status=answered)，任务恢复 submitted。"""
     db, dbname = _get_db()
@@ -98,7 +98,7 @@ async def resolve_problem(question_id, answer, answered_by="", answer_source="ma
             'id': question_id,
             'answer': answer or '',
             'answered_by': answered_by or '',
-            'answer_source': answer_source or 'main_agent',
+            'answer_source': answer_source or 'agent.main_agent',
             'status': S_ANSWERED,
         })
 

pipeline_service/questions.py

@@ -38,29 +38,30 @@ async def agent_ask(tenant_id: str, task_id: str, from_role: str, question: str,
 
     新代码请直接调 communication.raise_problem 显式指定首处理方。
     """
-    if from_role == "pm":
+    r = (from_role or "").lower()
-        ptype, hr = "review_reject", "develop"
+    if "pm" in r:
-    elif from_role == "cockpit":
+        ptype, hr = "review_reject", "agent.develop"
-        ptype, hr = "fault_report", "main_agent"
+    elif "main_agent" in r or "cockpit" in r:
+        ptype, hr = "fault_report", "agent.main_agent"
     else:
-        ptype, hr = "need_info", "main_agent"
+        ptype, hr = "need_info", "agent.main_agent"
     return await raise_problem(ptype, question, from_role, "",
                                tenant_id=tenant_id, task_id=task_id,
                                context=context, first_handler_role=hr)
 
 
 async def answer_question(question_id: str, answer: str, answered_by: str = "",
-                          answer_source: str = "main_agent", resume: bool = True):
+                          answer_source: str = "agent.main_agent", resume: bool = True):
     """回答问题（解决 → 停止冒泡，任务恢复 submitted）。"""
     return await resolve_problem(question_id, answer, answered_by, answer_source,
                                  resume_task=resume)
 
 
-async def forward_question(question_id: str, next_handler_role: str = "customer",
+async def forward_question(question_id: str, next_handler_role: str = "owner.superuser",
                            next_handler_agentid: str = ""):
-    """答不了 → 沿冒泡路径转给下一个处理方（默认客户）。"""
+    """答不了 → 沿冒泡路径转给下一个处理方（默认业主/客户）。"""
     return await escalate_problem(question_id, next_handler_role, next_handler_agentid,
-                                  by_role="main_agent")
+                                  by_role="agent.main_agent")
 
 
 async def get_question(question_id: str):
@@ -87,7 +88,7 @@ async def get_question(question_id: str):
 async def list_questions(tenant_id: str = None, task_id: str = None,
                          status: str = None, limit: int = 50) -> list:
     """客户视角：列出当前该「客户(人)」处理的问题。"""
-    rows = await list_problems_for("customer", tenant_id=tenant_id,
+    rows = await list_problems_for("owner.superuser", tenant_id=tenant_id,
                                    task_id=task_id, limit=limit)
     if status:
         rows = [r for r in rows if (r.get('status') or '') == status]

pipeline_service/sdlc_ability.py

@@ -142,7 +142,7 @@ SDL_PROMPT = """你是「开发产线」的驾驶舱 agent，负责软件项目
 
 SDL_ROLES = [
     RoleSpec(
-        name="requirement",
+        name="agent.requirement",
         description="需求分析师",
         aliases=["requirements", "requirement_analysis"],
         system_prompt="""你是需求分析师。按 SDLC 仓库标准产出文档。
@@ -153,10 +153,10 @@ SDL_ROLES = [
   - modules/<模块名>.md: 模块功能、仓库URL(先填写规划地址)、技术栈、依赖
   - 每个应用至少关联一个模块
 内容: 项目概述、用户角色及权限、功能列表(每个功能:输入/处理/输出/验收标准)、非功能需求、业务流程。""",
-        next_role="design",
+        next_role="agent.design",
     ),
     RoleSpec(
-        name="design",
+        name="agent.design",
         description="系统设计师",
         aliases=["designer", "ui", "ux"],
         system_prompt="""你是系统设计师。按 SDLC 仓库标准产出文档。
@@ -167,10 +167,10 @@ SDL_ROLES = [
   - api-design.md: 接口列表(method/path/request/response)
   - ui-design.md: 页面结构、组件树(如适用)
 产出后用 result 输出文档，files 列出所有文件路径。""",
-        next_role="develop",
+        next_role="agent.develop",
     ),
     RoleSpec(
-        name="develop",
+        name="agent.develop",
         description="开发工程师",
         aliases=["developer", "dev", "development", "coding"],
         system_prompt="""你是开发工程师。源码写入模块独立仓库，不在项目仓库。
@@ -187,10 +187,10 @@ SDL_ROLES = [
   8. write_file 写 docs/02-develop/dev-notes.md 记录开发内容
   9. deliver 交付，files 列出所有产出文件路径
 PM审核: git_status 检查模块仓库有提交记录。""",
-        next_role="test",
+        next_role="agent.test",
     ),
     RoleSpec(
-        name="test",
+        name="agent.test",
         description="测试工程师",
         aliases=["testing", "qa", "tester"],
         system_prompt="""你是测试工程师。按 SDLC 仓库标准产出文档。
@@ -200,10 +200,10 @@ PM审核: git_status 检查模块仓库有提交记录。""",
   - test-cases.md: 用例清单(编号/前置条件/步骤/预期结果)
   - test-report.md: 执行结果、Bug清单、覆盖率
 测试脚本放到 files。""",
-        next_role="deploy",
+        next_role="agent.deploy",
     ),
     RoleSpec(
-        name="deploy",
+        name="agent.deploy",
         description="部署运维工程师",
         aliases=["deployment", "release"],
         system_prompt="""你是部署运维工程师。按 SDLC 仓库标准产出文档和配置。
@@ -227,7 +227,7 @@ async def _h_create_task(sor, p, ctx):
     from appPublic.uniqueID import getID
 
     title = p.get("title", "").strip()
-    role = p.get("role", "develop")
+    role = p.get("role", "agent.develop")
     desc = p.get("description", "")
     if not title:
         return "需要任务标题"
@@ -315,7 +315,7 @@ async def _h_start_agents(sor, p, ctx):
     results = []
     for r in recs:
         tid = getattr(r, "id", "")
-        role = getattr(r, "role", "develop")
+        role = getattr(r, "role", "agent.develop")
         try:
             from pipeline_service.agent_loop import role_agent_run
             result = await role_agent_run(pid, role)
@@ -345,7 +345,7 @@ async def _h_diagnose_project(sor, p, ctx):
     try:
         from .communication import list_problems_for
         # 待回答问题 = 当前该「主 agent」处理的问题（不混入 PM 退回给角色 agent 的）
-        mine = await list_problems_for("main_agent", tenant_id=pid, limit=200)
+        mine = await list_problems_for("agent.main_agent", tenant_id=pid, limit=200)
         qc = len(mine)
     except Exception:
         pass
@@ -432,7 +432,7 @@ async def _h_list_questions(sor, p, ctx):
         return "请先切换到项目"
     from .communication import list_problems_for
     # 只列「当前该主 agent 处理」的问题（不混入 PM 退回给角色 agent 的退回意见）
-    recs = await list_problems_for("main_agent", tenant_id=pid, limit=10)
+    recs = await list_problems_for("agent.main_agent", tenant_id=pid, limit=10)
     if not recs:
         return "没有待回答问题"
     lines = []
@@ -459,10 +459,10 @@ async def _h_escalate_question(sor, p, ctx):
         if recs:
             full_qid = getattr(recs[0], "id", qid)
     # 主 agent 答不了 → 沿冒泡路径转给客户（下一个处理方由 team-communication skill 决定）
-    r = await escalate_problem(full_qid, "customer", by_role="main_agent")
+    r = await escalate_problem(full_qid, "owner.superuser", by_role="agent.main_agent")
     if r is None:
         return f"问题不存在: {qid}"
-    return f"OK: 已沿冒泡路径转给 {r.get('current_handler_role', 'customer')}"
+    return f"OK: 已沿冒泡路径转给 {r.get('current_handler_role', 'owner.superuser')}"
 
 
 async def _h_answer_question(sor, p, ctx):
@@ -482,8 +482,8 @@ async def _h_answer_question(sor, p, ctx):
     task_id = getattr(recs[0], "task_id", "") or ""
 
     await sor.sqlExe(
-        "UPDATE pipeline_agent_questions SET answer=${a}$, answer_source='main_agent', "
+        "UPDATE pipeline_agent_questions SET answer=${a}$, answer_source='agent.main_agent', "
-        "answered_by='main_agent', status='answered' WHERE id=${qid}$",
+        "answered_by='agent.main_agent', status='answered' WHERE id=${qid}$",
         {"a": answer, "qid": full_qid})
 
     resumed = False