feat(team-communication): 通用问题冒泡引擎（多agent感知）

- communication.py: raise/resolve/escalate/list_for，问题类型→冒泡路径，处理方=role+agentid
- questions.py: 向后兼容层，委托 communication
- agent_loop.py: 角色提问(need_info)/PM退回(review_reject)/故障(fault_report)走 raise_problem；
  _build_qna_section 按 handler 路由注入；PM approved 只答结 review_reject
- sdlc_ability.py: list_questions/answer_question/escalate_question 按 handler 路由
- models: pipeline_agent_questions 加 from_agentid/problem_type/escalation_path/escalation_pos

models/pipeline_agent_questions.json

@@ -12,8 +12,12 @@
         {"name": "tenant_id", "title": "租户ID", "type": "str", "length": 32, "nullable": "no"},
         {"name": "task_id", "title": "任务ID", "type": "str", "length": 32, "nullable": "no"},
         {"name": "from_role", "title": "提问角色", "type": "str", "length": 32, "nullable": "no"},
+        {"name": "from_agentid", "title": "提问Agent标识", "type": "str", "length": 64, "nullable": "yes"},
+        {"name": "problem_type", "title": "问题类型", "type": "str", "length": 50, "nullable": "yes"},
         {"name": "question", "title": "问题内容", "type": "text"},
         {"name": "context", "title": "问题上下文(JSON)", "type": "text"},
+        {"name": "escalation_path", "title": "冒泡路径(JSON角色/agent数组快照)", "type": "text"},
+        {"name": "escalation_pos", "title": "当前冒泡位置(处理方下标)", "type": "int", "default": "0"},
         {"name": "answer", "title": "回答内容", "type": "text"},
         {"name": "answered_by", "title": "回答人", "type": "str", "length": 64, "nullable": "yes"},
         {"name": "answer_source", "title": "回答来源", "type": "str", "length": 16, "nullable": "yes"},

pipeline_service/agent_loop.py

@@ -439,22 +439,20 @@ async def _get_workspace_dir(sor, project_id):
     return _resolve_workspace(ws)
 
 
-async def _build_qna_section(sor, task_id):
-    from .questions import get_task_qna
-    qna = await get_task_qna(task_id)  # 已回答问答（answered）
-    # 关键：额外查 pending 问题（PM 退回意见 / 待回答），角色 agent 重新认领时必须先响应这些再产出。
-    # 否则 PM review_reject 写的 pending 问题不会注入，角色反复产出同类不合格交付物，问题越堆越多。
-    pending = await sor.sqlExe(
-        "SELECT question, from_role FROM pipeline_agent_questions "
-        "WHERE task_id=${tid}$ AND status='pending' ORDER BY created_at ASC",
-        {"tid": task_id})
+async def _build_qna_section(sor, task_id, role, agent_id=None):
+    from .communication import get_task_qa
+    qa = await get_task_qa(task_id, include_pending_for=role,
+                           include_pending_agentid=agent_id)
+    qna = qa.get('answered', []) or [] if isinstance(qa, dict) else []
+    # pending = 当前该「本角色/本 agent」处理的退回意见，重新认领时必须先逐条响应再产出。
+    pending = qa.get('pending', []) or [] if isinstance(qa, dict) else []
     if not qna and not pending:
         return ""
     lines = []
     if pending:
         lines.append("⚠️ 待处理问题（审核/PM 退回意见，你必须先逐条响应这些再产出交付件）：")
         for p in pending:
-            lines.append(f"- [{getattr(p, 'from_role', '') or '审核'}] {getattr(p, 'question', '')}")
+            lines.append(f"- [{p.get('from_role', '') or '审核'}] {p.get('question', '')}")
     if qna:
         lines.append("历史问答：")
         for item in qna:
@@ -727,7 +725,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
             logger.warning(f"role_agent_run setup_repos failed: {e}")
 
         role_specific = _role_specific
