aps-agent/tests/golden/test_engine_constraint_semantics.py

# ============================================================
# matrix-113（Q 方向）：RULE/CP/HYBRID/GA 四引擎硬约束语义一致
# 黄金测试（可重生 ✅）
#   同一世界 + 同一 engine flags：硬约束开启 → 对应冲突出现且发布被拦；
#   关闭 → 冲突消失；软权重无法关闭硬约束（profile 门禁拒绝）。
# 审计结论（2026-08-02）：四引擎均经 RuleEngine.materialize_schedule 统一物化，
#   equipment(C4)/capacity(C7)/materialKit(C6) 开关行为一致，无需引擎修复；
#   personnel/tooling/freeze 开关只作用于 FLEX 引擎（柔性世界 flex*），
#   不在本 RULE/CP/HYBRID/GA 四引擎集合的可观察面内（差异见模块注释）。
# ============================================================
from __future__ import annotations

import copy

import pytest

from server.aps_domain.constraints import (
    ConstraintProfileDenied,
    apply_profile_save,
    engine_constraint_flags,
    hard_blocking_conflicts,
)
from server.engines import get_engine
from server.engines.base import EngineParams
from server.state.seed import seed_world
from server.timeutil import add_minutes, fmt_date, fmt_dt, parse_dt, today0

ENGINES = ["RULE", "CP", "HYBRID", "GA"]


def _next_id_factory():
    counters: dict[str, int] = {}

    def next_id(kind: str) -> int:
        counters[kind] = counters.get(kind, 0) + 1
        return counters[kind]

    return next_id


def _run(world, engine, flags):
    """在传入世界跑一次指定引擎（solve 会就地写版本/冲突）。"""
    start = fmt_date(add_minutes(today0(), 24 * 60))
    params = EngineParams(
        orderIds=[world["salesOrders"][0]["id"]], engineType=engine,
        strategyTemplate="COMPREHENSIVE", planningHorizonDays=14,
        startDate=start, constraints=flags, timeLimitSeconds=2.0,
    )
    return get_engine(engine).solve(world, params, _next_id_factory())


def _single_order_world():
    """只保留一个订单，放大数量并压缩班次 → 容量冲突对所有引擎都必然出现。"""
    world = seed_world()
    so = world["salesOrders"][0]
    so["items"] = [so["items"][0]]
    so["items"][0]["quantity"] = 3000
    keep_pid = so["items"][0]["productId"]
    world["lineProducts"] = [lp for lp in world["lineProducts"] if lp["productId"] == keep_pid]
    for s in world["shifts"]:
        start_hour = int(str(s["startTime"]).split(":", 1)[0])
        s["endTime"] = f"{start_hour + 2:02d}:00"
        s["breakPeriods"] = []
    return world


def _full_horizon_maintenance_world():
    """全时域维保覆盖全部设备 → 任何工单必然与维保重叠（EQUIPMENT 必现）。"""
    world = _single_order_world()
    world["maintenance"] = []
    start = parse_dt(fmt_date(add_minutes(today0(), 24 * 60)) + " 08:00")
    end = add_minutes(start, 15 * 24 * 60)
    for i, eq in enumerate(world["equipment"]):
        world["maintenance"].append({
            "id": 100 + i, "equipmentId": eq["id"], "status": "PLANNED",
            "plannedStart": fmt_dt(start), "plannedEnd": fmt_dt(end),
        })
    return world


def _zero_stock_world():
    """清空全部非成品库存 → 物料齐套冲突（C6）必现。"""
    world = seed_world()
    so = world["salesOrders"][0]
    so["items"] = [so["items"][0]]
    for m in world["materials"]:
        if m.get("type") != "FINISHED_PRODUCT":
            m["stock"] = 0
            m["inTransit"] = 0
    return world


def _conflicts_of(world, version_id, ctype):
    return [c for c in world["conflicts"]
            if c.get("conflictType") == ctype and c.get("versionId") == version_id]


def _blockers_of(world, version_id, ctype):
    return [c for c in hard_blocking_conflicts(world, version_id=version_id)
            if c["conflictType"] == ctype]


