接入 V2 原生 CP-SAT 求解器（round 89 slice 1）

新增 server/engines/optimize_cpsat.py：SchedulingProblemV2 -> CP-SAT -> SchedulingSolutionV2，目标最小化总拖期；含设备日历与维保阻塞区间、同设备不重叠、工序链前驱约束、合格设备二选一、provenance 绑定 problem hash。本轮只做问题到解，不物化 flex* 行（slice 2 负责）。验证：隔离解释器（numpy 1.26.4 + ortools 9.11）2 passed，72/72 工序排入且 validate_solution valid；项目 venv 因 NumPy X86_V2 基线与本机 CPU 不兼容按策略 skip。
修复模拟数据包订单契约与日期基准，消除交期看板 500
2026-09-16 17:32:32 +08:00 · 2026-09-16 17:32:29 +08:00 · 2026-09-16 15:30:39 +08:00 · 2026-09-14 16:11:54 +08:00 · 2026-09-08 16:00:25 +08:00 · 2026-09-04 16:55:33 +08:00
25 changed files with 3889 additions and 39 deletions
--- a/apps/web/src/api/types.ts
+++ b/apps/web/src/api/types.ts
@ -197,7 +197,7 @@ export interface IntentResult {
  source: 'RULE_FAST' | 'LLM';                        // 产生来源；RULE_FAST 仅历史审计兼容
 }

