from __future__ import annotations import pytest from pydantic import ValidationError from app.schemas.analysis_packets import ( AnalysisFocusObject, ExternalImpactAnalysisPacket, FollowupExplainAnalysisPacket, InternalExplainAnalysisPacket, ProjectFitAnalysisPacket, ) from app.services.analysis_packet_builder import build_analysis_packet def test_project_fit_analysis_packet_schema_contract() -> None: packet = ProjectFitAnalysisPacket( conclusion="放到采购经理每日实时库存上报服务里，它更像接在摘要卡片后面的解释层，不像主计算链。", focus_objects=[ AnalysisFocusObject( name="每日固定时段生成并推送结构化摘要卡片", kind="task", stance="better_fit", reason="这一步更像结果出来之后的解释和追问层。", ), AnalysisFocusObject( name="WMS库存快照自动拉取与清洗", kind="task", stance="not_fit", reason="这一步属于确定性取数主链，不适合直接塞进生成能力。", ), ], next_step="先确认它在当前项目里具体承接哪个节点。", fit_verdict="fit_later", current_subject="采购经理每日实时库存上报服务", candidate_position="每日固定时段生成并推送结构化摘要卡片", rejected_position="WMS库存快照自动拉取与清洗", preferred_layer="每日固定时段生成并推送结构化摘要卡片", rejected_layer="WMS库存快照自动拉取与清洗", reason_chain=[ "WMS库存快照自动拉取与清洗这一步属于确定性取数主链，结果要稳定、可复算。", "每日固定时段生成并推送结构化摘要卡片更接近结果出来之后的解释和追问层。", ], missing_mapping="还缺它在每日固定时段生成并推送结构化摘要卡片这一步到底承担什么角色的明确映射。", ) assert packet.packet_version == "v1" assert packet.packet_type == "project_fit_analysis" assert len(packet.focus_objects) == 2 assert packet.fit_verdict == "fit_later" assert packet.current_subject == "采购经理每日实时库存上报服务" assert packet.candidate_position == "每日固定时段生成并推送结构化摘要卡片" assert packet.rejected_position == "WMS库存快照自动拉取与清洗" assert len(packet.reason_chain) == 2 def test_analysis_packet_focus_objects_limit_is_two() -> None: with pytest.raises(ValidationError): ProjectFitAnalysisPacket( conclusion="结论", focus_objects=[ AnalysisFocusObject(name="A", kind="task", stance="better_fit", reason="A"), AnalysisFocusObject(name="B", kind="task", stance="not_fit", reason="B"), AnalysisFocusObject(name="C", kind="task", stance="relevant", reason="C"), ], fit_verdict="uncertain", missing_mapping="还缺映射。", ) def test_build_analysis_packet_for_project_fit_indirect_only_prefers_contrast_objects() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, project_operating_context={ "current_project_id": 123, "current_repo_scope": "pmflow-main", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, "external_evidence_required": True, "plan_handoff_mode": "offer_after_answer", }, fit_brief={ "project_summary": "采购经理每日实时库存上报服务", "repo_summary": "当前仓库负责统一项目上下文和 AI handoff。", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", "external_summary": "Responses API 适合多轮状态化交互。", "external_impact": "当前外部证据只支持背景相关性，暂不能直接映射到当前项目对象。", }, recommended_next_step="先确认它在当前项目里具体承接哪个节点。", project_snapshot={ "project_name": "采购经理每日实时库存上报服务", "task_titles": [ "WMS库存快照自动拉取与清洗", "TOP10商品缺货排序逻辑实现（基于当前库存与安全库存差值）", "每日固定时段生成并推送结构化摘要卡片", ], "signal_titles": [], "analysis_object_refs": { "fit_candidate_task_ref": {"kind": "task", "title": "每日固定时段生成并推送结构化摘要卡片"}, "fit_rejected_task_ref": {"kind": "task", "title": "WMS库存快照自动拉取与清洗"}, }, }, ) assert packet["packet_type"] == "project_fit_analysis" assert packet["fit_verdict"] == "fit_later" assert packet["current_subject"] == "采购经理每日实时库存上报服务" assert packet["candidate_position"] == "每日固定时段生成并推送结构化摘要卡片" assert packet["rejected_position"] == "WMS库存快照自动拉取与清洗" assert [item["stance"] for item in packet["focus_objects"]] == ["better_fit", "not_fit"] assert [item["name"] for item in packet["focus_objects"]] == [ "每日固定时段生成并推送结构化摘要卡片", "WMS库存快照自动拉取与清洗", ] assert packet["reason_chain"] == [ "WMS库存快照自动拉取与清洗这一步要稳，不适合一上来就塞新能力。", "每日固定时段生成并推送结构化摘要卡片这一步更适合先试，比直接改WMS库存快照自动拉取与清洗风险小。", ] assert packet["missing_mapping"] == "现在还说不准它会先帮到每日固定时段生成并推送结构化摘要卡片，还是会直接改到WMS库存快照自动拉取与清洗。" def test_build_analysis_packet_for_project_fit_prefers_structured_task_descriptions_over_generic_titles() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, project_operating_context={ "current_project_id": 123, }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, "external_evidence_required": True, "plan_handoff_mode": "offer_after_answer", }, fit_brief={ "project_summary": "采购经理每日实时库存上报服务", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", "external_summary": "Responses API 更适合放在结果出来之后的解释与追问层。", "external_impact": "当前只支持背景相关性，暂不能直接映射到当前项目对象。", }, recommended_next_step="先确认它是不是只用在结果出来之后的解释环节。", project_snapshot={ "project_name": "采购经理每日实时库存上报服务", "tasks": [ { "id": 901, "title": "任务A", "description": "每日凌晨生成结构化文本卡片并推送给采购经理，支持点击查看详情页。", }, { "id": 902, "title": "任务B", "description": "每小时从WMS拉取库存快照并清洗异常值后写入临时表。", }, ], "analysis_object_refs": { "fit_candidate_task_ref": {"kind": "task", "id": 901, "title": "任务A"}, "fit_rejected_task_ref": {"kind": "task", "id": 902, "title": "任务B"}, }, }, ) assert packet["packet_type"] == "project_fit_analysis" assert packet["candidate_position"] == "任务A" assert packet["rejected_position"] == "任务B" assert packet["reason_chain"] == [ "任务B现在做的是每小时从WMS拉取库存快照并清洗异常值后写入临时表，这一步要稳，不能随便把新能力塞进去。", "任务A现在做的是每日凌晨生成结构化文本卡片并推送给采购经理，支持点击查看详情页，先在这里试更稳，也更容易看出值不值。", ] assert packet["missing_mapping"] == "现在还说不准它会先帮到任务A，还是会直接改到任务B。" def test_build_analysis_packet_for_project_fit_fails_closed_without_explicit_object_refs() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, "external_evidence_required": True, "plan_handoff_mode": "offer_after_answer", }, fit_brief={ "project_summary": "采购经理每日实时库存上报服务", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", "external_summary": "Responses API 更适合放在结果出来之后的解释与追问层。", "external_impact": "当前只支持背景相关性，暂不能直接映射到当前项目对象。", }, recommended_next_step="先确认它是不是只用在结果出来之后的解释环节。", project_snapshot={ "project_name": "采购经理每日实时库存上报服务", "tasks": [ { "title": "任务A", "description": "根据前序结果生成给采购经理的日报文本，并保留查看入口。", }, { "title": "任务B", "description": "从WMS取库存样本写入中间表，补齐批次与仓位字段后供后续核算使用，并保留历史版本。", }, ], "signals": [ { "title": "信号一", "summary": "WMS每小时更新库存样本，是总水位核算的原始来源。", } ], "analysis_object_refs": { # no explicit refs on purpose }, }, ) assert packet["candidate_position"] is None assert packet["rejected_position"] is None assert packet["reason_chain"] == [ "现在项目里还没有足够具体的任务或信号源，硬把它挂到某个位置上，结论很容易说偏。" ] assert packet["missing_mapping"] == "至少还缺一个能对上的业务步骤或信号源。" def test_build_analysis_packet_for_project_fit_prefers_explicit_object_refs_over_ambiguous_task_names() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, "external_evidence_required": True, "plan_handoff_mode": "offer_after_answer", }, fit_brief={ "project_summary": "库存异常播报助手", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", "external_summary": "Responses API 更适合放在结果出来之后的解释与追问层。", "external_impact": "当前只支持背景相关性，暂不能直接映射到当前项目对象。", }, recommended_next_step="先确认它在项目里具体承接哪个节点。", project_snapshot={ "project_name": "库存异常播报助手", "tasks": [ { "id": 701, "title": "任务A", "description": "每 15 分钟从仓储中台拉取异常库存快照，清洗后写入临时表供后续核算使用。", }, { "id": 702, "title": "任务B", "description": "根据前序结果生成发给值班同学的异常播报文本，并支持继续追问异常原因。", }, ], "analysis_object_refs": { "fit_candidate_task_ref": {"kind": "task", "id": 702, "title": "任务B"}, "fit_rejected_task_ref": {"kind": "task", "id": 701, "title": "任务A"}, }, }, ) assert packet["candidate_position"] == "任务B" assert packet["rejected_position"] == "任务A" assert packet["reason_chain"] == [ "任务A现在做的是每 15 分钟从仓储中台拉取异常库存快照，清洗后写入临时表供后续核算使用，这一步要稳，不能随便把新能力塞进去。", "任务B现在做的是根据前序结果生成发给值班同学的异常播报文本，并支持继续追问异常原因，先在这里试更稳，也更容易看出值不值。", ] def test_build_analysis_packet_for_project_fit_fails_closed_when_tasks_and_signals_lack_explicit_refs() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, }, fit_brief={ "project_summary": "采购经理每日实时库存上报服务", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", "external_summary": "Responses API 更适合多轮、状态化的解释与追问。", "external_impact": "当前只支持背景相关性，暂不能直接映射到当前项目对象。", }, recommended_next_step="先确认它是不是只用在摘要卡片发出后的解释环节。", project_snapshot={ "project_name": "采购经理每日实时库存上报服务", "tasks": [ { "title": "每日固定时段生成并推送结构化摘要卡片", "description": "每日凌晨触发全量计算，生成含标题、总水位摘要、TOP10表格的结构化文本卡片，并在上午10点前推送。", }, { "title": "WMS库存快照自动拉取与清洗", "description": "从WMS系统每小时拉取库存快照，按可用/在途/冻结三态聚合，清洗异常值后写入中台临时表。", }, ], "signals": [ { "title": "交付形式 - 信息呈现方式", "summary": "以结构化文本卡片形式推送，含标题、总水位摘要、TOP10表格，并支持点击跳转至中台详情页。", }, { "title": "外部信号源 - WMS实时库存接口", "summary": "提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源。", }, ], }, ) assert packet["candidate_position"] is None assert packet["rejected_position"] is None assert packet["reason_chain"] == [ "现在项目里还没有足够具体的任务或信号源，硬把它挂到某个位置上，结论很容易说偏。" ] def test_build_analysis_packet_for_project_fit_sparse_project_avoids_fake_layers() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, "plan_handoff_mode": "offer_after_answer", }, fit_brief={ "project_summary": "自动开票系统", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", }, recommended_next_step="先把这个项目最关键的业务步骤补清楚，我再判断它更适合放哪里。", project_snapshot={ "project_name": "自动开票系统", "tasks": [], "signals": [], "analysis_object_refs": {}, }, ) assert packet["packet_type"] == "project_fit_analysis" assert packet["conclusion"] == "光看自动开票系统现在这些项目信息，我还不想直接说它该放哪一步。" assert packet["reason_chain"] == [ "现在项目里还没有足够具体的任务或信号源，硬把它挂到某个位置上，结论很容易说偏。" ] assert packet["missing_mapping"] == "至少还缺一个能对上的业务步骤或信号源。" assert packet["next_step"] == "先把这个项目最关键的业务步骤补清楚，我再判断它更适合放哪里。" assert "结果出来之后的解释层" not in packet["conclusion"] assert "当前主链" not in packet["conclusion"] assert packet["focus_objects"] == [] def test_build_analysis_packet_for_project_fit_sparse_project_uses_project_description_as_real_analysis() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, "plan_handoff_mode": "offer_after_answer", }, fit_brief={ "project_summary": "自动开票系统", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", }, recommended_next_step="先把这个项目最关键的业务步骤补清楚，我再判断它更适合放哪里。", project_snapshot={ "project_name": "自动开票系统", "project_description": "开发一套自动开票系统，实现开票接口到达后自动开票并发送发票图片的功能。", "tasks": [], "signals": [], "analysis_object_refs": {}, }, ) assert packet["packet_type"] == "project_fit_analysis" assert packet["reason_chain"] == [ "我现在只看得到这个项目的大致目标：开发一套自动开票系统，实现开票接口到达后自动开票并发送发票图片的功能。", "但还看不到这条链路里谁来触发、要接哪些外部输入、以及哪个结果算真的跑通。", ] assert packet["missing_mapping"] == "还缺能对上的实际流程、关键结果和外部输入。" assert packet["next_step"] == "至少先把实际流程拆出来：谁发起、系统接什么输入、最后产出什么结果。" assert packet["focus_objects"] == [] def test_build_analysis_packet_for_project_fit_engineering_tasks_avoids_false_business_placement() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, }, fit_brief={ "project_summary": "AI直播复盘", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", }, recommended_next_step="先定清它到底想改善哪个业务结果，再看该落到哪一步。", project_snapshot={ "project_name": "AI直播复盘", "tasks": [ { "title": "前端交互界面开发", "description": "基于React 18和Vite构建前端，开发视频上传组件及复盘数据可视化展示页面。", }, { "title": "容器化部署与ACK集群配置", "description": "编写Dockerfile，配置阿里云ACK集群，部署后端与AI服务，实现CI/CD自动化流程。", }, ], }, ) assert packet["conclusion"] == "光看AI直播复盘现在这些项目信息，我还不想直接说它该放哪一步。" assert packet["reason_chain"] == ["现在项目里还没有足够具体的任务或信号源，硬把它挂到某个位置上，结论很容易说偏。"] assert packet["missing_mapping"] == "至少还缺一个能对上的业务步骤或信号源。" assert packet["next_step"] == "先把这个项目最关键的业务步骤补清楚，我再判断它更适合放哪里。" def test_build_analysis_packet_for_project_fit_signal_placeholders_avoids_forced_contrast() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, }, fit_brief={ "project_summary": "健身食谱推荐系统", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", }, recommended_next_step="先定清这个项目最后要给谁看什么结果，再判断哪个信号只是输入、哪个位置更值得先试。", project_snapshot={ "project_name": "健身食谱推荐系统", "signals": [ { "title": "增加食谱卡路里计算信号源", "summary": "在健身食谱推荐系统中增加一个信号源，用于计算食物的卡路里。", }, { "title": "增加每日运动消耗卡路里信号源", "summary": "为健身食谱推荐系统增加一个信号源，用于计算每日运动消耗的卡路里。", }, ], }, ) assert packet["conclusion"] == "光看健身食谱推荐系统现在这些项目信息，我还不想直接说它该放哪一步。" assert packet["reason_chain"] == ["现在项目里还没有足够具体的任务或信号源，硬把它挂到某个位置上，结论很容易说偏。"] assert packet["missing_mapping"] == "至少还缺一个能对上的业务步骤或信号源。" assert packet["next_step"] == "先把这个项目最关键的业务步骤补清楚，我再判断它更适合放哪里。" def test_build_analysis_packet_for_project_fit_strips_embedded_context_from_task_detail() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, }, fit_brief={ "project_summary": "AI劝退助手", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", }, recommended_next_step="先定清它是给需求澄清与范围确认补说明，还是会直接改到核心方案落地，再决定要不要继续做。", project_snapshot={ "project_name": "AI劝退助手", "tasks": [ { "title": "需求澄清与范围确认", "description": "基于飞书信号补全需求边界、验收口径与优先级。上下文：请录入一个信号：AI劝退助手新增次日回访功能。", }, { "title": "核心方案落地", "description": "围绕需求信号推进可交付实现，拆分后端/前端/AI 子任务并排期。上下文：请录入一个信号：AI劝退助手新增次日回访功能。", }, ], }, ) assert all("上下文：" not in item for item in packet["reason_chain"]) def test_build_analysis_packet_for_external_impact_verifier_blocked() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, external_recommendation={ "retrieval_status": "verifier_blocked", "impact_status": "no_direct_impact", "external_summary": "", "external_impact": "", "affected_signal_sources": [], "external_action": "先修正证据质量，再决定是否继续外部推荐。", }, project_snapshot={ "project_name": "Alpha", "tasks": [ { "title": "每日固定时段生成并推送结构化摘要卡片", "description": "每日凌晨触发全量计算，生成含标题、总水位摘要、TOP10表格的结构化文本卡片，并在上午10点前推送。", } ], "signals": [ { "title": "外部信号源 - WMS实时库存接口", "summary": "提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源。", } ], }, ) assert packet["packet_type"] == "external_impact_analysis" assert packet["retrieval_status"] == "verifier_blocked" assert packet["blocked_reason_or_gap"] assert packet["focus_objects"] assert packet["current_subject"] == "Alpha" assert packet["reason_chain"] == [ "现在卡在外部信号源 - WMS实时库存接口这条线索上：它指向提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源，但还没通过校验。", "项目里现在更多还是实现层任务，我不想硬说它会直接改到哪一步，先把这条线索核实清楚更稳。", ] assert packet["blocked_reason_or_gap"] == "先把外部信号源 - WMS实时库存接口这条线索核实清楚，再确认它到底会影响这个项目里的哪个结果。" assert packet["next_step"] == "先把外部信号源 - WMS实时库存接口这条线索核实清楚，再确认它到底会影响这个项目里的哪个结果。" def test_build_analysis_packet_for_external_impact_verifier_blocked_with_engineering_tasks_stays_generic() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, external_recommendation={ "retrieval_status": "verifier_blocked", "impact_status": "no_direct_impact", "affected_signal_sources": ["外部信号源 - TikTok Shop 直播开放API（评论流"], }, project_snapshot={ "project_name": "AI直播复盘", "tasks": [ { "title": "前端交互界面开发", "description": "基于React 18和Vite构建前端，开发视频上传组件及复盘数据可视化展示页面。", }, { "title": "容器化部署与ACK集群配置", "description": "编写Dockerfile，配置阿里云ACK集群，部署后端与AI服务，实现CI/CD自动化流程。", }, ], "signals": [ { "title": "外部信号源 - TikTok Shop 直播开放API（评论流", "summary": "获取带毫秒级时间戳的原始评论序列，用于与录屏帧对齐。", } ], }, ) assert packet["conclusion"] == "这条外部变化现在还不能直接挂到AI直播复盘的项目结论上。" assert packet["reason_chain"] == [ "现在卡在外部信号源 - TikTok Shop 直播开放API（评论流这条线索上：它指向获取带毫秒级时间戳的原始评论序列，用于与录屏帧对齐，但还没通过校验。", "项目里现在更多还是实现层任务，我不想硬说它会直接改到哪一步，先把这条线索核实清楚更稳。", ] assert packet["next_step"] == "先把外部信号源 - TikTok Shop 直播开放API（评论流这条线索核实清楚，再确认它到底会影响这个项目里的哪个结果。" def test_build_analysis_packet_for_project_fit_direct_match_prefers_dependent_task_over_anchor_signal() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, "external_evidence_required": True, "plan_handoff_mode": "offer_after_answer", }, fit_brief={ "project_summary": "采购经理每日实时库存上报服务", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "direct_match", "affected_signal_sources": ["外部信号源 - WMS实时库存接口"], "external_action": "直接看它会影响哪条下游任务，再决定要不要进 plan。", }, project_snapshot={ "project_name": "采购经理每日实时库存上报服务", "tasks": [ { "title": "WMS库存快照自动拉取与清洗", "description": "从WMS系统每小时拉取库存快照，按可用/在途/冻结三态聚合，清洗异常值后写入中台临时表。", }, { "title": "每日固定时段生成并推送结构化摘要卡片", "description": "每日凌晨触发全量计算，生成含标题、总水位摘要、TOP10表格的结构化文本卡片，并在上午10点前推送。", }, ], "signals": [ { "title": "外部信号源 - WMS实时库存接口", "summary": "提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源。", } ], }, ) assert packet["packet_type"] == "project_fit_analysis" assert packet["fit_verdict"] == "fit_now" assert packet["candidate_position"] == "外部信号源 - WMS实时库存接口" assert packet["preferred_layer"] == "外部信号源 - WMS实时库存接口" assert [item["name"] for item in packet["focus_objects"]] == ["外部信号源 - WMS实时库存接口"] assert [item["stance"] for item in packet["focus_objects"]] == ["anchor"] assert packet["reason_chain"] == [ "外部信号源 - WMS实时库存接口是当前最直接会接住这条变化的项目对象。", "对应的正式锚点是外部信号源 - WMS实时库存接口，它指向提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源，已经和外部证据建立了直接映射。", ] def test_build_analysis_packet_for_external_impact_direct_match_prefers_business_task_before_anchor_signal() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "direct_match", "affected_signal_sources": ["外部信号源 - WMS实时库存接口"], "external_action": "直接看它会影响哪条下游任务，再决定要不要进 plan。", }, project_snapshot={ "project_name": "采购经理每日实时库存上报服务", "tasks": [ { "title": "WMS库存快照自动拉取与清洗", "description": "从WMS系统每小时拉取库存快照，按可用/在途/冻结三态聚合，清洗异常值后写入中台临时表。", }, { "title": "每日固定时段生成并推送结构化摘要卡片", "description": "每日凌晨触发全量计算，生成含标题、总水位摘要、TOP10表格的结构化文本卡片，并在上午10点前推送。", }, ], "signals": [ { "title": "外部信号源 - WMS实时库存接口", "summary": "提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源。", } ], }, ) assert packet["packet_type"] == "external_impact_analysis" assert packet["retrieval_status"] == "evidence_found" assert packet["impact_status"] == "direct_match" assert [item["name"] for item in packet["focus_objects"]] == ["外部信号源 - WMS实时库存接口"] assert [item["stance"] for item in packet["focus_objects"]] == ["anchor"] def test_build_analysis_packet_for_external_direct_match_engineering_tasks_stays_generic() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "direct_match", "affected_signal_sources": ["外部信号源 - TikTok Shop 直播开放API（评论流"], }, project_snapshot={ "project_name": "AI直播复盘", "tasks": [ { "title": "前端交互界面开发", "description": "基于React 18和Vite构建前端，开发视频上传组件及复盘数据可视化展示页面。", }, { "title": "容器化部署与ACK集群配置", "description": "编写Dockerfile，配置阿里云ACK集群，部署后端与AI服务，实现CI/CD自动化流程。", }, ], "signals": [ { "title": "外部信号源 - TikTok Shop 直播开放API（评论流", "summary": "获取带毫秒级时间戳的原始评论序列，用于与录屏帧对齐。", } ], }, ) assert packet["conclusion"] == "这条外部变化已经和AI直播复盘对上了，但现在还不适合硬挂到某个实现任务上。" assert packet["reason_chain"] == [ "能对上的外部来源是外部信号源 - TikTok Shop 直播开放API（评论流，它指向获取带毫秒级时间戳的原始评论序列，用于与录屏帧对齐，所以这次不是泛泛相关。", "项目里现在更多还是实现层任务，先别急着说它会直接改到哪一步，先看它到底影响哪个业务结果。", ] assert packet["next_step"] == "先确认这条变化会影响哪个业务结果，再决定要不要拆到具体任务。" def test_build_analysis_packet_for_external_direct_match_prefers_explicit_affected_signal_source_over_snapshot_anchor() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "direct_match", "affected_signal_sources": ["外部信号源 - WMS库存快照API"], }, recommended_next_step="直接看这条变化会不会先改到下游动作。", project_snapshot={ "project_name": "门店库存异常日报助手", "tasks": [ { "id": 180, "title": "三类异常的基础识别逻辑（缺货", "description": "三类异常的基础识别逻辑（缺货/负库存/90天滞销）。", } ], "signals": [ { "id": 411, "title": "配置灵活性 - 异常规则可低代码配置", "summary": "支持在后台界面调整三类异常的判定阈值。", }, { "id": 412, "title": "外部信号源 - WMS库存快照API", "summary": "提供T-1全量门店SKU级库存水位与状态，是异常识别的唯一原始数据源。", }, ], "analysis_object_refs": { "primary_anchor_signal_ref": {"kind": "signal", "id": 411, "title": "配置灵活性 - 异常规则可低代码配置"}, "primary_dependent_task_ref": {"kind": "task", "id": 180, "title": "三类异常的基础识别逻辑（缺货"}, }, }, ) assert [item["name"] for item in packet["focus_objects"]] == [ "三类异常的基础识别逻辑（缺货", "外部信号源 - WMS库存快照API", ] assert "配置灵活性 - 异常规则可低代码配置" not in "".join(packet["reason_chain"]) assert packet["next_step"] == "下一步就看这条变化会不会直接改到三类异常的基础识别逻辑（缺货后面的动作或规则。" def test_build_analysis_packet_for_project_fit_signal_only_project_uses_real_signal_objects() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "external", "effective_source_policy": "external", }, execution_recipe={ "primary_lane": "external", "fit_brief_required": True, "external_evidence_required": True, "plan_handoff_mode": "offer_after_answer", }, fit_brief={ "project_summary": "直播复盘辅助工具（速记版）", }, external_recommendation={ "retrieval_status": "evidence_found", "impact_status": "indirect_only", }, recommended_next_step="先确认它在当前项目里具体承接哪个节点。", project_snapshot={ "project_name": "直播复盘辅助工具（速记版）", "signals": [ { "id": 427, "title": "输入适配 - 双源时间对齐能力", "summary": "支持手动校准录屏与评论的时间偏移，并在此基础上完成全链路切片分析。", }, { "id": 428, "title": "输出规范 - 三栏速记清单格式", "summary": "输出纯文本文件，含高光、掉点、执行偏差三栏。", }, { "id": 430, "title": "外部信号源 - TikTok Shop 直播开放API（评论流", "summary": "获取带毫秒级时间戳的原始评论序列，用于与录屏帧对齐。", }, ], "analysis_object_refs": { "fit_candidate_signal_ref": {"kind": "signal", "id": 428, "title": "输出规范 - 三栏速记清单格式"}, "fit_rejected_signal_ref": {"kind": "signal", "id": 427, "title": "输入适配 - 双源时间对齐能力"}, "primary_anchor_signal_ref": {"kind": "signal", "id": 430, "title": "外部信号源 - TikTok Shop 直播开放API（评论流"}, }, }, ) assert packet["candidate_position"] == "输出规范 - 三栏速记清单格式" assert packet["rejected_position"] == "输入适配 - 双源时间对齐能力" assert packet["reason_chain"] == [ "输入适配 - 双源时间对齐能力现在代表的是支持手动校准录屏与评论的时间偏移，并在此基础上完成全链路切片分析，这类前面必须先跑通的条件，不适合直接拿来试新能力。", "输出规范 - 三栏速记清单格式现在代表的是输出纯文本文件，含高光、掉点、执行偏差三栏，先从这里试更稳，也更容易看出它有没有价值。", ] assert packet["missing_mapping"] == "现在还说不准它会先帮到输出规范 - 三栏速记清单格式，还是会直接改到输入适配 - 双源时间对齐能力。" assert "结果出来之后的解释层" not in packet["conclusion"] assert packet["conclusion"] == "放到直播复盘辅助工具（速记版）里，我更倾向先让它去帮输出规范 - 三栏速记清单格式，不太会先拿它改输入适配 - 双源时间对齐能力。" def test_build_analysis_packet_for_internal_explain_uses_internal_contract_and_target() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "internal", "effective_source_policy": "internal", "reply_contract": {"mode": "explicit_source", "head": "内部知识库查询："}, }, query_type="status", answer="当前活动引擎是 Qwen 低阶引擎。", evidence_pack={ "summary": { "conclusion": "当前活动引擎是 Qwen 低阶引擎。", "uncertainties": [], } }, internal_context_packet={ "operation_contract": { "query_type": "status", "answer_contract": "status_summary", }, "target_resolution": { "resolution_mode": "resolved", "primary_target_label": "模型路由配置", }, "query_plan": { "answer_contract": "status_summary", }, }, ) assert packet["packet_type"] == "internal_explain_analysis" assert packet["answer_contract"] == "status_summary" assert packet["resolved_target"] == "模型路由配置" assert packet["current_subject"] == "模型路由配置" assert packet["reason_chain"] == [ "这次就按模型路由配置的当前情况回答，先不展开方案。", ] def test_build_analysis_packet_for_internal_explain_can_resolve_target_from_project_snapshot_when_context_missing() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "internal", "effective_source_policy": "internal", }, query_type="status", answer="当前已经能检查团队成员负载并推荐可行 assignee。", evidence_pack={ "summary": { "conclusion": "当前已经能检查团队成员负载并推荐可行 assignee。", "uncertainties": [], } }, project_snapshot={ "project_name": "确认立项", "signals": [ { "title": "MVP 功能 - 检查团队成员负载并推荐可行assignee", "summary": "检查团队成员负载并推荐可行assignee", }, { "title": "技术架构 - 基于输入JSON结构做字段完整性校验", "summary": "基于输入JSON结构做字段完整性校验", }, ], }, ) assert packet["resolved_target"] is None assert packet["focus_objects"] == [] assert packet["reason_chain"] == ["这次先直接回答当前情况，不往方案上展开。"] def test_build_analysis_packet_for_internal_explain_ignores_none_conclusion_when_matching_snapshot_target() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "internal", "effective_source_policy": "internal", }, query_type="status", answer="当前已经能识别产能异常并生成提醒。", evidence_pack={ "summary": { "conclusion": None, "uncertainties": [], } }, project_snapshot={ "project_name": "测试产能警告", "signals": [ { "title": "MVP 功能 - 识别产能异常并生成提醒", "summary": "识别产能异常并生成提醒", }, ], }, ) assert packet["packet_type"] == "internal_explain_analysis" assert packet["resolved_target"] is None assert packet["current_subject"] is None assert packet["reason_chain"] == ["这次先直接回答当前情况，不往方案上展开。"] def test_build_analysis_packet_for_followup_explain_uses_anchor_and_clarification_state() -> None: packet = build_analysis_packet( route_decision={ "route_mode": "uncertain", "effective_source_policy": "uncertain", "reply_contract": {"mode": "clarify"}, }, answer="你是想问外部公开事实，还是系统内部配置/项目对象？", clarification_code="query_route_clarify", anchor_kind="reply_thread", meta_followup_kind="route_selection", evidence_pack={ "summary": { "conclusion": "需要先确认查询来源。", "uncertainties": ["内部还是外部"], } }, ) assert packet["packet_type"] == "followup_explain_analysis" assert packet["anchor_kind"] == "reply_thread" assert packet["continuation_kind"] == "route_selection" assert packet["clarification_state"] == "query_route_clarify"