aps-agent/server/engines/nsga2_engine.py

# ============================================================
# NSGA-II 多目标求解器（moduleId: engines-nsga2, round-40 方向 U，矩阵 85）
# 真实 NSGA-II 标准流程：快速非支配排序 + 拥挤度距离 + 锦标赛选择 + 精英保留，
# 不裁剪为单目标。输出 Pareto 解集，可被 scenario_selection.rank_scenarios 消费。
# 与 ga_engine/cp_engine 共享工序/资源数据模型（routingSteps / lineProducts /
# workstationOperations / shiftCalendar），染色体为工序级：
#   - sequence: 工序序列（job j 出现 op_count[j] 次，同单工序沿工艺路线保序）
#   - lines:    每订单一条产线（资源迁移 = 切换备选产线）
# 目标函数（3 个冲突目标，均最小化）：
#   f1 totalTardiness  加权总延迟（分钟 x 客户权重，与 ga_engine 口径一致）
#   f2 conflictCount   工作站容量冲突（连续解码下工作站占用超可用分钟的溢出小时）
#   f3 -loadBalance    产线负载均衡（CV 系数评分取负，与 scenario_selection 同口径）
# 引擎集成：get_engine("NSGA2") 返回 NSGA2Engine。ScheduleResult.engineType 的
# Literal 不含 NSGA2（contracts.py 不在本方向写范围），故复用 GA 契约槽位
# （engineType="GA"），solverMeta.backend="NSGA-II"、solverMeta.engineType="NSGA2"
# 如实标注算法身份。独立函数 solve_nsga2(world, params) 为本模块主入口。
# ============================================================
from __future__ import annotations

import math
from collections.abc import Callable
from dataclasses import dataclass
from itertools import pairwise
from random import Random
from time import monotonic
from typing import Any

from server.aps_domain.scenario_selection import load_balance_score
from server.contracts import ScheduleResult
from server.engines.base import EngineParams
from server.engines.cp_engine import _candidate_options, _due_minutes, _job_duration_min
from server.engines.queries import (
    find_product_lines,
    find_routing_steps,
    get_available_minutes,
)
from server.engines.rule_engine import RuleEngine
from server.timeutil import add_minutes, fmt_date, parse_dt, today0

World = dict[str, Any]

# ---------------- 参数模板 ----------------
# nsga2_defaults：种群/代数/交叉率/变异率/锦标赛/种子（确定性种子支持）。
# 与 server/agent_core/algolib.py 的 AlgorithmManifest 结构衔接（见 build_nsga2_manifest）；
# algolib 注册表文件级翻转（available=True）属方向 T（registry/rebuild）范围，本模块只提供模板。
nsga2_defaults: dict[str, Any] = {
    "populationSize": 48,       # 种群规模
    "generations": 80,          # 最大代数
    "crossoverRate": 0.90,      # 交叉率（OX/PMX）
    "mutationRate": 0.20,       # 工序交换变异率（每后代）
    "lineMutationRate": 0.35,   # 资源迁移变异率（每后代）
    "crossover": "OX",          # OX=顺序交叉（保工序保序）| PMX=部分映射交叉+修复
    "tournamentSize": 2,        # 锦标赛规模
    "seed": 42,                 # 确定性种子
    "timeLimitSeconds": 3.0,    # 兜底时限（EngineParams.timeLimitSeconds 优先）
}


def build_nsga2_manifest() -> dict[str, Any]:
    """NSGA-II 参数模板注册清单：字段与 algolib.AlgorithmManifest 对齐。

    供参数模板注册/选型展示使用；`parameters` 承载 nsga2_defaults。
    返回 dict 可直接 AlgorithmManifest(**manifest) 校验（pydantic extra=ignore
    容忍 parameters 扩展字段）。
    """
    return {
        "algo_id": "nsga2",
        "name": "NSGA-II 多目标遗传",
        "category": "C",
        "description": "多目标帕累托解集：快速非支配排序+拥挤度+锦标赛+精英保留，工序级染色体",
        "scale_limit": "<=10k 工单",
        "time_budget": "秒~分钟级",
        "input_schema": {
            "salesOrders": "array<订单>", "lines": "array<产线>",
            "objectives": ["totalTardiness", "conflictCount", "loadBalance"],
        },
        "output_schema": {
            "paretoSet": "array<方案>",
            "kpi": {"totalTardiness": "float", "conflictCount": "int",
                    "avgUtilization": "float", "loadBalance": "float"},
        },
        "golden_tests": ["tests/golden/test_nsga2_engine.py"],
        "deterministic": True,
        "random_seed": int(nsga2_defaults["seed"]),
        "version": "1.0",
        "regen_strategy": "hybrid",
        "available": True,
        "entrypoint": "server.engines.nsga2_engine:solve_nsga2",
        "parameters": dict(nsga2_defaults),
    }