def _semantics(engine, world_factory, ctype, flags_off_key, patch=None):
    """返回 (on_conflict, on_blocker, off_conflict, off_blocker) 四元组。"""
    base = world_factory()
    flags = engine_constraint_flags(base)
    if patch:
        apply_profile_save(base, {"constraints": patch})
        flags = engine_constraint_flags(base)
    w_on = copy.deepcopy(base)
    res = _run(w_on, engine, flags)
    on = (bool(_conflicts_of(w_on, res.versionId, ctype)),
          bool(_blockers_of(w_on, res.versionId, ctype)))
    w_off = copy.deepcopy(base)
    res = _run(w_off, engine, {**flags, flags_off_key: False})
    off = (bool(_conflicts_of(w_off, res.versionId, ctype)),
           bool(_blockers_of(w_off, res.versionId, ctype)))
    return (*on, *off)


@pytest.mark.parametrize("engine", ENGINES)
def test_equipment_hard_constraint_on_off_consistent_per_engine(engine):
    on_c, on_b, off_c, off_b = _semantics(engine, _full_horizon_maintenance_world,
                                          "EQUIPMENT", "equipment")
    assert on_c and on_b, f"{engine}: 开启 EQUIPMENT 硬约束应出现冲突并拦发布"
    assert not off_c and not off_b, f"{engine}: 关闭 EQUIPMENT 后冲突应消失"


@pytest.mark.parametrize("engine", ENGINES)
def test_capacity_hard_constraint_on_off_consistent_per_engine(engine):
    on_c, on_b, off_c, off_b = _semantics(engine, _single_order_world,
                                          "CAPACITY", "capacity")
    assert on_c and on_b, f"{engine}: 开启 CAPACITY 硬约束应出现冲突并拦发布"
    assert not off_c and not off_b, f"{engine}: 关闭 CAPACITY 后冲突应消失"


@pytest.mark.parametrize("engine", ENGINES)
def test_material_kit_hard_soft_off_consistent_per_engine(engine):
    # 硬：MAJOR 冲突且拦发布
    on_c, on_b, off_c, off_b = _semantics(
        engine, _zero_stock_world, "MATERIAL_SHORTAGE", "materialKit",
        patch={"C6_material_kit": {"kind": "hard", "enabled": True}})
    assert on_c and on_b, f"{engine}: C6 硬约束开启应出现 MAJOR 冲突并拦发布"
    assert not off_c and not off_b, f"{engine}: C6 关闭后缺料冲突应消失"


@pytest.mark.parametrize("engine", ENGINES)
def test_material_kit_soft_reports_risk_without_blocking(engine):
    base = _zero_stock_world()
    flags = engine_constraint_flags(base)          # C6 默认软
    w = copy.deepcopy(base)
    res = _run(w, engine, flags)
    conflicts = _conflicts_of(w, res.versionId, "MATERIAL_SHORTAGE")
    blockers = _blockers_of(w, res.versionId, "MATERIAL_SHORTAGE")
    assert conflicts and all(c["severity"] == "MINOR" for c in conflicts)
    assert not blockers


def test_all_engines_agree_on_hard_constraint_semantics():
    """跨引擎一致性：同一世界 + 同一 flags，四引擎开/关语义完全一致。"""
    for ctype, factory, flag_key, patch in [
        ("EQUIPMENT", _full_horizon_maintenance_world, "equipment", None),
        ("CAPACITY", _single_order_world, "capacity", None),
        ("MATERIAL_SHORTAGE", _zero_stock_world, "materialKit",
         {"C6_material_kit": {"kind": "hard", "enabled": True}}),
    ]:
        outcomes = {}
        for engine in ENGINES:
            outcomes[engine] = _semantics(engine, factory, ctype, flag_key, patch)
        first = outcomes["RULE"]
        assert first == (True, True, False, False), f"{ctype} RULE 基线异常: {first}"
        for engine in ENGINES[1:]:
            assert outcomes[engine] == first, (
                f"{ctype}: {engine} 与 RULE 语义不一致 {outcomes[engine]} != {first}")


def test_soft_weight_cannot_close_hard_constraint_end_to_end():
    """软权重归零被门禁拒绝 → flags 仍开启 → 四引擎仍拦发布。"""
    world = seed_world()
    with pytest.raises(ConstraintProfileDenied):
        apply_profile_save(world, {"constraints": {"C7_capacity": {"weight": 0}}})
    assert engine_constraint_flags(world)["capacity"] is True
    base = _single_order_world()
    results = {}
    for engine in ENGINES:
        w = copy.deepcopy(base)
        res = _run(w, engine, engine_constraint_flags(world))
        results[engine] = bool(_blockers_of(w, res.versionId, "CAPACITY"))
    assert results == {e: True for e in ENGINES}