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: