from __future__ import annotations import copy from pathlib import Path import pytest from server.agent_core import harness from server.aps_domain.workflow import execute_confirmed, stage_schedule_publish from server.aps_domain.mes import ( apply_dispatch, preview_dispatch, stage_dispatch, validate_dispatchable_version, ) from server.engines.pool_engine import PoolEngine from server.integrations.mes_stub import reset_mes_client from server.state.checkpoints import CheckpointStore from server.state.seed import build_demo_world BUSINESS_DATE = "2026-08-02" class _MemStore: def __init__(self, data: dict, checkpoint_path: Path): self.data = data self.checkpoints = CheckpointStore(str(checkpoint_path)) def next_id(self, kind: str) -> int: key = f"_test_{kind}" self.data[key] = self.data.get(key, 9000) + 1 return self.data[key] def save(self): pass def _scheduled_world() -> tuple[dict, int]: world = build_demo_world() order = copy.deepcopy(next(item for item in world["flexOrders"] if item["productCode"] == "HV-HARNESS")) world["flexOrders"] = [order] counters: dict[str, int] = {} def next_id(kind: str) -> int: counters[kind] = counters.get(kind, 0) + 1 return counters[kind] result = PoolEngine().solve( world, next_id, sort_mode="ASC", start_date=BUSINESS_DATE, ) return world, result["versionId"] def _evidence_refs(world: dict, version_id: int) -> list[str]: validation = validate_dispatchable_version(world, "flex", version_id) return [f"schedule-version:{version_id}", validation["evidenceRef"]] def _approved_grant(version_id: int, evidence_refs: list[str]) -> tuple[str, str]: from server.auth.context import IdentityContext, bind_identity, reset_identity block = harness.stage_confirmation( "test-flex-publish-mes", "mes.dispatch", {"track": "flex", "versionId": version_id, "evidenceRefs": evidence_refs}, title="MES dispatch", summary_lines=["test"], ) confirm_id = str(block.props["confirmId"]) first = harness.take_confirmation(confirm_id, approve=True) token = bind_identity(IdentityContext( 9202, "flex-approver-2", "Flex Approver 2", "platform", roles=("planner",), )) try: second = harness.take_confirmation(confirm_id, approve=True) finally: reset_identity(token) assert first and first["needsSecondConfirm"] is True assert second and second["executionGrant"] return confirm_id, str(second["executionGrant"]) def test_complete_draft_is_publish_ready_but_mes_preview_and_stage_fail_closed(tmp_path: Path): world, version_id = _scheduled_world() validation = validate_dispatchable_version(world, "flex", version_id) assert validation["structuralValid"] is True assert validation["publishReady"] is True assert validation["dispatchReady"] is False assert {item["code"] for item in validation["blockingReasons"]} == {"VERSION_NOT_PUBLISHED"} assert validation["publishBlockingReasons"] == [] preview = preview_dispatch(world, "flex", version_id=version_id) assert preview["items"] == [] assert preview["newCount"] == 0 assert preview["dispatchReady"] is False store = _MemStore(world, tmp_path / "draft-checkpoints.json") staged = stage_dispatch(store, "flex", session_id="draft-stage", actor="test") assert staged["staged"] is False assert staged["block"] is None assert {item["code"] for item in staged["blockingReasons"]} == {"VERSION_NOT_PUBLISHED"} def test_published_complete_version_is_dispatchable_and_has_stable_evidence(): world, version_id = _scheduled_world() version = next(item for item in world["flexScheduleVersions"] if item["id"] == version_id) version["status"] = "PUBLISHED" first = validate_dispatchable_version(world, "flex", version_id) second = validate_dispatchable_version(world, "flex", version_id) assert first["structuralValid"] is True assert first["dispatchReady"] is True assert first["blockingReasons"] == [] assert first["eligibleWorkOrderIds"] assert first["evidenceRef"] == second["evidenceRef"] assert first["evidenceRef"].startswith("schedule-evidence:") def test_orphan_work_order_and_incomplete_routing_block_publish_and_dispatch(): world, version_id = _scheduled_world() version = next(item for item in world["flexScheduleVersions"] if item["id"] == version_id) version["status"] = "PUBLISHED" work_order = next(item for item in world["flexWorkOrders"] if item["versionId"] == version_id) work_order["vlId"] = 999999 validation = validate_dispatchable_version(world, "flex", version_id) codes = {item["code"] for item in validation["blockingReasons"]} assert validation["structuralValid"] is False assert validation["publishReady"] is False assert validation["dispatchReady"] is False assert "ORPHAN_WORK_ORDER" in codes assert "INCOMPLETE_ROUTING" in codes assert preview_dispatch(world, "flex", version_id=version_id)["items"] == [] def test_unresolved_hard_conflict_blocks_publish_and_dispatch(): world, version_id = _scheduled_world() version = next(item for item in world["flexScheduleVersions"] if item["id"] == version_id) version["status"] = "PUBLISHED" world["flexConflicts"].append({ "id": 991, "versionId": version_id, "conflictType": "NO_CAPABILITY", "severity": "CRITICAL", "description": "No qualified equipment", "isResolved": False, }) version["conflictCount"] = len([ item for item in world["flexConflicts"] if item["versionId"] == version_id ]) validation = validate_dispatchable_version(world, "flex", version_id) assert validation["structuralValid"] is True assert validation["publishReady"] is False assert validation["dispatchReady"] is False assert "HARD_CONFLICT" in {item["code"] for item in validation["blockingReasons"]} def test_apply_dispatch_rejects_work_order_drift_after_checkpoint(tmp_path: Path): reset_mes_client(tmp_path / "mes-mirror.json") world, version_id = _scheduled_world() version = next(item for item in world["flexScheduleVersions"] if item["id"] == version_id) version["status"] = "PUBLISHED" store = _MemStore(world, tmp_path / "drift-checkpoints.json") evidence_refs = _evidence_refs(world, version_id) confirm_id, grant = _approved_grant(version_id, evidence_refs) checkpoint = store.checkpoints.create(world, label="approved", reason="test") work_order = next(item for item in world["flexWorkOrders"] if item["versionId"] == version_id) work_order["plannedEndTime"] = "2026-08-31 23:59" with pytest.raises(PermissionError, match="\u6f02\u79fb"): apply_dispatch( store, track="flex", actor="test", confirm_id=confirm_id, execution_grant=grant, version_id=version_id, before_snapshot=str(checkpoint["pairId"]), evidence_refs=evidence_refs, checkpoint_store=store.checkpoints, ) assert world.get("mesLinks") in (None, []) def test_flex_publish_uses_p2_confirmation_before_p3_dispatch(tmp_path: Path): world, version_id = _scheduled_world() store = _MemStore(world, tmp_path / "publish-checkpoints.json") reply = stage_schedule_publish( store, session_id="flex-publish", actor="planner", track="flex", version_id=version_id, ) block = next(block for block in reply.blocks if block.type == "confirm-card") confirm_id = str(block.props["confirmId"]) version = next(row for row in world["flexScheduleVersions"] if row["id"] == version_id) assert version["status"] == "DRAFT" assert world.get("mesLinks") in (None, []) text = execute_confirmed(store, confirm_id, True, actor="planner") assert "已发布" in text assert version["status"] == "PUBLISHED" assert world.get("mesLinks") in (None, []) assert validate_dispatchable_version(world, "flex", version_id)["dispatchReady"] is True def test_flex_publish_revalidates_exact_version_and_rejects_drift(tmp_path: Path): world, version_id = _scheduled_world() store = _MemStore(world, tmp_path / "publish-drift-checkpoints.json") reply = stage_schedule_publish( store, session_id="flex-publish-drift", actor="planner", track="flex", version_id=version_id, ) confirm_id = str(next(block for block in reply.blocks if block.type == "confirm-card").props["confirmId"]) version = next(row for row in world["flexScheduleVersions"] if row["id"] == version_id) work_order = next(row for row in world["flexWorkOrders"] if row["versionId"] == version_id) work_order["plannedEndTime"] = "2026-08-31 23:59" text = execute_confirmed(store, confirm_id, True, actor="planner") assert "漂移" in text assert version["status"] == "DRAFT" assert world.get("mesLinks") in (None, []) def test_trial_version_publish_reply_explains_unmet_conditions(tmp_path: Path): """试排版本被拒时，回复必须带结构化发布门禁：未满足条件、解除路径与边界。""" world, version_id = _scheduled_world() version = next(item for item in world["flexScheduleVersions"] if item["id"] == version_id) version["trialOnly"] = True store = _MemStore(world, tmp_path / "trial-publish-checkpoints.json") reply = stage_schedule_publish( store, session_id="flex-publish-trial", actor="planner", track="flex", version_id=version_id, ) assert not [block for block in reply.blocks if block.type == "confirm-card"] gate = next(block.props["publicationGate"] for block in reply.blocks if block.type == "text") assert gate["canPublish"] is False assert gate["statusLabel"] == "不可发布为执行基准" assert gate["sourceLabel"] == "历史资料试排" assert "TRIAL_ONLY" in {reason["code"] for reason in gate["reasons"]} assert all(reason["title"] and reason["detail"] for reason in gate["reasons"]) assert len(gate["nextSteps"]) >= 3 assert "MES" in gate["boundary"] assert version["status"] == "DRAFT" assert world.get("mesLinks") in (None, []) def test_publish_ready_reply_lists_passed_conditions(tmp_path: Path): """可发布版本同样给出门禁说明：已通过条件 + P2 审批边界。""" world, version_id = _scheduled_world() context = world.get("planningContext") if isinstance(context, dict): context["trialOnly"] = False store = _MemStore(world, tmp_path / "ready-publish-checkpoints.json") reply = stage_schedule_publish( store, session_id="flex-publish-ready", actor="planner", track="flex", version_id=version_id, ) gate = next(block.props["publicationGate"] for block in reply.blocks if block.type == "text") confirm = next(block for block in reply.blocks if block.type == "confirm-card") assert gate["canPublish"] is True assert gate["reasons"] == [] assert "发布校验：通过" in gate["checks"] assert "版本状态：草稿" in gate["checks"] assert gate["nextSteps"] assert "MES" in gate["boundary"] assert confirm.props["power"] == "P2"