# ---------------- 数据结构 ----------------
# 单道工序规格：(workstationId|None, dur分钟, gap分钟)
_OpSpec = tuple[int | None, int, int]


@dataclass(frozen=True)
class _Job:
    """工序级调度问题数据：备选产线规格、交期、权重、工序数。"""
    line_specs: dict[int, tuple[_OpSpec, ...]]
    line_options: tuple[int, ...]
    due: int
    weight: int
    op_count: int


@dataclass(frozen=True)
class _Individual:
    sequence: tuple[int, ...]   # 工序级染色体：job id 序列（同单保序）
    lines: tuple[int, ...]       # 每订单产线（按 job id 索引）


# ---------------- 目标评估 ----------------
def _build_jobs(world: World, entries: list[dict], params: EngineParams) -> list[_Job]:
    """把待排条目展开为工序级问题数据（与 ga_engine/cp_engine 同源数据模型）。

    每订单每备选产线 = 沿工艺路线的工序规格序列（含工序时长与间隔）；
    无任何可用产线时降级为哑产线 -1（单道整单时长，不参与容量/均衡统计）。
    """
    from server.aps_domain.params import level_weight

    base_start = (
        parse_dt(params.startDate + " 08:00")
        if params.startDate
        else add_minutes(today0(), 24 * 60)
    )
    due_buffer = (
        1.0
        if params.deliveryBufferRatio is None
        else max(0.5, min(1.0, float(params.deliveryBufferRatio)))
    )
    spec_cache: dict[int, dict[int, tuple[_OpSpec, ...]]] = {}
    jobs: list[_Job] = []
    for entry in entries:
        item = entry["item"]
        pid = int(item["productId"])
        if pid not in spec_cache:
            steps = find_routing_steps(world, pid)
            per_line: dict[int, tuple[_OpSpec, ...]] = {}
            for lp in find_product_lines(world, pid):
                line_id = int(lp["lineId"])
                line = next(l for l in world["lines"] if int(l["id"]) == line_id)
                specs = _candidate_options(world, item, line)
                if specs is None:
                    continue
                per_line[line_id] = tuple(
                    (int(s["workstationId"]), int(s["dur"]), int(s["gap"]))
                    for s in specs
                )
            if not per_line:
                total = _job_duration_min(world, item, {"efficiencyFactor": 1.0})
                op_count = max(1, len(steps))
                dummy: tuple[_OpSpec, ...] = ((None, total, 0),) + ((None, 0, 0),) * (op_count - 1)
                per_line[-1] = dummy
            spec_cache[pid] = per_line
        line_specs = spec_cache[pid]
        op_count = max(len(specs) for specs in line_specs.values())
        if -1 in line_specs and len(line_specs[-1]) < op_count:
            # 哑产线补齐到 op_count（整单时长在前，其余零时长占位），保持染色体长度与路由一致
            total = line_specs[-1][0][1]
            line_specs = dict(line_specs)
            line_specs[-1] = ((None, total, 0),) + ((None, 0, 0),) * (op_count - 1)
        options = tuple(lid for lid in line_specs if lid >= 0) or (-1,)
        weight = max(1, round(level_weight(world, entry["so"].get("customerLevel")) * 100))
        if entry["so"].get("isRush"):
            weight += 200
        if entry["so"].get("isForecast"):
            weight = max(1, weight // 2)
        jobs.append(_Job(
            line_specs=line_specs,
            line_options=options,
            due=_due_minutes(entry["so"], base_start, due_buffer),
            weight=weight,
            op_count=op_count,
        ))
    return jobs


def _ws_availability(world: World, params: EngineParams, jobs: list[_Job]) -> dict[int, int]:
    """每工作站在规划期内的可用分钟（按所属产线班次日历累计；哑产线无容量）。"""
    base = (
        parse_dt(params.startDate + " 08:00")
        if params.startDate
        else add_minutes(today0(), 24 * 60)
    )
    horizon_days = max(7, int(params.planningHorizonDays or 14))
    lines_in_use = {lid for job in jobs for lid in job.line_specs if lid >= 0}
    avail_by_line: dict[int, int] = {}
    for line_id in lines_in_use:
        total = 0
        for day in range(horizon_days):
            date_str = fmt_date(add_minutes(base, day * 24 * 60))
            total += get_available_minutes(world, line_id, date_str)
        avail_by_line[line_id] = total
    ws_line = {int(ws["id"]): int(ws["lineId"]) for ws in world["workstations"]}
    return {
        ws_id: avail_by_line.get(line_id, 0)
        for ws_id, line_id in ws_line.items()
        if line_id in avail_by_line
    }


def _evaluate(
    individual: _Individual,
    jobs: list[_Job],
    ws_avail: dict[int, int],
) -> tuple[float, int, float]:
    """解码工序序列（贪婪左移，工作站连续占用）并计算 3 个目标。

    返回 (totalTardiness, conflictCount, -loadBalance)，三者均最小化。
    - totalTardiness = Σ max(0, 完工-交期) x 客户权重（分钟口径）
    - conflictCount  = 工作站占用分钟超出可用容量的溢出小时（向上取整）
    - loadBalance    = scenario_selection.load_balance_score(产线负载)（0..1）
    """
    n = len(jobs)
    seq, lines = individual.sequence, individual.lines
    op_idx = [0] * n
    job_cursor = [0] * n
    job_end = [0] * n
    ws_cursor: dict[int, int] = {}
    line_load: dict[int, int] = {}
    for job_id in seq:
        j = job_id
        o = op_idx[j]
        op_idx[j] = o + 1
        line_id = lines[j]
        specs = jobs[j].line_specs[line_id]
        spec = specs[o] if o < len(specs) else specs[-1]
        ws, dur, gap = spec
        if dur <= 0:
            continue
        start = job_cursor[j]
        if ws is not None:
            busy = ws_cursor.get(ws, 0)
            start = max(start, busy)
        end = start + dur
        if ws is not None:
            ws_cursor[ws] = end
        job_cursor[j] = end + gap
        if line_id >= 0:
            line_load[line_id] = line_load.get(line_id, 0) + dur
        job_end[j] = max(job_end[j], end)
    tardiness = sum(max(0, job_end[j] - jobs[j].due) * jobs[j].weight for j in range(n))
    conflict_minutes = sum(
        max(0, busy - ws_avail.get(ws, 0)) for ws, busy in ws_cursor.items()
    )
    conflicts = math.ceil(conflict_minutes / 60.0)
    balance = load_balance_score(line_load.values()) if line_load else 1.0
    return (float(tardiness), conflicts, -balance)


def _decode_detail(
    individual: _Individual,
    jobs: list[_Job],
    ws_avail: dict[int, int],
) -> dict[str, Any]:
    """与 _evaluate 同口径的解码明细（供 Pareto 解卡 KPI/证据使用）。"""
    n = len(jobs)
    seq, lines = individual.sequence, individual.lines
    op_idx = [0] * n
    job_cursor = [0] * n
    job_end = [0] * n
    ws_cursor: dict[int, int] = {}
    line_load: dict[int, int] = {}
    for job_id in seq:
        j = job_id
        o = op_idx[j]
        op_idx[j] = o + 1
        line_id = lines[j]
        specs = jobs[j].line_specs[line_id]
        spec = specs[o] if o < len(specs) else specs[-1]
        ws, dur, gap = spec
        if dur <= 0:
            continue
        start = job_cursor[j]
        if ws is not None:
            busy = ws_cursor.get(ws, 0)
            start = max(start, busy)
        end = start + dur
        if ws is not None:
            ws_cursor[ws] = end
        job_cursor[j] = end + gap
        if line_id >= 0:
            line_load[line_id] = line_load.get(line_id, 0) + dur
        job_end[j] = max(job_end[j], end)
    conflict_minutes = sum(
        max(0, busy - ws_avail.get(ws, 0)) for ws, busy in ws_cursor.items()
    )
    used_ws = [ws for ws in ws_cursor if ws in ws_avail]
    busy_total = sum(ws_cursor[ws] for ws in used_ws)
    avail_total = sum(ws_avail[ws] for ws in used_ws)
    utilization = min(1.0, busy_total / avail_total) if avail_total > 0 else 0.0
    return {
        "jobEndMinutes": list(job_end),
        "conflictMinutes": conflict_minutes,
        "utilization": round(utilization, 6),
        "lineLoadMinutes": dict(line_load),
    }


# ---------------- 非支配排序 / 拥挤度 ----------------
def _dominates(a: tuple[float, ...], b: tuple[float, ...]) -> bool:
    """a 支配 b：所有目标不劣且至少一个严格更优（目标均最小化）。"""
    no_worse = True
    strictly_better = False
    for av, bv in zip(a, b):
        no_worse = no_worse and av <= bv
        strictly_better = strictly_better or av < bv
        if not no_worse:
            return False
    return strictly_better


def fast_non_dominated_sort(
    objectives: list[tuple[float, float, float]],
) -> list[list[int]]:
    """快速非支配排序（Deb et al. 2002）：返回前沿列表（每前沿为索引列表）。"""
    if not objectives:
        return [[]]
    n = len(objectives)
    dominated: list[set[int]] = [set() for _ in range(n)]
    dom_count = [0] * n
    fronts: list[list[int]] = [[]]
    for p in range(n):
        for q in range(n):
            if p == q:
                continue
            if _dominates(objectives[p], objectives[q]):
                dominated[p].add(q)
            elif _dominates(objectives[q], objectives[p]):
                dom_count[p] += 1
        if dom_count[p] == 0:
            fronts[0].append(p)
    front_idx = 0
    while fronts[front_idx]:
        nxt: list[int] = []
        for p in fronts[front_idx]:
            for q in dominated[p]:
                dom_count[q] -= 1
                if dom_count[q] == 0:
                    nxt.append(q)
        front_idx += 1
        fronts.append(nxt)
    fronts.pop()
    return fronts


def crowding_distance(
    objectives: list[tuple[float, float, float]],
    front: list[int],
) -> dict[int, float]:
    """拥挤度距离：边界点为无穷大，内部点按各目标归一化边长求和。"""
    if not front:
        return {}
    dist: dict[int, float] = {i: 0.0 for i in front}
    dimension = len(objectives[0])
    for k in range(dimension):
        ranked = sorted(front, key=lambda i: objectives[i][k])
        dist[ranked[0]] = math.inf
        dist[ranked[-1]] = math.inf
        span = objectives[ranked[-1]][k] - objectives[ranked[0]][k]
        if span <= 1e-12:
            continue
        for prev, cur in pairwise(ranked):
            delta = (objectives[cur][k] - objectives[prev][k]) / span
            dist[prev] += delta
            dist[cur] += delta
    return dist


def _assign_ranks_crowding(
    objectives: list[tuple[float, float, float]],
) -> tuple[list[list[int]], list[int], dict[int, float]]:
    """为给定目标集计算 (fronts, ranks, crowding)。"""
    fronts = fast_non_dominated_sort(objectives)
    ranks = [0] * len(objectives)
    crowding: dict[int, float] = {}
    for rank, front in enumerate(fronts):
        for idx in front:
            ranks[idx] = rank
        crowding.update(crowding_distance(objectives, front))
    return fronts, ranks, crowding


# ---------------- 交叉 / 变异（保持工序级可行性） ----------------
def _label_occurrences(sequence: tuple[int, ...]) -> list[tuple[int, int]]:
    """把 job id 序列唯一化为 (job, occurrence) 标记（occurrence 沿路由递增）。"""
    counter: dict[int, int] = {}
    tokens: list[tuple[int, int]] = []
    for job_id in sequence:
        occ = counter.get(job_id, 0)
        counter[job_id] = occ + 1
        tokens.append((job_id, occ))
    return tokens


def _order_crossover(
    a: tuple[int, ...], b: tuple[int, ...], rng: Random,
) -> tuple[int, ...]:
    """顺序交叉 OX：子代继承 a 的连续段，其余按 b 的相对顺序补齐。

    对工序级编码（job 出现多次但同单保序）保持可行性：两亲本均可行时子代可行。
    """
    length = len(a)
    if length < 2:
        return a
    left, right = sorted(rng.sample(range(length), 2))
    child: list[int | None] = [None] * length
    child[left:right + 1] = list(a[left:right + 1])
    # 工序级编码下 job 可多次出现：按段内已放置数量扣减，b 中其余副本按相对顺序补齐
    remaining_counts = {job_id: a[left:right + 1].count(job_id) for job_id in set(a[left:right + 1])}
    remaining: list[int] = []
    for gene in b:
        if remaining_counts.get(gene, 0) > 0:
            remaining_counts[gene] -= 1
        else:
            remaining.append(gene)
    for pos, idx in enumerate(list(range(right + 1, length)) + list(range(left))):
        child[idx] = remaining[pos]
    return tuple(int(gene) for gene in child if gene is not None)


def _partially_mapped_crossover(
    a: tuple[int, ...], b: tuple[int, ...], rng: Random,
) -> tuple[int, ...]:
    """部分映射交叉 PMX：把同 job 多次出现唯一化为 (job, occurrence) 标记，
    对标记排列做标准 PMX，再按 occurrence 重排修复为工序保序的可行子代。

    映射链带访问守卫并配计数回填兜底：a/b 段标记集合相同时链会在段内成环，
    此时对未解析位置按 b 相对顺序补齐（与 OX 同构），保证终止且子代可行。
    """
    length = len(a)
    if length < 2:
        return a
    tokens_a = _label_occurrences(a)
    tokens_b = _label_occurrences(b)
    left, right = sorted(rng.sample(range(length), 2))
    child: list[tuple[int, int] | None] = [None] * length
    child[left:right + 1] = tokens_a[left:right + 1]
    mapping: dict[tuple[int, int], tuple[int, int]] = {}
    for t in range(left, right + 1):
        av, bv = tokens_a[t], tokens_b[t]
        mapping[av] = bv
        mapping[bv] = av
    placed = set(tokens_a[left:right + 1])
    outside = list(range(left)) + list(range(right + 1, length))
    unresolved: list[int] = []
    for idx in outside:
        token = tokens_b[idx]
        walked: set[tuple[int, int]] = set()
        while token in placed and token not in walked:
            walked.add(token)
            token = mapping[token]
        if token in placed:
            unresolved.append(idx)
            continue
        child[idx] = token
        placed.add(token)
    if unresolved:
        # 兜底：段内已放置数量扣减后，按 b 相对顺序补齐未解析位置
        remaining_counts: dict[int, int] = {}
        for token in tokens_a:
            remaining_counts[token[0]] = remaining_counts.get(token[0], 0) + 1
        for token in placed:
            remaining_counts[token[0]] -= 1
        fill: list[tuple[int, int]] = []
        for token in tokens_b:
            if remaining_counts.get(token[0], 0) > 0:
                fill.append(token)
                remaining_counts[token[0]] -= 1
        for idx, token in zip(unresolved, fill):
            child[idx] = token
            placed.add(token)
    # 修复：同 job 的标记按 occurrence 升序重排（保证工序沿工艺路线保序）
    positions_by_job: dict[int, list[int]] = {}
    for idx, token in enumerate(child):
        if token is not None:
            positions_by_job.setdefault(token[0], []).append(idx)
    out: list[int] = [0] * length
    for job_id, positions in positions_by_job.items():
        positions.sort(key=lambda p: child[p][1])
        for pos in positions:
            out[pos] = job_id
    return tuple(out)


def _swap_mutation(sequence: tuple[int, ...], rng: Random) -> tuple[int, ...]:
    """工序交换变异：交换两个不同作业的工序位置（同单内不交换，保持保序可行）。"""
    length = len(sequence)
    if length < 2:
        return sequence
    for _ in range(4):  # 尝试数次避免同作业位置
        first, second = rng.sample(range(length), 2)
        if sequence[first] != sequence[second]:
            seq = list(sequence)
            seq[first], seq[second] = seq[second], seq[first]
            return tuple(seq)
    return sequence


def _resource_migration(
    lines: tuple[int, ...], jobs: list[_Job], rng: Random,
) -> tuple[int, ...]:
    """资源迁移变异：随机把某订单切到另一条备选产线。"""
    n = len(jobs)
    if n == 0:
        return lines
    gene = rng.randrange(n)
    alternatives = [option for option in jobs[gene].line_options if option != lines[gene]]
    if not alternatives:
        return lines
    new_lines = list(lines)
    new_lines[gene] = rng.choice(alternatives)
    return tuple(new_lines)


def _crossover(
    parent_a: _Individual, parent_b: _Individual,
    jobs: list[_Job], cfg: dict[str, Any], rng: Random,
) -> _Individual:
    """顺序/部分映射交叉 + 产线均匀交叉，输出保持可行性的子代。"""
    seq_a, seq_b = parent_a.sequence, parent_b.sequence
    if rng.random() < float(cfg["crossoverRate"]):
        if str(cfg.get("crossover", "OX")).upper() == "PMX":
            sequence = _partially_mapped_crossover(seq_a, seq_b, rng)
        else:
            sequence = _order_crossover(seq_a, seq_b, rng)
    else:
        sequence = seq_a
    lines = _repair_lines(
        tuple(
            parent_a.lines[idx] if rng.random() < 0.5 else parent_b.lines[idx]
            for idx in range(len(jobs))
        ),
        jobs,
    )
    individual = _Individual(sequence, lines)
    if rng.random() < float(cfg["mutationRate"]):
        individual = _Individual(_swap_mutation(individual.sequence, rng), individual.lines)
    if rng.random() < float(cfg["lineMutationRate"]):
        individual = _Individual(individual.sequence, _resource_migration(individual.lines, jobs, rng))
    return individual


# ---------------- 可行性修复 / 种子 ----------------
def _base_individual(jobs: list[_Job]) -> _Individual:
    """基准个体：按订单序展开工序序列 + 每订单首选产线（确定性起点）。"""
    sequence = tuple(j for j in range(len(jobs)) for _ in range(jobs[j].op_count))
    lines = tuple(jobs[j].line_options[0] for j in range(len(jobs)))
    return _Individual(sequence, lines)


def _repair_lines(lines: tuple[int, ...], jobs: list[_Job]) -> tuple[int, ...]:
    """把产线向量修复为合法备选产线（越界回退首选；按 job id 索引）。"""
    return tuple(
        line if line in jobs[idx].line_options else jobs[idx].line_options[0]
        for idx, line in enumerate(lines)
    )


def _job_order_to_sequence(order: tuple[int, ...], jobs: list[_Job]) -> tuple[int, ...]:
    """订单级排列（每订单一次，GA 口径）展开为工序级染色体。"""
    return tuple(j for j in order for _ in range(jobs[j].op_count))


def _expand_seed(
    job_order: tuple[int, ...], lines: tuple[int, ...], jobs: list[_Job],
) -> _Individual:
    """把外部种子（GA 等订单级排列 + 每订单产线）修复为合法工序级个体。"""
    n = len(jobs)
    seen: set[int] = set()
    unique_order: list[int] = []
    for j in job_order:
        if 0 <= j < n and j not in seen:
            seen.add(j)
            unique_order.append(j)
    for j in range(n):
        if j not in seen:
            unique_order.append(j)
    position = {j: idx for idx, j in enumerate(unique_order)}
    lines_r: list[int] = []
    for j in range(n):
        source = lines[position[j]] if position[j] < len(lines) else None
        lines_r.append(
            source if source in jobs[j].line_options else jobs[j].line_options[0]
        )
    return _Individual(_job_order_to_sequence(tuple(unique_order), jobs), tuple(lines_r))


# ---------------- 选择算子 ----------------
def _tournament(
    population: list[_Individual],
    ranks: list[int],
    crowding: dict[int, float],
    rng: Random,
    size: int,
) -> _Individual:
    """锦标赛选择：rank 小优先，同 rank 拥挤度大优先。"""
    best: int | None = None
    for _ in range(max(2, size)):
        idx = rng.randrange(len(population))
        if best is None or (
            ranks[idx] < ranks[best]
            or (ranks[idx] == ranks[best] and crowding.get(idx, 0.0) > crowding.get(best, 0.0))
        ):
            best = idx
    return population[best]


def _select_next_population(
    fronts: list[list[int]],
    crowding: dict[int, float],
    size: int,
) -> list[int]:
    """NSGA-II 环境选择：按前沿优先填充，最后一前沿按拥挤度降序截断。"""
    selected: list[int] = []
    for front in fronts:
        if len(selected) + len(front) <= size:
            selected.extend(front)
        else:
            need = size - len(selected)
            ordered = sorted(front, key=lambda idx: crowding.get(idx, 0.0), reverse=True)
            selected.extend(ordered[:need])
            break
    return selected


def _reserve_elites(
    selected: list[int],
    combined: list[_Individual],
    ranks: list[int],
    crowding: dict[int, float],
    elites: list[_Individual],
) -> list[int]:
    """精英保留：确保注入的基线/种子个体始终留在种群（替代最差占位）。

    这是 NSGA-II 精英策略的补充保证：外部基线（如单目标 GA 解）一旦注入，
    要么留在最终前沿上，要么被更优前沿成员支配——从而保证
    「Pareto 前沿存在成员在两个目标上均不劣于基线」的边界断言。
    """
    occupied = set(selected)
    replaced: set[int] = set()
    elite_genotypes = {(elite.sequence, elite.lines) for elite in elites}
    for elite in elites:
        present = any(
            combined[idx].sequence == elite.sequence and combined[idx].lines == elite.lines
            for idx in selected
        )
        if present:
            continue
        candidates = [
            idx for idx in selected
            if idx not in replaced
            and (combined[idx].sequence, combined[idx].lines) not in elite_genotypes
        ]
        if not candidates:
            break
        worst = max(candidates, key=lambda idx: (ranks[idx], -crowding.get(idx, 0.0)))
        elite_idx = next(
            (idx for idx, ind in enumerate(combined)
             if ind.sequence == elite.sequence and ind.lines == elite.lines),
            None,
        )
        if elite_idx is None or elite_idx in occupied:
            continue
        selected.remove(worst)
        selected.append(elite_idx)
        replaced.add(worst)
        occupied.add(elite_idx)
    return selected


# ---------------- 主求解器 ----------------
def solve_nsga2(
    world: World,
    params: EngineParams,
    *,
    entries: list[dict] | None = None,
    nsga2_params: dict[str, Any] | None = None,
    seed: int | None = None,
    seed_solutions: list[tuple[tuple[int, ...], tuple[int, ...]]] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """NSGA-II 多目标求解入口（独立函数，注册表 entrypoint）。

    Args:
        world:  世界状态字典。
        params: EngineParams（engineType 可为 NSGA2/GA；timeLimitSeconds 生效）。
        entries: 待排条目（[{so, item}]，None=按 RuleEngine.collect_and_order 收集）。
        nsga2_params: 覆盖 nsga2_defaults 的参数模板（种群/代数/交叉率/变异率等）。
        seed:   确定性种子（覆盖模板种子）。
        seed_solutions: 外部种子解 [(job_order, lines)]（如单目标 GA 基线），
                       精英保留保证前沿存在不劣于它的成员。
    Returns:
        (pareto_solutions, meta)：Pareto 前沿解卡列表 + 求解元信息。
    """
    started = monotonic()
    if entries is None:
        entries, _, _ = RuleEngine().collect_and_order(world, params)
    jobs = _build_jobs(world, entries, params)
    count = len(jobs)
    cfg = {**nsga2_defaults, **(nsga2_params or {})}
    if seed is not None:
        cfg["seed"] = int(seed)
    rng = Random(int(cfg["seed"]))
    pop_size = max(4, int(cfg["populationSize"]))
    max_generations = max(1, int(cfg["generations"]))
    time_limit = max(
        0.05,
        float(params.timeLimitSeconds or cfg.get("timeLimitSeconds") or 3.0),
    )
    generation_budget = max(1, min(max_generations, round(time_limit * 100)))

    meta: dict[str, Any] = {
        "backend": "NSGA-II",
        "engineType": "NSGA2",
        "pipeline": "NSGA2->shift-slot",
        "objectives": ["totalTardiness", "conflictCount", "loadBalance"],
        "crossover": str(cfg.get("crossover", "OX")).upper(),
        "seed": int(cfg["seed"]),
        "population": pop_size,
        "generationBudget": generation_budget,
    }
    if count == 0:
        meta.update({
            "status": "TRIVIAL", "wallTimeSec": round(monotonic() - started, 4),
            "generations": 0, "paretoSize": 0,
        })
        return [], meta

    ws_avail = _ws_availability(world, params, jobs)
    base = _base_individual(jobs)
    elites: list[_Individual] = [base]
    for job_order, lines in seed_solutions or []:
        individual = _expand_seed(tuple(job_order), tuple(lines), jobs)
        if all(
            (individual.sequence, individual.lines) != (e.sequence, e.lines)
            for e in elites
        ):
            elites.append(individual)

    population: list[_Individual] = []
    seen_genotypes: set[tuple[tuple[int, ...], tuple[int, ...]]] = set()
    for elite in elites:
        population.append(elite)
        seen_genotypes.add((elite.sequence, elite.lines))
    while len(population) < pop_size:
        order = list(range(count))
        rng.shuffle(order)
        lines = tuple(rng.choice(jobs[j].line_options) for j in range(count))
        individual = _Individual(_job_order_to_sequence(tuple(order), jobs), lines)
        genotype = (individual.sequence, individual.lines)
        if genotype not in seen_genotypes:
            population.append(individual)
            seen_genotypes.add(genotype)

    objectives = [_evaluate(ind, jobs, ws_avail) for ind in population]
    base_objs = _evaluate(base, jobs, ws_avail)
    seed_objs = [_evaluate(ind, jobs, ws_avail) for ind in elites[1:]]

    completed_generations = 0
    for generation in range(generation_budget):
        _, ranks, crowding = _assign_ranks_crowding(objectives)
        offspring: list[_Individual] = []
        while len(offspring) < pop_size:
            first = _tournament(population, ranks, crowding, rng, int(cfg["tournamentSize"]))
            second = _tournament(population, ranks, crowding, rng, int(cfg["tournamentSize"]))
            offspring.append(_crossover(first, second, jobs, cfg, rng))
        combined = population + offspring
        combined_objs = objectives + [_evaluate(ind, jobs, ws_avail) for ind in offspring]
        fronts_c, ranks_c, crowding_c = _assign_ranks_crowding(combined_objs)
        selected = _select_next_population(fronts_c, crowding_c, pop_size)
        selected = _reserve_elites(selected, combined, ranks_c, crowding_c, elites)
        population = [combined[idx] for idx in selected]
        objectives = [combined_objs[idx] for idx in selected]
        completed_generations = generation + 1
        # 墙钟兜底仅在大预算生效：小预算（<=200 代 ≈ <=2s）按确定性代数跑满，
        # 与 GA 引擎口径一致（时间预算 → 固定代数，可复现）；大预算保留墙钟保护。
        if generation_budget > 200 and monotonic() - started > time_limit:
            break

    final_front = fast_non_dominated_sort(objectives)[0]
    pareto: list[dict[str, Any]] = []
    for rank, idx in enumerate(final_front):
        individual = population[idx]
        objs = objectives[idx]
        detail = _decode_detail(individual, jobs, ws_avail)
        tardiness, conflicts, neg_balance = objs
        pareto.append({
            "solutionId": f"nsga2-{completed_generations:03d}-{rank:02d}",
            "strategy": "NSGA2",
            "objectives": {
                "totalTardiness": round(tardiness, 2),
                "conflictCount": int(conflicts),
                "loadBalance": round(-neg_balance, 6),
            },
            "kpi": {
                "totalTardiness": round(tardiness, 2),
                "conflictCount": int(conflicts),
                "totalCost": 0.0,
                "avgUtilization": round(detail["utilization"], 6),
                "loadBalance": round(-neg_balance, 6),
                "totalChangeoverMin": 0.0,
            },
            "hardFeasible": True,
            "robustness": 0.5,
            "jobOrder": list(dict.fromkeys(individual.sequence)),
            "lines": list(individual.lines),
            "jobEndMinutes": detail["jobEndMinutes"],
            "conflictMinutes": round(detail["conflictMinutes"], 2),
        })

    meta.update({
        "status": "FEASIBLE",
        "wallTimeSec": round(monotonic() - started, 4),
        "timeLimitSec": time_limit,
        "generations": completed_generations,
        "paretoSize": len(pareto),
        "baselineObjectives": {
            "totalTardiness": round(base_objs[0], 2),
            "conflictCount": int(base_objs[1]),
            "loadBalance": round(-base_objs[2], 6),
        },
        "seedObjectives": [
            {
                "totalTardiness": round(objs[0], 2),
                "conflictCount": int(objs[1]),
                "loadBalance": round(-objs[2], 6),
            }
            for objs in seed_objs
        ],
        "params": dict(cfg),
    })
    return pareto, meta


# ---------------- 引擎集成 ----------------
class NSGA2Engine(RuleEngine):
    """SC-03 增补：NSGA-II 多目标排产引擎（矩阵 85）。

    复用 RuleEngine 物化（班次占槽/PO/WO/冲突/KPI），解选择 = Pareto 前沿
    经 scenario_selection.rank_scenarios 加权推荐。ScheduleResult 契约
    engineType Literal 无 NSGA2，本引擎如实复用 GA 槽位并标注 solverMeta。
    """

    name = "NSGA2"
    supports_anytime = True

    def __init__(self) -> None:
        super().__init__(requested_type="GA")

    def solve(
        self,
        world: World,
        params: EngineParams,
        next_id: Callable[[str], int],
    ) -> ScheduleResult:
        entries, campaign_meta, source_count = self.collect_and_order(world, params)
        pareto, solver_meta = solve_nsga2(world, params, entries=entries)
        solver_meta = dict(solver_meta)
        solver_meta["paretoKpis"] = [
            {key: solution["kpi"][key] for key in (
                "totalTardiness", "conflictCount", "avgUtilization", "loadBalance",
            )}
            for solution in pareto
        ]
        if not pareto:
            solver_meta["status"] = "TRIVIAL"
            return self.materialize_schedule(
                world, params, next_id, entries, campaign_meta, source_count,
                solver_meta=solver_meta,
            )
        # 通过 scenario_selection 的 rank_scenarios 从 Pareto 前沿选推荐解（消费衔接）
        from server.aps_domain.scenario_selection import (
            nsga2_solutions_to_cards,
            rank_scenarios,
        )
        cards = rank_scenarios(nsga2_solutions_to_cards(pareto))
        recommended = next(
            (card for card in cards if card.get("isRecommended")),
            cards[0],
        )
        source = recommended.get("source") or {}
        job_order = source.get("order") or list(range(len(entries)))
        lines = source.get("lines") or []
        ordered: list[dict] = []
        for entry_idx in job_order:
            entry = dict(entries[entry_idx])
            if entry_idx < len(lines) and int(lines[entry_idx]) >= 0:
                entry["forcedLineId"] = int(lines[entry_idx])
            ordered.append(entry)
        return self.materialize_schedule(
            world,
            params,
            next_id,
            ordered,
            campaign_meta,
            source_count,
            solver_meta=solver_meta,
        )