diff --git a/docs/round-89-optimize-cpsat-native-plan.md b/docs/round-89-optimize-cpsat-native-plan.md
new file mode 100644
index 0000000..edab887
--- /dev/null
+++ b/docs/round-89-optimize-cpsat-native-plan.md
@@ -0,0 +1,74 @@
+# Round 89: optimize 原生 CP-SAT 求解器接入
+
+更新时间：2026-09-16
+
+## 1. 本轮目标
+
+把 `OptimizeEngine` 从「选规则 + 复用 PoolEngine 物化」推进到真正的 V2 原生求解：
+输入 `SchedulingProblemV2`，由 OR-Tools CP-SAT 决定资源分配与时序，输出
+`SchedulingSolutionV2`，并且必须通过 APS 独立的 `SchedulingValidator`。
+
+上游依据：`docs/round-85-optimize-integration-shape-plan.md` 第 14 节明确记着
+「CP-SAT 的 V2 原生求解器仍应作为后续轮次接入，本轮不把 PoolEngine 适配器冒充为
+CP-SAT 最优证明」。本轮就是那一轮。
+
+## 2. 现状（本轮开始时的事实）
+
+- `OptimizeEngine.solve_flex()` 只做算法选择 + provenance 标注，实际物化交给
+  `PoolEngine`（贪心占槽）。七种派工规则已跑通，但它们不是 CP-SAT。
+- `server/engines/cp_engine.py` 是**经典轨**（operations/routings/workstations）
+  的 CP-SAT，不是 flex 闭环 V2 原生；`solver_process.py` 是它的子进程安全边界。
+- 集成环境限制：项目 `.venv` 里的 NumPy 以 X86_V2 为基线构建，而本机 CPU
+  （sd-server，Family 6 Model 15）不支持该指令集，`ortools.sat.python.cp_model`
+  导入即失败。这与 `round-85` 记录的全量 CP/Excel 测试受限是同一个原因。
+
+## 3. 本轮范围
+
+### Slice 1（本轮完成）
+
+- 新增 `server/engines/optimize_cpsat.py`：`SchedulingProblemV2 -> CpsatOutcome`，
+  内含解与可审计元数据（solverId/solverVersion/OR-Tools 状态/耗时/种子/规模）。
+- 模型：每道工序在合格设备中选一台（`AddExactlyOne` + 可选定长区间），同设备
+  `AddNoOverlap`，设备日历与维保之外的时间作为阻塞区间一并进入 NoOverlap，工序链按
+  `predecessorActivityIds` 串行，目标为最小化总拖期。
+- provenance 与 problem hash 绑定，与 `flex_version_to_solution_v2` 同一套口径。
+- 新增 `tests/golden/test_optimize_cpsat_native.py`：解通过独立 V2 校验；总拖期不劣于
+  同口径 EDD 基线。
+
+### Slice 2（下一轮）
+
+- `OptimizeEngine.solve_flex` 增加 CP-SAT 算法路径（`algorithmId=optimize.cpsat`），
+  物化到 flex* 行，落版本、证据链，走 `/api/flex/schedule` 端到端。
+- 目标函数与 PoolEngine `totalTardiness` 口径对齐（当前两者定义不同，不能直接比较）。
+- 时间离散化（分钟 -> 5/15 分钟桶）、派工解热启动、缩短求解时间。
+- 冻结/在制/模具寿命/班组与工装累计容量接入模型。
+
+### 明确不做
+
+- 不改 WorldStore 权威语义、审批门禁和审计链。
+- 不新建第二套订单/资源/版本模型。
+- 不声称 PoolEngine 适配器结果是 CP-SAT 最优证明。
+
+## 4. 验收
+
+- `pytest tests/golden/test_optimize_cpsat_native.py`：在具备可用 OR-Tools 的环境
+  通过；在 NumPy/OR-Tools 不可用的环境按既有策略 skip，不得 error。
+- 解必须 `validate_solution(...).valid is True` 且无 hard violation。
+- 现有回归不受影响：`tests/golden/test_optimize_simulation_world_pack.py`、
+  `tests/golden/test_optimize_engine.py`。
+- 报告必须区分证据等级：代码、测试、真实运行；环境性跳过要写清原因。
+
+## 5. 停止条件
+
+- V2 契约无法表达所需约束，且无法通过局部兼容字段解决。
+- 需要放宽校验才能让测试通过。
+- 需要新增生产依赖或许可而没有明确运行环境。
+
+## 6. Slice 1 证据（2026-09-16）
+
+- 隔离解释器（CPython 3.11.15 + NumPy 1.26.4 + OR-Tools 9.11.4210）：
+  `2 passed`；模拟数据包 72/72 工序全部排入、`validate_solution` valid、
+  总拖期 1,782,876 分钟不劣于同口径 EDD 基线。
+- 项目 `.venv`：`1 skipped`（OR-Tools 因 NumPy 基线与 CPU 不兼容不可用）。
+- 已知限制：`INFEASIBLE` 曾因拖期变量上界未包含「历史欠交」而误判，已修（
+  交期早于计划起点的部分必须计入上界）；当前只做到 FEASIBLE，未证明最优。
diff --git a/server/engines/optimize_cpsat.py b/server/engines/optimize_cpsat.py
new file mode 100644
index 0000000..3628714
--- /dev/null
+++ b/server/engines/optimize_cpsat.py
@@ -0,0 +1,369 @@
+"""V2-native CP-SAT solver for the APS closed loop (round 89, slice 1).
+
+输入是 APS 的 `SchedulingProblemV2`，输出是 `SchedulingSolutionV2`：资源分配和
+时序由 OR-Tools CP-SAT 决定，准入、校验、版本物化和审计仍然全部归 APS。
+
+本切片只做「问题 -> 解」的原生求解：不写 flex* 行、不物化版本，也不改
+WorldStore。候选一旦物化，仍必须经过 `validate_solution` 才会成为 APS 版本。
+"""
+from __future__ import annotations
+
+import math
+import time
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+from typing import Any
+
+from server.aps_domain.scheduling_problem_v2 import (
+    PeggingAllocation,
+    ResourceKind,
+    ScheduledActivity,
+    ScheduledResourceAllocation,
+    SchedulingProblemV2,
+    SchedulingSolutionV2,
+    SolveStatus,
+    SolutionProvenance,
+    UnscheduledRequirement,
+    scheduling_problem_hash,
+)
+
+SOLVER_ID = "optimize-cpsat"
+SOLVER_VERSION = "0.1.0"
+DEFAULT_TIME_LIMIT_SECONDS = 20.0
+DEFAULT_SEED = 42
+
+
+@dataclass
+class CpsatOutcome:
+    """一次 CP-SAT 求解的解与可审计元数据。"""
+
+    solution: SchedulingSolutionV2
+    meta: dict[str, Any] = field(default_factory=dict)
+
+
+def _minutes_between(base: datetime, moment: datetime | None) -> int | None:
+    """把时间点换算成相对基准的整数分钟（向下取整）。"""
+
+    if moment is None:
+        return None
+    return int(math.floor((moment - base).total_seconds() / 60.0))
+
+
+def _minutes_ceil(base: datetime, moment: datetime | None) -> int | None:
+    """把时间点换算成相对基准的整数分钟（向上取整，用于日历右端点）。"""
+
+    if moment is None:
+        return None
+    return int(math.ceil((moment - base).total_seconds() / 60.0))
+
+
+def _merge_windows(windows: list[tuple[int, int]]) -> list[tuple[int, int]]:
+    merged: list[tuple[int, int]] = []
+    for start, end in sorted(windows):
+        if end <= start:
+            continue
+        if merged and start <= merged[-1][1]:
+            merged[-1] = (merged[-1][0], max(merged[-1][1], end))
+        else:
+            merged.append((start, end))
+    return merged
+
+
+def _blocked_windows(resource: Any, base: datetime, horizon: int) -> list[tuple[int, int]]:
+    """资源在 [0, horizon] 内不可排的分钟区间 = 日历与维保之外的补集。"""
+
+    open_windows: list[tuple[int, int]] = []
+    for interval in resource.calendarIntervals:
+        start = _minutes_between(base, interval.start)
+        end = _minutes_ceil(base, interval.end)
+        if start is None or end is None:
+            continue
+        start = max(0, start)
+        end = min(horizon, end)
+        if end <= start:
+            continue
+        open_windows.append((start, end))
+
+    maintenance: list[tuple[int, int]] = []
+    for interval in resource.maintenanceIntervals:
+        start = _minutes_between(base, interval.start)
+        end = _minutes_ceil(base, interval.end)
+        if start is None or end is None:
+            continue
+        start, end = max(0, start), min(horizon, end)
+        if end > start:
+            maintenance.append((start, end))
+
+    merged_open = _merge_windows(open_windows)
+    blocked: list[tuple[int, int]] = []
+    cursor = 0
+    for start, end in merged_open:
+        if start > cursor:
+            blocked.append((cursor, start))
+        cursor = max(cursor, end)
+    if cursor < horizon:
+        blocked.append((cursor, horizon))
+    return _merge_windows([*blocked, *maintenance])
+
+
+def _duration_minutes(value: float) -> int:
+    return max(1, int(math.ceil(float(value))))
+
+
+def _eligible_resources(activity: Any, resource_ids: set[str]) -> list[str]:
+    candidates: list[str] = []
+    for requirement in activity.resourceRequirements:
+        if requirement.kind != ResourceKind.EQUIPMENT:
+            continue
+        candidates.extend(requirement.eligibleResourceIds)
+    if not candidates:
+        candidates.extend(activity.eligibleResourceIds)
+    unique = [rid for rid in dict.fromkeys(candidates) if rid in resource_ids]
+    return unique
+
+
+def solve_problem_v2(
+    problem: SchedulingProblemV2,
+    *,
+    time_limit_seconds: float = DEFAULT_TIME_LIMIT_SECONDS,
+    seed: int = DEFAULT_SEED,
+    horizon_minutes: int | None = None,
+) -> CpsatOutcome:
+    """用 CP-SAT 求一个 `SchedulingProblemV2` 候选解。
+
+    目标：最小化总拖期（`objectivePolicy` 里 tardiness 权重）。资源分配为每个工序
+    在合格设备中选一台，同设备工序不重叠，工序链按前驱顺序串行，并且不允许落在
+    设备日历与维保之外。
+    """
+
+    from ortools.sat.python import cp_model
+
+    started = time.perf_counter()
+    base = problem.planningStart
+    span = int(math.floor((problem.planningEnd - base).total_seconds() / 60.0))
+    horizon = span if horizon_minutes is None else min(span, int(horizon_minutes))
+    if horizon <= 0:
+        raise ValueError("planning window must be positive")
+
+    resources_by_id = {resource.resourceId: resource for resource in problem.resources}
+    equipment_resources = [r for r in problem.resources if r.kind == ResourceKind.EQUIPMENT]
+    equipment_ids = {r.resourceId for r in equipment_resources}
+
+    model = cp_model.CpModel()
+    start_vars: dict[str, Any] = {}
+    end_vars: dict[str, Any] = {}
+    presence: dict[tuple[str, str], Any] = {}
+    intervals_by_resource: dict[str, list[Any]] = {r.resourceId: [] for r in equipment_resources}
+    unschedulable: list[str] = []
+
+    for activity in problem.activities:
+        duration = _duration_minutes(activity.durationMin)
+        eligible = _eligible_resources(activity, equipment_ids)
+        if not eligible or duration > horizon:
+            unschedulable.append(activity.activityId)
+            continue
+        release = int(max(0, _minutes_between(base, activity.materialReleaseAt) or 0))
+        if release + duration > horizon:
+            unschedulable.append(activity.activityId)
+            continue
+        start = model.NewIntVar(release, horizon - duration, f"start:{activity.activityId}")
+        end = model.NewIntVar(release + duration, horizon, f"end:{activity.activityId}")
+        model.Add(end == start + duration)
+        start_vars[activity.activityId] = start
+        end_vars[activity.activityId] = end
+
+        picks = []
+        for resource_id in eligible:
+            chosen = model.NewBoolVar(f"pick:{activity.activityId}:{resource_id}")
+            presence[(activity.activityId, resource_id)] = chosen
+            picks.append(chosen)
+            intervals_by_resource[resource_id].append(
+                model.NewOptionalFixedSizeIntervalVar(
+                    start, duration, chosen, f"interval:{activity.activityId}:{resource_id}"
+                )
+            )
+        model.AddExactlyOne(picks)
+
+    for activity in problem.activities:
+        target = start_vars.get(activity.activityId)
+        if target is None:
+            continue
+        for predecessor_id in activity.predecessorActivityIds:
+            predecessor_end = end_vars.get(predecessor_id)
+            if predecessor_end is not None:
+                model.Add(target >= predecessor_end)
+
+    blocked_total = 0
+    for resource in equipment_resources:
+        blocked = _blocked_windows(resource, base, horizon)
+        blocked_total += len(blocked)
+        for index, (start, end) in enumerate(blocked):
+            intervals_by_resource[resource.resourceId].append(
+                model.NewFixedSizeIntervalVar(start, end - start, f"closed:{resource.resourceId}:{index}")
+            )
+        if intervals_by_resource[resource.resourceId]:
+            model.AddNoOverlap(intervals_by_resource[resource.resourceId])
+
+    activities_by_requirement: dict[str, list[str]] = {}
+    for activity in problem.activities:
+        activities_by_requirement.setdefault(activity.requirementId, []).append(activity.activityId)
+
+    tardiness_weight = float((problem.objectivePolicy.weights or {}).get("tardiness", 1.0) or 1.0)
+    tardiness_terms = []
+    capacity = sum(_duration_minutes(a.durationMin) for a in problem.activities) or 1
+    due_offsets = [
+        offset
+        for offset in (_minutes_between(base, requirement.requiredAt) for requirement in problem.requirements)
+        if offset is not None
+    ]
+    # 交期可能早于计划起点（历史欠交），拖期上界必须把这段「已经迟到」的量算进去，
+    # 否则 tardy 变量的域会把模型判成不可行。
+    already_late = max(0, -(min(due_offsets) if due_offsets else 0))
+    tardy_upper = horizon + already_late + capacity + 1
+    for requirement in problem.requirements:
+        activity_ids = activities_by_requirement.get(requirement.requirementId) or []
+        if not activity_ids:
+            continue
+        ends = [end_vars[aid] for aid in activity_ids if aid in end_vars]
+        if not ends:
+            continue
+        completion = model.NewIntVar(0, horizon, f"completion:{requirement.requirementId}")
+        model.AddMaxEquality(completion, ends)
+        due = _minutes_between(base, requirement.requiredAt)
+        if due is None:
+            continue
+        tardy = model.NewIntVar(0, tardy_upper, f"tardy:{requirement.requirementId}")
+        model.Add(tardy >= completion - due)
+        tardiness_terms.append((requirement.requirementId, tardy))
+
+    if tardiness_terms:
+        model.Minimize(
+            sum(int(round(tardiness_weight * 1000)) * term for _, term in tardiness_terms)
+        )
+
+    solver = cp_model.CpSolver()
+    solver.parameters.max_time_in_seconds = float(time_limit_seconds)
+    solver.parameters.random_seed = int(seed)
+    solver.parameters.num_search_workers = 1  # 单线程保证可复现
+    status = solver.Solve(model)
+    elapsed = time.perf_counter() - started
+
+    status_name = solver.StatusName(status)
+    solve_status = {
+        cp_model.OPTIMAL: SolveStatus.OPTIMAL,
+        cp_model.FEASIBLE: SolveStatus.FEASIBLE,
+    }.get(status, SolveStatus.INFEASIBLE if status == cp_model.INFEASIBLE else SolveStatus.ERROR)
+
+    scheduled: list[ScheduledActivity] = []
+    if status in (cp_model.OPTIMAL, cp_model.FEASIBLE):
+        for activity in problem.activities:
+            if activity.activityId not in start_vars:
+                continue
+            start_minute = int(solver.Value(start_vars[activity.activityId]))
+            chosen_resource = None
+            for resource_id in _eligible_resources(activity, equipment_ids):
+                pick = presence.get((activity.activityId, resource_id))
+                if pick is not None and solver.Value(pick):
+                    chosen_resource = resource_id
+                    break
+            if chosen_resource is None:
+                continue
+            end_minute = start_minute + _duration_minutes(activity.durationMin)
+            units = float(activity.requiredResourceUnits or 1.0)
+            scheduled.append(
+                ScheduledActivity(
+                    activityId=activity.activityId,
+                    activityIdentity=activity.activityIdentity,
+                    requirementId=activity.requirementId,
+                    operationId=activity.operationId,
+                    sequence=activity.sequence,
+                    resourceId=chosen_resource,
+                    start=base + timedelta(minutes=start_minute),
+                    end=base + timedelta(minutes=end_minute),
+                    resourceUnits=units,
+                    resourceAllocations=(
+                        ScheduledResourceAllocation(
+                            resourceId=chosen_resource,
+                            kind=ResourceKind.EQUIPMENT,
+                            units=units,
+                        ),
+                    ),
+                )
+            )
+
+    scheduled_ids = {row.activityId for row in scheduled}
+    unscheduled_ids = set(unschedulable) | {
+        activity.activityId for activity in problem.activities if activity.activityId not in scheduled_ids
+    }
+    unscheduled = tuple(
+        UnscheduledRequirement(
+            requirementId=requirement.requirementId,
+            quantity=float(requirement.quantity),
+            reasonCode="UNSCHEDULED_ACTIVITY",
+            details="存在未落到候选解的制造活动",
+        )
+        for requirement in problem.requirements
+        if any(
+            activity_id in unscheduled_ids
+            for activity_id in activities_by_requirement.get(requirement.requirementId, [])
+        )
+    )
+
+    total_tardiness = 0.0
+    if tardiness_terms and status in (cp_model.OPTIMAL, cp_model.FEASIBLE):
+        total_tardiness = sum(float(solver.Value(term)) for _, term in tardiness_terms)
+
+    if solve_status in (SolveStatus.OPTIMAL, SolveStatus.FEASIBLE) and unscheduled:
+        solve_status = SolveStatus.PARTIAL
+
+    objective_values = {
+        "totalTardiness": total_tardiness,
+        "scheduledActivities": float(len(scheduled)),
+        "totalActivities": float(len(problem.activities)),
+    }
+    best_bound = None
+    objective_value = total_tardiness
+    gap = None
+    if status in (cp_model.OPTIMAL, cp_model.FEASIBLE) and tardiness_terms:
+        best_bound = float(solver.BestObjectiveBound()) / (tardiness_weight * 1000.0)
+        objective_value = float(solver.ObjectiveValue()) / (tardiness_weight * 1000.0)
+        if objective_value > 0:
+            gap = max(0.0, (objective_value - best_bound) / objective_value)
+
+    generated_at = datetime.now(tz=base.tzinfo or None)
+    solution = SchedulingSolutionV2(
+        problemId=problem.problemId,
+        solveStatus=solve_status,
+        objectiveValues=objective_values,
+        bestBound=best_bound,
+        gap=gap,
+        activities=tuple(scheduled),
+        pegging=(),
+        unscheduledRequirements=unscheduled,
+        provenance=SolutionProvenance(
+            runId=f"cpsat:{problem.problemId}:{int(started * 1000)}",
+            solverId=SOLVER_ID,
+            solverVersion=SOLVER_VERSION,
+            generatedAt=generated_at,
+            businessDate=problem.businessDate,
+            problemHash=scheduling_problem_hash(problem),
+            sourceRevision=problem.sourceRevision,
+            sourceFingerprints=problem.sourceFingerprints,
+        ),
+    )
+    meta = {
+        "solverId": SOLVER_ID,
+        "solverVersion": SOLVER_VERSION,
+        "ortoolsStatus": status_name,
+        "wallTimeSeconds": round(elapsed, 3),
+        "timeLimitSeconds": float(time_limit_seconds),
+        "seed": int(seed),
+        "horizonMinutes": horizon,
+        "activityCount": len(problem.activities),
+        "resourceCount": len(equipment_resources),
+        "blockedWindowCount": blocked_total,
+        "precedenceEdges": sum(len(a.predecessorActivityIds) for a in problem.activities),
+        "objective": "weightedTardiness",
+        "objectiveValue": objective_value,
+    }
+    return CpsatOutcome(solution=solution, meta=meta)
diff --git a/tests/golden/test_optimize_cpsat_native.py b/tests/golden/test_optimize_cpsat_native.py
new file mode 100644
index 0000000..ab9135a
--- /dev/null
+++ b/tests/golden/test_optimize_cpsat_native.py
@@ -0,0 +1,100 @@
+"""Round 89 slice 1: V2 原生 CP-SAT 求解器（SchedulingProblemV2 -> SchedulingSolutionV2）。
+
+本文件只验证「问题 -> 解」这一段：CP-SAT 自己决定资源分配和时序，解必须通过
+APS 的独立 V2 校验，且总拖期不劣于同口径的 EDD 基线。物化到 flex* 行仍由 APS
+既有通道负责（slice 2）。
+"""
+from __future__ import annotations
+
+import math
+from datetime import date, timedelta
+from pathlib import Path
+
+import pytest
+
+try:  # 环境缺 OR-Tools，或 venv 内 NumPy 基线与该机 CPU 不兼容（当前 sd-server 即如此）
+    from ortools.sat.python import cp_model  # noqa: F401
+except Exception as exc:  # pragma: no cover - 环境分支
+    pytest.skip(f"OR-Tools 在此环境不可用：{type(exc).__name__}: {exc}", allow_module_level=True)
+
+from server.aps_domain.closed_loop_problem import build_closed_loop_problem  # noqa: E402
+from server.aps_domain.closed_loop_runtime import closed_loop_to_problem_v2  # noqa: E402
+from server.aps_domain.scheduling_problem_v2 import SchedulingProblemV2, SolveStatus  # noqa: E402
+from server.aps_domain.scheduling_validator import validate_solution  # noqa: E402
+from server.engines.optimize_cpsat import solve_problem_v2  # noqa: E402
+from server.state.packs import load_pack  # noqa: E402
+
+PACK_PATH = Path(__file__).resolve().parents[2] / "server" / "data" / "packs" / "optimize-simulation-v1.json"
+
+
+def _pack_problem() -> tuple[dict, SchedulingProblemV2]:
+    world = load_pack(str(PACK_PATH))
+    closed_loop = build_closed_loop_problem(world, business_date=date.today().isoformat(), strict=True)
+    return world, closed_loop_to_problem_v2(world, closed_loop)
+
+
+def _edd_baseline_tardiness(problem: SchedulingProblemV2) -> float:
+    """同口径基线：完全按交期排序、每台设备串行占用的贪心解总拖期（分钟）。"""
+
+    base = problem.planningStart
+    order_index = {requirement.requirementId: idx for idx, requirement in enumerate(
+        sorted(problem.requirements, key=lambda row: row.requiredAt))}
+    activities_by_requirement: dict[str, list] = {}
+    for activity in problem.activities:
+        activities_by_requirement.setdefault(activity.requirementId, []).append(activity)
+
+    cursor: dict[str, timedelta] = {}
+    end_of: dict[str, timedelta] = {}
+    total = 0.0
+    for requirement_id in sorted(activities_by_requirement, key=lambda rid: order_index[rid]):
+        requirement = next(row for row in problem.requirements if row.requirementId == requirement_id)
+        for activity in sorted(activities_by_requirement[requirement_id], key=lambda row: row.sequence):
+            duration = timedelta(minutes=math.ceil(activity.durationMin))
+            earliest = max(
+                [timedelta(0)]
+                + [end_of[predecessor] for predecessor in activity.predecessorActivityIds if predecessor in end_of]
+            )
+            resource_id = sorted(activity.eligibleResourceIds)[0]
+            start = max(earliest, cursor.get(resource_id, timedelta(0)))
+            end = start + duration
+            cursor[resource_id] = end
+            end_of[activity.activityId] = end
+        completion = max(
+            (end_of[activity.activityId] for activity in activities_by_requirement[requirement_id]),
+            default=timedelta(0),
+        )
+        total += max(0.0, (base + completion - requirement.requiredAt).total_seconds() / 60.0)
+    return total
+
+
+def test_native_cpsat_schedules_every_pack_activity_and_passes_v2_validator():
+    world, problem = _pack_problem()
+
+    outcome = solve_problem_v2(problem, time_limit_seconds=15.0)
+    solution = outcome.solution
+
+    assert solution.solveStatus in (SolveStatus.OPTIMAL, SolveStatus.FEASIBLE)
+    assert len(solution.activities) == len(problem.activities) == 72
+    assert solution.unscheduledRequirements == ()
+    assert outcome.meta["solverId"] == "optimize-cpsat"
+
+    activity_by_id = {activity.activityId: activity for activity in problem.activities}
+    for row in solution.activities:
+        activity = activity_by_id[row.activityId]
+        assert row.resourceId in activity.eligibleResourceIds              # 只能用声明的合格设备
+        assert row.start >= problem.planningStart
+        span = (row.end - row.start).total_seconds() / 60.0
+        assert activity.durationMin <= span <= activity.durationMin + 1     # 工时按分钟向上取整
+
+    report = validate_solution(problem, solution, world=world)
+    assert report.valid is True
+    assert not report.hardViolations
+
+
+def test_native_cpsat_is_not_worse_than_the_edd_baseline_in_the_same_metric():
+    _, problem = _pack_problem()
+
+    outcome = solve_problem_v2(problem, time_limit_seconds=15.0)
+    baseline = _edd_baseline_tardiness(problem)
+
+    assert outcome.solution.objectiveValues["totalTardiness"] <= baseline + 0.001