-        qna_section = await _build_qna_section(sor, task_id)
+        qna_section = await _build_qna_section(sor, task_id, role, agent_id)
         tools_text = json.dumps(AGENT_TOOLS, ensure_ascii=False)
 
         system = (AGENT_SYSTEM_PROMPT
@@ -811,9 +809,10 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
                 break
 
         if ask_question:
-            from .questions import agent_ask
-            qid = await agent_ask(project_id, task_id, role, ask_question,
-                                  context={"title": title})
+            from .communication import raise_problem
+            qid = await raise_problem(project_id, "need_info", role, ask_question,
+                                      task_id=task_id, tenant_id=project_id,
+                                      from_agentid=agent_id, context={"title": title})
             return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": ask_question}
 
         if not deliverable:
@@ -989,12 +988,15 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
                 "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})
-            # 审核通过：该任务的 pending 问题（PM 退回意见）已由角色 agent 响应，标记 answered，避免问题永久堆积
+            # 审核通过：该任务 review_reject 类的 pending 退回意见已由角色 agent 响应 → 解决，避免永久堆积。
+            # 只答结 review_reject（归属角色 agent 的），不碰 need_info（归属主 agent）等其它 pending。
             await sor.sqlExe(
                 "UPDATE pipeline_agent_questions SET status='answered', answer=${a}$, "
-                "answer_source='main_agent', answered_by='pm' "
-                "WHERE task_id=${tid}$ AND status='pending'",
-                {"a": comment or "已响应并审核通过", "tid": task_id})
+                "answer_source='role_agent', answered_by=${role}$ "
+                "WHERE task_id=${tid}$ AND status='pending' "
+                "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})
             next_role = await _get_next_role(task_role, project_id)
             if next_role:
@@ -1004,9 +1006,15 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
                 return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}
 
         elif status == 'rejected':
-            from .questions import agent_ask
+            from .communication import raise_problem
             rejection_q = decision.get("questions") or comment or "交付件不满足要求"
-            await agent_ask(project_id, task_id, "pm", rejection_q, context={"pm_comment": comment, "deliverable_type": deliverable_type})
+            # 审核退回：路由到被退角色（该角色任意 agent 重新认领响应）。
+            # suspend_task=False —— 任务回 submitted（重新认领）而非 waiting（挂起等回答）。
+            await raise_problem(project_id, "review_reject", "pm", rejection_q,
+                                task_id=task_id, tenant_id=project_id,
+                                first_handler=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})
             return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q}
 
@@ -1088,11 +1096,12 @@ async def handle_failed_task(task_id: str, project_id: str) -> dict:
             return {"status": "retry", "task_id": task_id, "attempt": retry_count + 1}
 
         # 报故障：建问题通知用户，任务置 waiting 等待人工介入
-        from .questions import agent_ask
+        from .communication import raise_problem
         reporter = "cockpit" if role == "pm" else "pm"
