110 lines
4.5 KiB
Python
110 lines
4.5 KiB
Python
import copy
|
|
import inspect
|
|
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_answer_verifier_registry import PMFlowAIAnswerVerifierRegistry
|
|
from app.services.pmflow_ai_static_contract_registry import PMFlowAIStaticContractRegistry
|
|
from app.services.pmflow_ai_turn_runtime_service import (
|
|
PMFlowAITurnRuntimeService,
|
|
_render_source_authority_answer_from_turn_state,
|
|
)
|
|
|
|
|
|
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", "model_provider"],
|
|
"allowed_response_mode": "protected_refusal",
|
|
"contract_key": "protected_internal_refusal.v1",
|
|
},
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_packet_an_static_contract_is_answer_runtime_material_not_exact_answer() -> None:
|
|
state = _protected_internal_state()
|
|
contract = PMFlowAIStaticContractRegistry().resolve_for_turn_state(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(),
|
|
)
|
|
|
|
assert contract is not None
|
|
assert not hasattr(contract, "answer")
|
|
assert payload["answer_runtime_packet"]["plan"]["draft_sources"] == ["产品能力或披露边界合同"]
|
|
assert payload["answer_runtime_packet"]["draft"]["source_renderer"] == "产品能力或披露边界合同"
|
|
assert "不能披露" in payload["answer"]
|
|
assert ".answer" not in inspect.getsource(_render_source_authority_answer_from_turn_state)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_packet_an_verifier_accepts_contract_guided_rewording() -> 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(),
|
|
)
|
|
reworded = copy.deepcopy(payload)
|
|
reworded["answer"] = (
|
|
"这类内容属于 PMFlow 的内部实现与机密信息，用户侧不能披露。"
|
|
"我可以改为说明产品里可见、可使用的能力边界。"
|
|
)
|
|
reworded["reply_text"] = reworded["answer"]
|
|
|
|
failure = PMFlowAIAnswerVerifierRegistry().verify_payload(turn_state=state, payload=reworded)
|
|
|
|
assert failure is None
|
|
|
|
|
|
def test_packet_an_verifier_rejects_static_contract_without_answer_runtime_packet() -> None:
|
|
state = _protected_internal_state()
|
|
payload = {
|
|
"_runtime_payload_kind": "final_answer",
|
|
"answer": "PMFlow 的后端在 backend/app/services/pmflow_ai_turn_runtime_service.py 里组织运行。",
|
|
"customer_evidence_packet": {"source_summary": ["PMFlow 内部信息保护边界"]},
|
|
}
|
|
|
|
failure = PMFlowAIAnswerVerifierRegistry().verify_payload(turn_state=state, payload=payload)
|
|
|
|
assert failure is not None
|
|
assert failure.failure_type == "static_contract_mismatch"
|
|
assert failure.next_action_hint == "synthesize_answer"
|