aps-agent/tests/golden/test_optimize_engine.py

from __future__ import annotations

import hashlib
import os
from pathlib import Path

import pytest

from server.aps_domain.closed_loop_runtime import run_closed_loop_candidate
from server.aps_domain.kangni_intake import apply_site_payload_to_world, build_site_payload_from_data_dir
from server.engines import get_engine
from server.engines.optimize_engine import normalize_dispatch_rule
from server.engines.pool_engine import PoolEngine
from server.state.seed import seed_world

from tests.golden.test_closed_loop_runtime import BUSINESS_DATE, _ready_world


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 test_optimize_factory_and_rule_catalog_are_available():
    assert get_engine("OPTIMIZE").name == "OPTIMIZE"
    assert normalize_dispatch_rule("DELIVERY_FIRST") == "EDD"
    assert normalize_dispatch_rule("first-in-first-out") == "FIFO"
    assert normalize_dispatch_rule("unknown") == "EDD"


def test_dispatch_rules_have_stable_ordering_keys():
    routings = [
        {"productCode": "SHORT", "stdTimePerUnit": 2},
        {"productCode": "LONG", "stdTimePerUnit": 10},
    ]
    ops = {}
    short = {"productCode": "SHORT", "quantity": 1, "dueDate": "2026-08-05", "priority": 2, "orderNo": "SO-S"}
    long = {"productCode": "LONG", "quantity": 1, "dueDate": "2026-08-04", "priority": 1, "orderNo": "SO-L"}
    assert sorted((long, short), key=lambda row: PoolEngine._dispatch_key("SPT", row, routings, ops)) == [short, long]
    assert sorted((short, long), key=lambda row: PoolEngine._dispatch_key("LPT", row, routings, ops)) == [long, short]


def test_optimize_runs_through_closed_loop_v2_and_records_provenance():
    world = _ready_world()
    result = run_closed_loop_candidate(
        world,
        _next_id_factory(),
        business_date=BUSINESS_DATE,
        engine_type="OPTIMIZE",
        sort_mode="SPT",
    )

    assert result["solveStatus"] == "FEASIBLE"
    assert result["engineType"] == "OPTIMIZE"
    version = world["flexScheduleVersions"][-1]
    assert version["engineType"] == "OPTIMIZE"
    assert version["solverId"] == "optimize-dispatch"
    assert version["algorithmId"] == "optimize.spt"
    assert version["schedulingSolutionV2"]["assumptions"][0]["code"] == "OPTIMIZE_ENGINE_V1_ADAPTER"
    assert version["schedulingSolutionV2"]["provenance"]["solverId"] == "optimize-dispatch"


def test_optimize_blocker_keeps_engine_identity_and_zero_artifacts():
    world = _ready_world()
    world["materials"][1]["stock"] = 0
    world["routings"] = []
    result = run_closed_loop_candidate(
        world,
        _next_id_factory(),
        business_date=BUSINESS_DATE,
        engine_type="OPTIMIZE",
    )

    assert result["solveStatus"] == "BLOCKED"
    assert result["engineType"] == "OPTIMIZE"
    assert result["woCount"] == 0
    assert world["flexScheduleVersions"][-1]["engineType"] == "OPTIMIZE"
    assert world["flexWorkOrders"] == []


@pytest.mark.skipif(
    not os.environ.get("APS_KANGNI_DATA_DIR"),
    reason="set APS_KANGNI_DATA_DIR to run the external Kangni/MOM workbook test",
)
def test_real_kangni_workbooks_run_through_optimize_closed_loop_without_source_mutation():
    data_dir = Path(os.environ["APS_KANGNI_DATA_DIR"])
    files = sorted(data_dir.glob("*.xlsx"))
    assert files, f"no .xlsx workbooks found in {data_dir}"
    before = {path.name: hashlib.sha256(path.read_bytes()).hexdigest() for path in files}

    payload = build_site_payload_from_data_dir(data_dir, station_count=4)
    world = seed_world()
    intake = apply_site_payload_to_world(world, payload, clear_all=True)
    result = run_closed_loop_candidate(
        world,
        _next_id_factory(),
        business_date=BUSINESS_DATE,
        engine_type="OPTIMIZE",
        sort_mode="EDD",
        strict=True,
    )

    after = {path.name: hashlib.sha256(path.read_bytes()).hexdigest() for path in files}
    assert before == after
    assert intake["orderCount"] == 10
    assert intake["routingRecordCount"] == 72
    assert intake["bomCount"] == 823
    assert result["engineType"] == "OPTIMIZE"
    assert result["solveStatus"] == "REJECTED"
    assert result["solverId"] == "optimize-dispatch"
    assert result["algorithmId"] == "optimize.edd"
    assert result["vlCount"] == 0
    assert result["woCount"] == 0
    assert result["planning"]["summary"]["blockerCounts"]["SUPPLY_SHORTAGE"] > 0