337 lines
14 KiB
Python
337 lines
14 KiB
Python
# ============================================================
|
||
# OR-04 LNS 插单局部修复与影响半径黄金测试
|
||
# 固化：窗口内最小扰动（窗口外订单冻结）/ 超阈值升级全量重排 /
|
||
# 局部方案入 DRAFT 走确认 / 与现有 rush.evaluate|apply 兼容
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import copy
|
||
|
||
from server.aps_domain import workflow as wf
|
||
from server.aps_domain.lns import apply_lns_local, lns_local_repair
|
||
from server.aps_domain.rush import (
|
||
_engine_params,
|
||
_sandbox_counter,
|
||
apply_rush,
|
||
evaluate_rush,
|
||
)
|
||
from server.contracts import IntentResult
|
||
from server.engines import get_engine
|
||
from server.state.seed import seed_world
|
||
from server.timeutil import add_minutes, fmt_date, parse_dt, today0
|
||
|
||
|
||
def _next_id_factory(world):
|
||
tables = {
|
||
"audit": "auditEvents",
|
||
"conflict": "conflicts",
|
||
"log": "logs",
|
||
"productionOrder": "productionOrders",
|
||
"salesOrder": "salesOrders",
|
||
"scheduleVersion": "scheduleVersions",
|
||
"workOrder": "workOrders",
|
||
}
|
||
counters: dict[str, int] = {}
|
||
|
||
def next_id(kind: str) -> int:
|
||
if kind not in counters:
|
||
rows = world.get(tables.get(kind, kind + "s"), [])
|
||
counters[kind] = max((r.get("id", 0) for r in rows if isinstance(r, dict)), default=0)
|
||
counters[kind] += 1
|
||
return counters[kind]
|
||
|
||
return next_id
|
||
|
||
|
||
def _scheduled_world():
|
||
world = seed_world()
|
||
get_engine("RULE").solve(world, _engine_params(world, "DELIVERY_FIRST"), _sandbox_counter())
|
||
return world
|
||
|
||
|
||
def _small_rush_payload():
|
||
return {
|
||
"customerName": "华东装备客户",
|
||
"customerLevel": "VIP",
|
||
"productId": 1,
|
||
"quantity": 100,
|
||
"deliveryDate": fmt_date(add_minutes(today0(), 4 * 24 * 60)),
|
||
"priority": 1,
|
||
}
|
||
|
||
|
||
def _window_outside(world, win):
|
||
ws, we = parse_dt(win["start"]), parse_dt(win["end"])
|
||
return [p for p in world["productionOrders"]
|
||
if not (ws <= parse_dt(p["plannedStartDate"]) <= we
|
||
or ws <= parse_dt(p["plannedEndDate"]) <= we)]
|
||
|
||
|
||
def test_lns_local_repair_minimal_perturbation_window_outer_frozen():
|
||
"""窗口内最小扰动：LOCAL 方案不移动任何窗口外订单，急单落在窗口内，主干零接触。"""
|
||
world = _scheduled_world()
|
||
snapshot = copy.deepcopy(world)
|
||
|
||
lns = lns_local_repair(world, _small_rush_payload())
|
||
assert lns["status"] == "LOCAL"
|
||
assert lns["escalate"] is False
|
||
assert lns["escalateReasons"] == []
|
||
|
||
d = lns["disturbance"]
|
||
assert d["movedOrderCount"] <= lns["window"]["maxAffectedOrders"]
|
||
assert d["windowOuterMovedCount"] == 0
|
||
assert d["shiftedHours"] >= 0
|
||
assert "tardinessDelta" in d
|
||
|
||
# 急单落点必须在固定窗口内
|
||
ws, we = parse_dt(lns["window"]["start"]), parse_dt(lns["window"]["end"])
|
||
rs, re_ = parse_dt(lns["plan"]["rushStart"]), parse_dt(lns["plan"]["rushEnd"])
|
||
assert ws <= rs and re_ <= we
|
||
assert lns["plan"]["rushOrder"]["isNew"] is True
|
||
|
||
# 窗口外订单不被移动：方案移动集 ⊆ 窗口内订单
|
||
moved_sids = {m["salesOrderId"] for m in lns["plan"]["movedOrders"]}
|
||
assert all(p["salesOrderId"] not in moved_sids for p in _window_outside(snapshot, lns["window"]))
|
||
|
||
# P1：主干零接触
|
||
assert world["productionOrders"] == snapshot["productionOrders"]
|
||
assert world["salesOrders"] == snapshot["salesOrders"]
|
||
assert world["scheduleVersions"] == snapshot["scheduleVersions"]
|
||
|
||
|
||
def test_lns_escalate_when_move_count_exceeds_threshold():
|
||
"""超过影响阈值：受影响订单数 / 扰动率超限 → 升级全量重排信号。"""
|
||
world = _scheduled_world()
|
||
payload = {
|
||
"customerName": "华东装备客户",
|
||
"customerLevel": "VIP",
|
||
"productId": 1,
|
||
"quantity": 200,
|
||
"deliveryDate": fmt_date(add_minutes(today0(), 2 * 24 * 60)),
|
||
"priority": 1,
|
||
}
|
||
lns = lns_local_repair(world, payload,
|
||
window={"timeWindowHours": 48, "maxAffectedOrders": 1,
|
||
"disturbanceTolerance": 0.2})
|
||
assert lns["status"] == "ESCALATE"
|
||
assert lns["escalate"] is True
|
||
assert lns["escalation"] == {"mode": "FULL_RESCHEDULE"}
|
||
assert lns["escalateReasons"], "超阈值必须给出升级原因"
|
||
assert lns["disturbance"]["movedOrderCount"] > 1
|
||
assert any("上限" in r for r in lns["escalateReasons"])
|
||
|
||
|
||
def test_lns_escalate_when_duration_exceeds_window():
|
||
"""急单工时超过窗口宽度 → 窗口内不可行，升级全量重排。"""
|
||
world = _scheduled_world()
|
||
lns = lns_local_repair(world, {
|
||
"customerName": "华东装备客户",
|
||
"customerLevel": "VIP",
|
||
"productId": 1,
|
||
"quantity": 5000,
|
||
"deliveryDate": fmt_date(add_minutes(today0(), 2 * 24 * 60)),
|
||
"priority": 1,
|
||
}, window={"timeWindowHours": 48})
|
||
assert lns["status"] == "ESCALATE"
|
||
assert lns["escalate"] is True
|
||
assert lns["plan"] is None
|
||
assert any("窗口" in r for r in lns["escalateReasons"])
|
||
|
||
|
||
def test_lns_apply_local_creates_draft_and_keeps_window_frozen():
|
||
"""LNS 成功 → 局部方案入试排草稿（DRAFT）：急单入池 + 新版本 + 窗口外冻结。"""
|
||
world = _scheduled_world()
|
||
snapshot = copy.deepcopy(world)
|
||
baseline_versions = len(world["scheduleVersions"])
|
||
|
||
lns = lns_local_repair(world, _small_rush_payload())
|
||
assert lns["status"] == "LOCAL"
|
||
|
||
applied = apply_lns_local(world, _next_id_factory(world), {
|
||
"payload": lns["payload"],
|
||
"lns": lns,
|
||
"strategy": lns["strategy"],
|
||
})
|
||
order = applied["order"]
|
||
result = applied["result"]
|
||
|
||
assert order["isRush"] is True
|
||
assert order["status"] == "APPROVED"
|
||
assert any(so["id"] == order["id"] for so in world["salesOrders"])
|
||
assert len(world["scheduleVersions"]) == baseline_versions + 1
|
||
|
||
ver = world["scheduleVersions"][-1]
|
||
assert ver["status"] == "DRAFT"
|
||
assert ver["triggerType"] == "LNS_RUSH"
|
||
assert ver["versionNo"] == result.versionNo
|
||
assert result.solveStatus == "LOCAL_REPAIR"
|
||
assert applied["mode"] == "LNS_LOCAL"
|
||
|
||
# 急单 PO 按计划槽位落地，工单同步
|
||
rush_po = next(p for p in world["productionOrders"] if p.get("salesOrderId") == order["id"])
|
||
assert rush_po["plannedStartDate"] == lns["plan"]["rushStart"]
|
||
assert rush_po["plannedEndDate"] == lns["plan"]["rushEnd"]
|
||
rush_wos = [w for w in world["workOrders"] if w["productionOrderId"] == rush_po["id"]]
|
||
assert rush_wos
|
||
assert min(w["plannedStartTime"] for w in rush_wos) == lns["plan"]["rushStart"]
|
||
|
||
# 窗口外订单冻结：新旧版本中其计划时间均不变
|
||
for p in _window_outside(snapshot, lns["window"]):
|
||
copy_new = next((q for q in world["productionOrders"]
|
||
if q.get("salesOrderId") == p["salesOrderId"]
|
||
and q.get("schedulingVersionId") == ver["id"]), None)
|
||
assert copy_new is not None
|
||
assert copy_new["plannedStartDate"] == p["plannedStartDate"]
|
||
assert copy_new["plannedEndDate"] == p["plannedEndDate"]
|
||
|
||
# 回滚到采用前快照（模拟检查点恢复）
|
||
world.clear()
|
||
world.update(copy.deepcopy(snapshot))
|
||
assert len(world["salesOrders"]) == len(snapshot["salesOrders"])
|
||
assert len(world["scheduleVersions"]) == baseline_versions
|
||
|
||
|
||
|
||
|
||
def test_lns_existing_order_apply_moves_po_to_plan_slot():
|
||
"""已有订单加急（existing 模式）：PO 平移到 LNS 计划槽位，版本仍为 LNS_RUSH DRAFT。"""
|
||
world = _scheduled_world()
|
||
so = world["salesOrders"][3]
|
||
lns = lns_local_repair(world, {"orderNo": so["orderNo"], "strategy": "COMPREHENSIVE"})
|
||
assert lns["mode"] == "existing"
|
||
assert lns["status"] == "LOCAL"
|
||
assert lns["plan"]["rushOrder"]["isNew"] is False
|
||
|
||
baseline_versions = len(world["scheduleVersions"])
|
||
apply_lns_local(world, _next_id_factory(world), {
|
||
"payload": lns["payload"],
|
||
"lns": lns,
|
||
"strategy": lns["strategy"],
|
||
})
|
||
ver = world["scheduleVersions"][-1]
|
||
assert ver["status"] == "DRAFT" and ver["triggerType"] == "LNS_RUSH"
|
||
assert len(world["scheduleVersions"]) == baseline_versions + 1
|
||
po = next(p for p in world["productionOrders"]
|
||
if p.get("salesOrderId") == so["id"] and p.get("schedulingVersionId") == ver["id"])
|
||
assert po["plannedStartDate"] == lns["plan"]["rushStart"]
|
||
assert po["plannedEndDate"] == lns["plan"]["rushEnd"]
|
||
|
||
|
||
def test_lns_escalation_falls_back_to_full_reschedule_compat():
|
||
"""升级场景 → 复用 apply_rush 全量重排（走 P2 确认后执行），生成 DRAFT 版本。"""
|
||
world = _scheduled_world()
|
||
lns = lns_local_repair(world, {
|
||
"customerName": "华东装备客户",
|
||
"customerLevel": "VIP",
|
||
"productId": 1,
|
||
"quantity": 5000,
|
||
"deliveryDate": fmt_date(add_minutes(today0(), 2 * 24 * 60)),
|
||
"priority": 1,
|
||
}, window={"timeWindowHours": 48})
|
||
assert lns["status"] == "ESCALATE"
|
||
|
||
baseline_versions = len(world["scheduleVersions"])
|
||
applied = apply_rush(world, _next_id_factory(world), {
|
||
"payload": lns["payload"],
|
||
"strategy": lns["strategy"],
|
||
"lns": lns,
|
||
})
|
||
result = applied["result"]
|
||
assert len(world["scheduleVersions"]) == baseline_versions + 1
|
||
assert world["scheduleVersions"][-1]["status"] == "DRAFT"
|
||
assert result.orderCount == 8 # 种子 7 + 急单 1（全量重排口径）
|
||
assert result.versionNo == world["scheduleVersions"][-1]["versionNo"]
|
||
|
||
|
||
def test_evaluate_rush_attaches_lns_additively():
|
||
"""与现有 rush.evaluate 兼容：lns 字段附加，原有指标不变；可显式关闭。"""
|
||
world = _scheduled_world()
|
||
n_orders = len(world["salesOrders"])
|
||
|
||
impact = evaluate_rush(world, _small_rush_payload())
|
||
assert "lns" in impact
|
||
assert impact["lns"]["status"] == "LOCAL"
|
||
assert impact["lns"]["window"]["windowOrderCount"] > 0
|
||
# 原有字段不受影响
|
||
assert impact["after"]["orderCount"] == impact["baseline"]["orderCount"] + 1
|
||
assert "delayDelta" in impact and "conflictDelta" in impact
|
||
assert len(world["salesOrders"]) == n_orders
|
||
|
||
impact2 = evaluate_rush(world, {**_small_rush_payload(), "lns": False})
|
||
assert impact2["lns"] is None
|
||
|
||
|
||
# ------------------------------------------------------------
|
||
# 工作流接线：rush.evaluate 携带 LNS；rush.apply 按 LNS 状态路由
|
||
# ------------------------------------------------------------
|
||
class FakeStore:
|
||
"""最小化 WorldStore 替身（对齐 tests/golden/test_schedule_wizard.py 风格）。"""
|
||
|
||
def __init__(self):
|
||
self.data = seed_world()
|
||
self._counters: dict[str, int] = {}
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
tables = {
|
||
"audit": "auditEvents", "conflict": "conflicts", "log": "logs",
|
||
"productionOrder": "productionOrders", "salesOrder": "salesOrders",
|
||
"scheduleVersion": "scheduleVersions", "workOrder": "workOrders",
|
||
}
|
||
rows = self.data.get(tables.get(kind, kind + "s"), [])
|
||
cur = max((r.get("id", 0) for r in rows if isinstance(r, dict)), default=0)
|
||
self._counters[kind] = max(self._counters.get(kind, 0), cur) + 1
|
||
return self._counters[kind]
|
||
|
||
def save(self) -> None:
|
||
pass
|
||
|
||
|
||
def _run(intent_name, store, params):
|
||
return asyncio.run(wf.handle_intent(store, "s-lns", IntentResult(intent=intent_name, params=params, confidence=0.95), "tester"))
|
||
|
||
|
||
def test_workflow_rush_evaluate_attaches_lns_block():
|
||
"""rush.evaluate 接线：回复附带计划员可读的插单安排摘要与结果块。"""
|
||
wf._LAST_RUSH_EVAL.clear()
|
||
store = FakeStore()
|
||
get_engine("RULE").solve(store.data, _engine_params(store.data, "DELIVERY_FIRST"), store.next_id)
|
||
reply = _run("rush.evaluate", store, _small_rush_payload())
|
||
assert "插单安排评估完成" in reply.text
|
||
assert "本次只做试算，现有计划没有改动" in reply.text
|
||
assert "LNS" not in reply.text
|
||
assert "沙盒" not in reply.text
|
||
assert "DRAFT" not in reply.text
|
||
kinds = [b.props.get("kind") for b in reply.blocks]
|
||
assert "rush-impact" in kinds
|
||
assert "lns-repair" in kinds
|
||
wf._LAST_RUSH_EVAL.clear()
|
||
|
||
|
||
def test_workflow_rush_apply_local_routes_to_lns_draft():
|
||
"""rush.apply 增强：LNS=LOCAL → 局部方案入 DRAFT 草稿（P2 确认后执行）。"""
|
||
wf._LAST_RUSH_EVAL.clear()
|
||
store = FakeStore()
|
||
get_engine("RULE").solve(store.data, _engine_params(store.data, "DELIVERY_FIRST"), store.next_id)
|
||
|
||
_run("rush.evaluate", store, _small_rush_payload())
|
||
assert wf._LAST_RUSH_EVAL.get("lns", {}).get("status") == "LOCAL"
|
||
|
||
staged = _run("rush.apply", store, {})
|
||
confirm_blocks = [b for b in staged.blocks if b.type == "confirm-card"]
|
||
assert confirm_blocks, "rush.apply 必须出 P2 确认卡"
|
||
confirm_id = confirm_blocks[0].props["confirmId"]
|
||
assert any("局部调整" in line for line in (confirm_blocks[0].props.get("summary") or []))
|
||
assert not any("LNS" in line or "DRAFT" in line or "P2" in line
|
||
for line in (confirm_blocks[0].props.get("summary") or []))
|
||
|
||
baseline_versions = len(store.data["scheduleVersions"])
|
||
msg = wf.execute_confirmed(store, confirm_id, approve=True, actor="tester")
|
||
assert "插单方案已采用" in msg
|
||
assert "LNS" not in msg
|
||
assert "DRAFT" not in msg
|
||
assert len(store.data["scheduleVersions"]) == baseline_versions + 1
|
||
assert store.data["scheduleVersions"][-1]["triggerType"] == "LNS_RUSH"
|
||
assert store.data["scheduleVersions"][-1]["status"] == "DRAFT"
|
||
wf._LAST_RUSH_EVAL.clear()
|