From 7fa501d3597956e02230614adc7008bfdd101e29 Mon Sep 17 00:00:00 2001
From: ymq <ymq@opencomputing.net>
Date: Sun, 16 Aug 2026 23:55:14 +0800
Subject: [PATCH] =?UTF-8?q?feat(audit):=20question=20=E8=BF=81=E7=A7=BB?=
 =?UTF-8?q?=E6=8E=A5=20record=5Faudit?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

raise_problem(raise)/resolve_problem(resolve)/escalate_problem(escalate) 都写审计，
问题流转轨迹 append-only 可追溯（谁+何时+from→to）
---
 pipeline_service/communication.py | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/pipeline_service/communication.py b/pipeline_service/communication.py
index 7b7e4e8..a48c10c 100644
--- a/pipeline_service/communication.py
+++ b/pipeline_service/communication.py
@@ -18,6 +18,7 @@ import json
 import logging
 from sqlor.dbpools import DBPools
 from appPublic.uniqueID import getID
+from .audit import record_audit
 
 DBNAME = "pipeline"
 logger = logging.getLogger("pipeline.communication")
@@ -78,6 +79,9 @@ async def raise_problem(problem_type, question, from_role, from_agentid="",
             await sor.sqlExe(
                 "UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$",
                 {"tid": task_id})
+        await record_audit(tenant_id or '', 'pipeline_agent_questions', qid, 'raise',
+                           to_state=S_PENDING, who=from_role, agent_id=from_agentid,
+                           detail=problem_type, sor=sor)
         logger.info("raise_problem: type=%s from=%s/%s qid=%s -> handler=%s/%s",
                     problem_type, from_role, from_agentid, qid, hr, ha)
         return qid
@@ -93,6 +97,7 @@ async def resolve_problem(question_id, answer, answered_by="", answer_source="ag
             return None
         rec = recs[0]
         task_id = getattr(rec, 'task_id', '') or ''
+        tenant_id = getattr(rec, 'tenant_id', '') or ''
 
         await sor.U('pipeline_agent_questions', {
             'id': question_id,
@@ -101,6 +106,8 @@ async def resolve_problem(question_id, answer, answered_by="", answer_source="ag
             'answer_source': answer_source or 'agent.main_agent',
             'status': S_ANSWERED,
         })
+        await record_audit(tenant_id, 'pipeline_agent_questions', question_id, 'resolve',
+                           from_state=S_PENDING, to_state=S_ANSWERED, who=answered_by, sor=sor)
 
         resumed = False
         if resume_task and task_id:
@@ -128,11 +135,16 @@ async def escalate_problem(question_id, next_handler_role, next_handler_agentid=
         recs = await sor.R('pipeline_agent_questions', {'id': question_id})
         if not recs:
             return None
+        rec = recs[0]
+        tenant_id = getattr(rec, 'tenant_id', '') or ''
+        old_role = getattr(rec, 'current_handler_role', '') or ''
         await sor.U('pipeline_agent_questions', {
             'id': question_id,
             'current_handler_role': next_handler_role or '',
             'current_handler_agentid': next_handler_agentid or '',
         })
+        await record_audit(tenant_id, 'pipeline_agent_questions', question_id, 'escalate',
+                           from_state=old_role, to_state=next_handler_role, who=by_role, sor=sor)
         logger.info("escalate_problem: qid=%s by=%s/%s -> %s/%s",
                     question_id, by_role, by_agentid,
                     next_handler_role, next_handler_agentid)