"""Verifier policy registry for PMFlow AI turn runtime. Verifier policies decide whether a runtime payload is allowed to leave the turn loop. They consume canonical TurnState, authority decisions and registered contracts; they do not classify raw user text. """ from __future__ import annotations from typing import Any from ..schemas.pmflow_ai_turn_runtime import PMFlowAITurnState, PMFlowAIVerificationFailure from .pmflow_ai_evidence_authority import PMFlowAIEvidenceAuthority from .pmflow_ai_evidence_coverage_gate import PMFlowAIEvidenceCoverageGate from .pmflow_ai_static_contract_registry import PMFlowAIStaticContractRegistry class PMFlowAIAnswerVerifierRegistry: """Run ordered verifier policies against a runtime payload.""" def verify_payload( self, *, turn_state: PMFlowAITurnState, payload: dict[str, Any] | None, ) -> PMFlowAIVerificationFailure | None: if not isinstance(payload, dict): return None if _payload_is_final_answer(payload): failure = self._verify_final_answer(turn_state=turn_state, payload=payload) if failure is not None: return failure if isinstance(payload.get("interaction_request"), dict): failure = self._verify_interaction_request(turn_state=turn_state) if failure is not None: return failure failure = self._verify_collection_answer_shape(turn_state=turn_state) if failure is not None: return failure return None def _verify_final_answer( self, *, turn_state: PMFlowAITurnState, payload: dict[str, Any], ) -> PMFlowAIVerificationFailure | None: authority = PMFlowAIEvidenceAuthority().decide(turn_state) if authority.required_first_action is not None: return _verification_failure( failure_type=_wrong_source_failure_type(authority.authority_domain), reason=authority.reason or "Final answer attempted before reading required authority evidence.", state_patch={"can_answer_now": False, "needs_internal_evidence": True}, next_action_hint=authority.required_first_action, ) if authority.verifier_policy in {"static_contract", "protected_refusal"}: failure = self._verify_static_contract(turn_state=turn_state, payload=payload) if failure is not None: return failure if authority.verifier_policy == "protected_refusal": return None coverage = PMFlowAIEvidenceCoverageGate().evaluate(turn_state) if coverage.needs_probe and coverage.next_action_type is not None: return _verification_failure( failure_type="goal_fulfillment_missing_project_operating_facets", reason=coverage.reason, state_patch={"can_answer_now": False, "needs_internal_evidence": True}, next_action_hint=coverage.next_action_type, ) return None def _verify_static_contract( self, *, turn_state: PMFlowAITurnState, payload: dict[str, Any], ) -> PMFlowAIVerificationFailure | None: authority = PMFlowAIEvidenceAuthority().decide(turn_state) contract = PMFlowAIStaticContractRegistry().get(authority.contract_key) if contract is None: return _verification_failure( failure_type="static_contract_missing", reason="Source Authority selected a static contract, but the contract registry could not resolve it.", state_patch={"can_answer_now": False}, next_action_hint="render_or_fail_closed", customer_safe_summary="这轮缺少可公开说明的产品边界材料，不能把内部校验细节展示给用户。", customer_safe_effect={ "blocked_claims": ["不能把内部校验失败原因当成用户结论。"], "answerable_parts": [], }, retry_would_change_state=False, ) answer = str(payload.get("answer") or payload.get("reply_text") or "").strip() runtime_packet = payload.get("answer_runtime_packet") if not isinstance(runtime_packet, dict): return _verification_failure( failure_type="static_contract_mismatch", reason="Static contract answer did not pass through Answer Runtime objective/plan/draft packet.", state_patch={ "can_answer_now": True, "needs_internal_evidence": False, "needs_external_evidence": False, }, next_action_hint="synthesize_answer", customer_safe_summary="这轮答案没有正确承接客户可见的产品边界，不能把内部校验细节展示给用户。", customer_safe_effect={ "blocked_claims": ["不能把内部实现、合同校验或运行时诊断作为用户答案。"], "answerable_parts": ["可以说明客户侧可见的产品能力边界。"], }, retry_would_change_state=False, ) packet = payload.get("customer_evidence_packet") if isinstance(payload.get("customer_evidence_packet"), dict) else {} source_summary = { str(item or "").strip() for item in list(packet.get("source_summary") or []) if str(item or "").strip() } required_sources = {str(item or "").strip() for item in contract.source_summary if str(item or "").strip()} objective = runtime_packet.get("objective") if isinstance(runtime_packet.get("objective"), dict) else {} plan = runtime_packet.get("plan") if isinstance(runtime_packet.get("plan"), dict) else {} draft = runtime_packet.get("draft") if isinstance(runtime_packet.get("draft"), dict) else {} if ( answer and required_sources.issubset(source_summary) and _answer_runtime_supports_contract( turn_state=turn_state, contract_key=contract.contract_key, objective=objective, plan=plan, draft=draft, ) ): return None return _verification_failure( failure_type="static_contract_mismatch", reason="Static contract answer payload did not satisfy the registered customer-safe product contract and Answer Runtime plan.", state_patch={ "can_answer_now": True, "needs_internal_evidence": False, "needs_external_evidence": False, }, next_action_hint="synthesize_answer", customer_safe_summary="这轮答案没有正确承接客户可见的产品边界，不能把内部校验细节展示给用户。", customer_safe_effect={ "blocked_claims": ["不能把内部实现、合同校验或运行时诊断作为用户答案。"], "answerable_parts": ["可以说明客户侧可见的产品能力边界。"], }, retry_would_change_state=False, ) def _verify_interaction_request(self, *, turn_state: PMFlowAITurnState) -> PMFlowAIVerificationFailure | None: if _has_actionable_user_gap(turn_state): return None return _verification_failure( failure_type="interaction_not_user_actionable", reason="Generated interaction_request without a user-actionable gap.", state_patch={"must_ask_user": False, "can_answer_now": True}, next_action_hint="render_or_fail_closed", ) def _verify_collection_answer_shape(self, *, turn_state: PMFlowAITurnState) -> PMFlowAIVerificationFailure | None: selected = turn_state.selected_frame if selected is None or selected.operation_kind != "list_collection": return None if not any("list_collection" in observation.can_answer_operations for observation in turn_state.observations): return None return _verification_failure( failure_type="wrong_answer_shape", reason="Collection evidence can answer list_collection, so clarification is not allowed.", state_patch={"must_ask_user": False, "can_answer_now": True}, next_action_hint="render_collection_list", ) def _payload_is_final_answer(payload: dict[str, Any] | None) -> bool: if not isinstance(payload, dict): return False kind = str(payload.get("_runtime_payload_kind") or "").strip() if kind == "evidence": return False if kind == "final_answer": return True return bool(str(payload.get("answer") or payload.get("reply_text") or "").strip()) def _has_actionable_user_gap(turn_state: PMFlowAITurnState) -> bool: for gap in list(turn_state.operation_gaps or []): if gap.blocking and gap.user_actionable and gap.gap_type in {"user_clarification_gap", "user_authorization_gap"}: return True for gap in list(turn_state.evidence_state.evidence_gaps or []): if gap.severity == "blocking" and gap.user_actionable and gap.gap_type in {"user_clarification_gap", "user_authorization_gap"}: return True return False def _verification_failure( *, failure_type: str, reason: str, state_patch: dict[str, Any], next_action_hint: str, customer_safe_summary: str = "", customer_safe_effect: dict[str, Any] | None = None, retry_would_change_state: bool = True, ) -> PMFlowAIVerificationFailure: safe_effect = dict(customer_safe_effect or {}) delta = { "packet_version": "pmflow_ai_verification_delta.v1", "verification_ok": False, "failure_type": failure_type, "claim_deltas": [], "boundary_deltas": [], "attempt_delta": { "action_type": "verify_answer", "status": "failed", "customer_visible_summary": customer_safe_summary, }, "retry_would_change_state": bool(retry_would_change_state), "internal_diagnostic": reason, "customer_safe_effect": safe_effect, "next_action_hint": next_action_hint, "state_patch": dict(state_patch or {}), } return PMFlowAIVerificationFailure( failure_id="runtime_verification_failed", failure_type=failure_type, reason=reason, internal_diagnostic=reason, customer_safe_summary=customer_safe_summary, customer_safe_effect=safe_effect, verification_delta=delta, retry_would_change_state=retry_would_change_state, state_patch=state_patch, next_action_hint=next_action_hint, ) def _wrong_source_failure_type(authority_domain: str) -> str: if authority_domain == "pmflow_project_operating_state": return "wrong_source_or_missing_project_operating_state" return "wrong_source_or_missing_required_evidence" def _answer_runtime_supports_contract( *, turn_state: PMFlowAITurnState, contract_key: str, objective: dict[str, Any], plan: dict[str, Any], draft: dict[str, Any], ) -> bool: evidence_to_use = { str(item or "").strip() for item in list(objective.get("evidence_to_use") or []) if str(item or "").strip() } draft_sources = { str(item or "").strip() for item in list(plan.get("draft_sources") or []) if str(item or "").strip() } source_renderer = str(draft.get("source_renderer") or "").strip() if not evidence_to_use and not draft_sources and not source_renderer: return False if contract_key == "protected_internal_refusal.v1": must_include = { str(item or "").strip() for item in list(objective.get("must_include") or []) if str(item or "").strip() } must_not_include = { str(item or "").strip() for item in list(objective.get("must_not_include") or []) if str(item or "").strip() } protected_categories = { str(item or "").strip() for item in list(turn_state.disclosure_state.protected_categories or []) if str(item or "").strip() } return ( str(objective.get("required_decision") or "").strip() == "explain_customer_safe_boundary_and_allowed_alternative" and "说明用户侧信息披露边界" in must_include and protected_categories.issubset(must_not_include) and str(plan.get("answer_strategy") or "").strip() == "state_disclosure_boundary_then_offer_customer_safe_alternative" ) return bool(evidence_to_use or draft_sources or source_renderer)