aps-agent/tests/golden/test_cp_engine.py

# ============================================================
# SC-03 CP-SAT 引擎黄金测试
# ============================================================
from __future__ import annotations

import pytest

from server.engines import get_engine
from server.engines.base import EngineParams
from server.engines.cp_engine import CpSatEngine
from server.engines.rule_engine import RuleEngine
from server.state.seed import seed_world
from server.timeutil import add_minutes, fmt_date, parse_dt, today0


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, strategy="COMPREHENSIVE", engine="CP", time_limit=8.0):
    start = fmt_date(add_minutes(today0(), 24 * 60))
    params = EngineParams(
        orderIds=[], engineType=engine, strategyTemplate=strategy,
        planningHorizonDays=14, startDate=start, timeLimitSeconds=time_limit,
    )
    return get_engine(engine).solve(world, params, _next_id_factory())


def test_get_engine_cp_is_cpsat():
    eng = get_engine("CP")
    assert isinstance(eng, CpSatEngine)
    assert eng.name == "CP"
    assert eng.supports_anytime is True


def test_cp_engine_type_and_solver_meta():
    world = seed_world()
    result = _run(world)
    assert result.engineType == "CP"
    assert world["scheduleVersions"][0]["engineType"] == "CP"
    meta = world["scheduleVersions"][0].get("solverMeta") or {}
    assert meta.get("backend") == "OR-Tools CP-SAT"
    assert result.solveStatus in ("OPTIMAL", "FEASIBLE", "UNKNOWN", "INFEASIBLE")
    assert result.solveTimeSec is not None
    assert result.poCount == 7
    assert result.woCount == 30


def test_cp_no_workstation_overlap():
    world = seed_world()
    _run(world)
    by_ws: dict[int, list] = {}
    for wo in world["workOrders"]:
        by_ws.setdefault(wo["workstationId"], []).append(
            (parse_dt(wo["plannedStartTime"]), parse_dt(wo["plannedEndTime"])))
    for ws_id, ivs in by_ws.items():
        ivs.sort()
        for (s1, e1), (s2, e2) in zip(ivs, ivs[1:]):
            assert e1 <= s2, f"工位 {ws_id} 双占"


def test_cp_tardiness_not_worse_than_rule():
    """可行 CP 结果直接物化经父进程验证的 operation slots。"""
    start = fmt_date(add_minutes(today0(), 24 * 60))
    params_c = EngineParams(
        orderIds=[], engineType="CP", strategyTemplate="COMPREHENSIVE",
        planningHorizonDays=14, startDate=start, timeLimitSeconds=10.0,
    )
    w_c = seed_world()
    c = CpSatEngine().solve(w_c, params_c, _next_id_factory())
    version = w_c["scheduleVersions"][-1]
    solver_meta = version["solverMeta"]
    c3_validation = solver_meta["materializedC3Validation"]
    work_orders = [
        row for row in w_c["workOrders"]
        if row.get("schedulingVersionId") == c.versionId
    ]
    calendar_conflicts = [
        row for row in w_c["conflicts"]
        if row.get("versionId") == c.versionId and row.get("conflictType") == "CALENDAR"
    ]
    assert c.solveStatus in {"OPTIMAL", "FEASIBLE"}
    assert solver_meta["directlyConsumedByMaterializer"] is True
    assert solver_meta["operationTimingValidation"]["passed"] is True
    assert c3_validation["passed"] is True
    assert c3_validation["cpTimingApplied"] is True
    assert solver_meta["materializedC7Validation"]["passed"] is True
    assert len(work_orders) == c.woCount
    assert all(row.get("cpTimingSource") == "operationSlots" for row in work_orders)
    assert all(row.get("plannedSegments") for row in work_orders)
    assert version["totalTardiness"] == c.totalTardiness
    assert calendar_conflicts == []


def test_cp_timeout_returns_feasible_or_status():
    """极短时限：必须返回带 status 的结果，不得静默变 RULE。"""
    world = seed_world()
    result = _run(world, time_limit=0.5)
    assert result.engineType == "CP"
    assert result.solveStatus is not None
    meta = world["scheduleVersions"][0]["solverMeta"]
    assert meta["backend"] == "OR-Tools CP-SAT"
    # anytime：通常至少 FEASIBLE/OPTIMAL/UNKNOWN；若 INFEASIBLE 则有 fallback 留痕
    assert result.poCount >= 0


def test_cp_infeasible_solver_meta_exposes_native_assumption_core():
    world = seed_world()
    start = fmt_date(add_minutes(today0(), 24 * 60))
    params = EngineParams(
        orderIds=[], engineType="CP", strategyTemplate="COMPREHENSIVE",
        planningHorizonDays=14, startDate=start, timeLimitSeconds=3.0,
        freezeWindowHours=100_000,
    )
    result = CpSatEngine().solve(world, params, _next_id_factory())
    meta = world["scheduleVersions"][-1]["solverMeta"]
    assert result.solveStatus == "INFEASIBLE"
    core = meta["nativeIis"]
    assert core["native"] is True
    assert core["method"] == "SufficientAssumptionsForInfeasibility"
    assert core["minimality"] == "sufficient-assumption-core"
    assert core["isMinimalIis"] is False
    assert "C11_freeze" in core["constraintIds"]
    from server.aps_domain.constraints import get_constraint
    assert all(get_constraint(world, cid) is not None for cid in core["constraintIds"])


@pytest.mark.parametrize("requested", ["RULE", "GA"])
def test_non_cp_still_rule_proxy(requested):
    world = seed_world()
    result = _run(world, engine=requested)
    assert result.engineType == requested
    assert isinstance(get_engine(requested), RuleEngine)


def test_hybrid_is_real_pipeline():
    from server.engines.cp_engine import HybridEngine
    eng = get_engine("HYBRID")
    assert isinstance(eng, HybridEngine)
    world = seed_world()
    result = _run(world, engine="HYBRID")
    assert result.engineType == "HYBRID"
    meta = world["scheduleVersions"][0].get("solverMeta") or {}
    assert meta.get("warmStart") == "RULE"
    assert "RULE→CP-SAT" in str(meta.get("pipeline") or "")
    assert result.solveStatus is not None
    assert result.poCount == 7


def test_hybrid_tardiness_not_worse_than_rule():
    start = fmt_date(add_minutes(today0(), 24 * 60))
    w_h = seed_world()
    h = get_engine("HYBRID").solve(w_h, EngineParams(
        orderIds=[], engineType="HYBRID", strategyTemplate="COMPREHENSIVE",
        planningHorizonDays=14, startDate=start, timeLimitSeconds=10.0,
    ), _next_id_factory())
    version = w_h["scheduleVersions"][-1]
    solver_meta = version["solverMeta"]
    work_orders = [
        row for row in w_h["workOrders"]
        if row.get("schedulingVersionId") == h.versionId
    ]
    assert h.solveStatus in {"OPTIMAL", "FEASIBLE"}
    assert solver_meta["directlyConsumedByMaterializer"] is True
    assert solver_meta["operationTimingValidation"]["passed"] is True
    assert solver_meta["materializedC3Validation"]["passed"] is True
    assert solver_meta["materializedC7Validation"]["passed"] is True
    assert len(work_orders) == h.woCount
    assert all(row.get("cpTimingSource") == "operationSlots" for row in work_orders)
    assert all(row.get("plannedSegments") for row in work_orders)
    assert version["totalTardiness"] == h.totalTardiness