fix(flex): 对话新建急单排不进柔性排产——主干订单桥接 + 保留计划员单号

- 根因1：data.import/order.upsert 只写主干 salesOrders，PoolEngine 只读
  flexOrders，导致计划员对话新建的急单在重排时完全不可见
- 根因2：新建订单强制使用 SO+日期 自动单号，计划员指定的 RUSH-0007
  被丢弃，后续按单号查询/引用全部落空
- 新增 sync_sales_orders_to_flex：试排世界每次柔性排产前把可排主干单
  （APPROVED/CONFIRMED）幂等投影为 RELEASED 柔性单；主干取消/完成/
  退审同步回撤出可排集合；source=SALES 标记防止 flex→sales 回投重复
- intake/order.upsert 接受并保留指定 orderNo，双池查重 fail-closed
- 作用域收敛：仅 trial 试排世界桥接，演示厂维持经典/柔性双轨隔离
- 验证：锐扬资料世界五场景（排入/查重/幂等/回撤/兼容）+ golden 全量
  2378 通过 0 失败

server/aps_domain/flex.py

@@ -249,7 +249,7 @@ def run_flex_schedule(store, sort_mode: str | None = None, order_ids: list[int]
     from server.agent_core.audit import write_audit
     from server.aps_domain.closed_loop_runtime import run_closed_loop_candidate
     from server.aps_domain.mrp import decompose_orders
-    from server.aps_domain.orders import sync_flex_orders_to_sales
+    from server.aps_domain.orders import sync_flex_orders_to_sales, sync_sales_orders_to_flex
     from server.aps_domain.sourcing import annotate_world_sourcing
     from server.state.seed import ensure_flex_seed
 
@@ -279,6 +279,10 @@ def run_flex_schedule(store, sort_mode: str | None = None, order_ids: list[int]
 
     def execute_pipeline() -> dict:
         if explicit_demo or trial:
+            # 试排世界：主干对话/手工单（含紧急插单）桥接进柔性订单池，否则排产看不到它们；
+            # 演示厂保持经典轨/柔性轨双轨隔离，不桥接
+            if trial:
+                sync_sales_orders_to_flex(store.data)
             result = PoolEngine().solve(
                 store.data,
                 store.next_id,

server/aps_domain/intake.py

@@ -62,6 +62,12 @@ def _normalize_order_row(world: World, row: Any, index: int) -> tuple[dict[str,
     if level not in CUSTOMER_LEVELS:
         return None, f"第{index}行 customerLevel 必须是 VIP/A/B/C"
     is_rush = bool(row.get("isRush", False))
+    order_no = str(row.get("orderNo") or "").strip()
+    if order_no:
+        folded = order_no.casefold()
+        for pool in ("salesOrders", "flexOrders"):
+            if any(str(o.get("orderNo") or "").casefold() == folded for o in (world.get(pool) or [])):
+                return None, f"第{index}行 orderNo「{order_no}」已存在"
     priority = row.get("priority")
     if priority is None:
         priority = 1 if is_rush else 5
@@ -73,6 +79,7 @@ def _normalize_order_row(world: World, row: Any, index: int) -> tuple[dict[str,
     return {
         "customerName": customer,
         "customerLevel": level,
+        "orderNo": order_no,
         "productId": int(product["id"]),
         "productCode": product["code"],
         "productName": product["name"],

server/aps_domain/orders.py

@@ -120,7 +120,8 @@ def sync_flex_orders_to_sales(world: World) -> int:
     }
     flex: list[dict[str, Any]] = []
     for order in world.get("flexOrders") or []:
-        if str(order.get("source") or "").upper() in {"CLOSED_LOOP", "MRP"}:
+        # SALES 源柔性单来自主干销售订单投影，不反向回投，避免重复建单
+        if str(order.get("source") or "").upper() in {"CLOSED_LOOP", "MRP", "SALES"}:
             continue
         if str(order.get("status") or "").upper() in {"DONE", "CANCELLED", "COMPLETED"}:
             continue
@@ -257,6 +258,66 @@ def sync_flex_orders_to_sales(world: World) -> int:
         world["salesOrders"] = next_rows
     return changed
 
+
+SALES_FLEX_SOURCE = "SALES"  # 主干销售订单 → 柔性订单的桥接来源标记
+
+
+def sync_sales_orders_to_flex(world: World) -> int:
+    """把主干可排销售订单幂等投影为柔性订单（salesOrders → flexOrders）。
+
+    对话/手工新建的订单（含紧急插单）只落在主干 salesOrders；柔性 PoolEngine
+    只读 flexOrders，导致「订单已导入却排不进去」。每次柔性排产前调用本投影：
+    APPROVED/CONFIRMED 的主干单桥接为 RELEASED 柔性单；主干状态回撤
+    （取消/完成/退回审核）同步把柔性单移出可排集合。文件/SQL/闭环来源的
+    柔性单仍由各自通道维护，本函数只认领 source=SALES 的行。
+    """
+    flex_orders = world.setdefault("flexOrders", [])
+    by_no = {str(o.get("orderNo") or ""): o for o in flex_orders}
+    next_fid = max(
+        (int(o.get("id")) for o in flex_orders if isinstance(o.get("id"), int)),
+        default=0,
+    ) + 1
+    changed = 0
+    for so in world.get("salesOrders") or []:
+        order_no = str(so.get("orderNo") or "").strip()
+        if not order_no:
+            continue
+        existing = by_no.get(order_no)
+        if existing is not None and str(existing.get("source") or "").upper() != SALES_FLEX_SOURCE:
+            continue  # 同名柔性单由文件/SQL/闭环通道维护，不覆盖
+        status = str(so.get("status") or "").upper()
+        if is_schedulable(status):
+            item = (so.get("items") or [{}])[0]
+            if not item.get("productCode"):
+                continue
+            values = {
+                "productCode": item.get("productCode"),
+                "productName": item.get("productName"),
+                "quantity": item.get("quantity"),
+                "dueDate": str(so.get("deliveryDate") or "")[:10],
+                "priority": int(so.get("priority") or 5),
+                "customerName": so.get("customerName"),
+                "customerLevel": so.get("customerLevel") or "B",
+                "isRush": bool(so.get("isRush")),
+                "status": "RELEASED",
+                "source": SALES_FLEX_SOURCE,
+                "salesOrderNo": order_no,
+            }
+            if existing is None:
+                flex_orders.append({"id": next_fid, "orderNo": order_no, **values})
+                next_fid += 1
+                changed += 1
+            elif any(existing.get(k) != v for k, v in values.items()):
+                existing.update(values)
+                changed += 1
+        elif existing is not None:
+            # 主干状态回撤：柔性投影同步退出可排集合
+            target = {"CANCELLED": "CANCELLED", "COMPLETED": "DONE"}.get(status, "ON_HOLD")
+            if existing.get("status") != target:
+                existing["status"] = target
+                changed += 1
+    return changed
+
 def list_orders(world: World) -> list[dict[str, Any]]:
     """销售订单列表投影（含已投影的 SQL/柔性单；只读兜底不写盘）。"""
     rows: list[dict[str, Any]] = []
@@ -450,6 +511,7 @@ def normalize_order_payload(world: World, payload: dict[str, Any]) -> dict[str,
     out["customerName"] = str(out.get("customerName") or "").strip()
     if not out["customerName"]:
         raise ValueError("客户名称不能为空")
+    out["orderNo"] = str(out.get("orderNo") or "").strip()
     out["customerLevel"] = str(out.get("customerLevel") or "A").upper()
     if out["customerLevel"] not in CUSTOMER_LEVELS:
         raise ValueError("客户等级必须是 VIP/A/B/C")
@@ -649,9 +711,22 @@ def apply_order_action(world: World, next_id, action: str, payload: dict[str, An
         order_id = next_id("salesOrder")
         order_date = today0().strftime("%Y-%m-%d")
         product = _product(world, p["productId"])
+        order_no = p.get("orderNo") or ""
+        if order_no:
+            # 计划员指定单号（如急单 RUSH-0007）：保留并查重，失败即拒绝
+            folded = order_no.casefold()
+            clash = any(
+                str(o.get("orderNo") or "").casefold() == folded
+                for pool in ("salesOrders", "flexOrders")
+                for o in (world.get(pool) or [])
+            )
+            if clash:
+                raise ValueError(f"订单号 {order_no} 已存在")
+        else:
+            order_no = "SO" + today0().strftime("%Y%m%d") + f"{order_id:03d}"
         order = {
             "id": order_id,
-            "orderNo": "SO" + today0().strftime("%Y%m%d") + f"{order_id:03d}",
+            "orderNo": order_no,
             "customerId": f"CUST{order_id:03d}",
             "customerName": p["customerName"],
             "customerLevel": p["customerLevel"],