-        qid = await agent_ask(
-            project_id, task_id, reporter,
+        qid = await raise_problem(
+            project_id, "fault_report", reporter,
             f"任务「{title}」已失败 {retry_count} 次，自动重试无法解决，请人工介入。失败原因：{last_error or '未知'}",
+            task_id=task_id, tenant_id=project_id,
             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

@@ -0,0 +1,329 @@
+"""团队沟通引擎 — 通用问题冒泡（多 agent 感知）。
+
+问题类型(pipeline_problem_types)定义冒泡路径(escalation_path，处理方序列)。
+处理方 = role + agentid（同一角色可能有多个 agent，必须精确到具体 agent；
+人类角色 agentid 为空表示"该角色的任意人"）。
+
+问题产生后从路径第一个处理方开始，每个处理方要么解决(resolve)要么未解决
+继续冒泡(escalate)到下一个：
+- 解决了 → status=answered，不再冒泡
+- 没解决 → escalation_pos+1，沿路径继续走
+- 路径尽头通常是"人"(human 角色)兜底
+
+冒泡路径条目格式：
+- "role_name"                    → 该角色任意 agent/人
+- {"role":"x", "agentid":"y"}    → 指定角色 + 指定 agent
+"""
+
+import json
+import logging
+from sqlor.dbpools import DBPools
+from appPublic.uniqueID import getID
+
+DBNAME = "pipeline"
+logger = logging.getLogger("pipeline.communication")
+
+S_PENDING = "pending"    # 冒泡中（未解决）
+S_ANSWERED = "answered"  # 已解决，停止冒泡
+
+DEFAULT_PIPELINE_ID = "sdlc_general"
+
+
+def _get_db():
+    db = DBPools()
+    if not db.databases:
+        from appPublic.jsonConfig import getConfig
+        config = getConfig()
+        if config.databases:
+            db.databases = config.databases
+    return db, DBNAME
+
+
+def _parse_path(raw):
+    """解析冒泡路径为 [{"role":..., "agentid":...}] 列表（统一归一化）。"""
+    if not raw:
+        return []
+    if isinstance(raw, (list, tuple)):
+        items = raw
+    else:
+        try:
+            items = json.loads(raw)
+        except (json.JSONDecodeError, TypeError):
+            return []
+        if not isinstance(items, list):
+            return []
+    result = []
+    for it in items:
+        if isinstance(it, str):
+            result.append({"role": it, "agentid": ""})
+        elif isinstance(it, dict):
+            result.append({
+                "role": str(it.get("role", "") or ""),
+                "agentid": str(it.get("agentid", "") or ""),
+            })
+    return [x for x in result if x["role"]]
+
+
+def current_handler_of(rec) -> dict:
+    """当前该处理这个问题的处理方 {"role", "agentid"}（由 escalation_path + escalation_pos 决定）。"""
+    path = _parse_path(getattr(rec, 'escalation_path', ''))
+    try:
+        pos = int(getattr(rec, 'escalation_pos', 0) or 0)
+    except (TypeError, ValueError):
+        pos = 0
+    if not path:
+        # 旧数据/无路径兜底：待主 agent 处理
+        return {"role": "main_agent", "agentid": ""}
+    if pos < 0 or pos >= len(path):
+        return path[-1]
+    return path[pos]
+
+
+def _handler_matches(handler: dict, role: str, agentid: str = None) -> bool:
+    """处理方是否匹配目标(role, agentid)。agentid 为空=任意 agent。"""
+    if handler.get("role") != role:
+        return False
+    if agentid and handler.get("agentid") and handler.get("agentid") != agentid:
+        return False  # 指定了具体 agent，但处理方是别的 agent
+    return True
+
+
+async def _resolve_pipeline_id(sor, project_id):
+    """从项目解析产线 id；查不到用默认 sdlc_general。"""
+    if project_id:
+        recs = await sor.sqlExe(
+            "SELECT pipeline_id FROM sd_projects WHERE id=${p}$", {"p": project_id})
+        if recs:
+            pid = getattr(recs[0], 'pipeline_id', '') or ''
+            if pid:
+                return pid
+    return DEFAULT_PIPELINE_ID
+
+
+async def get_problem_type(sor, pipeline_id, name):
+    """查问题类型定义（含冒泡路径，已归一化）。找不到返回 None。"""
+    recs = await sor.sqlExe(
+        "SELECT * FROM pipeline_problem_types "
+        "WHERE pipeline_id=${p}$ AND name=${n}$ LIMIT 1",
+        {"p": pipeline_id, "n": name})
+    if not recs:
+        return None
+    rec = recs[0]
+    return {
+        'id': getattr(rec, 'id', ''),
+        'pipeline_id': pipeline_id,
+        'name': name,
+        'title': getattr(rec, 'title', '') or name,
+        'escalation_path': _parse_path(getattr(rec, 'escalation_path', '')),
+    }
+
+
+async def raise_problem(pipeline_id, problem_type, from_role, question,
+                        task_id=None, tenant_id=None, context=None,
+                        from_agentid=None, first_handler=None,
+                        first_handler_agentid=None, suspend_task=True) -> str:
+    """提出问题。按问题类型的冒泡路径路由到第一个处理方。
+
+    Args:
+        pipeline_id: 产线 id（为空则从 tenant_id/project 解析）
+        problem_type: 问题类型标识（pipeline_problem_types.name）
+        from_role: 谁提的（角色名）
+        from_agentid: 提问的具体 agent 标识（多 agent 场景必传）
+        first_handler: 可选，指定第一个处理角色（如审核退回指定被退角色）
+        first_handler_agentid: 可选，第一个处理方具体 agent
+        suspend_task: 是否把任务置 waiting（角色 agent 提问时挂起任务等回答）
+
+    Returns:
+        question_id
+    """
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        if not pipeline_id:
+            pipeline_id = await _resolve_pipeline_id(sor, tenant_id)
+        pt = await get_problem_type(sor, pipeline_id, problem_type)
+        path = pt['escalation_path'] if pt else []
+        if first_handler and first_handler not in [p["role"] for p in path]:
+            path = [{"role": first_handler, "agentid": first_handler_agentid or ""}] + path
+        if not path:
+            path = [{"role": from_role or "main_agent", "agentid": from_agentid or ""}]
+
+        qid = getID()
+        ctx_json = json.dumps(context, ensure_ascii=False, default=str) if context else None
+        await sor.C('pipeline_agent_questions', {
+            'id': qid,
+            'tenant_id': tenant_id or '',
+            'task_id': task_id or '',
+            'from_role': from_role or '',
+            'from_agentid': from_agentid or '',
+            'problem_type': problem_type or '',
+            'question': question or '',
+            'context': ctx_json,
+            'escalation_path': json.dumps(path, ensure_ascii=False),
+            'escalation_pos': 0,
+            'status': S_PENDING,
+        })
+        if suspend_task and task_id:
+            # 挂起任务并清 claimed_by（否则 poller 要求 claimed_by IS NULL 才重新认领，任务僵尸）
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$",
+                {"tid": task_id})
+        logger.info("raise_problem: pipeline=%s type=%s from=%s/%s qid=%s path=%s",
+                    pipeline_id, problem_type, from_role, from_agentid, qid, path)
+        return qid
+
+
+async def resolve_problem(question_id, answer, answered_by='', answer_source='main_agent',
+                          resume_task=True):
+    """当前处理方解决了问题 → 停止冒泡(status=answered)，任务恢复 submitted。"""
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.R('pipeline_agent_questions', {'id': question_id})
+        if not recs:
+            return None
+        rec = recs[0]
+        task_id = getattr(rec, 'task_id', '') or ''
+
+        await sor.U('pipeline_agent_questions', {
+            'id': question_id,
+            'answer': answer or '',
+            'answered_by': answered_by or '',
+            'answer_source': answer_source or 'main_agent',
+            'status': S_ANSWERED,
+        })
+
+        resumed = False
+        if resume_task and task_id:
+            trecs = await sor.R('pipeline_tasks', {'id': task_id})
+            if trecs:
+                tstate = getattr(trecs[0], 'state', '')
+                if tstate == 'waiting':
+                    await sor.sqlExe(
+                        "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL "
+                        "WHERE id=${tid}$", {"tid": task_id})
+                    resumed = True
+        logger.info("resolve_problem: qid=%s task=%s by=%s resumed=%s",
+                    question_id, task_id, answered_by, resumed)
+        return {'id': question_id, 'status': S_ANSWERED, 'resumed': resumed}
+
+
+async def escalate_problem(question_id, by_role='', by_agentid='', note=''):
+    """当前处理方没解决 → 冒泡到路径下一个处理方（pos+1）。"""
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.R('pipeline_agent_questions', {'id': question_id})
+        if not recs:
+            return None
+        rec = recs[0]
+        path = _parse_path(getattr(rec, 'escalation_path', ''))
+        try:
+            pos = int(getattr(rec, 'escalation_pos', 0) or 0)
+        except (TypeError, ValueError):
+            pos = 0
+        new_pos = pos + 1
+        if path and new_pos >= len(path):
+            new_pos = len(path) - 1  # 尽头：停在最后一个处理方（通常是"人"兜底）
+        elif not path:
+            new_pos = 0
+        await sor.U('pipeline_agent_questions', {
+            'id': question_id,
+            'escalation_pos': new_pos,
+        })
+        handler = path[new_pos] if path else {}
+        logger.info("escalate_problem: qid=%s by=%s/%s -> %s (pos %d)",
+                    question_id, by_role, by_agentid, handler, new_pos)
+        return {'id': question_id, 'escalation_pos': new_pos, 'current_handler': handler}
+
+
+async def forward_to_customer(question_id, by_role='main_agent', by_agentid=''):
+    """便捷封装：主 agent 答不了，直接冒泡给客户(人)。等价于 escalate。"""
+    return await escalate_problem(question_id, by_role=by_role, by_agentid=by_agentid)
+
+
+async def list_problems_for(role, agentid=None, tenant_id=None, task_id=None, limit=50) -> list:
+    """列出「当前该 (role, agentid) 处理」的 pending 问题。
+
+    多 agent 感知：agentid 指定时，只返回"该 agent 处理"或"该角色任意 agent 处理"
+    的问题；不指定 agentid 时返回该角色所有待处理问题。
+
+    替代旧版「按 status 无差别列全部 pending」——把归属不同的问题混给同一个处理者。
+    """
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        conditions = ["status='pending'"]
+        params = {}
+        if tenant_id:
+            conditions.append("tenant_id=${tenant_id}$")
+            params["tenant_id"] = tenant_id
+        if task_id:
+            conditions.append("task_id=${task_id}$")
+            params["task_id"] = task_id
+        where = " AND ".join(conditions)
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError):
+            limit = 50
+        sql = f"SELECT * FROM pipeline_agent_questions WHERE {where} ORDER BY created_at DESC LIMIT {limit}"
+        recs = await sor.sqlExe(sql, params)
+        # 释放 SELECT 元数据锁
+        await sor.sqlExe("COMMIT", {})
+
+        result = []
+        for rec in (recs or []):
+            h = current_handler_of(rec)
+            if not _handler_matches(h, role, agentid):
+                continue
+            d = _rec_to_dict(rec)
+            d['current_handler'] = h
+            result.append(d)
+        return result
+
+
+def _rec_to_dict(rec):
+    if hasattr(rec, '__dict__'):
+        return {k: getattr(rec, k) for k in dir(rec)
+                if not k.startswith('_') and not callable(getattr(rec, k))}
+    return dict(rec)
+
+
+async def get_task_qa(task_id, include_pending_for=None,
+                      include_pending_agentid=None) -> dict:
+    """取任务的历史问答（已回答）+ 可选「待某处理方处理」的 pending 退回意见。
+
+    Args:
+        include_pending_for: 角色名，额外返回当前该角色应响应的 pending 问题
+        include_pending_agentid: 具体 agent 标识（多 agent 场景）
+
+    Returns:
+        {"answered": [...], "pending": [...]}
+    """
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.sqlExe(
+            "SELECT * FROM pipeline_agent_questions "
+            "WHERE task_id=${tid}$ AND status='answered' ORDER BY created_at ASC",
+            {"tid": task_id})
+        answered = []
+        for rec in (recs or []):
+            answered.append({
+                "question": getattr(rec, 'question', '') or '',
+                "answer": getattr(rec, 'answer', '') or '',
+                "answer_source": getattr(rec, 'answer_source', '') or '',
+                "from_role": getattr(rec, 'from_role', '') or '',
+            })
+
+        pending = []
+        if include_pending_for:
+            precs = await sor.sqlExe(
+                "SELECT * FROM pipeline_agent_questions "
+                "WHERE task_id=${tid}$ AND status='pending' ORDER BY created_at ASC",
+                {"tid": task_id})
+            for rec in (precs or []):
+                h = current_handler_of(rec)
+                if _handler_matches(h, include_pending_for, include_pending_agentid):
+                    pending.append({
+                        "question": getattr(rec, 'question', '') or '',
+                        "from_role": getattr(rec, 'from_role', '') or '',
+                        "problem_type": getattr(rec, 'problem_type', '') or '',
+                        "current_handler": h,
+                    })
+        return {"answered": answered, "pending": pending}

