From 6dac18e1b1ebe11609b4045b5da406db5e16cdc3 Mon Sep 17 00:00:00 2001
From: ymq <ymq@opencomputing.net>
Date: Tue, 15 Sep 2026 16:52:43 +0800
Subject: [PATCH] =?UTF-8?q?fix(comms):=20raise=5Fproblem=E5=90=8C=E4=BB=BB?=
 =?UTF-8?q?=E5=8A=A1=E5=90=8C=E7=B1=BB=E5=9E=8Bpending=E9=97=AE=E9=A2=98?=
 =?UTF-8?q?=E5=8E=BB=E9=87=8D=E2=80=94=E2=80=94=E9=87=8D=E5=A4=8D=E5=86=92?=
 =?UTF-8?q?=E6=B3=A1=E6=9B=B4=E6=96=B0=E6=97=A2=E6=9C=89=E8=A1=8C(?=
 =?UTF-8?q?=E6=AD=A3=E6=96=87/context=E5=88=B7=E6=96=B0+=E4=BB=BB=E5=8A=A1?=
 =?UTF-8?q?=E7=85=A7=E6=8C=82=E8=B5=B7)=E8=BF=94=E5=9B=9E=E5=8E=9Fqid,?=
 =?UTF-8?q?=E4=B8=8D=E5=86=8D=E6=AF=8F=E8=BD=AE=E6=96=B0=E5=BB=BA(?=
 =?UTF-8?q?=E5=AE=9E=E6=B5=8B=E5=90=8Ctask=5Fid=20qc=5Freject/review=5Frej?=
 =?UTF-8?q?ect/fault=5Freport=E5=A0=862-3=E6=9D=A1pending,=E8=A7=92?=
 =?UTF-8?q?=E6=A0=87=E8=99=9A=E9=AB=98=E9=87=8D=E5=A4=8D=E8=BD=B0=E7=82=B8?=
 =?UTF-8?q?owner.superuser,2026-09-15=E7=94=A8=E6=88=B7=E6=8C=87=E4=BB=A4:?=
 =?UTF-8?q?=E9=81=BF=E5=85=8D=E9=87=8D=E5=A4=8D)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pipeline_service/communication.py | 31 +++++++++++++++++++++++++++++++
 1 file changed, 31 insertions(+)

diff --git a/pipeline_service/communication.py b/pipeline_service/communication.py
index 0ab01b2..727f7f7 100644
--- a/pipeline_service/communication.py
+++ b/pipeline_service/communication.py
@@ -82,6 +82,37 @@ async def raise_problem(problem_type, question, from_role, from_agentid="",
                 return ""
         hr = first_handler_role or "agent.main_agent"
         ha = first_handler_agentid or ""
+        # 重复冒泡去重（2026-09-15 用户指令「避免重复」）：同任务+同问题类型已有
+        # pending 问题 → 不新建，把正文/context 更新到既有行并返回既有 qid。
+        # 实测历史数据：同 task_id 的 qc_reject/review_reject/fault_report 各堆 2~3 条
+        # pending（每轮退回都新发一条，旧的不答结），角标虚高、用户重复看到同一事项。
+        _tk = (task_id or '').strip()
+        if _tk and problem_type:
+            _dup = await sor.sqlExe(
+                "SELECT id FROM pipeline_agent_questions WHERE task_id=${tk}$ "
+                "AND problem_type=${pt}$ AND status='pending' "
+                "ORDER BY created_at DESC LIMIT 1",
+                {"tk": _tk, "pt": problem_type})
+            await sor.sqlExe("COMMIT", {})
+            if _dup:
+                old_qid = getattr(_dup[0], 'id', '')
+                _ctx2 = dict(context or {})
+                if form_schema:
+                    _ctx2['form_schema'] = form_schema
+                await sor.sqlExe(
+                    "UPDATE pipeline_agent_questions SET question=${q}$, context=${c}$, "
+                    "updated_at=NOW() WHERE id=${i}$",
+                    {"q": question or '', "c": json.dumps(_ctx2, ensure_ascii=False, default=str)
+                     if _ctx2 else None, "i": old_qid})
+                await sor.sqlExe("COMMIT", {})
+                if suspend_task:
+                    await sor.sqlExe(
+                        "UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$",
+                        {"tid": _tk})
+                    await sor.sqlExe("COMMIT", {})
+                logger.info("raise_problem dedup: task=%s type=%s reuse qid=%s",
+                            _tk, problem_type, old_qid)
+                return old_qid
         qid = getID()
         # form_schema 并入 context 存储（待办详情按它渲染动态表单）
         if form_schema: