223 lines
8.4 KiB
Python
223 lines
8.4 KiB
Python
from __future__ import annotations
|
|
|
|
from app.schemas.pmflow_ai_turn_runtime import (
|
|
PMFlowAIEvidenceGap,
|
|
PMFlowAIGoalFrame,
|
|
PMFlowAIOperationGap,
|
|
PMFlowAIObservation,
|
|
PMFlowAIObservationItem,
|
|
PMFlowAIStateRevision,
|
|
PMFlowAITargetCollection,
|
|
PMFlowAITargetScope,
|
|
PMFlowAITurnState,
|
|
PMFlowAIVerificationFailure,
|
|
)
|
|
from app.services.pmflow_ai_turn_contract import (
|
|
canonical_action_hint_for_gap,
|
|
normalize_turn_hypothesis_contract,
|
|
validate_goal_frame_contract,
|
|
)
|
|
|
|
|
|
def test_working_state_contract_represents_current_space_project_collection() -> None:
|
|
scope = PMFlowAITargetScope(scope_type="current_space", binding_strength="surface", source="space_runtime_context")
|
|
frame = PMFlowAIGoalFrame(
|
|
frame_id="projects-in-current-space",
|
|
operation_kind="list_collection",
|
|
target_scope=scope,
|
|
target_collection=PMFlowAITargetCollection(object_type="project", required=True, scope=scope),
|
|
target_object_requirement="none",
|
|
answer_shape="fact_list",
|
|
confidence=0.91,
|
|
)
|
|
observation = PMFlowAIObservation(
|
|
observation_id="obs-projects",
|
|
observation_type="collection_snapshot",
|
|
action_phase="read",
|
|
scope=scope,
|
|
object_type="project",
|
|
count=1,
|
|
can_answer_operations=["list_collection"],
|
|
cannot_answer_operations=["inspect_object"],
|
|
confidence="high",
|
|
items=[
|
|
PMFlowAIObservationItem(
|
|
object_type="project",
|
|
object_ref="project:141",
|
|
title="童析一学龄前儿童行为特质 AI 评估助手",
|
|
status="进行中",
|
|
href="/pmflow-os?projectId=141",
|
|
)
|
|
],
|
|
)
|
|
state = PMFlowAITurnState(
|
|
candidate_frames=[frame],
|
|
selected_frame=frame,
|
|
observations=[observation],
|
|
customer_packet_draft={"operation_kind": "list_collection", "object_type": "project"},
|
|
)
|
|
|
|
assert state.selected_frame is not None
|
|
assert state.selected_frame.operation_kind == "list_collection"
|
|
assert state.selected_frame.target_object_requirement == "none"
|
|
assert state.observations[0].items[0].object_ref == "project:141"
|
|
assert state.observations[0].can_answer_operations == ["list_collection"]
|
|
|
|
|
|
def test_working_state_contract_represents_assistant_identity_without_project_binding() -> None:
|
|
frame = PMFlowAIGoalFrame(
|
|
frame_id="assistant-profile",
|
|
operation_kind="explain_assistant",
|
|
target_scope=PMFlowAITargetScope(scope_type="current_space", binding_strength="surface"),
|
|
target_object_requirement="none",
|
|
answer_shape="general_answer",
|
|
confidence=0.88,
|
|
)
|
|
observation = PMFlowAIObservation(
|
|
observation_id="obs-assistant-profile",
|
|
observation_type="assistant_profile",
|
|
action_phase="probe",
|
|
can_answer_operations=["explain_assistant"],
|
|
confidence="high",
|
|
summary="PMFlow AI 是当前空间的操作伙伴，只做建议、解释、聚合和受控路由。",
|
|
)
|
|
state = PMFlowAITurnState(candidate_frames=[frame], selected_frame=frame, observations=[observation])
|
|
|
|
assert state.object_state.primary_object is None
|
|
assert state.selected_frame is not None
|
|
assert state.selected_frame.operation_kind == "explain_assistant"
|
|
assert state.observations[0].observation_type == "assistant_profile"
|
|
|
|
|
|
def test_gap_contract_keeps_system_gaps_out_of_user_interaction() -> None:
|
|
gap = PMFlowAIEvidenceGap(
|
|
key="project_collection_reader_unavailable",
|
|
label="项目集合读取器暂不可用",
|
|
severity="blocking",
|
|
gap_type="system_evidence_gap",
|
|
user_actionable=True,
|
|
customer_question="请补充内部项目证据",
|
|
)
|
|
|
|
assert gap.user_actionable is False
|
|
assert gap.customer_question == ""
|
|
|
|
|
|
def test_gap_contract_marks_user_authorization_as_actionable() -> None:
|
|
gap = PMFlowAIEvidenceGap(
|
|
key="write_memory_authorization_required",
|
|
label="写入项目记忆前需要确认授权",
|
|
severity="blocking",
|
|
gap_type="user_authorization_gap",
|
|
customer_question="是否确认把这组候选变更写入项目记忆？",
|
|
)
|
|
|
|
assert gap.user_actionable is True
|
|
assert "项目记忆" in gap.customer_question
|
|
|
|
|
|
def test_working_state_records_revision_and_verification_feedback() -> None:
|
|
state = PMFlowAITurnState(
|
|
operation_gaps=[
|
|
PMFlowAIOperationGap(
|
|
key="collection_items_missing",
|
|
gap_type="system_evidence_gap",
|
|
blocking=True,
|
|
system_action_hint="retry_collection_reader",
|
|
)
|
|
],
|
|
state_revisions=[
|
|
PMFlowAIStateRevision(
|
|
revision_id="rev-collection",
|
|
source_observation_id="obs-projects",
|
|
selected_frame_id="projects-in-current-space",
|
|
reason="collection observation can answer list_collection",
|
|
state_patch={"must_ask_user": False},
|
|
)
|
|
],
|
|
verification_failures=[
|
|
PMFlowAIVerificationFailure(
|
|
failure_id="vf-wrong-surface",
|
|
failure_type="wrong_answer_shape",
|
|
reason="collection evidence exists, clarification is not allowed",
|
|
next_action_hint="render_collection_list",
|
|
)
|
|
],
|
|
)
|
|
|
|
assert state.operation_gaps[0].gap_type == "system_evidence_gap"
|
|
assert state.state_revisions[0].state_patch["must_ask_user"] is False
|
|
assert state.verification_failures[0].next_action_hint == "render_collection_list"
|
|
|
|
|
|
def test_turn_contract_normalizes_external_strategy_frame_and_freeform_hint() -> None:
|
|
normalized = normalize_turn_hypothesis_contract(
|
|
{
|
|
"normalized_goal": "获取 AI 漫剧行业的宏观趋势信息",
|
|
"answer_shape": "strategic_judgement",
|
|
"goal_frames": [
|
|
{
|
|
"frame_id": "industry_trend_research",
|
|
"operation_kind": "list_collection",
|
|
"target_scope": {"scope_type": "unspecified", "binding_strength": "none"},
|
|
"target_object_requirement": "none",
|
|
"answer_shape": "strategic_judgement",
|
|
"confidence": 0.95,
|
|
}
|
|
],
|
|
"operation_gaps": [
|
|
{
|
|
"key": "evidence_source_authority",
|
|
"gap_type": "system_evidence_gap",
|
|
"blocking": False,
|
|
"user_actionable": False,
|
|
"system_action_hint": "fetch latest industry reports and public announcements",
|
|
"reason": "需要实时聚合多源证据。",
|
|
}
|
|
],
|
|
"primary_object": {"binding_strength": "none"},
|
|
}
|
|
)
|
|
|
|
frame = normalized["goal_frames"][0]
|
|
gap = normalized["operation_gaps"][0]
|
|
|
|
assert frame["operation_kind"] == "judge_direction"
|
|
assert frame["source_policy"] == "external"
|
|
assert frame["target_scope"]["scope_type"] == "external_public"
|
|
assert gap["gap_type"] == "freshness_gap"
|
|
assert gap["system_action_hint"] == "external_search"
|
|
assert validate_goal_frame_contract(frame) == []
|
|
|
|
|
|
def test_turn_contract_rejects_external_project_collection_shape() -> None:
|
|
frame = PMFlowAIGoalFrame(
|
|
frame_id="bad-external-project-list",
|
|
operation_kind="list_collection",
|
|
target_scope=PMFlowAITargetScope(scope_type="external_public", binding_strength="surface"),
|
|
target_collection=PMFlowAITargetCollection(object_type="project", required=True),
|
|
answer_shape="strategic_judgement",
|
|
source_policy="external",
|
|
)
|
|
|
|
issues = validate_goal_frame_contract(frame)
|
|
|
|
assert issues
|
|
assert any("source_policy" in item or "answer_shape" in item for item in issues)
|
|
|
|
|
|
def test_freeform_system_hint_without_external_contract_is_not_canonical() -> None:
|
|
hint = canonical_action_hint_for_gap(
|
|
hint="read whatever seems useful",
|
|
gap_type="system_evidence_gap",
|
|
frame=PMFlowAIGoalFrame(
|
|
frame_id="internal-status",
|
|
operation_kind="inspect_object",
|
|
target_scope=PMFlowAITargetScope(scope_type="current_project", binding_strength="explicit"),
|
|
answer_shape="status_answer",
|
|
source_policy="internal",
|
|
),
|
|
)
|
|
|
|
assert hint == ""
|