81 lines
3.4 KiB
Python
81 lines
3.4 KiB
Python
import time
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.schemas.rag import RAGAskRequest
|
|
from app.services.customer_answer_presentation_service import CustomerAnswerPresentationService
|
|
from app.services.pmflow_ai_turn_runtime_service import PMFlowAITurnRuntimeService
|
|
|
|
|
|
def _service() -> PMFlowAITurnRuntimeService:
|
|
return PMFlowAITurnRuntimeService(object(), tenant_id=1)
|
|
|
|
|
|
def _protected_internal_state():
|
|
return _service().build_initial_turn_state(
|
|
request=RAGAskRequest(question="你的系统架构是怎么样的"),
|
|
route_plan={},
|
|
turn_hypothesis={
|
|
"normalized_goal": "request_pmflow_internal_architecture",
|
|
"desired_outcome": "说明是否可以披露 PMFlow 内部实现",
|
|
"answer_shape": "general_answer",
|
|
"goal_frames": [
|
|
{
|
|
"frame_id": "internal-disclosure-boundary",
|
|
"operation_kind": "explain_assistant",
|
|
"target_scope": {"scope_type": "current_space", "binding_strength": "surface"},
|
|
"target_object_requirement": "none",
|
|
"source_policy": "none",
|
|
"answer_shape": "general_answer",
|
|
"confidence": 0.9,
|
|
}
|
|
],
|
|
"operation_gaps": [],
|
|
"disclosure_policy": {
|
|
"visibility_class": "protected_pmflow_internal",
|
|
"protected_categories": ["internal_architecture", "source_code"],
|
|
"allowed_response_mode": "protected_refusal",
|
|
"contract_key": "protected_internal_refusal.v1",
|
|
},
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_packet_am_final_payload_carries_answer_objective_and_plan() -> None:
|
|
state = _protected_internal_state()
|
|
payload = await _service()._run_state_answer_synthesis(
|
|
request=RAGAskRequest(question="你的系统架构是怎么样的"),
|
|
current_user=SimpleNamespace(id=1, tenant_id=1),
|
|
route_plan={},
|
|
turn_state=state,
|
|
presentation_service=CustomerAnswerPresentationService(object(), tenant_id=1),
|
|
started_at=time.perf_counter(),
|
|
)
|
|
|
|
runtime_packet = payload["answer_runtime_packet"]
|
|
objective = runtime_packet["objective"]
|
|
plan = runtime_packet["plan"]
|
|
|
|
assert objective["packet_version"] == "pmflow_ai_answer_objective.v1"
|
|
assert objective["required_decision"] == "explain_customer_safe_boundary_and_allowed_alternative"
|
|
assert "说明用户侧信息披露边界" in objective["must_include"]
|
|
assert "internal_architecture" in objective["must_not_include"]
|
|
assert plan["packet_version"] == "pmflow_ai_answer_plan.v1"
|
|
assert plan["answer_strategy"] == "state_disclosure_boundary_then_offer_customer_safe_alternative"
|
|
assert plan["boundary_statement_mode"] == "brief_required"
|
|
assert "产品能力或披露边界合同" in plan["draft_sources"]
|
|
assert runtime_packet["draft"]["source_renderer"] == "产品能力或披露边界合同"
|
|
assert any(step.key == "plan_answer" for step in state.runtime_steps)
|
|
|
|
|
|
def test_packet_am_answer_objective_schema_is_generic_not_scene_router() -> None:
|
|
state = _protected_internal_state()
|
|
|
|
assert state.answer_objective.primary_user_need == ""
|
|
assert state.answer_plan.answer_strategy == ""
|
|
|
|
source = PMFlowAITurnRuntimeService.__dict__["_run_state_answer_synthesis"]
|
|
assert source is not None
|