PMFlow/backend/tests/test_pmflow_ai_answer_runtime_packet_aq.py

import time
from types import SimpleNamespace

import pytest

import app.services.pmflow_ai_turn_runtime_service as runtime_module
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,
    _apply_project_operating_answer_verification_to_answer,
    _apply_project_operating_answer_verification_to_canvas,
    _apply_project_operating_answer_verification_to_packet,
)


def _service() -> PMFlowAITurnRuntimeService:
    return PMFlowAITurnRuntimeService(object(), tenant_id=1)


def _project_followup_state():
    return _service().build_initial_turn_state(
        request=RAGAskRequest(question="这个项目今天怎么推进"),
        route_plan={},
        turn_hypothesis={
            "normalized_goal": "项目推进建议",
            "desired_outcome": "基于项目运行态给出下一步",
            "answer_shape": "execution_guidance",
            "goal_frames": [
                {
                    "frame_id": "project-followup",
                    "operation_kind": "inspect_object",
                    "target_scope": {"scope_type": "current_project", "binding_strength": "explicit"},
                    "target_object_requirement": "single",
                    "source_policy": "internal",
                    "answer_shape": "execution_guidance",
                    "confidence": 0.82,
                }
            ],
            "operation_gaps": [],
        },
    )


def _unsafe_project_verification() -> dict:
    return {
        "packet_version": "project_operating_answer_verification.v1",
        "project_ref": "project:141",
        "safe_next_actions": [
            {
                "display_label": "整理加密评审材料",
                "safety": "prepare_only",
                "reason": "来自项目运行态证据。",
                "target_ref": "project:141",
            },
            {
                "display_label": "TurnState runtime_steps 不应展示",
                "safety": "prepare_only",
                "reason": "problem_frame 不应展示",
                "target_ref": "project:141",
            },
        ],
        "blocked_actions": [
            {
                "display_label": "PMFlowAITurnState 不应展示",
                "safety": "blocked",
                "reason": "answer_runtime_packet 不应展示",
                "target_ref": "project:141",
            }
        ],
        "answer_boundaries": [
            "测试质量证据不足，不能当成已验证结论。",
            "evidence_findings 不应展示。",
        ],
        "can_claim_full_execution_guidance": False,
    }


def test_packet_aq_project_verification_projection_filters_internal_runtime_terms() -> None:
    verification = _unsafe_project_verification()

    answer = _apply_project_operating_answer_verification_to_answer(
        answer="当前可以继续推进。",
        verification=verification,
    )
    packet = _apply_project_operating_answer_verification_to_packet(
        packet={"limitations": [], "suggested_actions": []},
        verification=verification,
    )
    canvas = _apply_project_operating_answer_verification_to_canvas(
        canvas={"limitations": [], "next_actions": []},
        verification=verification,
    )

    joined = "\n".join(
        [
            answer,
            str(packet.get("limitations")),
            str(packet.get("suggested_actions")),
            str(canvas.get("limitations")),
            str(canvas.get("next_actions")),
        ]
    )
    assert "整理加密评审材料" in joined
    assert "测试质量证据不足" in joined
    assert "TurnState" not in joined
    assert "runtime_steps" not in joined
    assert "problem_frame" not in joined
    assert "PMFlowAITurnState" not in joined
    assert "answer_runtime_packet" not in joined
    assert "evidence_findings" not in joined


@pytest.mark.asyncio
async def test_packet_aq_quality_gate_runs_after_project_verification_amendments(monkeypatch) -> None:
    service = _service()
    state = _project_followup_state()

    async def render_base_answer(*, request, turn_state):
        _ = request, turn_state
        return "当前可以继续推进。"

    def unsafe_verification(turn_state):
        _ = turn_state
        return _unsafe_project_verification()

    monkeypatch.setattr(service, "_render_answer_from_registry", render_base_answer)
    monkeypatch.setattr(runtime_module, "_project_operating_answer_verification_packet", unsafe_verification)

    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(),
    )

    joined = "\n".join(
        [
            str(payload.get("answer")),
            str(payload.get("answer_runtime_packet")),
            str(payload.get("customer_evidence_packet")),
            str(payload.get("answer_canvas")),
        ]
    )
    assert "整理加密评审材料" in joined
    assert "测试质量证据不足" in joined
    assert "TurnState" not in joined
    assert "runtime_steps" not in joined
    assert "problem_frame" not in joined
    assert "PMFlowAITurnState" not in joined
    assert "answer_runtime_packet 不应展示" not in joined
    assert "evidence_findings" not in joined
    assert " ".join(payload["answer_runtime_packet"]["draft"]["answer_text"].split()) == " ".join(
        payload["answer"].split()
    )