128 lines
5.2 KiB
Python
128 lines
5.2 KiB
Python
"""Evidence coverage gate for PMFlow AI turn runtime."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Literal
|
|
|
|
from ..schemas.pmflow_ai_turn_runtime import PMFlowAITurnState, RuntimeActionKind
|
|
|
|
|
|
CoverageStatus = Literal["not_applicable", "needs_probe", "sufficient", "insufficient_after_probe"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PMFlowAIEvidenceCoverageDecision:
|
|
status: CoverageStatus = "not_applicable"
|
|
missing: list[str] = field(default_factory=list)
|
|
next_action_type: RuntimeActionKind | None = None
|
|
reason: str = ""
|
|
|
|
@property
|
|
def needs_probe(self) -> bool:
|
|
return self.status == "needs_probe" and self.next_action_type is not None
|
|
|
|
@property
|
|
def sufficient(self) -> bool:
|
|
return self.status in {"not_applicable", "sufficient", "insufficient_after_probe"}
|
|
|
|
|
|
class PMFlowAIEvidenceCoverageGate:
|
|
"""Require minimum structured evidence before final synthesis."""
|
|
|
|
def evaluate(self, turn_state: PMFlowAITurnState) -> PMFlowAIEvidenceCoverageDecision:
|
|
if not _requires_project_execution_coverage(turn_state):
|
|
return PMFlowAIEvidenceCoverageDecision()
|
|
|
|
missing = _missing_project_execution_signals(turn_state)
|
|
if not missing:
|
|
return PMFlowAIEvidenceCoverageDecision(status="sufficient", reason="项目执行态证据覆盖已满足。")
|
|
|
|
if not _action_performed(turn_state, "inspect_project_status"):
|
|
return PMFlowAIEvidenceCoverageDecision(
|
|
status="needs_probe",
|
|
missing=missing,
|
|
next_action_type="inspect_project_status",
|
|
reason="当前回答需要项目执行态覆盖，必须先读取项目阶段、当前执行、阻塞和任务队列。",
|
|
)
|
|
return PMFlowAIEvidenceCoverageDecision(
|
|
status="insufficient_after_probe",
|
|
missing=missing,
|
|
reason="已读取项目执行态，但证据覆盖仍不完整；最终回答必须显式保留缺口边界。",
|
|
)
|
|
|
|
|
|
def _requires_project_execution_coverage(turn_state: PMFlowAITurnState) -> bool:
|
|
selected = turn_state.selected_frame
|
|
if selected is None:
|
|
return False
|
|
if turn_state.answer_state.answer_shape not in {"execution_guidance", "status_answer"}:
|
|
return False
|
|
if selected.operation_kind not in {"inspect_object", "guide_execution"}:
|
|
return False
|
|
primary = turn_state.object_state.primary_object
|
|
if primary is None:
|
|
return False
|
|
if str(primary.object_type or "").lower() != "project":
|
|
return False
|
|
return str(primary.binding_strength or "").lower() not in {"", "none", "context", "visible_context"}
|
|
|
|
|
|
def _missing_project_execution_signals(turn_state: PMFlowAITurnState) -> list[str]:
|
|
packet = _project_operating_packet_for_turn(turn_state)
|
|
if not packet:
|
|
return list(_PROJECT_EXECUTION_COVERAGE_PROFILE.keys())
|
|
missing: list[str] = []
|
|
for gap_key, facet_keys in _PROJECT_EXECUTION_COVERAGE_PROFILE.items():
|
|
if not _any_facet_covered(packet=packet, facet_keys=facet_keys):
|
|
missing.append(gap_key)
|
|
return missing
|
|
|
|
|
|
_PROJECT_EXECUTION_COVERAGE_PROFILE: dict[str, tuple[str, ...]] = {
|
|
"project_stage": ("command_state",),
|
|
"current_execution_state": ("execution_state",),
|
|
"active_or_next_tasks": ("workbench_state",),
|
|
"blockers_or_gate": ("risk_blocker_state", "decision_review_state", "release_growth_state"),
|
|
}
|
|
|
|
|
|
def _project_operating_packet_for_turn(turn_state: PMFlowAITurnState) -> dict[str, Any]:
|
|
structured = dict(turn_state.evidence_state.structured_evidence or {})
|
|
packets = structured.get("project_operating_evidence_packets")
|
|
rows = packets if isinstance(packets, list) else []
|
|
primary = turn_state.object_state.primary_object
|
|
if primary is None:
|
|
return {}
|
|
primary_ref = str(primary.object_ref or "").strip()
|
|
primary_title = str(primary.title or "").strip()
|
|
for packet in rows:
|
|
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 packet
|
|
single = structured.get("project_operating_evidence_packet")
|
|
if isinstance(single, dict):
|
|
project_ref = str(single.get("project_ref") or "").strip()
|
|
project_title = str(single.get("project_title") or "").strip()
|
|
if (primary_ref and project_ref == primary_ref) or (primary_title and project_title == primary_title):
|
|
return single
|
|
return {}
|
|
|
|
|
|
def _any_facet_covered(*, packet: dict[str, Any], facet_keys: tuple[str, ...]) -> bool:
|
|
return any(_facet_covered(packet.get(key)) for key in facet_keys)
|
|
|
|
|
|
def _facet_covered(value: Any) -> bool:
|
|
if not isinstance(value, dict):
|
|
return False
|
|
status = str(value.get("coverage_status") or "").strip()
|
|
return status in {"covered", "partial"}
|
|
|
|
|
|
def _action_performed(turn_state: PMFlowAITurnState, action_type: RuntimeActionKind) -> bool:
|
|
return any(action.action_type == action_type for action in list(turn_state.action_state.performed_actions or []))
|