"""AI call trace read model for platform AI Ops.""" from __future__ import annotations import json import re from datetime import datetime, timedelta from decimal import Decimal from typing import Any from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from ..models import AICallLog, Integration, PMFlowLaunchDraftJob from ..services.ai_scene_runtime_policy import AISceneRuntimePolicyService from ..services.ai_routing_scene_contracts import AI_SCENE_REGISTRY LLM_ENGINE_INTEGRATION_TYPE = "llm_engine" PROVIDER_LABELS = { "dashscope": "Qwen / DashScope", "openai": "OpenAI 兼容网关", "zhipu": "智谱 BigModel", "deepseek": "DeepSeek", "ollama": "本地 Ollama", } EXTRA_SCENE_LABELS = { "platform_ai_ops": "平台 AI 运营", } FUNCTION_LABELS = { "probe_engine_capabilities": "供应商能力探测汇总（历史兼容）", "test_engine_connection": "供应商连接测试", "pmflow_launch_draft_v1": "新项目草案生成", "pmflow_change_review_extraction_v1": "项目主链变更抽取", "space_saved_idea_enrichment_v1": "空间想法结构化", "pmflow_ai_route_planner": "旧 PMFlow AI 路由规划（已下线）", "rag_query_semantics": "问答语义与来源策略", "pmflow_semantic_context_decision": "上下文与主链边界决策", "rag_followup_anchor_semantics": "续问锚点归位", "rag_pending_clarification_selection_semantics": "待澄清项选择", "rag_answer_scene_semantics": "问答场景选择", "customer_answer_composer": "客户答案表达生成", "customer_answer_verifier": "客户答案质检", "pmflow_ai_operating_mission": "Space AI Operating Mission 生成", } PROBE_TARGET_LABELS = { "basic_chat": "基础聊天", "model_self_report": "模型自报", "structured_output": "结构化输出", "json_schema": "JSON Schema", "tool_calling": "工具调用", "streaming": "流式响应", "long_context": "长上下文", "long_context_32k": "长上下文 32K", "long_context_128k": "长上下文 128K", "vision_input": "视觉输入", "audio_input": "音频输入", } SENSITIVE_PATTERNS = ( re.compile(r"(?i)(sk-[a-z0-9][a-z0-9_\-]{12,})"), re.compile(r"(?i)(api[_-]?key|authorization|access[_-]?token|refresh[_-]?token|secret)(\s*[=:]\s*)([^\s,;，；]+)"), re.compile(r"(?i)(bearer\s+)([a-z0-9._\-]{16,})"), re.compile(r"(? None: self.db = db async def list_traces( self, *, page: int = 1, page_size: int = 30, status: str | None = None, provider: str | None = None, model: str | None = None, function_name: str | None = None, scene_key: str | None = None, slot_key: str | None = None, prompt_key: str | None = None, prompt_source: str | None = None, prompt_release_id: int | None = None, trace_id: str | None = None, tenant_id: int | 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._build_filters( status=status, provider=provider, model=model, function_name=function_name, scene_key=scene_key, slot_key=slot_key, prompt_key=prompt_key, prompt_source=prompt_source, prompt_release_id=prompt_release_id, trace_id=trace_id, tenant_id=tenant_id, keyword=keyword, since_hours=since_hours, ) where_clause = and_(*filters) if filters else None count_stmt = select(func.count(AICallLog.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(AICallLog).order_by(AICallLog.created_at.desc(), AICallLog.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()) runtime_index = await self._runtime_display_index(rows) operator_user_index = await self._operator_user_index(rows) return { "summary": await self._summary(filters=filters), "items": [ self._row_summary(row, runtime_index=runtime_index, operator_user_index=operator_user_index) for row in rows ], "page": page, "page_size": page_size, "total": total, "has_next": page * page_size < total, } async def get_trace(self, trace_id_or_log_id: str) -> dict[str, Any]: text = str(trace_id_or_log_id or "").strip() stmt = select(AICallLog) if text.isdigit(): stmt = stmt.where(AICallLog.id == int(text)) else: stmt = stmt.where(AICallLog.trace_id == text) row = (await self.db.execute(stmt.order_by(AICallLog.created_at.desc(), AICallLog.id.desc()).limit(1))).scalar_one_or_none() if not row: raise ValueError("AI 调用记录不存在") runtime_index = await self._runtime_display_index([row]) operator_user_index = await self._operator_user_index([row]) return self._row_detail(row, runtime_index=runtime_index, operator_user_index=operator_user_index) def _build_filters( self, *, status: str | None, provider: str | None, model: str | None, function_name: str | None, scene_key: str | None, slot_key: str | None, prompt_key: str | None, prompt_source: str | None, prompt_release_id: int | None, trace_id: str | None, tenant_id: int | None, keyword: str | None, since_hours: int | None, ) -> list[Any]: filters: list[Any] = [] if since_hours and int(since_hours) > 0: filters.append(AICallLog.created_at >= datetime.now() - timedelta(hours=min(int(since_hours), 24 * 365))) if not function_name: filters.append(AICallLog.function_name != "probe_engine_capabilities") if status and status != "all": filters.append(AICallLog.status == status) if provider: filters.append(AICallLog.provider == provider) if model: filters.append(AICallLog.model.ilike(f"%{model.strip()}%")) if function_name: filters.append(AICallLog.function_name.ilike(f"%{function_name.strip()}%")) if scene_key: scene_functions = [ str(function_name) for function_name in (AI_SCENE_REGISTRY.get(scene_key, {}).get("function_names") or []) if str(function_name).strip() ] if scene_functions: filters.append(AICallLog.function_name.in_(scene_functions)) else: filters.append(AICallLog.function_name == "__no_scene_match__") if slot_key: filters.append(or_(AICallLog.executed_tier == slot_key, AICallLog.planned_tier == slot_key)) if prompt_key: filters.append(AICallLog.prompt_key == str(prompt_key).strip()) if prompt_source: filters.append(AICallLog.prompt_source == str(prompt_source).strip()) if prompt_release_id is not None: filters.append(AICallLog.prompt_release_id == int(prompt_release_id)) if trace_id: filters.append(AICallLog.trace_id.ilike(f"%{trace_id.strip()}%")) if tenant_id: filters.append(AICallLog.tenant_id == int(tenant_id)) if keyword: value = f"%{keyword.strip()}%" filters.append( or_( AICallLog.function_name.ilike(value), AICallLog.prompt_key.ilike(value), AICallLog.prompt.ilike(value), AICallLog.response.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(AICallLog.id), func.sum(AICallLog.tokens_used), func.sum(AICallLog.cost), func.avg(AICallLog.latency_ms), ) if where_clause is not None: stmt = stmt.where(where_clause) total, tokens, cost, latency = (await self.db.execute(stmt)).one() status_rows_stmt = select(AICallLog.status, func.count(AICallLog.id)).group_by(AICallLog.status) provider_rows_stmt = select(AICallLog.provider, func.count(AICallLog.id)).group_by(AICallLog.provider).order_by(func.count(AICallLog.id).desc()).limit(5) if where_clause is not None: status_rows_stmt = status_rows_stmt.where(where_clause) provider_rows_stmt = provider_rows_stmt.where(where_clause) statuses = {str(status or "unknown"): int(count or 0) for status, count in (await self.db.execute(status_rows_stmt)).all()} top_providers = [ { "provider": str(provider or "unknown"), "provider_code": str(provider or "unknown"), "provider_label": self._provider_label(provider), "count": int(count or 0), } for provider, count in (await self.db.execute(provider_rows_stmt)).all() ] total_count = int(total or 0) failed_count = sum(count for status, count in statuses.items() if self._is_failed_status(status)) return { "total_calls": total_count, "success_calls": max(total_count - failed_count, 0), "failed_calls": failed_count, "success_rate": round((total_count - failed_count) / total_count, 4) if total_count else 0, "tokens_used": int(tokens or 0), "estimated_cost": self._number(cost), "average_latency_ms": round(float(latency), 2) if latency is not None else None, "status_counts": statuses, "top_providers": top_providers, } def _row_summary( self, row: AICallLog, *, runtime_index: dict[str, dict[str, str]] | None = None, operator_user_index: dict[int, int] | None = None, ) -> dict[str, Any]: meta = self._row_meta(row) scene_key = self._scene_key(row, meta) slot_key = self._slot_key(row, meta) runtime = self._runtime_display(row, meta, runtime_index or {}) provider_usage = self._provider_usage(row, meta) user_id = row.user_id if row.user_id is not None else (operator_user_index or {}).get(int(row.id or 0)) return { "id": row.id, "trace_id": row.trace_id, "tenant_id": row.tenant_id, "user_id": user_id, "created_at": row.created_at.isoformat() if row.created_at else None, "function_name": row.function_name, "function_label": self._function_label(row.function_name), "scene_key": scene_key, "scene_label": self._scene_label(scene_key), "status": row.status, "status_label": self._status_label(row.status), "provider": runtime["provider_code"] or None, "provider_code": runtime["provider_code"] or None, "provider_label": runtime["provider_label"] or None, "runtime_label": runtime["runtime_label"] or None, "engine_id": runtime["engine_id"] or None, "engine_name": runtime["engine_name"] or None, "model": row.model or str(meta.get("model") or "") or None, "slot_key": slot_key, "slot_label": self._slot_label(slot_key), "prompt_key": row.prompt_key or str(meta.get("prompt_key") or "") or None, "prompt_release_id": row.prompt_release_id if row.prompt_release_id is not None else meta.get("prompt_release_id"), "prompt_source": row.prompt_source or str(meta.get("prompt_source") or "") or None, "routing_decision": row.routing_decision, "route_source": row.route_source, "route_rule_name": row.route_rule_name, "tokens_used": int(row.tokens_used or 0), "prompt_tokens": provider_usage.get("prompt_tokens"), "completion_tokens": provider_usage.get("completion_tokens"), "total_tokens": provider_usage.get("total_tokens"), "request_max_tokens": self._request_max_tokens(row), "finish_reason": self._finish_reason(row), "function_runtime_max_tokens": meta.get("function_runtime_max_tokens"), "function_runtime_profile_version": meta.get("function_runtime_profile_version") or None, "estimated_cost": self._number(row.cost), "latency_ms": row.latency_ms, "prompt_preview": self._preview(row.prompt), "response_preview": self._preview(row.response or ""), "error_summary": self._error_summary(row), "error_diagnosis": self._error_diagnosis(row, meta), } def _row_detail( self, row: AICallLog, *, runtime_index: dict[str, dict[str, str]] | None = None, operator_user_index: dict[int, int] | None = None, ) -> dict[str, Any]: meta = self._row_meta(row) summary = self._row_summary(row, runtime_index=runtime_index, operator_user_index=operator_user_index) return { **summary, "request": { "prompt_redacted": self._redact(row.prompt), "prompt_chars": len(row.prompt or ""), "provider_request_payload_redacted": self._redact_payload(row.provider_request_payload), "provider_request_payload_chars": self._payload_chars(row.provider_request_payload), "provider_payload_missing_reason": str(meta.get("upstream_missing_reason") or "").strip() or None, "upstream_request_sent": meta.get("upstream_request_sent") if isinstance(meta.get("upstream_request_sent"), bool) else None, }, "response": { "response_redacted": self._redact(row.response or ""), "response_chars": len(row.response or ""), "looks_truncated": "[ai_call_log_truncated]" in str(row.response or ""), "provider_response_payload_redacted": self._redact_payload(row.provider_response_payload), "provider_response_payload_chars": self._payload_chars(row.provider_response_payload), "provider_attempts_redacted": self._redact_payload(row.provider_attempts), "provider_attempts_count": len(row.provider_attempts or []) if isinstance(row.provider_attempts, list) else 0, }, "routing": { "planned_tier": row.planned_tier, "executed_tier": row.executed_tier, "routing_decision": row.routing_decision, "route_source": row.route_source, "route_rule_name": row.route_rule_name, "routing_config_version": row.routing_config_version, "routing_config_updated_at": row.routing_config_updated_at, "routing_meta": meta, }, "probe": self._probe_detail(meta), "audit": { "raw_payload_available": bool(row.provider_request_payload or row.provider_response_payload or row.provider_attempts), "raw_payload_policy": "已落库真实上游 provider request/response payload；平台审计详情默认展示完整脱敏视图，不在客户侧暴露。", }, "runtime_policy": self._runtime_policy_detail(row, meta), } @staticmethod def _provider_usage(row: AICallLog, meta: dict[str, Any]) -> dict[str, int]: payload = row.provider_response_payload if isinstance(row.provider_response_payload, dict) else {} usage = payload.get("usage") if isinstance(payload.get("usage"), dict) else {} if not usage and isinstance(meta.get("provider_usage"), dict): usage = meta.get("provider_usage") or {} result: dict[str, int] = {} for source_key, target_key in ( ("prompt_tokens", "prompt_tokens"), ("completion_tokens", "completion_tokens"), ("total_tokens", "total_tokens"), ("input_tokens", "prompt_tokens"), ("output_tokens", "completion_tokens"), ): if source_key not in usage: continue try: value = int(usage.get(source_key) or 0) except (TypeError, ValueError): continue if value > 0: result[target_key] = value if "total_tokens" not in result and (result.get("prompt_tokens") or result.get("completion_tokens")): result["total_tokens"] = int(result.get("prompt_tokens") or 0) + int(result.get("completion_tokens") or 0) return result @staticmethod def _request_max_tokens(row: AICallLog) -> int | None: payload = row.provider_request_payload if isinstance(row.provider_request_payload, dict) else {} try: value = int(payload.get("max_tokens") or 0) except (TypeError, ValueError): return None return value if value > 0 else None @staticmethod def _finish_reason(row: AICallLog) -> str | None: payload = row.provider_response_payload if isinstance(row.provider_response_payload, dict) else {} choices = payload.get("choices") if isinstance(payload.get("choices"), list) else [] choice = choices[0] if choices and isinstance(choices[0], dict) else {} text = str(choice.get("finish_reason") or "").strip() return text or None @staticmethod def _probe_detail(meta: dict[str, Any]) -> dict[str, Any] | None: if not isinstance(meta, dict): return None probe_run_id = str(meta.get("probe_run_id") or "").strip() probe_target = str(meta.get("probe_target") or "").strip() probe_targets = meta.get("probe_targets") if isinstance(meta.get("probe_targets"), list) else [] if not probe_run_id and not probe_target and not probe_targets: return None return { "probe_run_id": probe_run_id or None, "mode": str(meta.get("probe_mode") or "").strip() or None, "target": probe_target or None, "target_label": PROBE_TARGET_LABELS.get(probe_target, probe_target) if probe_target else None, "targets": probe_targets, "method": str(meta.get("probe_method") or "").strip() or None, "status": str(meta.get("probe_status") or "").strip() or None, "evidence_strength": str(meta.get("evidence_strength") or "").strip() or None, "failure_code": str(meta.get("failure_code") or "").strip() or None, "summary": meta.get("probe_summary") if isinstance(meta.get("probe_summary"), dict) else {}, "direct_operation": bool(meta.get("direct_operation")), } @staticmethod def _row_meta(row: AICallLog) -> dict[str, Any]: if isinstance(row.routing_meta, dict): return row.routing_meta response = str(row.response or "") if response.strip().startswith("{"): try: parsed = json.loads(response) if isinstance(parsed, dict) and isinstance(parsed.get("routing_meta"), dict): return parsed["routing_meta"] except Exception: return {} return {} def _redact_payload(self, payload: Any) -> str: if payload is None: return "" try: text = json.dumps(payload, ensure_ascii=False, indent=2, default=str) except Exception: text = str(payload) return self._redact(text) @staticmethod def _payload_chars(payload: Any) -> int: if payload is None: return 0 try: return len(json.dumps(payload, ensure_ascii=False, default=str)) except Exception: return len(str(payload)) async def _runtime_display_index(self, rows: list[AICallLog]) -> dict[str, dict[str, str]]: tenant_ids = {int(row.tenant_id) for row in rows if row.tenant_id is not None} if not tenant_ids: return {} stmt = select(Integration).where( Integration.type == LLM_ENGINE_INTEGRATION_TYPE, Integration.tenant_id.in_(tenant_ids), ) result = await self.db.execute(stmt) entries: dict[str, dict[str, str]] = {} provider_model_candidates: dict[str, list[dict[str, str]]] = {} for integration in result.scalars().all(): config = integration.config if isinstance(integration.config, dict) else {} for engine in config.get("engines") or []: if not isinstance(engine, dict): continue provider_code = str(engine.get("provider") or "").strip().lower() model = str(engine.get("model") or "").strip() engine_id = str(engine.get("id") or "").strip() item = { "provider_code": provider_code, "provider_label": self._provider_label(provider_code), "model": model, "engine_id": engine_id, "engine_name": str(engine.get("name") or "").strip(), } if engine_id: entries[self._runtime_key(integration.tenant_id, "engine", engine_id)] = item if provider_code and model: key = self._runtime_key(integration.tenant_id, "provider_model", f"{provider_code}:{model}") provider_model_candidates.setdefault(key, []).append(item) for key, candidates in provider_model_candidates.items(): names = {item.get("engine_name") or "" for item in candidates} if len(candidates) == 1 or len(names) == 1: entries[key] = candidates[0] return entries async def _operator_user_index(self, rows: list[AICallLog]) -> dict[int, int]: log_jobs: list[tuple[int, str]] = [] for row in rows: if row.user_id is not None: continue meta = self._row_meta(row) launch_job_id = str(meta.get("launch_job_id") or "").strip() if row.id is not None and launch_job_id: log_jobs.append((int(row.id), launch_job_id)) if not log_jobs: return {} job_ids = sorted({job_id for _, job_id in log_jobs}) result = await self.db.execute( select(PMFlowLaunchDraftJob.job_id, PMFlowLaunchDraftJob.user_id).where( PMFlowLaunchDraftJob.job_id.in_(job_ids) ) ) job_users = {str(job_id): int(user_id) for job_id, user_id in result.all() if user_id is not None} return {log_id: job_users[job_id] for log_id, job_id in log_jobs if job_id in job_users} def _runtime_display(self, row: AICallLog, meta: dict[str, Any], runtime_index: dict[str, dict[str, str]]) -> dict[str, str]: provider_code = str(row.provider or meta.get("provider_code") or meta.get("provider") or "").strip().lower() model = str(row.model or meta.get("model") or "").strip() engine_id = str(meta.get("engine_id") or "").strip() matched = {} if engine_id: matched = runtime_index.get(self._runtime_key(row.tenant_id, "engine", engine_id), {}) if not matched and provider_code and model: matched = runtime_index.get(self._runtime_key(row.tenant_id, "provider_model", f"{provider_code}:{model}"), {}) provider_label = str(meta.get("provider_label") or matched.get("provider_label") or self._provider_label(provider_code)).strip() engine_name = str(meta.get("engine_name") or matched.get("engine_name") or "").strip() return { "provider_code": provider_code, "provider_label": provider_label, "runtime_label": engine_name or provider_label or provider_code, "engine_id": engine_id or str(matched.get("engine_id") or "").strip(), "engine_name": engine_name, } @staticmethod def _runtime_key(tenant_id: Any, kind: str, value: str) -> str: return f"{int(tenant_id or 0)}:{kind}:{value}" @staticmethod def _provider_label(provider: Any) -> str: value = str(provider or "").strip().lower() if not value or value == "unknown": return "unknown" if value == "unknown" else "" return PROVIDER_LABELS.get(value, value) @staticmethod def _number(value: Any) -> float: if value is None: return 0.0 if isinstance(value, Decimal): return float(value) try: return float(value) except (TypeError, ValueError): return 0.0 @staticmethod def _scene_key(row: AICallLog, meta: dict[str, Any]) -> str | None: scene_key = str(meta.get("scene_key") or meta.get("business_scene") or "").strip() if scene_key: return scene_key return SCENE_BY_FUNCTION.get(str(row.function_name or "")) @staticmethod def _scene_label(scene_key: str | None) -> str: key = str(scene_key or "").strip() if not key: return "未归类" return EXTRA_SCENE_LABELS.get(key) or SCENE_LABELS.get(key, key) @staticmethod def _function_label(function_name: str | None) -> str: value = str(function_name or "").strip() if value.startswith("probe_capability_"): target = value.removeprefix("probe_capability_") return f"能力探测：{PROBE_TARGET_LABELS.get(target, target)}" return FUNCTION_LABELS.get(value, value or "未记录") @staticmethod def _slot_key(row: AICallLog, meta: dict[str, Any]) -> str | None: return ( str(meta.get("slot_key") or "").strip() or str(row.executed_tier or "").strip() or str(row.planned_tier or "").strip() or None ) @staticmethod def _slot_label(slot_key: str | None) -> str: return { "low": "快速槽", "high": "深度槽", "structured": "结构化槽", "long_context": "长上下文槽", "tool": "工具槽", "external_query": "外部查询槽", }.get(str(slot_key or ""), str(slot_key or "未标注")) @staticmethod def _status_label(status: str | None) -> str: text = str(status or "").strip() if text in {"success", "ok"}: return "成功" if text == "structured_timeout": return "模型超时" if text in {"structured_json_invalid", "structured_invalid"}: return "结构化 JSON 无效" if text == "structured_empty": return "模型空响应" if text == "structured_http_error": return "供应商 HTTP 异常" if "structured" in text: return "结构化失败" if "capability_probe" in text or text.startswith("capability_"): return "能力探测" if text in {"failed", "error"} or "fail" in text: return "失败" return text or "未知" @staticmethod def _is_failed_status(status: str) -> bool: text = str(status or "").strip().lower() return text not in {"success", "ok"} and ( "fail" in text or "error" in text or "invalid" in text or "empty" in text or "timeout" in text ) @classmethod def _error_summary(cls, row: AICallLog) -> str | None: if not cls._is_failed_status(str(row.status or "")): return None response = str(row.response or "").strip() if response.startswith("[error]"): return cls._redact(response[:300]) return cls._redact(response[:180]) if response else cls._status_label(row.status) @classmethod def _error_diagnosis(cls, row: AICallLog, meta: dict[str, Any]) -> dict[str, Any] | None: status = str(row.status or "").strip().lower() attempts = row.provider_attempts if isinstance(row.provider_attempts, list) else [] attempt_errors = [ { "status": str(item.get("status") or ""), "error_type": str(item.get("error_type") or ""), "error": cls._redact(str(item.get("error") or "")[:300]), } for item in attempts if isinstance(item, dict) and (item.get("error_type") or item.get("error") or item.get("status") == "request_error") ] if status == "structured_timeout": recommendation = AISceneRuntimePolicyService.timeout_recommendation(status=status, routing_meta=meta) or {} return { "code": "model_read_timeout", "label": "模型读超时", "summary": "上游模型在规定时间内没有返回结构化草案或结构化结果。", "attempt_count": len(attempts), "attempt_errors": attempt_errors, "recommendation": recommendation, } if status in {"structured_json_invalid", "structured_invalid", "structured_empty", "structured_http_error"}: return { "code": status, "label": cls._status_label(status), "summary": cls._error_summary(row) or cls._status_label(status), "attempt_count": len(attempts), "attempt_errors": attempt_errors, } return None @staticmethod def _runtime_policy_detail(row: AICallLog, meta: dict[str, Any]) -> dict[str, Any]: return { "policy_id": str(meta.get("scene_timeout_policy_id") or "").strip() or None, "policy_version": str(meta.get("scene_timeout_policy_version") or "").strip() or None, "input_bucket": str(meta.get("input_bucket") or "").strip() or None, "input_tokens_estimate": meta.get("input_tokens_estimate"), "foreground_wait_seconds": meta.get("foreground_wait_seconds"), "provider_timeout_seconds": meta.get("provider_timeout_seconds"), "job_timeout_seconds": meta.get("job_timeout_seconds"), "max_retries": meta.get("max_retries"), "allow_low_to_high_escalation": meta.get("allow_low_to_high_escalation"), "disable_thinking": meta.get("disable_thinking"), "scene_fallback_policy": str(meta.get("scene_fallback_policy") or "").strip() or None, "large_input_policy": str(meta.get("large_input_policy") or "").strip() or None, "timeout_policy_source": str(meta.get("timeout_policy_source") or "").strip() or None, "timeout_policy_reason": str(meta.get("timeout_policy_reason") or "").strip() or None, "latency_ms": row.latency_ms, } @classmethod def _preview(cls, text: str) -> str: redacted = cls._redact(text) return redacted[:180] + ("..." if len(redacted) > 180 else "") @staticmethod def _redact(text: str) -> str: value = str(text or "") for pattern in SENSITIVE_PATTERNS: if pattern.pattern.startswith("([A-Za-z0-9"): value = pattern.sub(lambda match: f"{match.group(1)[:2]}***@{match.group(2)}", value) elif "api" in pattern.pattern.lower() or "authorization" in pattern.pattern.lower(): value = pattern.sub(lambda match: f"{match.group(1)}{match.group(2)}[REDACTED]", value) elif "bearer" in pattern.pattern.lower(): value = pattern.sub(lambda match: f"{match.group(1)}[REDACTED]", value) else: value = pattern.sub("[REDACTED]", value) return value