pipeline_service/init.py

@@ -37,6 +37,10 @@ from .questions import (
     agent_ask, answer_question, forward_question,
     get_question, list_questions, get_task_qna,
 )
+from .communication import (
+    raise_problem, resolve_problem, escalate_problem,
+    forward_to_customer, list_problems_for, get_task_qa,
+)
 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
 
@@ -539,6 +543,14 @@ def load_pipeline_service():
     env.question_list = list_questions
     env.question_qna = get_task_qna
 
+    # 团队沟通（通用问题冒泡引擎，多 agent 感知）
+    env.raise_problem = raise_problem
+    env.resolve_problem = resolve_problem
+    env.escalate_problem = escalate_problem
+    env.problem_forward = forward_to_customer
+    env.problem_list_for = list_problems_for
+    env.problem_qa = get_task_qa
+
     # DevOps: git/shell operations (v3.2.1)
     env.shell_exec = shell_exec
     env.skill_import_git = skill_import_git

pipeline_service/questions.py

@@ -1,28 +1,28 @@
-"""问题回路 — 角色agent缺信息时提问，主agent路由回答。
+"""问题回路 — 向后兼容层，委托 communication.py 通用冒泡引擎。
 
-流程：
-1. 角色agent执行任务时发现缺关键信息 → agent_ask() 写入 pipeline_agent_questions
-   (status=pending)，任务置 waiting，退出本轮执行。
-2. 主agent轮询 pending 问题：
-   - 能直接回答 → answer_question(answer_source='main_agent')，任务回 submitted，
-     角色agent下轮重新认领并带着问答历史继续。
-   - 答不了 → forward_question() 转客户 (status=forwarded)；
-     客户在会话中回答后，主agent调 answer_question(answer_source='customer')。
+历史状态 pending/forwarded/answered 收敛为通用模型：
+- 问题类型(pipeline_problem_types)定义冒泡路径(escalation_path)
+- 问题沿路径冒泡：解决→answered(停)，没解决→pos+1(沿路径走)，尽头=人兜底
+- 归属不再靠 status 无差别路由，而靠 current_handler(escalation_path+pos)
+
+本文件保留旧签名供 DSPY(env.question_*) 与既有调用方使用，内部全部走 communication.py。
 """
 
 import json
 import logging
 from sqlor.dbpools import DBPools
-from appPublic.uniqueID import getID
+
+from .communication import (
+    raise_problem,
+    resolve_problem,
+    forward_to_customer,
+    list_problems_for,
+    get_task_qa,
+)
 
 DBNAME = "pipeline"
 logger = logging.getLogger("pipeline.questions")
 
-# 问题状态
-Q_PENDING = "pending"      # 待主agent回答
-Q_FORWARDED = "forwarded"  # 主agent已转客户，等客户回复
-Q_ANSWERED = "answered"    # 已回答
-
 
 def _get_db():
     db = DBPools()
@@ -34,86 +34,27 @@ def _get_db():
     return db, DBNAME
 
 
-async def agent_ask(tenant_id: str, task_id: str, from_role: str, question: str, context: dict = None) -> str:
-    """角色agent提问。任务置 waiting，等待主agent/客户回答。返回 question_id。"""
-    db, dbname = _get_db()
-    async with db.sqlorContext(dbname) as sor:
-        qid = getID()
-        ctx_json = json.dumps(context, ensure_ascii=False, default=str) if context else None
-        await sor.C('pipeline_agent_questions', {
-            "id": qid,
-            "tenant_id": tenant_id or "",
-            "task_id": task_id or "",
-            "from_role": from_role or "",
-            "question": question or "",
-            "context": ctx_json,
-            "status": Q_PENDING,
-        })
-        if task_id:
-            # 置 waiting 同时清 claimed_by，否则任务僵尸（poll器要求 claimed_by IS NULL 才会重新认领）
-            await sor.sqlExe(
-                "UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$",
-                {"tid": task_id})
-        logger.info(f"agent_ask: task={task_id} role={from_role} qid={qid}")
-        return qid
+async def agent_ask(tenant_id: str, task_id: str, from_role: str, question: str,
+                    context: dict = None) -> str:
+    """角色 agent / PM 提问。按 from_role 推断问题类型，走通用冒泡引擎。"""
+    # 审核退回(PM) vs 故障报告(cockpit/pm) vs 缺信息(角色)：
+    ptype = "review_reject" if from_role == "pm" else (
+        "fault_report" if from_role == "cockpit" else "need_info")
+    return await raise_problem(
+        "", ptype, from_role, question,
+        task_id=task_id, tenant_id=tenant_id, context=context)
 
 
 async def answer_question(question_id: str, answer: str, answered_by: str = "",
                           answer_source: str = "main_agent", resume: bool = True):
-    """回答问题。任务从 waiting 回 submitted，让角色agent重新认领继续执行。
-
-    Args:
-        answer_source: 'main_agent' 或 'customer'
-        resume: 是否把任务恢复为 submitted
-
-    Returns:
-        更新后的问题记录 dict；问题不存在返回 None。
-    """
-    db, dbname = _get_db()
-    async with db.sqlorContext(dbname) as sor:
-        recs = await sor.R('pipeline_agent_questions', {'id': question_id})
-        if not recs:
-            return None
-        rec = recs[0]
-        task_id = rec.task_id if hasattr(rec, 'task_id') else rec['task_id']
-
-        await sor.U('pipeline_agent_questions', {
-            "id": question_id,
-            "answer": answer or "",
-            "answered_by": answered_by or "",
-            "answer_source": answer_source or "main_agent",
-            "status": Q_ANSWERED,
-        })
-
-        resumed = False
-        if resume and task_id:
-            trecs = await sor.R('pipeline_tasks', {'id': task_id})
-            if trecs:
-                t = trecs[0]
-                tstate = t.state if hasattr(t, 'state') else t['state']
-                if tstate == "waiting":
-                    # 恢复 submitted 同时清 claimed_by，否则 poll 器永不重新认领该任务
-                    await sor.sqlExe(
-                        "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL WHERE id=${tid}$",
-                        {"tid": task_id})
-                    resumed = True
-
-        logger.info(f"answer_question: qid={question_id} task={task_id} source={answer_source} resumed={resumed}")
-        return {
-            "id": question_id,
-            "task_id": task_id,
-            "status": Q_ANSWERED,
-            "answer_source": answer_source,
-            "resumed": resumed,
-        }
+    """回答问题（解决 → 停止冒泡，任务恢复 submitted）。"""
+    return await resolve_problem(question_id, answer, answered_by, answer_source,
+                                 resume_task=resume)
 
 
 async def forward_question(question_id: str):
-    """主agent答不了 → 转给客户。status: pending → forwarded。"""
-    db, dbname = _get_db()
-    async with db.sqlorContext(dbname) as sor:
-        await sor.U('pipeline_agent_questions', {"id": question_id, "status": Q_FORWARDED})
-        logger.info(f"forward_question: qid={question_id}")
+    """主 agent 答不了 → 沿路径冒泡给客户(人)。"""
+    return await forward_to_customer(question_id)
 
 
 async def get_question(question_id: str):
@@ -139,58 +80,18 @@ 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:
-    """问题列表，支持租户/任务/状态过滤。"""
-    db, dbname = _get_db()
-    async with db.sqlorContext(dbname) as sor:
-        conditions = []
-        params = {}
-        if tenant_id:
-            conditions.append("tenant_id=${tenant_id}$")
-            params["tenant_id"] = tenant_id
-        if task_id:
-            conditions.append("task_id=${task_id}$")
-            params["task_id"] = task_id
-        if status:
-            conditions.append("status=${status}$")
-            params["status"] = status
-        where = " AND ".join(conditions) if conditions else "1=1"
-        try:
-            limit = int(limit)
-        except (TypeError, ValueError):
-            limit = 50
-        sql = f"SELECT * FROM pipeline_agent_questions WHERE {where} ORDER BY created_at DESC LIMIT {limit}"
-        recs = await sor.sqlExe(sql, params)
-        result = []
-        for rec in (recs or []):
-            if hasattr(rec, '__dict__'):
-                d = {k: getattr(rec, k) for k in dir(rec)
-                     if not k.startswith('_') and not callable(getattr(rec, k))}
-            else:
-                d = dict(rec)
-            ctx = d.get('context')
-            if ctx and isinstance(ctx, str):
-                try:
-                    d['context'] = json.loads(ctx)
-                except (json.JSONDecodeError, TypeError):
-                    pass
-            result.append(d)
-        return result
+    """客户视角：列出当前该「客户(人)」处理的问题。
+
+    旧签名兼容 status 过滤；无 status 时列客户待处理(pending)。
+    """
+    rows = await list_problems_for("customer", tenant_id=tenant_id,
+                                   task_id=task_id, limit=limit)
+    if status:
+        rows = [r for r in rows if (r.get('status') or '') == status]
+    return rows
 
 
 async def get_task_qna(task_id: str) -> list:
-    """取任务的全部已回答问答（按时间序），用于角色agent继续执行时注入prompt。"""
-    db, dbname = _get_db()
-    async with db.sqlorContext(dbname) as sor:
-        recs = await sor.sqlExe(
-            "SELECT question, answer, answer_source, from_role FROM pipeline_agent_questions "
-            "WHERE task_id=${tid}$ AND status='answered' ORDER BY created_at ASC",
-            {"tid": task_id})
-        result = []
-        for rec in (recs or []):
-            result.append({
-                "question": getattr(rec, 'question', '') or '',
-                "answer": getattr(rec, 'answer', '') or '',
-                "answer_source": getattr(rec, 'answer_source', '') or '',
-                "from_role": getattr(rec, 'from_role', '') or '',
-            })
-        return result
+    """取任务的全部已回答问答（旧返回结构：list），供角色 agent 注入 prompt。"""
+    qa = await get_task_qa(task_id)
+    return qa.get('answered', []) if isinstance(qa, dict) else []

pipeline_service/sdlc_ability.py

@@ -75,16 +75,22 @@ SDL_TOOLS = [
     ),
     ToolDefinition(
         name="list_questions",
-        description="查看待回答的问题",
+        description="查看待我(主agent)处理的问题",
         parameters={},
         category="agent",
     ),
     ToolDefinition(
         name="answer_question",
-        description="回答agent提出的问题",
+        description="回答agent提出的问题（解决→停止冒泡）",
         parameters={"question_id": "问题ID", "answer": "回答内容"},
         category="agent",
     ),
+    ToolDefinition(
+        name="escalate_question",
+        description="答不了的问题沿冒泡路径转给下一个处理方(通常转客户)",
+        parameters={"question_id": "问题ID"},
+        category="agent",
+    ),
     ToolDefinition(
         name="add_repo",
         description="添加项目关联的Git仓库",
@@ -129,7 +135,7 @@ SDL_PROMPT = """你是「开发产线」的驾驶舱 agent，负责软件项目
 - 用户报告异常/故障 → 先用 diagnose_project 定位根因，再用 task_detail 查详情。
 - 仓库管理 → add_repo / list_repos / clone_repo。
 - 发现卡点（任务卡死、审核超时、失败、僵尸 claimed_by）时，主动定位根因并推动修复。
