"""Platform read model for full PMFlow AI run audits.""" from __future__ import annotations from datetime import datetime, timedelta, timezone from typing import Any from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from ..models import ( AICallLog, PMFlowAgentArtifact, PMFlowAgentMemory, PMFlowAgentObservation, PMFlowAgentRun, PMFlowAgentStep, PMFlowAgentVerification, PMFlowAIMessage, PMFlowAIThread, ) from .ai_call_trace_service import AICallTraceService def _utc_iso(value: datetime | None) -> str | None: if value is None: return None if value.tzinfo is None: value = value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc).isoformat() def _derived_started_at(completed_at: datetime | None, duration_ms: int | None) -> datetime | None: if completed_at is None or duration_ms is None: return None try: duration = int(duration_ms) except (TypeError, ValueError): return None if duration < 0: return None return completed_at - timedelta(milliseconds=duration) def _local_tz() -> timezone: tz = datetime.now().astimezone().tzinfo return tz if isinstance(tz, timezone) else timezone.utc def _coerce_time(value: Any) -> datetime | None: if value is None: return None if isinstance(value, datetime): parsed = value else: try: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) except (TypeError, ValueError): return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=_local_tz()) return parsed def _local_iso(value: Any) -> str | None: parsed = _coerce_time(value) return parsed.isoformat() if parsed else None def _derived_started_iso(completed_at: Any, duration_ms: Any) -> str | None: completed = _coerce_time(completed_at) if completed is None or duration_ms is None: return None try: duration = int(duration_ms) except (TypeError, ValueError): return None if duration < 0: return None return (completed - timedelta(milliseconds=duration)).isoformat() def _timeline_sort_timestamp(value: Any) -> float: parsed = _coerce_time(value) if parsed is None: return float("inf") return parsed.timestamp() def _timeline_description(value: Any, *, fallback: str = "") -> str: text = " ".join(str(value or "").strip().split()) return text[:500] or fallback class AgentRunAuditService: def __init__(self, db: AsyncSession) -> None: self.db = db async def list_runs( self, *, page: int = 1, page_size: int = 30, tenant_id: int | None = None, user_id: int | None = None, status: str | None = None, source_channel: str | None = None, keyword: str | None = None, since_hours: int | None = 168, ) -> dict[str, Any]: page = max(1, int(page or 1)) page_size = min(max(1, int(page_size or 30)), 100) filters = self._run_filters( tenant_id=tenant_id, user_id=user_id, status=status, source_channel=source_channel, keyword=keyword, since_hours=since_hours, ) where_clause = and_(*filters) if filters else None count_stmt = select(func.count(PMFlowAgentRun.id)) if where_clause is not None: count_stmt = count_stmt.where(where_clause) total = int((await self.db.execute(count_stmt)).scalar() or 0) stmt = ( select(PMFlowAgentRun) .order_by(PMFlowAgentRun.started_at.desc(), PMFlowAgentRun.id.desc()) .offset((page - 1) * page_size) .limit(page_size) ) if where_clause is not None: stmt = stmt.where(where_clause) rows = list((await self.db.execute(stmt)).scalars().all()) run_ids = [str(row.public_run_id) for row in rows if row.public_run_id] counts = await self._counts_for_runs(run_ids) return { "summary": await self._summary(filters=filters), "items": [self._run_summary(row, counts=counts.get(str(row.public_run_id), {})) for row in rows], "page": page, "page_size": page_size, "total": total, "has_next": page * page_size < total, } async def get_run(self, public_run_id: str) -> dict[str, Any]: run_id = str(public_run_id or "").strip() row = ( await self.db.execute( select(PMFlowAgentRun) .where(PMFlowAgentRun.public_run_id == run_id) .order_by(PMFlowAgentRun.id.desc()) .limit(1) ) ).scalar_one_or_none() if row is None: raise ValueError("链路审计记录不存在") steps = list( ( await self.db.execute( select(PMFlowAgentStep) .where(and_(PMFlowAgentStep.tenant_id == int(row.tenant_id), PMFlowAgentStep.run_id == int(row.id))) .order_by(PMFlowAgentStep.step_index.asc(), PMFlowAgentStep.id.asc()) ) ) .scalars() .all() ) observations = list( ( await self.db.execute( select(PMFlowAgentObservation) .where(and_(PMFlowAgentObservation.tenant_id == int(row.tenant_id), PMFlowAgentObservation.run_id == int(row.id))) .order_by(PMFlowAgentObservation.observed_at.asc(), PMFlowAgentObservation.id.asc()) ) ) .scalars() .all() ) artifacts = list( ( await self.db.execute( select(PMFlowAgentArtifact) .where(and_(PMFlowAgentArtifact.tenant_id == int(row.tenant_id), PMFlowAgentArtifact.run_id == int(row.id))) .order_by(PMFlowAgentArtifact.id.asc()) ) ) .scalars() .all() ) verifications = list( ( await self.db.execute( select(PMFlowAgentVerification) .where(and_(PMFlowAgentVerification.tenant_id == int(row.tenant_id), PMFlowAgentVerification.run_id == int(row.id))) .order_by(PMFlowAgentVerification.verified_at.asc(), PMFlowAgentVerification.id.asc()) ) ) .scalars() .all() ) memories = list( ( await self.db.execute( select(PMFlowAgentMemory) .where(PMFlowAgentMemory.run_id == int(row.id)) .order_by(PMFlowAgentMemory.id.asc()) ) ) .scalars() .all() ) messages = await self._messages_for_run(row) ai_calls = await self._ai_calls_for_run(str(row.public_run_id)) counts = (await self._counts_for_runs([str(row.public_run_id)])).get(str(row.public_run_id), {}) return { **self._run_summary(row, counts=counts), "messages": messages, "steps": [self._step_detail(item, ai_calls=ai_calls) for item in steps], "observations": [self._observation_detail(item) for item in observations], "artifacts": [self._artifact_detail(item) for item in artifacts], "verifications": [self._verification_detail(item) for item in verifications], "memories": [self._memory_detail(item) for item in memories], "ai_calls": ai_calls, "timeline": self._timeline(row=row, steps=steps, observations=observations, verifications=verifications, ai_calls=ai_calls), "audit_policy": { "visibility": "platform_only", "payload_policy": "展示平台脱敏后的链路参数、上游 API 请求/响应、耗时和质量门禁；客户侧不暴露这些内部诊断。", }, } def _run_filters( self, *, tenant_id: int | None, user_id: int | None, status: str | None, source_channel: str | None, keyword: str | None, since_hours: int | None, ) -> list[Any]: filters: list[Any] = [] if since_hours and int(since_hours) > 0: filters.append(PMFlowAgentRun.started_at >= datetime.utcnow() - timedelta(hours=min(int(since_hours), 24 * 365))) if tenant_id: filters.append(PMFlowAgentRun.tenant_id == int(tenant_id)) if user_id: filters.append(PMFlowAgentRun.user_id == int(user_id)) if status and status != "all": filters.append(PMFlowAgentRun.status == str(status).strip()) if source_channel: filters.append(PMFlowAgentRun.source_channel == str(source_channel).strip()) if keyword: value = f"%{str(keyword).strip()}%" filters.append( or_( PMFlowAgentRun.public_run_id.ilike(value), PMFlowAgentRun.title.ilike(value), PMFlowAgentRun.goal.ilike(value), PMFlowAgentRun.final_answer.ilike(value), PMFlowAgentRun.source_ref.ilike(value), ) ) return filters async def _summary(self, *, filters: list[Any]) -> dict[str, Any]: where_clause = and_(*filters) if filters else None stmt = select(func.count(PMFlowAgentRun.id)) if where_clause is not None: stmt = stmt.where(where_clause) total = int((await self.db.execute(stmt)).scalar() or 0) status_stmt = select(PMFlowAgentRun.status, func.count(PMFlowAgentRun.id)).group_by(PMFlowAgentRun.status) if where_clause is not None: status_stmt = status_stmt.where(where_clause) status_counts = {str(status or "unknown"): int(count or 0) for status, count in (await self.db.execute(status_stmt)).all()} run_id_stmt = select(PMFlowAgentRun.public_run_id) if where_clause is not None: run_id_stmt = run_id_stmt.where(where_clause) call_stmt = select(func.count(AICallLog.id)).where(AICallLog.agent_run_id.in_(run_id_stmt)) return { "total_runs": total, "status_counts": status_counts, "linked_ai_calls": int((await self.db.execute(call_stmt)).scalar() or 0), } async def _counts_for_runs(self, run_ids: list[str]) -> dict[str, dict[str, int]]: if not run_ids: return {} result: dict[str, dict[str, int]] = {run_id: {} for run_id in run_ids} call_rows = await self.db.execute( select(AICallLog.agent_run_id, func.count(AICallLog.id), func.sum(AICallLog.latency_ms)) .where(AICallLog.agent_run_id.in_(run_ids)) .group_by(AICallLog.agent_run_id) ) for run_id, count, latency in call_rows.all(): result.setdefault(str(run_id), {})["ai_call_count"] = int(count or 0) result[str(run_id)]["ai_call_latency_ms"] = int(latency or 0) return result def _run_summary(self, row: PMFlowAgentRun, *, counts: dict[str, int]) -> dict[str, Any]: duration_ms = None if row.started_at and row.completed_at: duration_ms = max(0, int((row.completed_at - row.started_at).total_seconds() * 1000)) metadata = row.run_metadata if isinstance(row.run_metadata, dict) else {} return { "id": int(row.id), "public_run_id": row.public_run_id, "tenant_id": int(row.tenant_id), "user_id": int(row.user_id) if row.user_id is not None else None, "thread_id": int(row.thread_id) if row.thread_id is not None else None, "qa_session_id": int(row.qa_session_id) if row.qa_session_id is not None else None, "workspace_scope": row.workspace_scope, "surface": row.surface, "presentation_mode": row.presentation_mode, "source_channel": row.source_channel, "source_ref": row.source_ref, "title": row.title, "goal": row.goal, "status": row.status, "status_label": self._status_label(row.status), "active_object_refs": row.active_object_refs if isinstance(row.active_object_refs, list) else [], "final_answer_preview": self._preview(row.final_answer), "last_error": row.last_error, "started_at": _utc_iso(row.started_at), "completed_at": _utc_iso(row.completed_at), "duration_ms": duration_ms, "step_count": int(metadata.get("step_count") or 0), "observation_count": int(metadata.get("observation_count") or 0), "artifact_count": int(metadata.get("artifact_count") or 0), "verification_count": int(metadata.get("verification_count") or 0), "tool_call_count": int(metadata.get("tool_call_count") or 0), "iterations": int(metadata.get("iterations") or 0), "ai_call_count": int(counts.get("ai_call_count") or 0), "ai_call_latency_ms": int(counts.get("ai_call_latency_ms") or 0), "run_metadata": metadata, } async def _messages_for_run(self, row: PMFlowAgentRun) -> list[dict[str, Any]]: if not row.public_run_id: return [] result = await self.db.execute( select(PMFlowAIMessage, PMFlowAIThread.public_thread_id) .join(PMFlowAIThread, PMFlowAIThread.id == PMFlowAIMessage.thread_id) .where( and_( PMFlowAIMessage.tenant_id == int(row.tenant_id), PMFlowAIMessage.run_public_id == str(row.public_run_id), ) ) .order_by(PMFlowAIMessage.message_index.asc(), PMFlowAIMessage.id.asc()) ) return [ { "id": int(message.id), "public_thread_id": public_thread_id, "public_message_id": message.public_message_id, "role": message.role, "message_index": int(message.message_index or 0), "content": message.content, "payload": message.message_payload if isinstance(message.message_payload, dict) else {}, "sent_at": _utc_iso(message.sent_at), } for message, public_thread_id in result.all() ] async def _ai_calls_for_run(self, public_run_id: str) -> list[dict[str, Any]]: rows = list( ( await self.db.execute( select(AICallLog) .where(AICallLog.agent_run_id == public_run_id) .order_by(AICallLog.created_at.asc(), AICallLog.id.asc()) ) ) .scalars() .all() ) if not rows: return [] trace_service = AICallTraceService(self.db) runtime_index = await trace_service._runtime_display_index(rows) operator_index = await trace_service._operator_user_index(rows) return [ { **trace_service._row_detail(row, runtime_index=runtime_index, operator_user_index=operator_index), "agent_run_id": row.agent_run_id, "agent_step_key": row.agent_step_key, } for row in rows ] def _step_detail(self, row: PMFlowAgentStep, *, ai_calls: list[dict[str, Any]]) -> dict[str, Any]: payload = row.step_payload if isinstance(row.step_payload, dict) else {} linked_calls = [ self._ai_call_ref(item) for item in ai_calls if self._ai_call_matches_step(item, row=row, payload=payload) ] return { "id": int(row.id), "public_step_id": row.public_step_id, "step_index": int(row.step_index or 0), "step_kind": row.step_kind, "status": row.status, "input_summary": row.input_summary, "output_summary": row.output_summary, "tool_name": row.tool_name, "duration_ms": row.duration_ms, "step_payload": payload, "linked_ai_calls": linked_calls, "started_at": _utc_iso(row.started_at), "completed_at": _utc_iso(row.completed_at), } @staticmethod def _ai_call_matches_step(item: dict[str, Any], *, row: PMFlowAgentStep, payload: dict[str, Any]) -> bool: key = str(item.get("agent_step_key") or "").strip() if not key: return False key_kind_map = { "agent_goal_frame": "understand_goal", "agent_step_planner": "planner", "agent_evidence_normalizer": "extract_claims", "agent_state_critic": "critic", "agent_answer_runtime": "answer", "agent_answer_reviser": "answer", } candidates = { str(row.step_kind or "").strip(), str(row.tool_name or "").strip(), str(payload.get("kind") or "").strip(), str(payload.get("label") or "").strip(), } mapped = key_kind_map.get(key) if mapped and mapped in candidates: return True return key in candidates or any(candidate and (candidate in key or key in candidate) for candidate in candidates) @staticmethod def _ai_call_ref(item: dict[str, Any]) -> dict[str, Any]: return { "id": item.get("id"), "trace_id": item.get("trace_id"), "function_name": item.get("function_name"), "function_label": item.get("function_label"), "status": item.get("status"), "status_label": item.get("status_label"), "provider_label": item.get("provider_label"), "model": item.get("model"), "latency_ms": item.get("latency_ms"), "tokens_used": item.get("tokens_used"), "request_max_tokens": item.get("request_max_tokens"), "function_runtime_max_tokens": item.get("function_runtime_max_tokens"), "function_runtime_profile_version": item.get("function_runtime_profile_version"), "provider_timeout_seconds": (item.get("routing") or {}).get("routing_meta", {}).get("provider_timeout_seconds") if isinstance(item.get("routing"), dict) else None, "job_timeout_seconds": (item.get("routing") or {}).get("routing_meta", {}).get("job_timeout_seconds") if isinstance(item.get("routing"), dict) else None, } @staticmethod def _observation_detail(row: PMFlowAgentObservation) -> dict[str, Any]: return { "id": int(row.id), "step_id": int(row.step_id) if row.step_id is not None else None, "public_observation_id": row.public_observation_id, "source_kind": row.source_kind, "customer_safe_summary": row.customer_safe_summary, "confidence": row.confidence, "facts": row.facts if isinstance(row.facts, list) else [], "gaps": row.gaps if isinstance(row.gaps, list) else [], "raw_payload": row.raw_payload if isinstance(row.raw_payload, dict) else {}, "observed_at": _utc_iso(row.observed_at), } @staticmethod def _artifact_detail(row: PMFlowAgentArtifact) -> dict[str, Any]: return { "id": int(row.id), "step_id": int(row.step_id) if row.step_id is not None else None, "public_artifact_id": row.public_artifact_id, "artifact_kind": row.artifact_kind, "title": row.title, "status": row.status, "payload": row.payload if isinstance(row.payload, dict) else {}, "customer_safe_payload": row.customer_safe_payload if isinstance(row.customer_safe_payload, dict) else {}, "created_by_step_id": row.created_by_step_id, } @staticmethod def _verification_detail(row: PMFlowAgentVerification) -> dict[str, Any]: return { "id": int(row.id), "step_id": int(row.step_id) if row.step_id is not None else None, "public_verification_id": row.public_verification_id, "target_ref": row.target_ref, "result": row.result, "retry_would_change_state": bool(row.retry_would_change_state), "findings": row.findings if isinstance(row.findings, list) else [], "verification_payload": row.verification_payload if isinstance(row.verification_payload, dict) else {}, "verified_at": _utc_iso(row.verified_at), } @staticmethod def _memory_detail(row: PMFlowAgentMemory) -> dict[str, Any]: return { "id": int(row.id), "memory_kind": row.memory_kind, "scope_ref": row.scope_ref, "outcome_signal": row.outcome_signal, "content": row.content if isinstance(row.content, dict) else {}, "memory_payload": row.memory_payload if isinstance(row.memory_payload, dict) else {}, } @staticmethod def _timeline( *, row: PMFlowAgentRun, steps: list[PMFlowAgentStep], observations: list[PMFlowAgentObservation], verifications: list[PMFlowAgentVerification], ai_calls: list[dict[str, Any]], ) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] run_duration_ms = None if row.started_at and row.completed_at: run_duration_ms = max(0, int((row.completed_at - row.started_at).total_seconds() * 1000)) if row.started_at: items.append( { "type": "run_started", "label": "触发链路", "description": _timeline_description(row.goal, fallback="用户端触发后，平台创建本轮 Agent Run。"), "at": _utc_iso(row.started_at), "started_at": _utc_iso(row.started_at), "completed_at": None, "duration_ms": None, "status": row.status, } ) for step in steps: step_started_at = step.started_at or _derived_started_at(step.completed_at, step.duration_ms) step_description = _timeline_description( step.output_summary or step.input_summary, fallback="该节点已执行，未写入额外摘要。", ) items.append( { "type": "step", "label": step.step_payload.get("label") if isinstance(step.step_payload, dict) else step.step_kind, "description": step_description, "ref": step.public_step_id, "at": _utc_iso(step.completed_at), "started_at": _utc_iso(step_started_at), "completed_at": _utc_iso(step.completed_at), "duration_ms": step.duration_ms, "status": step.status, } ) for observation in observations: items.append( { "type": "observation", "label": observation.customer_safe_summary or observation.source_kind, "description": _timeline_description( observation.customer_safe_summary, fallback=f"观察记录包含 {len(observation.facts if isinstance(observation.facts, list) else [])} 条事实和 {len(observation.gaps if isinstance(observation.gaps, list) else [])} 个缺口。", ), "ref": observation.public_observation_id, "at": _utc_iso(observation.observed_at), "started_at": None, "completed_at": _utc_iso(observation.observed_at), "duration_ms": None, "status": observation.confidence or "unknown", } ) for call in ai_calls: call_latency = call.get("latency_ms") call_created_at = _local_iso(call.get("created_at")) call_label = call.get("function_label") or call.get("function_name") provider_label = call.get("provider_label") or call.get("provider_code") or call.get("provider") or "未知供应商" call_status = call.get("status_label") or call.get("status") or "未知状态" call_preview = _timeline_description( call.get("response_preview") or call.get("error_summary"), fallback="", ) call_description = ( f"{call_label or '外部 API'} 输出摘要：{call_preview}" if call_preview else f"{call_label or '外部 API'} 没有记录响应摘要，本条只保留供应商、模型、状态、耗时和 Trace。" ) items.append( { "type": "external_api", "label": call_label, "description": _timeline_description( f"{call_description} 调用信息：{provider_label} / {call.get('model') or '未知模型'}，状态：{call_status}，Trace：{call.get('trace_id') or call.get('id') or '-'}。", ), "ref": call.get("trace_id") or call.get("id"), "at": call_created_at, "started_at": _derived_started_iso(call_created_at, call_latency), "completed_at": call_created_at, "duration_ms": call_latency, "status": call.get("status"), } ) for verification in verifications: findings = verification.findings if isinstance(verification.findings, list) else [] finding_summary = "" if findings: first = findings[0] finding_summary = first.get("summary") if isinstance(first, dict) else str(first) items.append( { "type": "verification", "label": verification.target_ref or verification.public_verification_id, "description": _timeline_description( finding_summary, fallback=f"质量检查结果：{verification.result or '未知'}；retry_would_change_state={bool(verification.retry_would_change_state)}。", ), "ref": verification.public_verification_id, "at": _utc_iso(verification.verified_at), "started_at": None, "completed_at": _utc_iso(verification.verified_at), "duration_ms": None, "status": verification.result, } ) if row.completed_at: items.append( { "type": "run_completed", "label": "链路完成", "description": _timeline_description( row.final_answer, fallback=f"本轮 Agent Run 已停止，最终状态：{row.status or '未知'}。", ), "at": _utc_iso(row.completed_at), "started_at": _utc_iso(row.started_at), "completed_at": _utc_iso(row.completed_at), "duration_ms": run_duration_ms, "status": row.status, } ) def sort_key(item: dict[str, Any]) -> tuple[float, int]: item_type = str(item.get("type") or "") if item_type == "run_started": return (_timeline_sort_timestamp(item.get("started_at") or item.get("at")), 0) if item_type == "run_completed": return (float("inf"), 9) return (_timeline_sort_timestamp(item.get("started_at") or item.get("at") or item.get("completed_at")), 1) return sorted(items, key=sort_key) @staticmethod def _preview(value: Any, *, limit: int = 500) -> str: return " ".join(str(value or "").strip().split())[:limit] @staticmethod def _status_label(status: str | None) -> str: text = str(status or "").strip() return { "completed": "已完成", "answered": "已回答", "failed": "失败", "blocked": "已阻断", "active": "运行中", "waiting_for_user": "等待用户", }.get(text, text or "未知") __all__ = ["AgentRunAuditService"]