112 lines
4.9 KiB
Python
112 lines
4.9 KiB
Python
"""Source authority decisions for PMFlow AI turn runtime.
|
|
|
|
This layer decides which source domain has authority for the current turn.
|
|
It does not classify user text by keywords. It only consumes structured turn
|
|
state: object binding, selected goal frame, requested source policy and already
|
|
performed actions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
from ..schemas.pmflow_ai_turn_runtime import PMFlowAITurnState, RuntimeActionKind
|
|
from .pmflow_ai_static_contract_registry import PMFlowAIStaticContractRegistry
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PMFlowAIEvidenceAuthorityDecision:
|
|
authority_domain: str = "unrestricted"
|
|
required_evidence: str = "unrestricted"
|
|
contract_key: str = ""
|
|
required_first_action: RuntimeActionKind | None = None
|
|
forbidden_first_actions: tuple[RuntimeActionKind, ...] = field(default_factory=tuple)
|
|
can_answer_without_tool: bool = False
|
|
renderer_key: str = ""
|
|
verifier_policy: str = ""
|
|
process_label: str = ""
|
|
reason: str = ""
|
|
|
|
@property
|
|
def needs_probe(self) -> bool:
|
|
return bool(self.required_first_action)
|
|
|
|
|
|
class PMFlowAISourceAuthority:
|
|
"""Protect source selection before the action selector runs."""
|
|
|
|
def decide(self, turn_state: PMFlowAITurnState) -> PMFlowAIEvidenceAuthorityDecision:
|
|
static_contract = PMFlowAIStaticContractRegistry().resolve_for_turn_state(turn_state)
|
|
if static_contract is not None:
|
|
return PMFlowAIEvidenceAuthorityDecision(
|
|
authority_domain=static_contract.authority_domain,
|
|
required_evidence=static_contract.required_evidence,
|
|
contract_key=static_contract.contract_key,
|
|
forbidden_first_actions=static_contract.forbidden_first_actions,
|
|
can_answer_without_tool=True,
|
|
renderer_key=static_contract.renderer_key,
|
|
verifier_policy=static_contract.verifier_policy,
|
|
process_label=static_contract.process_label,
|
|
reason=static_contract.reason,
|
|
)
|
|
|
|
primary = turn_state.object_state.primary_object
|
|
if primary is None:
|
|
return PMFlowAIEvidenceAuthorityDecision()
|
|
if str(primary.object_type or "").strip().lower() != "project":
|
|
return PMFlowAIEvidenceAuthorityDecision()
|
|
if not _is_strong_project_binding(turn_state):
|
|
return PMFlowAIEvidenceAuthorityDecision()
|
|
if _project_status_already_read(turn_state) or _project_operating_evidence_available(turn_state):
|
|
return PMFlowAIEvidenceAuthorityDecision(
|
|
authority_domain="pmflow_project_operating_state",
|
|
required_evidence="project_operating_state",
|
|
forbidden_first_actions=("external_search", "external_fetch"),
|
|
reason="已读取绑定项目运行态，后续可按证据缺口继续。",
|
|
)
|
|
return PMFlowAIEvidenceAuthorityDecision(
|
|
authority_domain="pmflow_project_operating_state",
|
|
required_evidence="project_operating_state",
|
|
required_first_action="inspect_project_status",
|
|
forbidden_first_actions=("external_search", "external_fetch"),
|
|
reason="本轮已绑定 PMFlow 项目对象，必须先读取项目运行态证据，不能用外部公开来源替代项目进展。",
|
|
)
|
|
|
|
|
|
class PMFlowAIEvidenceAuthority(PMFlowAISourceAuthority):
|
|
"""Backward-compatible name for existing runtime imports."""
|
|
|
|
|
|
def _is_strong_project_binding(turn_state: PMFlowAITurnState) -> bool:
|
|
primary = turn_state.object_state.primary_object
|
|
if primary is None:
|
|
return False
|
|
binding = str(primary.binding_strength or "").strip().lower()
|
|
if binding in {"", "none", "context", "visible_context"}:
|
|
return False
|
|
return bool(str(primary.object_ref or primary.title or "").strip())
|
|
|
|
|
|
def _project_status_already_read(turn_state: PMFlowAITurnState) -> bool:
|
|
return any(action.action_type == "inspect_project_status" for action in list(turn_state.action_state.performed_actions or []))
|
|
|
|
|
|
def _project_operating_evidence_available(turn_state: PMFlowAITurnState) -> bool:
|
|
structured = dict(turn_state.evidence_state.structured_evidence or {})
|
|
primary = turn_state.object_state.primary_object
|
|
if primary is None:
|
|
return False
|
|
primary_ref = str(primary.object_ref or "").strip()
|
|
primary_title = str(primary.title or "").strip()
|
|
packets = structured.get("project_operating_evidence_packets")
|
|
rows = packets if isinstance(packets, list) else []
|
|
single = structured.get("project_operating_evidence_packet")
|
|
for packet in [*rows, single]:
|
|
if not isinstance(packet, dict):
|
|
continue
|
|
project_ref = str(packet.get("project_ref") or "").strip()
|
|
project_title = str(packet.get("project_title") or "").strip()
|
|
if (primary_ref and project_ref == primary_ref) or (primary_title and project_title == primary_title):
|
|
return True
|
|
return False
|