-- 发现「待回答问题」> 0 或角色提问时，加载 question-escalation 技能按问题升级链处理（列问题 → 能答则答 → 不能答抛给客户）。"""
+- 发现「待回答问题」> 0 或角色提问时，加载 team-communication 技能按冒泡链处理（list_questions 列问题 → 能答则 answer_question → 不能答 escalate_question 沿冒泡路径转下一个处理方）。"""
 
 
 # ── 产线角色集（可插拔，替代硬编码 ROLE_SPECIFICS/ROLE_ALIASES/ROLE_CHAIN） ──
@@ -337,10 +343,10 @@ async def _h_diagnose_project(sor, p, ctx):
 
     qc = 0
     try:
-        questions = await sor.sqlExe(
-            "SELECT COUNT(*) as c FROM pipeline_agent_questions WHERE tenant_id=${pid}$ AND status='pending'",
-            {"pid": pid})
-        qc = getattr(questions[0], "c", 0) if questions else 0
+        from .communication import list_problems_for
+        # 待回答问题 = 当前该「主 agent」处理的问题（不混入 PM 退回给角色 agent 的）
+        mine = await list_problems_for("main_agent", tenant_id=pid, limit=200)
+        qc = len(mine)
     except Exception:
         pass
 
@@ -424,27 +430,41 @@ async def _h_list_questions(sor, p, ctx):
     pid = ctx.get("project_id", "")
     if not pid:
         return "请先切换到项目"
