yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(audit): question 迁移接 record_audit

Browse Source 
raise_problem(raise)/resolve_problem(resolve)/escalate_problem(escalate) 都写审计，
问题流转轨迹 append-only 可追溯（谁+何时+from→to）
This commit is contained in:
ymq 2026-08-16 23:55:14 +08:00
parent bf819f3b79
commit 7fa501d359
1 changed files with 12 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
pipeline_service/communication.py
View File

@ -18,6 +18,7 @@ import json
import logging import logging
from sqlor.dbpools import DBPools from sqlor.dbpools import DBPools
from appPublic.uniqueID import getID from appPublic.uniqueID import getID
from .audit import record_audit
DBNAME = "pipeline" DBNAME = "pipeline"
logger = logging.getLogger("pipeline.communication") logger = logging.getLogger("pipeline.communication")
@ -78,6 +79,9 @@ async def raise_problem(problem_type, question, from_role, from_agentid="",
await sor.sqlExe( await sor.sqlExe(
"UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$", "UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$",
{"tid": task_id}) {"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", logger.info("raise_problem: type=%s from=%s/%s qid=%s -> handler=%s/%s",
problem_type, from_role, from_agentid, qid, hr, ha) problem_type, from_role, from_agentid, qid, hr, ha)
return qid return qid
@ -93,6 +97,7 @@ async def resolve_problem(question_id, answer, answered_by="", answer_source="ag
return None return None
rec = recs[0] rec = recs[0]
task_id = getattr(rec, 'task_id', '') or '' task_id = getattr(rec, 'task_id', '') or ''
tenant_id = getattr(rec, 'tenant_id', '') or ''
await sor.U('pipeline_agent_questions', { await sor.U('pipeline_agent_questions', {
'id': question_id, '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', 'answer_source': answer_source or 'agent.main_agent',
'status': S_ANSWERED, '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 resumed = False
if resume_task and task_id: 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}) recs = await sor.R('pipeline_agent_questions', {'id': question_id})
if not recs: if not recs:
return None 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', { await sor.U('pipeline_agent_questions', {
'id': question_id, 'id': question_id,
'current_handler_role': next_handler_role or '', 'current_handler_role': next_handler_role or '',
'current_handler_agentid': next_handler_agentid 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", logger.info("escalate_problem: qid=%s by=%s/%s -> %s/%s",
question_id, by_role, by_agentid, question_id, by_role, by_agentid,
next_handler_role, next_handler_agentid) next_handler_role, next_handler_agentid)