from __future__ import annotations from datetime import datetime from types import SimpleNamespace import pytest import pytest_asyncio from fastapi import FastAPI from httpx import ASGITransport, AsyncClient from sqlalchemy import JSON from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.api import ai_call_traces_api from app.core.database import get_db from app.dependencies.permissions import require_platform_ai_ops from app.models import PMFlowLaunchDraftJob, User from app.models.ai_call_log import AICallLog from app.models.base import Base from app.models.integration import Integration def _use_sqlite_json_types(*tables) -> None: for table in tables: for column in table.columns: if column.type.__class__.__name__ in {"JSONB", "ARRAY", "Vector"}: column.type = JSON() @pytest_asyncio.fixture async def db_session(): engine = create_async_engine("sqlite+aiosqlite:///:memory:") _use_sqlite_json_types(AICallLog.__table__, Integration.__table__, PMFlowLaunchDraftJob.__table__, User.__table__) async with engine.begin() as conn: await conn.run_sync( lambda sync_conn: Base.metadata.create_all( sync_conn, tables=[User.__table__, AICallLog.__table__, Integration.__table__, PMFlowLaunchDraftJob.__table__], ) ) maker = async_sessionmaker(engine, expire_on_commit=False) async with maker() as session: yield session await engine.dispose() @pytest.mark.asyncio async def test_ai_call_trace_list_and_detail_redact_payload(db_session) -> None: db_session.add( Integration( tenant_id=1, type="llm_engine", is_active=True, config={ "engines": [ { "id": "engine-qwen-plus", "name": "阿里云-qwen-plus", "provider": "dashscope", "model": "qwen-plus", } ] }, ) ) db_session.add( AICallLog( tenant_id=1, user_id=7, function_name="pmflow_launch_draft_v1", prompt="用户手机号 13800138000，key=sk-secretsecretsecretsecret，请生成项目草案", response='{"ok":true,"routing_meta":{"scene_key":"project_launch","slot_key":"structured"}}', tokens_used=128, cost=0.001, status="success", trace_id="trace-1", provider="dashscope", model="qwen-plus", latency_ms=345, routing_meta={"scene_key": "project_launch", "slot_key": "structured", "route_source": "active_slot"}, prompt_key="PROMPT_PM_FLOW_LAUNCH_DRAFT", prompt_release_id=12, prompt_source="active_release", provider_request_payload={ "model": "qwen-plus", "messages": [ {"role": "system", "content": "你是项目助手"}, {"role": "user", "content": "手机号 13800138000，key=sk-secretsecretsecretsecret"}, ], "temperature": 0.0, "max_tokens": 5600, }, provider_response_payload={ "id": "cmpl-test", "choices": [{"message": {"content": "{\"ok\":true}"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 90, "completion_tokens": 38, "total_tokens": 128}, }, provider_attempts=[ { "provider": "dashscope", "model": "qwen-plus", "request_payload": {"model": "qwen-plus", "messages": [{"role": "user", "content": "sk-secretsecretsecretsecret"}]}, "response_payload": {"choices": [{"message": {"content": "{\"ok\":true}"}}]}, "status": "success", } ], planned_tier="structured", executed_tier="structured", routing_decision="slot", route_source="active_slot", route_rule_name="project_launch", created_at=datetime(2026, 4, 29, 10, 0, 0), ) ) await db_session.commit() app = FastAPI() app.include_router(ai_call_traces_api.router) async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db app.dependency_overrides[require_platform_ai_ops] = lambda: SimpleNamespace(id=1, role="OWNER") async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: list_response = await client.get("/ai-call-traces?provider=dashscope&scene_key=project_launch&prompt_key=PROMPT_PM_FLOW_LAUNCH_DRAFT&since_hours=8760") detail_response = await client.get("/ai-call-traces/trace-1") assert list_response.status_code == 200 payload = list_response.json()["data"] assert payload["summary"]["total_calls"] == 1 assert payload["items"][0]["provider"] == "dashscope" assert payload["items"][0]["provider_code"] == "dashscope" assert payload["items"][0]["provider_label"] == "Qwen / DashScope" assert payload["items"][0]["runtime_label"] == "阿里云-qwen-plus" assert payload["items"][0]["engine_name"] == "阿里云-qwen-plus" assert payload["items"][0]["function_label"] == "新项目草案生成" assert payload["items"][0]["scene_key"] == "project_launch" assert payload["items"][0]["slot_key"] == "structured" assert payload["items"][0]["prompt_key"] == "PROMPT_PM_FLOW_LAUNCH_DRAFT" assert payload["items"][0]["prompt_release_id"] == 12 assert payload["items"][0]["prompt_source"] == "active_release" assert payload["items"][0]["prompt_tokens"] == 90 assert payload["items"][0]["completion_tokens"] == 38 assert payload["items"][0]["request_max_tokens"] == 5600 assert payload["items"][0]["finish_reason"] == "stop" assert "13800138000" not in payload["items"][0]["prompt_preview"] assert "sk-secret" not in payload["items"][0]["prompt_preview"] assert detail_response.status_code == 200 detail = detail_response.json()["data"] assert detail["request"]["prompt_chars"] > 0 assert detail["runtime_label"] == "阿里云-qwen-plus" assert "[REDACTED]" in detail["request"]["prompt_redacted"] assert detail["request"]["provider_request_payload_chars"] > 0 assert "qwen-plus" in detail["request"]["provider_request_payload_redacted"] assert "13800138000" not in detail["request"]["provider_request_payload_redacted"] assert "sk-secret" not in detail["request"]["provider_request_payload_redacted"] assert detail["response"]["provider_response_payload_chars"] > 0 assert detail["response"]["provider_attempts_count"] == 1 assert detail["routing"]["routing_meta"]["slot_key"] == "structured" assert detail["audit"]["raw_payload_available"] is True @pytest.mark.asyncio async def test_ai_call_trace_falls_back_to_launch_job_operator(db_session) -> None: db_session.add( PMFlowLaunchDraftJob( tenant_id=1, user_id=42, job_id="launch_job_1", revision_id="revision_1", input_text="我想做一个 AI 会议纪要工具", input_type="text", source="web", attachments=[], mode="create", status="completed", changed_fields=[], expires_at=datetime(2026, 5, 9, 10, 0, 0), ) ) db_session.add( AICallLog( tenant_id=1, user_id=None, function_name="pmflow_launch_draft_v1", prompt="prompt", response="{}", tokens_used=10, cost=0.0001, status="success", trace_id="trace-launch-job", provider="dashscope", model="qwen-plus", routing_meta={"launch_job_id": "launch_job_1"}, created_at=datetime(2026, 5, 8, 10, 0, 0), ) ) await db_session.commit() app = FastAPI() app.include_router(ai_call_traces_api.router) async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db app.dependency_overrides[require_platform_ai_ops] = lambda: SimpleNamespace(id=1, role="OWNER") async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: list_response = await client.get("/ai-call-traces?trace_id=trace-launch-job&since_hours=8760") detail_response = await client.get("/ai-call-traces/trace-launch-job") assert list_response.status_code == 200 assert list_response.json()["data"]["items"][0]["user_id"] == 42 assert detail_response.status_code == 200 assert detail_response.json()["data"]["user_id"] == 42 @pytest.mark.asyncio async def test_ai_call_trace_classifies_pmflow_ai_prompt_governance_functions(db_session) -> None: rows = [ AICallLog( tenant_id=1, user_id=7, function_name="pmflow_semantic_context_decision", prompt="semantic packet", response="{}", tokens_used=10, cost=0.0001, status="success", trace_id="trace-semantic", provider="dashscope", model="qwen-plus", prompt_key="PMFLOW_SEMANTIC_CONTEXT_DECISION_PROMPT", created_at=datetime(2026, 5, 8, 10, 0, 0), ), AICallLog( tenant_id=1, user_id=7, function_name="pmflow_ai_route_planner", prompt="route plan", response="{}", tokens_used=10, cost=0.0001, status="success", trace_id="trace-route", provider="dashscope", model="qwen-plus", prompt_key="PROMPT_PMFLOW_AI_ROUTE_PLANNER", created_at=datetime(2026, 5, 8, 10, 1, 0), ), AICallLog( tenant_id=1, user_id=7, function_name="customer_answer_verifier", prompt="verify answer", response="{}", tokens_used=10, cost=0.0001, status="success", trace_id="trace-verifier", provider="dashscope", model="qwen-plus", prompt_key="PROMPT_CUSTOMER_ANSWER_VERIFIER", created_at=datetime(2026, 5, 8, 10, 2, 0), ), AICallLog( tenant_id=1, user_id=7, function_name="pmflow_ai_operating_mission", prompt="operating mission", response="{}", tokens_used=10, cost=0.0001, status="success", trace_id="trace-operating-mission", provider="dashscope", model="qwen-plus", prompt_key="PROMPT_PMFLOW_AI_OPERATING_MISSION", routing_meta={ "scene_key": "space_pmflow_ai_operating_runtime", "provider_timeout_seconds": 60, "function_runtime_profile_version": "profile-default-v1", }, created_at=datetime(2026, 5, 8, 10, 3, 0), ), ] db_session.add_all(rows) await db_session.commit() app = FastAPI() app.include_router(ai_call_traces_api.router) async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db app.dependency_overrides[require_platform_ai_ops] = lambda: SimpleNamespace(id=1, role="OWNER") async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.get("/ai-call-traces?provider=dashscope&since_hours=8760") assert response.status_code == 200 items = {item["function_name"]: item for item in response.json()["data"]["items"]} assert items["pmflow_semantic_context_decision"]["scene_key"] == "pmflow_ai_context_governance" assert items["pmflow_semantic_context_decision"]["scene_label"] == "上下文与语义治理" assert items["pmflow_semantic_context_decision"]["function_label"] == "上下文与主链边界决策" assert items["pmflow_ai_route_planner"]["scene_key"] == "pmflow_ai_route_planning" assert items["pmflow_ai_route_planner"]["scene_label"] == "旧 PMFlow AI 路由规划（已下线）" assert items["pmflow_ai_route_planner"]["function_label"] == "旧 PMFlow AI 路由规划（已下线）" assert items["customer_answer_verifier"]["scene_key"] == "pmflow_ai_answer_presentation" assert items["customer_answer_verifier"]["scene_label"] == "客户答案呈现与质检" assert items["pmflow_ai_operating_mission"]["scene_key"] == "space_pmflow_ai_operating_runtime" assert items["pmflow_ai_operating_mission"]["scene_label"] == "Space AI Operating Mission" @pytest.mark.asyncio async def test_ai_call_trace_list_shows_real_capability_probe_calls_by_default(db_session) -> None: db_session.add_all( [ AICallLog( tenant_id=1, user_id=7, function_name="probe_engine_capabilities", prompt="basic_chat,structured_output", response='{"basic_chat":{"status":"verified"}}', tokens_used=100, cost=0.001, status="capability_probe_partial", trace_id="trace-probe-summary", provider="dashscope", model="qwen-plus", routing_meta={ "scene_key": "platform_ai_ops", "routing_decision": "direct_capability_probe", "route_source": "provider_console", "probe_run_id": "cap-1", "probe_targets": ["basic_chat", "structured_output"], }, provider_attempts=[ {"probe_target": "basic_chat", "status": "success"}, {"probe_target": "structured_output", "status": "success"}, ], created_at=datetime(2026, 4, 29, 10, 1, 0), ), AICallLog( tenant_id=1, user_id=7, function_name="probe_capability_basic_chat", prompt="发送最小连通 prompt，要求模型只返回 OK。", response="OK", tokens_used=6, cost=0.00001, status="capability_verified", trace_id="trace-probe-basic-chat", provider="dashscope", model="qwen-plus", routing_meta={ "scene_key": "platform_ai_ops", "routing_decision": "direct_capability_probe", "route_source": "provider_console", "probe_run_id": "cap-1", "probe_target": "basic_chat", }, created_at=datetime(2026, 4, 29, 10, 1, 1), ), ] ) await db_session.commit() app = FastAPI() app.include_router(ai_call_traces_api.router) async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db app.dependency_overrides[require_platform_ai_ops] = lambda: SimpleNamespace(id=1, role="OWNER") async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: default_response = await client.get("/ai-call-traces?provider=dashscope&since_hours=8760") legacy_summary_response = await client.get("/ai-call-traces?provider=dashscope&function_name=probe_engine_capabilities&since_hours=8760") assert default_response.status_code == 200 default_payload = default_response.json()["data"] assert default_payload["total"] == 1 assert default_payload["summary"]["total_calls"] == 1 assert default_payload["items"][0]["function_name"] == "probe_capability_basic_chat" assert default_payload["items"][0]["function_label"] == "能力探测：基础聊天" assert legacy_summary_response.status_code == 200 legacy_summary_payload = legacy_summary_response.json()["data"] assert legacy_summary_payload["total"] == 1 assert legacy_summary_payload["items"][0]["function_name"] == "probe_engine_capabilities" assert legacy_summary_payload["items"][0]["function_label"] == "供应商能力探测汇总（历史兼容）" @pytest.mark.asyncio async def test_ai_call_trace_structured_timeout_has_specific_diagnosis(db_session) -> None: db_session.add( AICallLog( tenant_id=1, user_id=7, function_name="pmflow_launch_draft_v1", prompt="请生成项目草案", response="模型调用失败：function=pmflow_launch_draft_v1, reason=upstream_empty_or_invalid_response", tokens_used=128, cost=0.001, status="structured_timeout", trace_id="trace-timeout", provider="dashscope", model="qwen-plus", routing_meta={ "scene_key": "project_launch", "slot_key": "structured", "provider_timeout_seconds": 90, "job_timeout_seconds": 180, "input_bucket": "normal", "scene_timeout_policy_id": "project_launch:pmflow_launch_draft_v1:normal", "scene_timeout_policy_version": "scene-timeout-v1", }, provider_attempts=[ {"status": "request_error", "error_type": "ReadTimeout", "error": ""}, {"status": "request_error", "error_type": "ReadTimeout", "error": ""}, ], planned_tier="structured", executed_tier="structured", route_source="slot_binding", created_at=datetime(2026, 5, 1, 19, 37, 42), ) ) await db_session.commit() app = FastAPI() app.include_router(ai_call_traces_api.router) async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db app.dependency_overrides[require_platform_ai_ops] = lambda: SimpleNamespace(id=1, role="OWNER") async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: list_response = await client.get("/ai-call-traces?status=structured_timeout&since_hours=8760") detail_response = await client.get("/ai-call-traces/trace-timeout") assert list_response.status_code == 200 payload = list_response.json()["data"] assert payload["summary"]["failed_calls"] == 1 assert payload["items"][0]["status_label"] == "模型超时" assert payload["items"][0]["error_diagnosis"]["code"] == "model_read_timeout" assert payload["items"][0]["error_diagnosis"]["attempt_count"] == 2 assert detail_response.status_code == 200 detail = detail_response.json()["data"] assert detail["runtime_policy"]["provider_timeout_seconds"] == 90 assert detail["runtime_policy"]["job_timeout_seconds"] == 180 assert detail["runtime_policy"]["input_bucket"] == "normal" assert detail["error_diagnosis"]["attempt_errors"][0]["error_type"] == "ReadTimeout" assert "fallback runtime" in " ".join(detail["error_diagnosis"]["recommendation"]["recommended_actions"])