-    try:
-        recs = await sor.sqlExe(
-            "SELECT id, question, answer_source FROM pipeline_agent_questions "
-            "WHERE tenant_id=${pid}$ AND status='pending' ORDER BY created_at DESC LIMIT 10",
-            {"pid": pid})
-    except Exception:
-        try:
-            recs = await sor.sqlExe(
-                "SELECT id, question, answer_source FROM pipeline_agent_questions "
-                "WHERE tenant_id=${pid}$ ORDER BY created_at DESC LIMIT 10",
-                {"pid": pid})
-        except Exception:
-            return "暂无问题数据"
+    from .communication import list_problems_for
+    # 只列「当前该主 agent 处理」的问题（不混入 PM 退回给角色 agent 的退回意见）
+    recs = await list_problems_for("main_agent", tenant_id=pid, limit=10)
     if not recs:
         return "没有待回答问题"
     lines = []
     for r in recs:
-        lines.append(f"- [{getattr(r, 'id', '')}] {getattr(r, 'question', '')[:200]}")
+        qid = r.get('id', '') or ''
+        q = (r.get('question', '') or '')[:200]
+        ptype = r.get('problem_type', '') or ''
+        tag = f"[{ptype}] " if ptype else ""
+        lines.append(f"- [{qid}] {tag}{q}")
     return "\n".join(lines)
 
 
