479 lines
20 KiB
Python
479 lines
20 KiB
Python
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from app.schemas.pmflow_ai_turn_runtime import (
|
||
PMFlowAIAnswerState,
|
||
PMFlowAIEvidenceFinding,
|
||
PMFlowAIGoalFrame,
|
||
PMFlowAIRuntimeAction,
|
||
PMFlowAITargetScope,
|
||
PMFlowAITurnActionState,
|
||
PMFlowAITurnEvidenceState,
|
||
PMFlowAITurnObjectRef,
|
||
PMFlowAITurnObjectState,
|
||
PMFlowAITurnState,
|
||
)
|
||
from app.schemas.rag import RAGAskRequest
|
||
from app.services.pmflow_ai_evidence_absorber import PMFlowAIEvidenceAbsorber
|
||
from app.services import pmflow_ai_evidence_tool_registry as tool_registry_module
|
||
from app.services.pmflow_ai_evidence_tool_registry import PMFlowAIEvidenceToolRegistry
|
||
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,
|
||
_project_operating_answer_verification_packet,
|
||
_render_project_operating_answer_from_turn_state,
|
||
)
|
||
|
||
|
||
class _FakeProjectionPayload:
|
||
def __init__(self, payload: dict):
|
||
self.payload = payload
|
||
|
||
def model_dump(self, *, mode: str = "json") -> dict:
|
||
_ = mode
|
||
return self.payload
|
||
|
||
|
||
class _FakeProjectionService:
|
||
def __init__(self, db, *, tenant_id: int):
|
||
_ = db, tenant_id
|
||
|
||
async def get_truth_state(self, *, project_id: int, current_user_id: int):
|
||
_ = current_user_id
|
||
return _FakeProjectionPayload(
|
||
{
|
||
"loopStage": "command_idle",
|
||
"project": {
|
||
"id": str(project_id),
|
||
"name": "童析",
|
||
"stage": "development",
|
||
"status": "active",
|
||
},
|
||
"projectMemory": {
|
||
"truthNodeFacts": [
|
||
{"summary": "MVP聚焦单次视频到单次报告闭环"},
|
||
],
|
||
},
|
||
"changeSets": [
|
||
{"id": "cs-1", "title": "补齐加密方案评审", "status": "pending_review"},
|
||
],
|
||
"changeItems": [
|
||
{"id": "ci-1", "title": "传输加密验收标准", "riskLevel": "high"},
|
||
],
|
||
"approvals": [
|
||
{"id": "approval-1", "state": "requested"},
|
||
],
|
||
"reviewerRouting": {
|
||
"summary": {
|
||
"totalPendingReview": 1,
|
||
"highRiskCount": 1,
|
||
"missingReviewerCount": 0,
|
||
},
|
||
"items": [
|
||
{
|
||
"reviewReason": "高风险儿童数据传输方案需要技术审查",
|
||
"suggestedReviewer": {
|
||
"displayName": "安全负责人",
|
||
"reason": "需要确认数据保护方案",
|
||
},
|
||
}
|
||
],
|
||
},
|
||
"feed": [
|
||
{
|
||
"title": "加密方案进入评审",
|
||
"summary": "等待测试证据和技术审查",
|
||
"kind": "review",
|
||
}
|
||
],
|
||
"risks": [
|
||
{"title": "儿童数据合规风险", "summary": "需要补齐加密和匿名化证据"},
|
||
],
|
||
}
|
||
)
|
||
|
||
async def get_execution_summary(self, *, project_id: int, current_user_id: int):
|
||
_ = project_id, current_user_id
|
||
return _FakeProjectionPayload(
|
||
{
|
||
"execution": {
|
||
"taskId": "194",
|
||
"status": "pr_open",
|
||
"prompt": "",
|
||
},
|
||
"acceptance": {
|
||
"status": "waiting_test_evidence",
|
||
"requiredAction": "run_required_tests",
|
||
"blockingReasons": ["缺少 test_passed 或 checks_passed 证据。"],
|
||
},
|
||
"releaseGate": {
|
||
"status": "needs_confirmation",
|
||
"title": "需要人工确认",
|
||
"summary": "已有执行回写，但缺少通过测试证据。",
|
||
"blockingReasons": ["缺少 PR 证据。"],
|
||
"nextActions": ["补充测试回写。"],
|
||
},
|
||
"qualityContract": {
|
||
"summary": {"releaseBlockingMissing": 4},
|
||
"blockingReasons": ["代码仓库联调与 PR 证据"],
|
||
},
|
||
"releaseSignals": {
|
||
"feedback": [{"title": "家长样本反馈"}],
|
||
"recommendations": [{"title": "优先补齐安全声明"}],
|
||
},
|
||
"recentProgressEvents": [
|
||
{"title": "PR 已打开", "status": "pr_open"},
|
||
{"title": "测试证据待回写", "status": "waiting"},
|
||
],
|
||
"recentDeliveries": [
|
||
{"title": "加密方案评审包", "status": "submitted"},
|
||
],
|
||
"deliveryArtifacts": {
|
||
"test_outputs": [
|
||
{"name": "pytest", "status": "passed"},
|
||
],
|
||
},
|
||
"artifactReview": {
|
||
"items": [
|
||
{"title": "PR 证据", "status": "needs_review"},
|
||
],
|
||
"blockingReasons": ["缺少截图证据"],
|
||
},
|
||
"activeDevTasks": [
|
||
{
|
||
"id": "194",
|
||
"title": "完成儿童数据匿名化与传输加密方案评审",
|
||
"status": "in_progress",
|
||
},
|
||
{
|
||
"id": "188",
|
||
"title": "设计视频上传-分析-报告三步信息架构与低保真原型",
|
||
"status": "todo",
|
||
},
|
||
],
|
||
}
|
||
)
|
||
|
||
|
||
def _project_execution_state() -> PMFlowAITurnState:
|
||
frame = PMFlowAIGoalFrame(
|
||
frame_id="project-execution",
|
||
operation_kind="inspect_object",
|
||
target_scope=PMFlowAITargetScope(scope_type="current_project", binding_strength="explicit"),
|
||
answer_shape="execution_guidance",
|
||
target_object_requirement="single",
|
||
)
|
||
return PMFlowAITurnState(
|
||
selected_frame=frame,
|
||
candidate_frames=[frame],
|
||
answer_state=PMFlowAIAnswerState(answer_shape="execution_guidance"),
|
||
object_state=PMFlowAITurnObjectState(
|
||
primary_object=PMFlowAITurnObjectRef(
|
||
object_type="project",
|
||
object_ref="project:141",
|
||
title="童析",
|
||
binding_strength="strong",
|
||
)
|
||
),
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_packet_k_project_status_tool_preserves_execution_projection(monkeypatch) -> None:
|
||
monkeypatch.setattr(tool_registry_module, "PMFlowOSProjectionService", _FakeProjectionService)
|
||
state = _project_execution_state()
|
||
|
||
payload = await PMFlowAIEvidenceToolRegistry(object(), tenant_id=12).run(
|
||
action_type="inspect_project_status",
|
||
request=RAGAskRequest(
|
||
question="现在我的项目怎么推进好呢",
|
||
surface_context={
|
||
"space_runtime_context": {
|
||
"workbench": {
|
||
"projects": [
|
||
{"id": 141, "object_ref": "project:141", "title": "童析", "status": "active"},
|
||
]
|
||
}
|
||
}
|
||
},
|
||
),
|
||
current_user=object(),
|
||
turn_state=state,
|
||
)
|
||
|
||
result = payload["runtime_evidence_tool_result"]
|
||
facts = "\n".join(item["fact"] for item in result["findings"])
|
||
packet = result["structured_evidence"]["project_operating_evidence_packet"]
|
||
|
||
assert "项目概况：童析，阶段：development" in facts
|
||
assert "当前执行任务：完成儿童数据匿名化与传输加密方案评审" in facts
|
||
assert "发布/验收门禁：门禁状态：needs_confirmation" in facts
|
||
assert "阻塞项：缺少 test_passed 或 checks_passed 证据。" in facts
|
||
assert "候选任务：当前有 2 个执行/待执行任务" in facts
|
||
assert "项目状态 active" not in facts
|
||
assert packet["packet_version"] == "project_operating_evidence.v1"
|
||
assert packet["project_ref"] == "project:141"
|
||
assert packet["command_state"]["coverage_status"] == "covered"
|
||
assert packet["execution_state"]["coverage_status"] == "covered"
|
||
assert packet["quality_test_state"]["coverage_status"] == "covered"
|
||
assert packet["release_growth_state"]["coverage_status"] == "covered"
|
||
assert packet["team_state"]["coverage_status"] == "covered"
|
||
assert packet["recent_timeline_state"]["coverage_status"] == "covered"
|
||
assert packet["project_memory_state"]["coverage_status"] == "covered"
|
||
assert packet["risk_blocker_state"]["facts"][0] == "缺少 test_passed 或 checks_passed 证据。"
|
||
assert "team_state" not in packet["coverage"]["missing_facets"]
|
||
assert any("待审路由" in fact for fact in packet["team_state"]["facts"])
|
||
assert any("进展事件" in fact for fact in packet["execution_state"]["facts"])
|
||
assert any("反馈输入" in fact for fact in packet["release_growth_state"]["facts"])
|
||
assert packet["coverage"]["coverage_score"] > 0.5
|
||
assert payload["runtime_observation"]["items"][0]["business_values"]["coverage_score"] == packet["coverage"]["coverage_score"]
|
||
assert payload["runtime_observation"]["can_answer_operations"] == ["inspect_object"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_packet_q_absorber_consumes_project_operating_evidence_packet(monkeypatch) -> None:
|
||
monkeypatch.setattr(tool_registry_module, "PMFlowOSProjectionService", _FakeProjectionService)
|
||
state = _project_execution_state()
|
||
payload = await PMFlowAIEvidenceToolRegistry(object(), tenant_id=12).run(
|
||
action_type="inspect_project_status",
|
||
request=RAGAskRequest(
|
||
question="现在我的项目怎么推进好呢",
|
||
surface_context={
|
||
"space_runtime_context": {
|
||
"workbench": {
|
||
"projects": [
|
||
{"id": 141, "object_ref": "project:141", "title": "童析", "status": "active"},
|
||
]
|
||
}
|
||
}
|
||
},
|
||
),
|
||
current_user=object(),
|
||
turn_state=state,
|
||
)
|
||
|
||
absorption = PMFlowAIEvidenceAbsorber().absorb(
|
||
turn_state=state,
|
||
action=PMFlowAIRuntimeAction(
|
||
action_id="inspect_project_status",
|
||
action_type="inspect_project_status",
|
||
tool_name="project_status_evidence_tool",
|
||
),
|
||
payload=payload,
|
||
)
|
||
absorbed_facts = "\n".join(item.fact for item in absorption.findings)
|
||
|
||
assert "童析 / 测试与质量" in absorbed_facts
|
||
assert "童析 / 发布与增长" in absorbed_facts
|
||
assert "童析 / 团队层" in absorbed_facts
|
||
assert "team_state_missing" not in absorption.remaining_gaps
|
||
assert absorption.structured_evidence["project_operating_evidence_packet"]["project_ref"] == "project:141"
|
||
|
||
|
||
def test_packet_l_coverage_gate_forces_project_execution_probe_before_answer() -> None:
|
||
state = _project_execution_state()
|
||
state.sufficiency_state.can_answer_now = True
|
||
state.action_state = PMFlowAITurnActionState(
|
||
candidate_actions=[
|
||
PMFlowAIRuntimeAction(
|
||
action_id="synthesize_answer",
|
||
action_type="synthesize_answer",
|
||
tool_name="turn_state_answer_synthesizer",
|
||
)
|
||
]
|
||
)
|
||
|
||
service = PMFlowAITurnRuntimeService(object(), tenant_id=12)
|
||
decision = service.assess_sufficiency(state)
|
||
action = service.select_next_minimal_action(state)
|
||
|
||
assert decision.needs_action is True
|
||
assert decision.next_action_type == "inspect_project_status"
|
||
assert action is not None
|
||
assert action.action_type == "inspect_project_status"
|
||
|
||
|
||
def test_packet_s_coverage_gate_ignores_legacy_text_findings_without_profile() -> None:
|
||
state = _project_execution_state()
|
||
state.sufficiency_state.can_answer_now = True
|
||
state.evidence_state = PMFlowAITurnEvidenceState(
|
||
known_facts=[
|
||
PMFlowAIEvidenceFinding(
|
||
fact="项目概况：童析，阶段：development，状态：active。",
|
||
source="PMFlow 项目执行态证据",
|
||
),
|
||
PMFlowAIEvidenceFinding(
|
||
fact="当前执行任务：完成儿童数据匿名化与传输加密方案评审，任务状态：in_progress，执行状态：pr_open。",
|
||
source="PMFlow 项目执行态证据",
|
||
),
|
||
PMFlowAIEvidenceFinding(
|
||
fact="发布/验收门禁：门禁状态：needs_confirmation，需要人工确认。",
|
||
source="PMFlow 项目执行态证据",
|
||
),
|
||
PMFlowAIEvidenceFinding(
|
||
fact="候选任务：当前有 2 个执行/待执行任务。",
|
||
source="PMFlow 项目执行态证据",
|
||
),
|
||
]
|
||
)
|
||
|
||
decision = PMFlowAITurnRuntimeService(object(), tenant_id=12).assess_sufficiency(state)
|
||
|
||
assert decision.needs_action is True
|
||
assert decision.next_action_type == "inspect_project_status"
|
||
|
||
|
||
def test_packet_s_coverage_gate_allows_answer_from_structured_profile() -> None:
|
||
state = _project_execution_state()
|
||
state.sufficiency_state.can_answer_now = True
|
||
state.evidence_state = PMFlowAITurnEvidenceState(
|
||
structured_evidence={
|
||
"project_operating_evidence_packets": [_covered_project_operating_packet()],
|
||
"project_operating_evidence_packet": _covered_project_operating_packet(),
|
||
}
|
||
)
|
||
|
||
decision = PMFlowAITurnRuntimeService(object(), tenant_id=12).assess_sufficiency(state)
|
||
|
||
assert decision.can_answer_now is True
|
||
assert decision.needs_action is False
|
||
|
||
|
||
def test_packet_t_project_operating_verification_bounds_final_answer_surface() -> None:
|
||
packet = _verification_project_operating_packet()
|
||
state = _project_execution_state()
|
||
state.answer_state = PMFlowAIAnswerState(answer_shape="execution_guidance")
|
||
state.object_state = PMFlowAITurnObjectState(
|
||
primary_object=PMFlowAITurnObjectRef(
|
||
object_type="project",
|
||
object_ref="project:141",
|
||
title="童析",
|
||
binding_strength="explicit",
|
||
)
|
||
)
|
||
state.evidence_state = PMFlowAITurnEvidenceState(
|
||
structured_evidence={
|
||
"project_operating_evidence_packets": [packet],
|
||
"project_operating_evidence_packet": packet,
|
||
}
|
||
)
|
||
|
||
verification = _project_operating_answer_verification_packet(state)
|
||
|
||
assert verification["packet_version"] == "project_operating_answer_verification.v1"
|
||
assert verification["can_claim_full_execution_guidance"] is False
|
||
assert verification["missing_or_partial_facets"][0]["display_label"] == "测试质量"
|
||
assert verification["safe_next_actions"][0]["display_label"] == "整理加密评审材料"
|
||
assert verification["blocked_actions"][0]["display_label"] == "直接发布"
|
||
|
||
customer_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,
|
||
)
|
||
answer = _apply_project_operating_answer_verification_to_answer(
|
||
answer="当前可以继续推进，但要先补齐证据。",
|
||
verification=verification,
|
||
)
|
||
|
||
assert any("测试质量" in item for item in customer_packet["limitations"])
|
||
assert customer_packet["suggested_actions"][0]["title"] == "整理加密评审材料"
|
||
assert canvas["next_actions"][0]["title"] == "整理加密评审材料"
|
||
assert not any(action.get("title") == "直接发布" for action in canvas["next_actions"])
|
||
assert "可推进：整理加密评审材料" in answer
|
||
assert "不能直接做：直接发布" in answer
|
||
assert "需要注意：测试质量" in answer
|
||
|
||
|
||
def test_packet_ab_project_operating_packet_renders_concrete_project_answer() -> None:
|
||
packet = _verification_project_operating_packet()
|
||
state = _project_execution_state()
|
||
state.answer_state = PMFlowAIAnswerState(answer_shape="execution_guidance")
|
||
state.evidence_state = PMFlowAITurnEvidenceState(
|
||
structured_evidence={
|
||
"project_operating_evidence_packets": [packet],
|
||
"project_operating_evidence_packet": packet,
|
||
}
|
||
)
|
||
|
||
answer = _render_project_operating_answer_from_turn_state(
|
||
question="现在这个项目怎么样了，详细跟我说下",
|
||
turn_state=state,
|
||
)
|
||
|
||
assert "童析" in answer
|
||
assert "开发执行：PR 已打开" in answer
|
||
assert "风险阻塞：发布前需解除加密评审阻塞" in answer
|
||
assert "整理加密评审材料" in answer
|
||
assert "直接发布" in answer
|
||
assert "无法提供" not in answer
|
||
assert "请补充" not in answer
|
||
|
||
|
||
def _covered_project_operating_packet() -> dict:
|
||
return {
|
||
"packet_version": "project_operating_evidence.v1",
|
||
"project_ref": "project:141",
|
||
"project_title": "童析",
|
||
"command_state": {"coverage_status": "covered", "summary": "阶段：development"},
|
||
"execution_state": {"coverage_status": "covered", "summary": "执行状态：pr_open"},
|
||
"workbench_state": {"coverage_status": "covered", "summary": "执行/待执行任务 2 个"},
|
||
"risk_blocker_state": {"coverage_status": "covered", "summary": "缺少测试证据"},
|
||
"decision_review_state": {"coverage_status": "missing", "summary": ""},
|
||
"release_growth_state": {"coverage_status": "missing", "summary": ""},
|
||
}
|
||
|
||
|
||
def _verification_project_operating_packet() -> dict:
|
||
def facet(key: str, label: str, status: str, summary: str = "", gaps: list[str] | None = None) -> dict:
|
||
return {
|
||
"facet_key": key,
|
||
"display_label": label,
|
||
"coverage_status": status,
|
||
"summary": summary,
|
||
"facts": [summary] if summary else [],
|
||
"remaining_gaps": gaps or [],
|
||
}
|
||
|
||
return {
|
||
"packet_version": "project_operating_evidence.v1",
|
||
"project_id": 141,
|
||
"project_ref": "project:141",
|
||
"project_title": "童析",
|
||
"project_status": "active",
|
||
"command_state": facet("command_state", "指挥台", "covered", "阶段：development"),
|
||
"workbench_state": facet("workbench_state", "项目工作台", "covered", "有 2 个下一步"),
|
||
"decision_review_state": facet("decision_review_state", "决策与待确认", "covered", "有高风险评审"),
|
||
"execution_state": facet("execution_state", "开发执行", "covered", "PR 已打开"),
|
||
"quality_test_state": facet("quality_test_state", "测试质量", "partial", "缺少测试运行记录", ["缺少测试运行记录"]),
|
||
"release_growth_state": facet("release_growth_state", "发布增长", "missing", "", ["缺少发布门禁结果"]),
|
||
"team_state": facet("team_state", "团队层", "covered", "建议安全负责人评审"),
|
||
"project_memory_state": facet("project_memory_state", "项目记忆", "covered", "已记录 MVP 边界"),
|
||
"risk_blocker_state": facet("risk_blocker_state", "风险阻塞", "covered", "发布前需解除加密评审阻塞"),
|
||
"recent_timeline_state": facet("recent_timeline_state", "最近动态", "covered", "最近进入评审"),
|
||
"safe_next_actions": [
|
||
{
|
||
"action_key": "prepare_review",
|
||
"display_label": "整理加密评审材料",
|
||
"safety": "prepare_only",
|
||
"reason": "来自项目执行态的准备动作，执行前仍需确认。",
|
||
"target_ref": "project:141",
|
||
}
|
||
],
|
||
"blocked_actions": [
|
||
{
|
||
"action_key": "release_now",
|
||
"display_label": "直接发布",
|
||
"safety": "blocked",
|
||
"reason": "当前证据显示发布门禁未解除。",
|
||
"target_ref": "project:141",
|
||
}
|
||
],
|
||
}
|