# ============================================================ # AG-07 主动引导黄金测试 # ============================================================ from __future__ import annotations from pathlib import Path from server.agent_core.intent import parse_fast from server.aps_domain.guidance import build_guidance, scheduling_data_guide from server.aps_domain.workflow import handle_intent from server.contracts import IntentResult from server.state.seed import empty_world, seed_world class _MemStore: def __init__(self, data): self.data = data def next_id(self, kind: str) -> int: key = f"_c_{kind}" self.data[key] = self.data.get(key, 9000) + 1 return self.data[key] def save(self): pass def test_guidance_empty_schedule(): world = seed_world() world["scheduleVersions"] = [] g = build_guidance(world) cmds = {s["command"] for s in g["suggestions"]} assert any(s["id"] == "no_version" for s in g["signals"]) assert "生成一版交期优先试排方案" in cmds def test_guidance_knowledge_miss_context(): world = seed_world() g = build_guidance(world, context="knowledge_miss") cmds = {s["command"] for s in g["suggestions"]} assert "知识库" in cmds assert any(s["id"] == "knowledge_miss" for s in g["signals"]) def test_guidance_high_conflict(): world = seed_world() world["scheduleVersions"] = [{ "id": 1, "versionNo": "V1", "status": "DRAFT", "conflictCount": 8, "poCount": 1, "woCount": 1, "totalTardiness": 0, "avgUtilization": 0.5, "createdAt": "2026-07-22 10:00", "engineType": "RULE", "strategy": "COMPREHENSIVE", }] g = build_guidance(world, context="high_conflict") cmds = {s["command"] for s in g["suggestions"]} assert "查看冲突" in cmds assert "多策略方案对比" in cmds def test_intent_guidance_next(): hit = parse_fast("下一步建议", seed_world()) assert hit and hit.intent == "guidance.next" def test_workflow_guidance_and_kb_miss(tmp_path: Path, monkeypatch): """知识未命中不编造（隔离知识库：只含种子资产，规避 server/data/knowledge.json 库容漂移）。""" import asyncio import server.knowledge.assets as _kb_assets monkeypatch.setenv("APS_KNOWLEDGE_PATH", str(tmp_path / "knowledge.json")) monkeypatch.setattr(_kb_assets, "_stores", {}) # 清全局缓存，强制按新路径重载（种子播种） store = _MemStore(seed_world()) store.data["scheduleVersions"] = [] async def _run(): reply = await handle_intent( store, "test", IntentResult(intent="guidance.next", params={}, confidence=1.0, source="RULE_FAST")) assert reply.blocks and reply.blocks[0].type == "guidance" assert "试排" in reply.text or any( "试排" in s.get("command", "") for s in reply.blocks[0].props.get("suggestions", [])) miss = await handle_intent( store, "test", IntentResult(intent="knowledge.query", params={"query": "完全不存在的zzz知识xyz"}, confidence=1.0, source="RULE_FAST")) assert "不编造" in miss.text assert miss.blocks and miss.blocks[0].type == "guidance" assert miss.blocks[0].props.get("context") == "knowledge_miss" asyncio.run(_run()) def test_folder_analyze_and_schedule_cards_forward_skipped_rows(monkeypatch): """folder.analyze 与 folder.schedule 富卡都必须透传 round-61 skippedRows。""" import asyncio report = { "ok": True, "projectName": "锐扬", "workDir": r"D:\ItemSpace\14.工业智核\锐扬", "files": [], "coverage": {}, "coverageDetail": [], "missing": [], "softMissing": [], "canSchedule": True, "totalOk": 1357, "totalErrors": 53, "skippedRows": 974, "batches": [], "markdown": "另有 974 行来自非排产块，已跳过，不计入问题行。", "sqlApplied": False, "sqlStats": {}, } deep = { "folder": report, "projectName": report["projectName"], "workDir": report["workDir"], "markdown": report["markdown"], "sources": [], "summary": {}, "orders": [], "materials": [], "routings": [], "equipment": [], "plan": [], "kbActions": [], "knowledgeIngest": {}, } monkeypatch.setattr( "server.aps_domain.project_analyze.analyze_project_deep", lambda *_args, **_kwargs: deep, ) monkeypatch.setattr( "server.aps_domain.folder_pack.analyze_work_dir", lambda *_args, **_kwargs: report, ) store = _MemStore(seed_world()) async def _run(): replies = [] for intent_name in ("folder.analyze", "folder.schedule"): replies.append( await handle_intent( store, "s-r61", IntentResult( intent=intent_name, params={"query": "分析并排产"}, confidence=1.0, source="RULE_FAST", ), actor="test", ) ) return replies for reply in asyncio.run(_run()): block = next(item for item in reply.blocks if item.type == "folder-pack") assert block.props["totalErrors"] == 53 assert block.props["skippedRows"] == 974 assert block.props["totalErrors"] + block.props["skippedRows"] == 1027 def test_scheduling_data_guide_empty_world_lists_missing_steps(): world = empty_world() # 空世界：无订单/物料/工艺/设备 guide = scheduling_data_guide(world) assert guide["mode"] == "data-missing" ids = {step["id"] for step in guide["steps"]} assert {"orders", "products", "routing", "equipment"} <= ids assert all(step["actions"] for step in guide["steps"]) assert any(s["command"] == "带我排一版" for s in guide["suggestions"]) def test_scheduling_data_guide_ready_when_minimal_data_complete(): world = seed_world() world["flexOrders"] = [{ "id": 1, "orderNo": "SO-1", "productCode": "P1", "quantity": 10, "dueDate": "2026-08-20", "status": "APPROVED", }] world["flexMaterials"] = [{"code": "P1", "name": "测试产品", "stock": 100, "inTransit": 0}] world["flexRoutings"] = [{ "productCode": "P1", "seq": 10, "operationCode": "CUT", "stdTimePerUnit": 5, "stdTimeSource": "实测", }] world["flexEquipment"] = [{ "code": "EQ1", "name": "切割机", "status": "RUNNING", "capabilities": ["CUT"], }] world["flexCalendar"] = [{ "shiftCode": "D", "startTime": "08:00", "endTime": "17:00", "workdays": [1, 2, 3, 4, 5], }] guide = scheduling_data_guide(world) assert guide["mode"] == "ready" assert guide["steps"] == [] assert any(s["command"] == "直接排一版" for s in guide["suggestions"]) def test_guidance_next_returns_data_missing_steps(): import asyncio store = _MemStore(empty_world()) async def _run(): reply = await handle_intent( store, "test", IntentResult(intent="guidance.next", params={}, confidence=1.0, source="RULE_FAST")) block = next(b for b in reply.blocks if b.type == "guidance") assert block.props.get("mode") == "data-missing" assert block.props.get("steps") assert "还差" in reply.text or "还差" in (block.props.get("hint") or "") asyncio.run(_run())