+async def _h_escalate_question(sor, p, ctx):
+    qid = p.get("question_id", "")
+    if not qid:
+        return "需要问题ID"
+    from .communication import forward_to_customer
+    # 模糊匹配短 id
+    full_qid = qid
+    if len(qid) < 32:
+        recs = await sor.sqlExe(
+            "SELECT id FROM pipeline_agent_questions WHERE id LIKE ${prefix}$ LIMIT 1",
+            {"prefix": qid + "%"})
+        if recs:
+            full_qid = getattr(recs[0], "id", qid)
+    r = await forward_to_customer(full_qid, by_role="main_agent")
+    if r is None:
+        return f"问题不存在: {qid}"
+    h = r.get("current_handler", {}) or {}
+    return f"OK: 已沿冒泡路径转给 {h.get('role', '?')}"
+
+
 async def _h_answer_question(sor, p, ctx):
     qid = p.get("question_id", "")
     answer = p.get("answer", "")
@@ -594,6 +614,7 @@ SDL_HANDLERS = {
     "view_deliverable": _h_view_deliverable,
     "list_questions": _h_list_questions,
     "answer_question": _h_answer_question,
+    "escalate_question": _h_escalate_question,
     "add_repo": _h_add_repo,
     "list_repos": _h_list_repos,
     "clone_repo": _h_clone_repo,