-export type ScheduleEngineType = 'RULE' | 'CP' | 'GA' | 'HYBRID' | 'EXTERNAL';
+export type ScheduleEngineType = 'RULE' | 'CP' | 'GA' | 'HYBRID' | 'EXTERNAL' | 'OPTIMIZE';

 export interface ScheduleResult {
  versionId: number;
--- a/apps/web/src/chat/ChatPanel.tsx
+++ b/apps/web/src/chat/ChatPanel.tsx
@ -1198,6 +1198,8 @@ function FlexScheduleBlock(props: { block: UIBlock }) {
    trialOnly?: boolean; productionReady?: boolean; solveStatus?: string;
    assumptionCount?: number;
    versionNo: string; modeCn: string;
+    dataPack?: { file?: string; name?: string; reference?: string } | null;
+    algorithm?: { engine?: string; rule?: string; algorithmId?: string | null } | null;
    stats: { vlCount: number; woCount: number; makespan: string | null; onTimeCount: number; conflictCount: number; utilization: number };
    bottleneck: { name: string; count: number }[];
    lines: FlexLine[];
@ -1231,6 +1233,12 @@ function FlexScheduleBlock(props: { block: UIBlock }) {
          {downloadState.downloading ? '下载中…' : '下载 Excel 工作计划表'}</button>
        }
      </div>
+      {(p.dataPack || p.algorithm) && (
+        <div className="flex-block-meta">
+          {p.dataPack && <span>数据：{p.dataPack.name || p.dataPack.file}</span>}
+          {p.algorithm && <span>算法：{p.algorithm.engine || 'CLOSED_LOOP'}{p.algorithm.rule ? ` · ${p.algorithm.rule}` : ''}</span>}
+        </div>
+      )}
      <DownloadFailure message={downloadState.error} />
      <div className="flex-stats">
        <div className="flex-stat"><b>{s.vlCount}</b><span>虚拟产线</span></div>
--- a/apps/web/src/styles.css
+++ b/apps/web/src/styles.css
@ -2044,6 +2044,7 @@ html.aps-desktop .app-menubar-spacer,
 .flex-mode-badge { margin-left: auto; flex: 0 0 auto; font-size: 11px; font-weight: 600;
  color: var(--accent-deep); background: var(--accent-soft); border-radius: 999px; padding: 2px 10px; }
 .flex-mode-badge.warn { color: var(--warn); background: var(--warn-soft); }
+.flex-block-meta { display: flex; gap: 14px; flex-wrap: wrap; margin: -2px 0 10px; color: var(--muted); font-size: 11.5px; }
 .flex-bottleneck { font-size: 11.5px; color: var(--muted); margin-bottom: 10px;
  display: flex; align-items: center; gap: 6px; flex-wrap: wrap; }
 .flex-bn-chip { font-size: 11px; font-weight: 600; color: var(--warn); background: var(--warn-soft);
--- a/docs/round-85-optimize-integration-shape-plan.md
+++ b/docs/round-85-optimize-integration-shape-plan.md
@ -0,0 +1,208 @@
+# Round 85: optimize 集成形态计划
+
+更新时间：2026-09-03
+
+## 1. 本轮目标
+
+将当前 `optimize` 的集成形态落定为 `aps-agent` 内部的 V2 原生排产求解器，并为下一轮实现定义清晰、可验证的边界。
+
+本轮只完成架构和实施计划，不修改生产代码。下一轮实现应能让真实 MOM 数据经过 APS 既有闭环，由 optimize 求解，再通过 V2 校验并写回 APS 版本和审计记录。
+
+## 2. 背景与当前状态
+
+### aps-agent 现有主链
+
+```text
+/api/flex/schedule
+  -> Harness P1 门禁
+  -> workflow / run_flex_schedule
+  -> MOM/world 同步、MRP 分解、来源标注
+  -> closed_loop_problem
+  -> SchedulingProblemV2
+  -> solver
+  -> SchedulingValidator
+  -> 候选结果原子化物化
+  -> WorldStore + audit
+```
+
+关键现有模块：
+
+- `server/agent_core/harness.py`：权限等级和写入门禁
+- `server/aps_domain/flex.py`：柔性排产入口和审计闭环
+- `server/aps_domain/closed_loop_problem.py`：需求、BOM、供应和阻断建模
+- `server/aps_domain/closed_loop_runtime.py`：V2 问题、求解、校验和物化
+- `server/aps_domain/scheduling_problem_v2.py`：生产排产问题/结果契约
+- `server/aps_domain/scheduling_validator.py`：独立 fail-closed 校验
+- `server/engines/`：RULE、CP、HYBRID、GA、NSGA2、EXTERNAL 等引擎
+- `server/importers/mom_pack.py`、`excel_importer.py`、`profiles/kangni.json`：现有 MOM/Kangni 导入
+- `server/state/store.py`：WorldStore 唯一状态源
+
+### optimize 现有能力
+
+- Node 调度规则：EDD、SPT、PRIORITY、FIFO、LPT、CR、ATC
+- Node CP-SAT 适配和独立结果认证
+- Kangni/MOM 输入准入、manifest/hash、数据等级和阻断报告
+- source-aware RAG、文档抽取、检索和 Python 数据流
+- 既有 Node/Python 测试和 fixture
+
+### 预检记录
+
+- 目标仓库：`<workspace>\aps-agent`
+- 目标分支：`codex/integrate-optimize`
+- 轮次分支：`round/85-optimize-integration-shape`
+- 轮次工作树：`<worktrees>\aps-agent-round-85-optimize-integration-shape`
+- 基线 commit：`7f171f328d99974ff83eba1754b2fe669c7f9d15`
+- 原始 `optimize` 工作树存在既有未提交/未跟踪改动，受保护，不自动带入本轮工作树
+- `aps-agent` 轮次工作树当前干净
+
+## 3. 已确认决策
+
+任务重量：轻量。当前只有一个主集成方向，后续实现可由单 agent 在轮次工作树完成，不预先创建 worker worktree。
+
+P0/P1 决策：
+
+1. `optimize` 的最终身份是 V2 原生求解器。
+2. 第一阶段直接进入生产 `closed-loop V2`，不建立 legacy 主链。
+3. 第一轮纳入排产核心、APS 适配和 provenance；复用 aps-agent 已有 MOM/Kangni 导入；RAG 不在本轮重做。
+4. APS WorldStore 是唯一权威状态源；optimize 不拥有订单、物料、资源或排产版本状态。
+5. 成功标准包含真实 MOM 数据端到端验收；数据被 admission 阻断时必须准确报告阻断原因。
+6. 排产核心迁移为 Python；Node 版本仅作为迁移期间的差分基线。
+7. APS 与 optimize 直接使用完整 V2 问题/结果格式，不保留简化格式作为生产接口。
+8. 七种规则和 CP-SAT 一起迁移；规则作为稳定基线，CP-SAT 处理复杂约束。
+
+## 4. 集成形态
+
+```text
+APS WorldStore / MOM importer
+  -> closed_loop_problem
+  -> SchedulingProblemV2
+  -> OptimizeEngine (Python)
+      - dispatch rules: EDD/SPT/PRIORITY/FIFO/LPT/CR/ATC
+      - CP-SAT adapter/certification
+  -> SchedulingValidator
+  -> closed_loop_runtime materialization
+  -> APS schedule version / audit / evidence
+```
+
+optimize 只能产生候选排产解。Harness、Workflow、WorldStore、版本发布、确认卡、MES 写入和审计仍由 aps-agent 负责。
+
+现有 MOM 导入直接复用，不创建第二套 `records` 主模型。optimize 当前 manifest/hash 能力只在必要处转换成 APS 的 source revision、problem hash、solverMeta 和 evidenceRefs。
+
+## 5. 下一轮实现范围
+
+### In scope
+
+- 在 `server/engines/` 增加 Python `OptimizeEngine`，接入 `get_engine("OPTIMIZE")`。
+- 将七种 dispatch 规则迁移到统一的 V2 求解输入和结果结构。
+- 将当前 CP-SAT 认证规则迁移或接入 APS 现有 solver isolation/validator 边界。
+- 将 `SchedulingProblemV2` 转成算法内部只读视图，并将结果转换成 `SchedulingSolutionV2`。
+- 接入算法版本、随机种子、输入 hash、运行 ID、solverMeta 和 evidenceRefs。
+- 复用现有 `server/importers/mom_pack.py` 和 Kangni world fixture，完成真实 MOM 闭环测试。
+- 保留 Node 结果对照脚本或 fixture，验证迁移前后的算法关键指标一致性。
+
+### Out of scope
+
+- 不复制或重写 aps-agent 的 MOM/Kangni importer。
+- 不创建第二套 WorldStore、订单模型、排产版本或冲突模型。
+- 不迁移 optimize 的前端、独立 Gateway 或独立审批流程。
+- 不在本轮重做 APS 已有 `server/knowledge/` 的知识资产、检索和权限平台。
+- 不把 Node 运行时作为 APS 生产容器的必要依赖。
+- 不在本轮扩展新的 GA、NSGA-II 或其他算法；先完成已确认的七种规则和 CP-SAT。
+
+## 6. 实施任务与写入边界
+
+本轮后续采用单 agent 轻量实现，所有写入只发生在轮次工作树。任务顺序如下：
+
+1. **契约与引擎骨架**
+   - 写入范围：`server/engines/`、必要的 `server/aps_domain/` 适配文件
+   - 结果：`OptimizeEngine` 可被工厂选择，并明确 V2 输入/输出边界
+   - 停止条件：发现 V2 字段不足以表达当前算法所需约束时，先回报，不绕过契约
+
+2. **规则算法迁移**
+   - 写入范围：`server/engines/` 下 optimize 专属实现和算法目录注册
+   - 结果：七种规则在同一 V2 只读问题上运行，返回可校验的候选解
+   - 停止条件：规则语义无法在 V2 中保持，或必须修改 WorldStore 才能运行
+
+3. **CP-SAT 认证接入**
+   - 写入范围：`server/engines/`、必要的 solver isolation 适配和测试
+   - 结果：CP-SAT 结果带独立目标/完成时间检查，不声称未经证明的 optimal
+   - 停止条件：需要改变现有 solver 子进程安全边界，或出现无法解释的目标不一致
+
+4. **真实 MOM 闭环与差分验证**
+   - 写入范围：`tests/golden/`、`tests/e2e/` 或明确的 round fixture；不修改原始 MOM 数据
+   - 结果：真实 MOM world 能完成 admission、求解、V2 校验和版本物化；被阻断时有稳定 blocker
+   - 停止条件：真实数据缺少业务前置条件，必须记录为数据阻断，不通过放宽校验解决
+
+## 7. 成功标准
+
+- `get_engine("OPTIMIZE")` 能稳定选择 Python OptimizeEngine。
+- 七种规则和 CP-SAT 均使用 `SchedulingProblemV2`，不依赖旧简化生产接口。
+- 结果必须通过 `SchedulingValidator`；非法结果不得写入 APS 生产版本。
+- 真实 MOM 数据从 APS 现有 importer/world 进入闭环，产生以下之一：
+  - 合法、可追溯的排产版本；或
+  - 明确、可复现的 admission blocker。
+- 版本中包含算法 ID/版本、problem hash、run ID、solverMeta 和 evidenceRefs。
+- Node 对照结果用于发现迁移差异，但不参与生产写回。
+
+## 8. 验证方式
+
+下一轮至少执行：
+
+- `pytest tests/golden/test_rule_engine.py tests/golden/test_cp_engine.py -q`
+- 相关 `SchedulingProblemV2` / `scheduling_validator` golden tests
+- 新增 `test_optimize_engine.py`：工厂选择、V2 字段、规则结果和非法结果拒绝
+- 新增真实 MOM 闭环测试：导入或加载现有 MOM world -> `flex.schedule` -> blocker 或合法版本
+- Node/Python 差分检查：同一固定输入的订单完成时间、总延迟、资源分配和状态语义
+- 运行 `git diff --check`，确认没有无关文件和临时数据
+
+真实数据门禁必须在集成点之前安排：先确认 MOM world 的 admission 状态，再判断求解器和物化结果。不能只在最终测试阶段才发现数据本身不可排。
+
+## 9. 关键风险与控制
+
+| 风险 | 影响 | 控制方式 |
+|---|---|---|
+| V2 与 optimize 旧模型字段不一致 | 迁移时丢失资源/物料/来源信息 | 先做只读 V2 adapter；缺字段时停下扩展契约，不静默丢弃 |
+| 重复维护 MOM 输入模型 | 数据含义和 hash 漂移 | 复用 aps-agent importer/world，optimize 不写业务输入 |
+| Node/Python 结果差异 | 迁移后业务行为变化 | 固定 fixture 做差分；记录算法版本、种子和时间语义 |
+| 外部/不完整 MOM 数据被误判为算法失败 | 错误业务结论 | 保留 admission blocker，阻断时不物化版本 |
+| CP-SAT 认证被绕过 | 产生虚假的 optimal/feasible 声明 | 统一经过现有 solver isolation 和独立 Validator |
+| 迁移范围扩展到 RAG/UI/MES | 本轮失控 | 明确排除；通过现有接口接入，不改主流程 |
+
+## 10. 停止条件
+
+遇到以下情况暂停并回报，不继续扩大改动：
+
+- 必须修改 WorldStore 的权威语义或审批/审计门禁才能接入。
+- 需要引入第二套生产订单、物料、资源或版本数据源。
+- V2 契约无法表达真实 MOM 约束，且无法通过局部兼容字段解决。
+- 真实 MOM 数据的阻断原因尚未明确，却要求通过放宽校验让测试通过。
+- Node/Python 差分出现未解释的完成时间、资源分配或可行性变化。
+- 需要新增生产依赖、许可证或外部服务而没有明确运行环境。
+
+## 11. 计划可行性检查
+
+PLAN AUDIT: PASS
+
+## 14. 本轮实施结果
+
+- 已新增 `server/engines/optimize_engine.py`，并通过 `get_engine("OPTIMIZE")` 接入 APS。
+- 七种派工规则（EDD/SPT/PRIORITY/FIFO/LPT/CR/ATC）共用 APS 的能力池、日历、班组、工装和物化逻辑；结果统一进入 `SchedulingSolutionV2` 校验。
+- `/api/flex/schedule` 增加 `engine=OPTIMIZE`；Optimize 版本记录 `solverId`、`solverVersion`、`algorithmId`、`algorithmVersion`，并保留 V2 provenance 和 adapter assumption。
+- 准入阻断时仍由 APS 记录零工单 DRAFT 版本，同时保留请求的引擎身份和 blocker，不绕过 admission。
+- 新增 `tests/golden/test_optimize_engine.py`；Optimize/V2/算法注册表相关定向测试共 20 项通过，相关 APS 回归共 34 项通过。
+- 当前 APS 轮次工作树未包含 `server/data/world-proj_712276ba.json`，因此真实 MOM world 用例只能按既有测试策略跳过；全量 CP/Excel 测试还受到当前环境 NumPy（X86_V2）二进制不兼容影响。CP-SAT 的 V2 原生求解器仍应作为后续轮次接入，本轮不把 PoolEngine 适配器冒充为 CP-SAT 最优证明。
+
+阻塞问题：无。
+
+检查结论：目标、写入范围、验收标准、验证命令、真实 MOM 门禁和停止条件均已明确；单 agent 任务没有并行写入冲突，也没有依赖未讨论的产品决策。
+
+## 12. 目标分支合并前确认
+
+本轮计划分支为 `round/85-optimize-integration-shape`，基于 `codex/integrate-optimize`。后续实现、验证和自检通过后，主 agent 必须先报告：主要结论、关键洞察、仍需特别留意的风险和未覆盖环境，再请求用户确认是否合并到 `codex/integrate-optimize`。未获得确认前，不合并目标分支。
+
+## 13. 当前轮次完成定义
+
+- 本计划文件已提交到轮次分支。
+- 用户确认后才能进入目标模式实现；当前不修改生产代码。
+- 实现完成后必须通过自身检查；如形成集成结果，补充统一集成审计和真实 MOM 验证。
+- 轮次结束时保留计划、验证证据和未合并分支，直到用户明确决定是否合并。
--- a/docs/round-86-generic-world-pack-plan.md
+++ b/docs/round-86-generic-world-pack-plan.md
@ -0,0 +1,19 @@
+# Round 86: 通用 Optimize 模拟数据包
+
+## 目标
+
+把 `optimize/deliverables/kangni-simulation-package-v1` 的 10 个模拟订单、72 条工序转换成 APS 已有的通用 `world pack`，用于对话式 Optimize 调试和回归测试。
+
+## 设计约束
+
+- 不在 APS 生产代码中增加康尼专用 loader，也不让运行时识别客户名或外部包目录。
+- 数据包只填充现有 `flexOrders`、`flexRoutings`、`flexOperations`、`flexEquipment`、`flexMaterials`、`salesOrders` 和 `flexCalendar` 契约。
+- `flexParams.dataGrade=synthetic`，并保留来源包名；该数据只用于开发调试，不作为生产排产输入。
+- 原始数据的 8:00-17:00 班次保存在 `flexParams.sourceCalendar`。由于当前 PoolEngine 按分钟落盘且不拆跨班次工序，调试日历使用全天可排，保证严格 V2 校验能验证结果。
+- 小数分钟工时向上取整到分钟，原值写入 `sourceStdTimePerUnit`/`sourceStandardTime`。
+
+## 验收
+
+- `load_pack` 能加载数据包并出现在 `list_packs()`。
+- 7 个 Optimize 调度规则（EDD、SPT、PRIORITY、FIFO、LPT、CR、ATC）均产生 10 条虚拟产线、72 条工单，且 V2 严格校验通过。
+- 目标分支仍保持为 `codex/integrate-optimize`；本轮只在 `round/86-generic-world-pack` 开发，合并前单独确认。
--- a/docs/round-88-chat-data-algorithm-selection-plan.md
+++ b/docs/round-88-chat-data-algorithm-selection-plan.md
@ -0,0 +1,61 @@
+# Round 88：对话指定数据包与排产算法
+
+## 目标
+
+让用户可以用一条自然语言指令同时指定数据来源和算法，例如：
+
+> 使用 Optimize synthetic debug V1 数据，用 EDD 算法生成排产方案
+
+也支持指定本机目录里的文件：
+
+> 使用 D:\\aps-data 目录下的 orders.xlsx 数据用 EDD 算法生成排产方案
+
+系统应解析出 `dataPackRef`、`engine=OPTIMIZE` 和 `sortMode=EDD`，从已登记的数据包加载完整世界，再通过现有柔性闭环流程生成方案。
+
+## 设计
+
+1. 已登记数据包使用 `server/state/packs.py` 的通用注册表解析，不增加某个客户或某个文件的专用 loader。目录/文件指令复用 `analyze_project_deep` 的通用 Excel/CSV/SQL 读取器；路径不存在时不回退到旧世界。
+2. 对话解析器识别 Optimize 算法及 EDD/SPT/PRIORITY/FIFO/LPT/CR/ATC 等排产规则，并保留原有未点名算法时的默认行为。
+3. `_run_flex` 在明确指定数据包时替换当前会话世界、重置 ID 计数器并记录所选包；随后把算法交给现有 `run_flex_schedule(..., engine_type="OPTIMIZE")`，不复制调度逻辑。
+4. 结果块增加数据包和算法证据，文本明确显示“使用哪个数据包、哪个算法、生成多少虚拟产线/工单”。
+
+## 实施范围
+
+- `server/state/packs.py`：包别名和通用引用解析。
+- `server/agent_core/intent.py`：数据包引用、算法槽位和自然语言参数。
+- `server/aps_domain/workflow.py`：加载包、重置计数器、结果证据和未找到包的错误提示。
+- `server/data/packs/optimize-simulation-v1.json`：补充可读 aliases（仅包元数据）。
+- `tests/golden/test_optimize_chat_data_selection.py`：解析和端到端柔性排产验证。
+
+## 验收标准
+
+- 指令 `使用 Optimize synthetic debug V1 数据，用 EDD 算法生成排产方案` 命中 `flex.schedule`，参数包含 `dataPackRef`、`engine=OPTIMIZE`、`sortMode=EDD`。
+- 指令 `使用 D:\\aps-data 目录下的 orders.xlsx 数据用 EDD 算法生成排产方案` 命中 `flex.schedule`，参数包含 `dataDir`、`dataFile`、`dataPath`、`engine=OPTIMIZE`、`sortMode=EDD`。
+- 在干净会话中执行后得到 `FEASIBLE`、10 条虚拟产线、72 条工单，结果块能看到数据包名称和算法。
+- 未知数据包不会创建排产版本，并返回可用数据包提示。
+- 不存在的目录或文件不会使用当前旧数据排产，而是直接返回路径错误。
+- 普通“生成一版柔性排产方案”仍走原来的 CLOSED_LOOP 默认路径。
+
+## 验证命令
+
+```text
+pytest -q tests/golden/test_optimize_chat_data_selection.py tests/golden/test_optimize_chat_trigger.py tests/golden/test_optimize_simulation_world_pack.py tests/golden/test_optimize_engine.py
+```
+
+再使用独立 `APS_HOME` 启动后端和前端，通过 `/api/chat` 发送上述完整中文指令，检查 SSE 中的 intent 参数、`FEASIBLE` 结果及 10/72 统计。
+
+## 风险与边界
+
+- 数据包选择是显式的会话世界替换，只允许来自注册表的路径；本轮不开放任意路径加载。
+- 规则名由 Optimize 适配器支持的 dispatch rule 白名单约束，未知算法不会被静默当成 EDD。
+- 当前轮次只在 `round/88-chat-data-algorithm-selection` 验证；完成后停在合并前，等待用户确认是否合并到 `codex/integrate-optimize`。
+
+## 计划审计
+
+PLAN AUDIT: PASS
+
+Blocking issues: none
+
+Clarification needed: none
+
+Non-blocking improvements: 后续可再把算法白名单从代码抽到能力描述接口；目录导入仍沿用现有 P2 确认流程的审计策略。
--- a/docs/round-89-optimize-cpsat-native-plan.md
+++ 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，未证明最优。
--- a/server/agent_core/algolib.py
+++ b/server/agent_core/algolib.py
@ -120,6 +120,63 @@ def _builtin_catalog() -> list[AlgorithmManifest]:
        "constraints": "dict<bool 约束开关>",
    }
    items = [
+        # ---- A. Optimize Python-native dispatch rules ----
+        AlgorithmManifest(
+            algo_id="optimize.edd", name="Optimize EDD 最早交期",
+            category="A", description="Optimize V2 适配器：按最早交期派工",
+            scale_limit="<=50k 工单", time_budget="毫秒级",
+            input_schema=rule_in, output_schema=kpi_out,
+            golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True,
+            entrypoint="OPTIMIZE:EDD", regen_strategy="manual",
+        ),
+        AlgorithmManifest(
+            algo_id="optimize.spt", name="Optimize SPT 最短工时",
+            category="A", description="Optimize V2 适配器：短工时优先",
+            scale_limit="<=50k 工单", time_budget="毫秒级",
+            input_schema=rule_in, output_schema=kpi_out,
+            golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True,
+            entrypoint="OPTIMIZE:SPT", regen_strategy="manual",
+        ),
+        AlgorithmManifest(
+            algo_id="optimize.priority", name="Optimize PRIORITY 优先级",
+            category="A", description="Optimize V2 适配器：订单优先级优先",
+            scale_limit="<=50k 工单", time_budget="毫秒级",
+            input_schema=rule_in, output_schema=kpi_out,
+            golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True,
+            entrypoint="OPTIMIZE:PRIORITY", regen_strategy="manual",
+        ),
+        AlgorithmManifest(
+            algo_id="optimize.fifo", name="Optimize FIFO 先来先服务",
+            category="A", description="Optimize V2 适配器：按释放时间派工",
+            scale_limit="<=50k 工单", time_budget="毫秒级",
+            input_schema=rule_in, output_schema=kpi_out,
+            golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True,
+            entrypoint="OPTIMIZE:FIFO", regen_strategy="manual",
+        ),
+        AlgorithmManifest(
+            algo_id="optimize.lpt", name="Optimize LPT 最长工时",
+            category="A", description="Optimize V2 适配器：长工时优先",
+            scale_limit="<=50k 工单", time_budget="毫秒级",
+            input_schema=rule_in, output_schema=kpi_out,
+            golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True,
+            entrypoint="OPTIMIZE:LPT", regen_strategy="manual",
+        ),
+        AlgorithmManifest(
+            algo_id="optimize.cr", name="Optimize CR 临界比",
+            category="A", description="Optimize V2 适配器：交期紧迫度优先",
+            scale_limit="<=50k 工单", time_budget="毫秒级",
+            input_schema=rule_in, output_schema=kpi_out,
+            golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True,
+            entrypoint="OPTIMIZE:CR", regen_strategy="manual",
+        ),
+        AlgorithmManifest(
+            algo_id="optimize.atc", name="Optimize ATC 逾期成本",
+            category="A", description="Optimize V2 适配器：逾期成本代理排序",
+            scale_limit="<=50k 工单", time_budget="毫秒级",
+            input_schema=rule_in, output_schema=kpi_out,
+            golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True,
+            entrypoint="OPTIMIZE:ATC", regen_strategy="manual",
+        ),
        # ---- A. 启发式（RULE 引擎各策略模板）----
        AlgorithmManifest(
            algo_id="rule.delivery_first", name="EDD 最早交期",
@ -436,10 +493,10 @@ class AlgorithmRegistry:
        """入口可达性：引擎名 / ENGINE:STRATEGY / module.path:attr。"""
        if ":" not in entrypoint:
            return True, ""                             # 纯引擎名（RULE/CP/GA/HYBRID）由 get_engine 工厂保证
-        if entrypoint.startswith("RULE:"):
+        if entrypoint.startswith(("RULE:", "OPTIMIZE:")):
            from server.engines import get_engine
            try:
-                get_engine("RULE")
+                get_engine(entrypoint.split(":", 1)[0])
                return True, ""
            except (ImportError, AttributeError, RuntimeError, ValueError, TypeError) as exc:
                return False, str(exc)
--- a/server/aps_domain/closed_loop_runtime.py
+++ b/server/aps_domain/closed_loop_runtime.py
@ -43,6 +43,7 @@ from server.aps_domain.scheduling_problem_v2 import (
 )
 from server.aps_domain.scheduling_validator import validate_solution
 from server.engines.pool_engine import PoolEngine
+from server.engines.optimize_engine import OptimizeEngine

 World = dict[str, Any]
 _TZ = ZoneInfo("Asia/Shanghai")
@ -624,7 +625,13 @@ def persist_closed_loop_projection(world: World, closed_loop: ClosedLoopProblem,
    }


-def record_blocked_flex_version(world: World, next_id, closed_loop: ClosedLoopProblem) -> dict[str, Any]:
+def record_blocked_flex_version(
+    world: World,
+    next_id,
+    closed_loop: ClosedLoopProblem,
+    *,
+    engine_type: str = "CLOSED_LOOP",
+) -> dict[str, Any]:
    """Record an honest zero-WO DRAFT version when manufacturing admission is blocked."""

    world.setdefault("flexScheduleVersions", [])
@ -632,13 +639,18 @@ def record_blocked_flex_version(world: World, next_id, closed_loop: ClosedLoopPr
    world.setdefault("flexWorkOrders", [])
    world.setdefault("flexConflicts", [])
    version_id = next_id("flexScheduleVersion")
+    normalized_engine = (
+        "OPTIMIZE"
+        if str(engine_type or "CLOSED_LOOP").strip().upper() == "OPTIMIZE"
+        else "CLOSED_LOOP"
+    )
    version_no = f"FV{closed_loop.business_date.replace('-', '')}-{len(world['flexScheduleVersions']) + 1:03d}"
    version = {
        "id": version_id,
        "versionNo": version_no,
        "versionName": f"闭环排产准入阻断 {closed_loop.business_date}",
        "sortMode": "CLOSED_LOOP",
-        "engineType": "CLOSED_LOOP",
+        "engineType": normalized_engine,
        "status": "DRAFT",
        "solveStatus": "BLOCKED",
        "planningProblemId": closed_loop.problem_id,
@ -682,7 +694,7 @@ def record_blocked_flex_version(world: World, next_id, closed_loop: ClosedLoopPr
    return {
        "versionId": version_id,
        "versionNo": version_no,
-        "engineType": "CLOSED_LOOP",
+        "engineType": normalized_engine,
        "status": "DRAFT",
        "solveStatus": "BLOCKED",
        "orderCount": version["orderCount"],
@ -1003,15 +1015,19 @@ def flex_version_to_solution_v2(
        unscheduledRequirements=unscheduled,
        hardViolations=hard_conflicts,
        assumptions=(Assumption(
-            code="POOL_ENGINE_V1_ADAPTER",
-            message="PoolEngine 候选结果已映射到闭环 V2 契约并执行独立校验",
+            code=("OPTIMIZE_ENGINE_V1_ADAPTER"
+                  if str(version.get("engineType") or "").upper() == "OPTIMIZE"
+                  else "POOL_ENGINE_V1_ADAPTER"),
+            message=("OptimizeEngine 候选结果已映射到闭环 V2 契约并执行独立校验"
+                     if str(version.get("engineType") or "").upper() == "OPTIMIZE"
+                     else "PoolEngine 候选结果已映射到闭环 V2 契约并执行独立校验"),
            sourceRef=f"flex-version:{version_id}",
            confidence=1.0,
        ),),
        provenance=SolutionProvenance(
            runId=f"flex-version:{version_id}",
-            solverId="pool-engine",
-            solverVersion="closed-loop-v1",
+            solverId=str(version.get("solverId") or "pool-engine"),
+            solverVersion=str(version.get("solverVersion") or "closed-loop-v1"),
            generatedAt=generated_at,
            businessDate=business_day,
            problemHash=scheduling_problem_hash(problem),
@ -1027,6 +1043,8 @@ def _record_rejected_candidate(
    closed_loop: ClosedLoopProblem,
    report: Any,
    candidate_result: Mapping[str, Any],
+    *,
+    engine_type: str = "CLOSED_LOOP",
 ) -> dict[str, Any]:
    """Persist validation diagnostics without leaking candidate WO/VL artifacts."""

@ -1037,12 +1055,21 @@ def _record_rejected_candidate(
    version_id = next_id("flexScheduleVersion")
    version_no = f"FV{closed_loop.business_date.replace('-', '')}-{len(world['flexScheduleVersions']) + 1:03d}"
    violations = list(report.hardViolations)
+    normalized_engine = (
+        "OPTIMIZE"
+        if str(engine_type or "CLOSED_LOOP").strip().upper() == "OPTIMIZE"
+        else "CLOSED_LOOP"
+    )
    version = {
        "id": version_id,
        "versionNo": version_no,
        "versionName": f"闭环排产校验失败 {closed_loop.business_date}",
        "sortMode": "CLOSED_LOOP",
-        "engineType": "CLOSED_LOOP",
+        "engineType": normalized_engine,
+        "solverId": candidate_result.get("solverId"),
+        "solverVersion": candidate_result.get("solverVersion"),
+        "algorithmId": candidate_result.get("algorithmId"),
+        "algorithmVersion": candidate_result.get("algorithmVersion"),
        "status": "DRAFT",
        "solveStatus": "REJECTED",
        "planningProblemId": closed_loop.problem_id,
@ -1078,7 +1105,11 @@ def _record_rejected_candidate(
    return {
        "versionId": version_id,
        "versionNo": version_no,
-        "engineType": "CLOSED_LOOP",
+        "engineType": normalized_engine,
+        "solverId": version.get("solverId"),
+        "solverVersion": version.get("solverVersion"),
+        "algorithmId": version.get("algorithmId"),
+        "algorithmVersion": version.get("algorithmVersion"),
        "status": "DRAFT",
        "solveStatus": "REJECTED",
        "orderCount": version["orderCount"],
@ -1104,6 +1135,7 @@ def run_closed_loop_candidate(
    sort_mode: str | None = None,
    window: str | None = None,
    name: str | None = None,
+    engine_type: str = "CLOSED_LOOP",
    strict: bool = True,
 ) -> dict[str, Any]:
    """Build, solve and validate one closed-loop candidate with version-level atomicity."""
@ -1134,7 +1166,12 @@ def run_closed_loop_candidate(
        }
    }
    if not admitted:
-        result = record_blocked_flex_version(world, next_id, closed_loop)
+        result = record_blocked_flex_version(
+            world,
+            next_id,
+            closed_loop,
+            engine_type=engine_type or "CLOSED_LOOP",
+        )
        return {**result, **base, "validation": None}

    candidate = deepcopy(world)
@ -1146,20 +1183,40 @@ def run_closed_loop_candidate(
        candidate.setdefault(key, [] if key != "flexParams" else {})
    projected = project_admitted_demands_to_flex_orders(candidate, closed_loop)
    schedule_start = schedule_start_date or (_as_day(business_date) + timedelta(days=1)).isoformat()
-    solved = PoolEngine().solve(
-        candidate,
-        next_id,
-        sort_mode=sort_mode,
-        order_ids=projected["orderIds"],
-        start_date=schedule_start,
-        name=name or f"闭环排产 {business_date}",
-        window=window,
+    normalized_engine = (
+        "OPTIMIZE"
+        if str(engine_type or "CLOSED_LOOP").strip().upper() == "OPTIMIZE"
+        else "CLOSED_LOOP"
    )
+    if normalized_engine == "OPTIMIZE":
+        solved = OptimizeEngine().solve_flex(
+            candidate,
+            next_id,
+            dispatch_rule=sort_mode,
+            order_ids=projected["orderIds"],
+            start_date=schedule_start,
+            name=name or f"Optimize 闭环排产 {business_date}",
+            window=window,
+        )
+    else:
+        solved = PoolEngine().solve(
+            candidate,
+            next_id,
+            sort_mode=sort_mode,
+            order_ids=projected["orderIds"],
+            start_date=schedule_start,
+            name=name or f"闭环排产 {business_date}",
+            window=window,
+        )
    version_id = int(solved["versionId"])
    version = next(row for row in candidate["flexScheduleVersions"] if row.get("id") == version_id)
    version["versionNo"] = f"FV{business_date.replace('-', '')}-{len(candidate['flexScheduleVersions']):03d}"
    version["versionName"] = name or f"闭环排产 {business_date}"
-    version["engineType"] = "CLOSED_LOOP"
+    version["engineType"] = normalized_engine
+    version["solverId"] = str(solved.get("solverId") or ("pool-engine" if normalized_engine != "OPTIMIZE" else "optimize-dispatch"))
+    version["solverVersion"] = str(solved.get("solverVersion") or ("closed-loop-v1" if normalized_engine != "OPTIMIZE" else "1.0.0"))
+    version["algorithmId"] = solved.get("algorithmId")
+    version["algorithmVersion"] = solved.get("algorithmVersion")
    version["planningProblemId"] = closed_loop.problem_id
    version["planningSourceHash"] = closed_loop.source_revision
    version["demandCount"] = len(closed_loop.manufacturing_demands)
@ -1167,7 +1224,7 @@ def run_closed_loop_candidate(
    version["unscheduledDemandCount"] = max(0, len(admitted) - int(version.get("vlCount") or 0))
    version["createdAt"] = f"{business_date} 00:00"
    solved["versionNo"] = version["versionNo"]
-    solved["engineType"] = "CLOSED_LOOP"
+    solved["engineType"] = normalized_engine

    solution = flex_version_to_solution_v2(candidate, closed_loop, problem, version_id)
    report = validate_solution(problem, solution, world=candidate)
@ -1184,7 +1241,14 @@ def run_closed_loop_candidate(
    solved["projectedFlexOrders"] = projected

    if not report.valid or solution.solveStatus != SolveStatus.FEASIBLE:
-        rejected = _record_rejected_candidate(world, next_id, closed_loop, report, solved)
+        rejected = _record_rejected_candidate(
+            world,
+            next_id,
+            closed_loop,
+            report,
+            solved,
+            engine_type=normalized_engine,
+        )
        return {**rejected, **base, "projectedFlexOrders": projected}

    for key in (
--- a/server/aps_domain/flex.py
+++ b/server/aps_domain/flex.py
@ -233,7 +233,8 @@ def run_flex_schedule(store, sort_mode: str | None = None, order_ids: list[int]
                      start_date: str | None = None, name: str | None = None,
                      actor: str = "web", window: str | None = None,
                      enforce_teams: bool | None = None,
-                      trial: bool = False) -> dict:
+                      trial: bool = False,
+                      engine_type: str | None = None) -> dict:
    """Run the governed closed-loop scheduling pipeline as one P1 action.

    Real/site worlds always use the closed-loop requirement, supply, routing and
@ -307,9 +308,14 @@ def run_flex_schedule(store, sort_mode: str | None = None, order_ids: list[int]
                sort_mode=sort_mode,
                window=window,
                name=name,
+                engine_type=engine_type or "CLOSED_LOOP",
                strict=True,
            )
-            result["executionMode"] = "CLOSED_LOOP_V1"
+            result["executionMode"] = (
+                "OPTIMIZE_CLOSED_LOOP_V1"
+                if str(engine_type or "").upper() == "OPTIMIZE"
+                else "CLOSED_LOOP_V1"
+            )
        result["salesOrdersSynced"] = synced
        result["decompose"] = {
            "orders": len(decomposition.get("orders") or []),
--- a/server/aps_domain/params.py
+++ b/server/aps_domain/params.py
@ -96,8 +96,8 @@ def normalize_params_payload(payload: dict[str, Any]) -> dict[str, Any]:

    if "defaultEngine" in payload and payload["defaultEngine"] is not None:
        eng = str(payload["defaultEngine"]).upper()
-        if eng not in ("RULE", "CP", "GA", "HYBRID"):
-            raise ValueError("defaultEngine 须为 RULE/CP/GA/HYBRID")
+        if eng not in ("RULE", "CP", "GA", "HYBRID", "OPTIMIZE"):
+            raise ValueError("defaultEngine 须为 RULE/CP/GA/HYBRID/OPTIMIZE")
        out["defaultEngine"] = eng

    if "cpTimeLimitSeconds" in payload and payload["cpTimeLimitSeconds"] is not None:
--- a/server/aps_domain/workflow.py
+++ b/server/aps_domain/workflow.py
@ -154,6 +154,10 @@ def _run_schedule(store: WorldStore, intent: IntentResult, actor: str) -> AgentR
        constraints=engine_constraint_flags(store.data),           # SC-04 约束剖面 → 引擎开关
        timeLimitSeconds=float(sp["cpTimeLimitSeconds"]) if sp.get("cpTimeLimitSeconds") is not None else 8.0,
    )
+    if params.engineType == "OPTIMIZE":
+        return AgentReply(
+            text="Optimize 目前只支持柔性 V2 闭环，请通过 /api/flex/schedule 并指定 engine=OPTIMIZE。"
+        )
    engine = get_engine(params.engineType)             # CP / HYBRID 真管线；GA 仍 RULE 代跑
    result: ScheduleResult = engine.solve(store.data, params, store.next_id)  # 求解（写内存世界）
    # 可追溯链：run-id / 算法版本 / 种子 / 知识版本 / 用户确认 串成一条链（§8.4）
@ -2086,7 +2090,8 @@ def _external_audit_evidence(summary: dict, trace: dict) -> list[str]:
    return refs


-def _run_flex(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply:
+def _run_flex(store: WorldStore, intent: IntentResult, actor: str,
+              session_id: str | None = None) -> AgentReply:
    """触发柔性排产并回执（短文案 + 结构化 flex-schedule 块：KPI + 虚拟产线表 + 瓶颈）。"""
    from server.aps_domain.flex import run_flex_schedule
    from server.importers.workbook_profiles import has_adoption_flow
@ -2109,6 +2114,94 @@ def _run_flex(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply
    if not mode:
        return AgentReply(text="请先选择本次排产目标，再生成方案。")
    window = intent.params.get("window")
+    data_pack = None
+    data_source = None
+    data_path = str(intent.params.get("dataPath") or "").strip()
+    data_dir = str(intent.params.get("dataDir") or "").strip()
+    data_file = str(intent.params.get("dataFile") or "").strip()
+    if data_path or data_dir:
+        source_query = data_path or (os.path.join(data_dir, data_file) if data_file else data_dir)
+        if data_file and not os.path.isfile(source_query):
+            return AgentReply(text=f"数据文件不存在：{source_query}，没有执行排产。")
+        if not data_file and data_dir and not os.path.isdir(data_dir):
+            return AgentReply(text=f"数据目录不存在：{data_dir}，没有执行排产。")
+        if data_path and not data_file and not os.path.isfile(data_path) and not os.path.isdir(data_path):
+            return AgentReply(text=f"数据路径不存在：{data_path}，没有执行排产。")
+        if os.path.isfile(source_query) and source_query.lower().endswith(".json"):
+            from server.state.packs import load_pack
+            try:
+                candidate_world = load_pack(source_query)
+            except (OSError, ValueError, TypeError) as exc:
+                return AgentReply(text=f"APS 数据包读取失败：{exc}，没有执行排产。")
+            required = ("flexOrders", "flexRoutings", "flexEquipment")
+            if not isinstance(candidate_world, dict) or not all(candidate_world.get(key) for key in required):
+                return AgentReply(
+                    text="该 JSON 不是可排产的 APS 数据包：至少需要订单、工艺和设备数据，没有执行排产。",
+                )
+            store.data = candidate_world
+            store._reset_counters()
+            data_pack = {
+                "file": os.path.basename(source_query),
+                "name": os.path.splitext(os.path.basename(source_query))[0],
+                "path": source_query,
+            }
+            data_source = {
+                "directory": os.path.dirname(source_query),
+                "file": os.path.basename(source_query),
+                "paths": [os.path.abspath(source_query)],
+            }
+        else:
+            from server.aps_domain.project_analyze import analyze_project_deep
+            from server.state.seed import empty_world
+            candidate_world = empty_world()
+            try:
+                source_report = analyze_project_deep(
+                    candidate_world, session_id, apply_sql=True,
+                    query=source_query, next_id=store.next_id,
+                )
+            except (OSError, PermissionError, ValueError) as exc:
+                return AgentReply(text=f"数据文件读取失败：{exc}")
+            if not source_report.get("ok"):
+                return AgentReply(text=f"数据目录分析失败：{source_report.get('error') or '目录不可用'}")
+            source_paths = source_report.get("sourcePaths") or []
+            if not source_paths:
+                return AgentReply(
+                    text="指定位置没有找到可读取的排产数据文件，没有执行排产。",
+                )
+            if not source_report.get("canSchedule"):
+                missing = (source_report.get("plan") or [])[:3]
+                detail = "；".join(str(item) for item in missing)
+                return AgentReply(
+                    text=(f"数据已读取，但暂时不能排产：缺少可排所需的订单、工艺或设备信息。"
+                          + (f"\n{detail}" if detail else "")),
+                )
+            store.data = candidate_world
+            store._reset_counters()
+            data_source = {
+                "directory": source_report.get("workDir") or data_dir,
+                "file": data_file or (os.path.basename(data_path) if data_path else None),
+                "paths": source_paths,
+            }
+    data_pack_ref = str(intent.params.get("dataPackRef") or "").strip()
+    if data_pack_ref:
+        from server.state.packs import load_pack, resolve_pack_reference
+        data_pack = resolve_pack_reference(data_pack_ref)
+        if not data_pack:
+            from server.state.packs import list_packs
+            available = "、".join(str(item.get("name") or item.get("file")) for item in list_packs())
+            return AgentReply(
+                text=(f"找不到数据包「{data_pack_ref}」，没有执行排产。"
+                      f"可用数据包：{available or '暂无'}。"),
+            )
+        # An explicit pack reference intentionally starts this run from that
+        # registered world, making repeated debug commands deterministic.
+        store.data = load_pack(data_pack["path"])
+        store._reset_counters()
+        store.data.setdefault("flexParams", {})["selectedPack"] = {
+            "file": data_pack.get("file"),
+            "name": data_pack.get("name"),
+            "reference": data_pack_ref,
+        }
    order_ids = intent.params.get("orderIds")
    if isinstance(order_ids, list):
        order_ids = [int(x) for x in order_ids if str(x).isdigit() or isinstance(x, int)]
@ -2139,9 +2232,16 @@ def _run_flex(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply
        store.save()
    else:
        result = run_flex_schedule(store, sort_mode=mode, order_ids=order_ids or None,
-                                   actor=actor, window=window)  # P1
-    mode_key = str(result.get("sortMode") or result.get("engineType") or mode or "CLOSED_LOOP")
+                                   actor=actor, window=window,
+                                   engine_type=engine or None)  # P1
+    mode_key = str(result.get("dispatchRule") or result.get("sortMode") or result.get("engineType") or mode or "CLOSED_LOOP")
    mode_cn = _FLEX_MODE_CN.get(mode_key, mode_key)
+    if engine == "OPTIMIZE":
+        optimize_rule_cn = {
+            "EDD": "EDD 最早交期", "SPT": "SPT 最短工时", "PRIORITY": "PRIORITY 优先级",
+            "FIFO": "FIFO 先进先出", "LPT": "LPT 最长工时", "CR": "CR 临界比", "ATC": "ATC 逾期成本",
+        }
+        mode_cn = f"Optimize · {optimize_rule_cn.get(mode_key.upper(), mode_cn)}"
    solve_status = str(result.get("solveStatus") or "FEASIBLE").upper()
    if result.get("sortMode") == "EXTERNAL" or use_ext:
        mode_cn = f"外部算法({result.get('skillId') or 'skill'})"
@ -2243,6 +2343,14 @@ def _run_flex(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply
        blockId=f"flex-sched-{result['versionId']}", type="flex-schedule",
        props={
            "versionNo": result["versionNo"], "modeCn": mode_cn,
+            "algorithm": {
+                "engine": result.get("engineType") or engine or "CLOSED_LOOP",
+                "rule": result.get("dispatchRule") or mode,
+                "algorithmId": result.get("algorithmId"),
+            },
+            "dataPack": ({"file": data_pack.get("file"), "name": data_pack.get("name"),
+                           "reference": data_pack_ref} if data_pack else None),
+            "dataSource": data_source,
            "stats": {"vlCount": result["vlCount"], "woCount": result["woCount"],
                      "makespan": result.get("makespan"), "onTimeCount": result.get("onTimeCount", 0),
                      "conflictCount": result["conflictCount"],
@ -2293,8 +2401,11 @@ def _run_flex(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply
            + ("\n\n别担心，我把还缺的东西列在下面，你按顺序补就行。" if guide_block else "")
        )
    else:
+        pack_hint = (f"数据包「{data_pack.get('name')}」· " if data_pack else
+                     f"数据文件「{data_source.get('file') or data_source.get('directory')}」· "
+                     if data_source else "")
        text = (
-            f"柔性排产完成。{focus_hint}版本 {result['versionNo']}（{mode_cn}·{win_cn}）："
+            f"柔性排产完成。{pack_hint}{focus_hint}版本 {result['versionNo']}（{mode_cn}·{win_cn}）："
            f"{result['vlCount']} 条虚拟产线 / {result['woCount']} 个工单，冲突 {result['conflictCount']} 项"
            + (f"，窗外延期 {deferred}" if deferred else "")
            + f"。{dl_hint}"
@ -3457,7 +3568,7 @@ async def handle_intent(store: WorldStore, session_id: str, intent: IntentResult
        return AgentReply(text=f"{title} 属于 P2 写操作，需要你确认。", blocks=[block])
    # ---- 柔性排产（P1：能力池动态组虚拟产线，写草稿版本 §M5） ----
    if name == "flex.schedule":
-        return _run_flex(store, intent, actor)
+        return _run_flex(store, intent, actor, session_id=session_id)
    if name == "flex.reschedule":
        return _flex_reschedule(store, intent, session_id, actor)
    if name == "flex.swap":
--- a/server/contracts.py
+++ b/server/contracts.py
@ -125,7 +125,7 @@ class ScheduleResult(BaseModel):
    """排产结果摘要：引擎 solve() 的标准输出（§9.1），回复/审计/KPI 共用。"""
    versionId: int                                     # 版本 ID
    versionNo: str                                     # 版本号（V+日期+序号）
-    engineType: Literal["RULE", "CP", "GA", "HYBRID", "EXTERNAL"]  # 引擎类型
+    engineType: Literal["RULE", "CP", "GA", "HYBRID", "EXTERNAL", "OPTIMIZE"]  # 引擎类型
    strategy: str                                      # 策略模板
    status: Literal["DRAFT", "PUBLISHED", "ARCHIVED"] = "DRAFT"  # 版本状态
    orderCount: int                                    # 参与排产的订单项数
--- a/server/data/packs/optimize-simulation-v1.json
+++ b/server/data/packs/optimize-simulation-v1.json
--- a/server/engines/init.py
+++ b/server/engines/init.py
@ -8,6 +8,7 @@ from server.engines.cp_engine import CpSatEngine, HybridEngine
 from server.engines.external_engine import ExternalEngine
 from server.engines.ga_engine import GeneticAlgorithmEngine
 from server.engines.nsga2_engine import NSGA2Engine, nsga2_defaults, solve_nsga2
+from server.engines.optimize_engine import OptimizeEngine
 from server.engines.pool_engine import PoolEngine
 from server.engines.rule_engine import RuleEngine

@ -19,6 +20,7 @@ __all__ = [
    "HybridEngine",
    "ISchedulingEngine",
    "NSGA2Engine",
+    "OptimizeEngine",
    "PoolEngine",
    "RuleEngine",
    "get_engine",
@ -28,7 +30,7 @@ __all__ = [


 def get_engine(engine_type: str) -> ISchedulingEngine:
-    """引擎工厂：RULE / CP / GA / HYBRID / NSGA2 / EXTERNAL。"""
+    """引擎工厂：RULE / CP / GA / HYBRID / NSGA2 / OPTIMIZE / EXTERNAL。"""
    kind = (engine_type or "RULE").upper()
    if kind.startswith("EXTERNAL"):
        skill_id = None
@ -43,4 +45,6 @@ def get_engine(engine_type: str) -> ISchedulingEngine:
        return GeneticAlgorithmEngine()
    if kind == "NSGA2":
        return NSGA2Engine()
+    if kind == "OPTIMIZE":
+        return OptimizeEngine()
    return RuleEngine(requested_type="RULE")
--- a/server/engines/base.py
+++ b/server/engines/base.py
@ -14,7 +14,7 @@ from server.contracts import ScheduleResult            # 引擎输出契约
 class EngineParams(BaseModel):
    """引擎入参：一次排产请求的全部参数（与 legacy runScheduling params 对齐）。"""
    orderIds: list[int] = Field(default_factory=list)  # 目标订单 ID（空=全部待排）
-    engineType: str = "RULE"                           # 请求的引擎类型（RULE/CP/GA/HYBRID）
+    engineType: str = "RULE"                           # 请求的引擎类型（RULE/CP/GA/HYBRID/OPTIMIZE）
    strategyTemplate: str = "COMPREHENSIVE"            # 策略模板（排序规则）
    planningHorizonDays: int = 14                      # 计划展望期（天）
    startDate: str | None = None                       # 排产起始日 YYYY-MM-DD（None=明天）
--- a/server/engines/optimize_cpsat.py
+++ 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)
--- a/server/engines/optimize_engine.py
+++ b/server/engines/optimize_engine.py
@ -0,0 +1,114 @@
+"""Optimize scheduling engine integrated with the APS V2 closed loop.
+
+The engine owns algorithm selection and provenance. APS still owns the world,
+admission, candidate validation, version materialization, and audit trail.
+"""
+from __future__ import annotations
+
+from typing import Any, Callable
+
+from server.contracts import ScheduleResult
+from server.engines.base import EngineParams, ISchedulingEngine
+from server.engines.pool_engine import PoolEngine
+
+
+DISPATCH_RULES = frozenset({"EDD", "SPT", "PRIORITY", "FIFO", "LPT", "CR", "ATC"})
+
+
+def normalize_dispatch_rule(value: str | None) -> str:
+    rule = str(value or "EDD").strip().upper().replace("-", "_")
+    aliases = {
+        "DELIVERY_FIRST": "EDD",
+        "EARLIEST_DUE_DATE": "EDD",
+        "FIRST_IN_FIRST_OUT": "FIFO",
+        "APPARENT_TARDINESS_COST": "ATC",
+    }
+    rule = aliases.get(rule, rule)
+    return rule if rule in DISPATCH_RULES else "EDD"
+
+
+class OptimizeEngine(ISchedulingEngine):
+    """Python-native Optimize entry point for APS scheduling.
+
+    The first integration reuses PoolEngine's already validated flex
+    materializer. Its ordering policy is supplied by ``dispatch_rule`` so the
+    seven optimize rules share APS calendars, teams, tooling, and rollback
+    semantics while the V2 runtime remains the authority for validation.
+    """
+
+    name = "OPTIMIZE"
+    supports_anytime = False
+
+    def solve(
+        self,
+        world: dict[str, Any],
+        params: EngineParams,
+        next_id: Callable[[str], int],
+    ) -> ScheduleResult:
+        rule = normalize_dispatch_rule(params.strategyTemplate)
+        solved = PoolEngine().solve(
+            world,
+            next_id,
+            sort_mode="ASC",
+            order_ids=params.orderIds or None,
+            start_date=params.startDate,
+            name=params.name,
+            window=None,
+            enforce_teams=params.constraints.get("personnel") if params.constraints else None,
+            dispatch_rule=rule,
+        )
+        return _summary_to_result(solved, rule)
+
+    def solve_flex(
+        self,
+        world: dict[str, Any],
+        next_id: Callable[[str], int],
+        *,
+        dispatch_rule: str | None = None,
+        order_ids: list[int] | None = None,
+        start_date: str | None = None,
+        name: str | None = None,
+        window: str | None = None,
+        enforce_teams: bool | None = None,
+    ) -> dict[str, Any]:
+        """Materialize an Optimize candidate for the closed-loop V2 adapter."""
+
+        rule = normalize_dispatch_rule(dispatch_rule)
+        solved = PoolEngine().solve(
+            world,
+            next_id,
+            sort_mode="ASC",
+            order_ids=order_ids,
+            start_date=start_date,
+            name=name,
+            window=window,
+            enforce_teams=enforce_teams,
+            dispatch_rule=rule,
+        )
+        solved.update({
+            "engineType": "OPTIMIZE",
+            "algorithmId": f"optimize.{rule.lower()}",
+            "algorithmVersion": "1.0.0",
+            "solverId": "optimize-dispatch",
+            "solverVersion": "1.0.0",
+            "dispatchRule": rule,
+        })
+        return solved
+
+
+def _summary_to_result(solved: dict[str, Any], rule: str) -> ScheduleResult:
+    return ScheduleResult(
+        versionId=int(solved["versionId"]),
+        versionNo=str(solved["versionNo"]),
+        engineType="OPTIMIZE",
+        strategy=rule,
+        status="DRAFT",
+        orderCount=int(solved.get("orderCount") or 0),
+        poCount=int(solved.get("vlCount") or 0),
+        woCount=int(solved.get("woCount") or 0),
+        conflictCount=int(solved.get("conflictCount") or 0),
+        totalTardiness=float(solved.get("totalTardiness") or 0),
+        avgUtilization=float(solved.get("avgUtilization") or 0),
+        evidenceRefs=[f"algorithm:optimize.{rule.lower()}", f"run:{solved['versionId']}"],
+        solveStatus="FEASIBLE" if not solved.get("conflictCount") else "PARTIAL",
+    )
--- a/server/engines/pool_engine.py
+++ b/server/engines/pool_engine.py
@ -99,6 +99,7 @@ class PoolEngine:
              window: str | None = None,
              seed_busy: dict[int, list[tuple[datetime, datetime]]] | None = None,
              enforce_teams: bool | None = None,
+              dispatch_rule: str | None = None,
              trial: bool = False) -> dict[str, Any]:
        """执行一次柔性排产，返回结果摘要 dict。

@ -163,7 +164,10 @@ class PoolEngine:
            orders.append(o)

        # ---- ③ 派工排序（吸收排产逻辑 PPT：正排 EDD / 倒排最晚优先 / 瓶颈锚）----
-        if mode == SORT_DESC:
+        dispatch = str(dispatch_rule or "").strip().upper()
+        if dispatch in {"SPT", "LPT", "CR", "ATC", "PRIORITY", "FIFO", "EDD"}:
+            orders.sort(key=lambda o: self._dispatch_key(dispatch, o, routings, ops_by_code))
+        elif mode == SORT_DESC:
            # 倒排：交期最晚的订单先占资源（自交期向前的派工近似）
            orders.sort(key=lambda o: (o["dueDate"], o["priority"]), reverse=True)
        elif mode == SORT_BOTTLENECK:
@ -652,6 +656,38 @@ class PoolEngine:
               if strict_inputs else {}),
        }

+    @staticmethod
+    def _dispatch_key(rule: str, order: dict[str, Any], routings: list[dict], ops_by_code: dict[str, dict]) -> tuple:
+        """Return stable dispatch keys for the optimize rule catalog."""
+        steps = [step for step in routings if step.get("productCode") == order.get("productCode")]
+        duration = sum(float(step.get("stdTimePerUnit") or 1) for step in steps) * float(order.get("quantity") or 1)
+        due = str(order.get("dueDate") or "9999-12-31")
+        release = str(order.get("releaseDate") or order.get("releaseAt") or "0000-01-01")
+        priority = -int(order.get("priority") or 0)
+        def _day_number(value: str, fallback: float) -> float:
+            try:
+                return datetime.fromisoformat(value[:10]).toordinal()
+            except (TypeError, ValueError):
+                return fallback
+        due_day = _day_number(due, 3652059.0)
+        release_day = _day_number(release, 1.0)
+        slack = max(0.0, (due_day - release_day) * 24 * 60 - duration)
+        if rule == "SPT":
+            return (duration, due, priority, str(order.get("orderNo") or ""))
+        if rule == "LPT":
+            return (-duration, due, priority, str(order.get("orderNo") or ""))
+        if rule == "PRIORITY":
+            return (priority, due, release, str(order.get("orderNo") or ""))
+        if rule == "FIFO":
+            return (release, due, priority, str(order.get("orderNo") or ""))
+        if rule == "CR":
+            return ((due_day - release_day) / max(duration, 1e-9), priority, str(order.get("orderNo") or ""))
+        if rule == "ATC":
+            score = (abs(priority) or 1) / max(duration, 1e-9)
+            score *= pow(2.718281828, -slack / max(4 * duration, 1.0))
+            return (-score, due, priority, str(order.get("orderNo") or ""))
+        return (due, priority, release, str(order.get("orderNo") or ""))
+
    # ---------------- 占槽：设备级 + 可选班组并发（SC-11） ----------------
    def _place(self, cursor: datetime, duration_min: float, eq_id: int,
               eq_busy: dict[int, list[tuple[datetime, datetime]]], world: World,
--- a/server/gateway/app.py
+++ b/server/gateway/app.py
@ -220,6 +220,7 @@ class FlexScheduleRequest(BaseModel):
    orderIds: list[int] = Field(default_factory=list)
    window: str | None = None                          # short/mid/long/full（SC-12）
    enforceTeams: bool | None = None                   # SC-11 班组约束
+    engine: str | None = None                           # CLOSED_LOOP / OPTIMIZE


 class TimeUpdateRequest(BaseModel):
@ -2575,6 +2576,7 @@ def create_app() -> FastAPI:
                actor=req.sessionId or "web",
                window=req.window,
                enforce_teams=req.enforceTeams,
+                engine_type=req.engine,
            )
        except (ValueError, PermissionError) as exc:
            return {"error": str(exc)}
--- a/server/state/packs.py
+++ b/server/state/packs.py
@ -3,7 +3,7 @@
 # 把「一套完整世界数据」封装为可重放的 JSON 包：
 #   - 演示厂 = server/data/packs/demo.json（由脚本从种子生成）
 #   - 行业种子 / 客户现场数据都可做成包，data.reset 即「重放数据包」
-# 日期重基：包内记录 baseDate（生成日），加载时把所有 YYYY-MM-DD
+# 日期重基：包内记录 baseDate（世界业务基准日：演示包=生成日，回放包=案例业务日），加载时把所有 YYYY-MM-DD
 # 字面日期整体平移到「今天」，保证任意运行日行为一致（黄金测试稳定）。
 # ============================================================
 from __future__ import annotations
@ -80,7 +80,40 @@ def list_packs() -> list[dict[str, Any]]:
                head = json.load(f)
            out.append({"file": fn, "name": head.get("name") or fn,
                        "description": head.get("description") or "",
+                        "aliases": list(head.get("aliases") or []),
                        "baseDate": head.get("baseDate") or "", "path": path})
        except (json.JSONDecodeError, OSError):
            continue
    return out
+
+
+def _pack_ref_key(value: str) -> str:
+    """Normalize a user-facing pack reference without imposing a customer name."""
+    value = str(value or "").strip().lower()
+    value = re.sub(r"(?:数据包?|data\s*pack|数据)$", "", value).strip()
+    return re.sub(r"[^0-9a-z\u4e00-\u9fff]+", "", value)
+
+
+def resolve_pack_reference(reference: str) -> dict[str, Any] | None:
+    """Resolve a natural-language pack reference against the registered pack catalog.
+
+    Matching is deliberately limited to metadata (file/name/aliases) so a chat
+    command can never turn an arbitrary path into a data load operation.
+    """
+    key = _pack_ref_key(reference)
+    if not key:
+        return None
+    packs = list_packs()
+    exact: list[dict[str, Any]] = []
+    loose: list[dict[str, Any]] = []
+    for pack in packs:
+        labels = [pack.get("file"), os.path.splitext(str(pack.get("file") or ""))[0],
+                  pack.get("name"), *(pack.get("aliases") or [])]
+        keys = {_pack_ref_key(label) for label in labels if label}
+        if key in keys:
+            exact.append(pack)
+            continue
+        if any(key in label_key or label_key in key for label_key in keys if label_key):
+            loose.append(pack)
+    matches = exact if exact else loose
+    return matches[0] if len(matches) == 1 else None
--- a/shared/schemas/schedule_result.schema.json
+++ b/shared/schemas/schedule_result.schema.json
@ -8,7 +8,7 @@
  "properties": {
    "versionId": { "description": "排产版本 ID", "type": "integer" },
    "versionNo": { "description": "版本号（如 V20260716-003）", "type": "string" },
-    "engineType": { "description": "引擎类型", "type": "string", "enum": ["RULE", "CP", "GA", "HYBRID", "EXTERNAL"] },
+    "engineType": { "description": "引擎类型", "type": "string", "enum": ["RULE", "CP", "GA", "HYBRID", "EXTERNAL", "OPTIMIZE"] },
    "strategy": { "description": "策略模板", "type": "string" },
    "status": { "description": "版本状态", "type": "string", "enum": ["DRAFT", "PUBLISHED", "ARCHIVED"] },
    "orderCount": { "description": "参与排产的订单项数", "type": "integer" },
--- a/tests/golden/test_optimize_cpsat_native.py
+++ 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
--- a/tests/golden/test_optimize_engine.py
+++ b/tests/golden/test_optimize_engine.py
@ -0,0 +1,119 @@
+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
--- a/tests/golden/test_optimize_simulation_world_pack.py
+++ b/tests/golden/test_optimize_simulation_world_pack.py
@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import copy
+import json
+from datetime import date
+from pathlib import Path
+
+from server.aps_domain.closed_loop_runtime import run_closed_loop_candidate
+from server.aps_domain.views import due_view
+from server.state.packs import load_pack, list_packs
+
+
+PACK_PATH = Path(__file__).resolve().parents[2] / "server" / "data" / "packs" / "optimize-simulation-v1.json"
+RULES = ("EDD", "SPT", "PRIORITY", "FIFO", "LPT", "CR", "ATC")
+
+
+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_simulation_world_pack_is_a_generic_aps_fixture():
+    pack = json.loads(PACK_PATH.read_text(encoding="utf-8"))
+    world = load_pack(str(PACK_PATH))
+
+    assert pack["packVersion"] == 1
+    assert pack["name"] == "Optimize synthetic debug V1"
+    assert world["flexParams"]["dataGrade"] == "synthetic"
+    assert len(world["flexOrders"]) == 10
+    assert len(world["salesOrders"]) == 10
+    assert len(world["flexRoutings"]) == 72
+    assert len(world["flexEquipment"]) == 6
+    assert all(row["source"] == "SYNTHETIC" for row in world["flexOrders"])
+    assert {row["productCode"] for row in world["flexRoutings"]} == {
+        row["productCode"] for row in world["flexOrders"]
+    }
+    assert any(item["file"] == PACK_PATH.name for item in list_packs())
+
+
+def test_simulation_world_pack_runs_all_optimize_dispatch_rules():
+    source_world = load_pack(str(PACK_PATH))
+
+    for rule in RULES:
+        world = copy.deepcopy(source_world)
+        result = run_closed_loop_candidate(
+            world,
+            _next_id_factory(),
+            business_date="2026-08-02",
+            engine_type="OPTIMIZE",
+            sort_mode=rule,
+            strict=True,
+        )
+
+        assert result["solveStatus"] == "FEASIBLE", rule
+        assert result["engineType"] == "OPTIMIZE", rule
+        assert result["algorithmId"] == f"optimize.{rule.lower()}", rule
+        assert result["orderCount"] == 10, rule
+        assert result["vlCount"] == 10, rule
+        assert result["woCount"] == 72, rule
+        assert result["validation"]["valid"] is True, rule
+        assert result["validation"]["hardViolations"] == [], rule
+
+
+def test_simulation_world_pack_keeps_sales_order_fields_the_due_board_reads():
+    """交期承诺看板直取订单客户字段，数据包必须按订单契约补齐（回归：/api/world/due 500）。"""
+    world = load_pack(str(PACK_PATH))
+
+    rows = due_view(world)
+
+    assert len(rows) == len(world["salesOrders"]) == 10
+    assert all(row["customerName"] and row["customerLevel"] for row in rows)
+    assert all(isinstance(row["isRush"], bool) for row in rows)
+    assert any(row["isRush"] for row in rows)                  # 至少一张插单，加急标记在 UI 有数据可验
+    assert {row["customerLevel"] for row in rows} >= {"VIP", "A", "B", "C"}  # 等级过滤有数据可筛
+
+
+def test_simulation_world_pack_timeline_follows_its_case_business_day():
+    """回归：baseDate 必须是案例业务日，否则世界日期随运行日漂移、黄金测试输入每天变。
+
+    源案例（kangni-simulation-package-v1）的业务日交期结构固定为「9 张逾期 + 最晚交期在业务日 +6 天」，
+    世界日期整体贴着 baseDate 平移后，这个结构在任意运行日都应原样落在"今天"上。
+    """
+    world = load_pack(str(PACK_PATH))
+    due_dates = sorted(date.fromisoformat(order["dueDate"]) for order in world["flexOrders"])
+    today = date.today()
+
+    assert (due_dates[-1] - today).days == 6           # 最晚交期 = 案例业务日 +6 天
+    assert sum(1 for due in due_dates if due < today) == 9   # 案例本身是 9 张逾期单
Author	SHA1	Message	Date
ssk	65b539e010	接入 V2 原生 CP-SAT 求解器（round 89 slice 1）新增 server/engines/optimize_cpsat.py：SchedulingProblemV2 -> CP-SAT -> SchedulingSolutionV2，目标最小化总拖期；含设备日历与维保阻塞区间、同设备不重叠、工序链前驱约束、合格设备二选一、provenance 绑定 problem hash。本轮只做问题到解，不物化 flex* 行（slice 2 负责）。验证：隔离解释器（numpy 1.26.4 + ortools 9.11）2 passed，72/72 工序排入且 validate_solution valid；项目 venv 因 NumPy X86_V2 基线与本机 CPU 不兼容按策略 skip。	2026-09-16 17:32:32 +08:00
ssk	7de23a0ccd	修复模拟数据包订单契约与日期基准，消除交期看板 500 /api/world/due 直取订单客户字段，而模拟数据包只填了 8 个字段，导致 KeyError 500。补齐 10 张订单的订单契约字段（客户/等级/加急/下单日等），并把 baseDate 从转换日 2026-09-03 改为案例业务日 2026-08-02，避免世界日期随运行日漂移、黄金测试输入每天变化。验证：隔离实例 GET /api/world/due 返回 200（10 行）；定向测试 8 passed, 1 skipped。	2026-09-16 17:32:29 +08:00
ssk	cc151c6fc3	Merge remote-tracking branch 'origin/main' into codex/integrate-optimize	2026-09-16 15:30:39 +08:00
ssk	9071602cbf	合并最新主线更新，放弃规则解析器路径合入 origin/main@88930a1（第87轮主数据与排产入口）。冲突解决：intent.py 按主线删除，对话入口统一走 Pi 通道，并移除依赖该模块的 round-87/88 规则测试；pool_engine.py 同时保留主线 trial 与本分支 dispatch_rule；ChatPanel.tsx 同时保留数据/算法来源徽标与主线下载失败提示。验证：前端 tsc+vite build 通过；golden 套件与未改动主线失败集合逐条一致（76 项均为 CP-SAT/DXF/sidecar 环境失败），无新增失败，通过数 2246 高于主线 2236。	2026-09-14 16:11:54 +08:00
ssk	3781b67881	合并最新主线更新合入 origin/main@f9873ee，包含 Pi Agent 完整接入、闭环浏览器 E2E 稳定化及生成文件清理。同步修正第85轮文档中的本机绝对路径，以满足新主线可移植性门禁。验证：核心与 Pi Agent 回归 564 通过、2 跳过，仅路径门禁初次失败；修复后可移植性与 Optimize 回归 478 通过、1 跳过；前端构建通过。	2026-09-08 16:00:25 +08:00
ssk	b8fb217c09	修复显式数据源串用旧数据支持通过目录和文件名加载标准 APS JSON 数据包；表格目录改为先导入独立候选世界，未读取到文件或数据不完整时拒绝排产，避免沿用当前会话数据。补充 JSON 解析、成功排产和空目录防串数据测试。	2026-09-04 16:55:33 +08:00
ssk	cd8806f649	合并第88轮对话排产集成支持在对话中指定目录或文件数据，并选择 Optimize、EDD、SPT、FIFO 等算法生成排产方案；同时补充通用数据加载、结果展示和回归测试。GitNexus 提交前检查为 LOW：11 个文件、无额外受影响流程。	2026-09-04 16:42:46 +08:00
ssk	9df02cbac4	合并最新主线更新合入 main@35131cd，包含 Pi Agent 回退通道、功能开关、产品手册、康尼材料和锐扬演示数据等更新。合并无文本冲突；GitNexus 因 104 个代码文件及 64 条关联流程将影响范围评为 CRITICAL，统一验证在后续集成完成后执行。	2026-09-04 16:41:59 +08:00
ssk	f03596b134	支持目录文件指定算法排产扩展对话解析以识别目录、文件和算法三个槽位，复用项目深度分析读取 Excel/CSV/SQL，并在路径不存在时阻断旧数据回退。关键洞察是显式文件输入必须先做路径校验，再进入现有 Optimize 闭环。已验证 29 个回归测试通过，并用 HTTP SSE 实测不存在路径会直接返回错误。	2026-09-03 22:24:48 +08:00
ssk	1790a23a2d	支持对话指定数据包与排产算法为柔性排产增加数据包引用和 Optimize dispatch rule 解析，使用注册表安全加载世界数据并在结果卡片展示数据与算法证据。关键洞察是把数据源和算法作为两个独立槽位复用现有闭环管线，避免客户专用 loader 和重复调度逻辑。已验证 12 个定向测试、19 个闭环回归、前端构建及独立 HTTP SSE 聊天链路，实测产出 10 条虚拟产线和 72 个工单。	2026-09-03 21:28:19 +08:00
ssk	1e64e9c0ae	规划对话指定数据包与算法新增本轮对话式数据包与 Optimize 算法选择计划，明确通用注册表解析、闭环调用、验收命令和合并前风险边界。关键洞察是把数据与算法作为同一条指令的两个独立槽位，复用现有柔性排产管线。	2026-09-03 21:11:32 +08:00
ssk	5485f6de62	支持对话触发 Optimize 柔性排产	2026-09-03 21:04:23 +08:00
ssk	6478e8c512	修正模拟订单的通用投影来源	2026-09-03 19:27:39 +08:00
ssk	ac6026089e	加入 Optimize 模拟 world pack	2026-09-03 19:18:54 +08:00
ssk	c64fca09e1	整理通用 Optimize 模拟数据包	2026-09-03 19:18:42 +08:00
ssk	e1fd636790	合并第85轮 Optimize 集成将 OptimizeEngine、V2 闭环适配、算法元数据和验证资产合并到 codex/integrate-optimize。集成继续复用 APS WorldStore 与校验边界，真实 MOM 文件缺失和 CP-SAT 环境依赖仍需后续补齐。	2026-09-03 17:56:15 +08:00
ssk	a5ecdf3820	校正第85轮集成目标分支将轮次计划中的目标分支统一为 codex/integrate-optimize，与当前实际合并方向一致，避免把 round/85 分支误合并到 main。	2026-09-03 17:48:08 +08:00
ssk	abc9a3037c	修正 Optimize 引擎文件格式移除文件末尾多余空行，保持本轮集成代码通过 diff 检查。Optimize 引擎逻辑未改变；相关 Optimize 与闭环测试已通过。	2026-09-03 17:45:58 +08:00
ssk	818885951c	补充真实 MOM 数据验证	2026-09-03 13:27:27 +08:00
ssk	9aec353c73	接入 Optimize 柔性 V2 闭环新增 Python OptimizeEngine 和七种派工规则，复用 APS 能力池与物化逻辑。通过 flex.schedule 的 engine=OPTIMIZE 进入闭环 V2，记录 solver、algorithm、provenance，并在准入阻断时保留引擎身份且不产生工单。新增 Optimize/V2/注册表黄金测试；CP-SAT 原生 V2 求解与真实 MOM world 留待后续环境和轮次验证。	2026-09-03 13:14:04 +08:00
ssk	6ebf90393f	规划 optimize 集成形态明确以 aps-agent WorldStore 和 closed-loop V2 为唯一业务主链，optimize 作为 Python 原生求解器进入 Solver 层；计划覆盖真实 MOM 验收、V2 校验、provenance 和迁移停止条件。已完成 git diff --check。	2026-09-03 12:37:12 +08:00