refactor: 试排轨道判定收敛为 params.is_trial_world 唯一入口

run_flex_schedule 与 rush.apply 不再各自读 planningContext.trialOnly，
统一经 params.is_trial_world 分流，消除双处字段判断漂移的隐患。
rush/flex/params golden 全过。

server/aps_domain/flex.py

@@ -298,8 +298,10 @@ def run_flex_schedule(store, sort_mode: str | None = None, order_ids: list[int]
     from server.state.seed import ensure_flex_seed
 
     ensure_flex_seed(store.data)
+    from server.aps_domain.params import is_trial_world
+
     planning_context = store.data.get("planningContext") or {}
-    if planning_context.get("trialOnly"):
+    if is_trial_world(store.data):
         trial = True
     if start_date is None and planning_context.get("planStart"):
         start_date = str(planning_context["planStart"])[:10]

server/aps_domain/params.py

@@ -26,6 +26,16 @@ def default_schedule_params() -> dict[str, Any]:
     }
 
 
+def is_trial_world(world: World) -> bool:
+    """是否试排世界（文件夹快试导入）：排产/重排统一走柔性能力池草稿轨。
+
+    轨道判定唯一入口：run_flex_schedule、rush.apply 等所有排产触发点
+    都经此函数分流，避免各处自写字段判断导致同一世界走出两条轨道。
+    """
+    planning_context = world.get("planningContext") or {}
+    return bool(planning_context.get("trialOnly"))
+
+
 def get_schedule_params(world: World) -> dict[str, Any]:
     """合并默认值后的排产参数投影（P0）。"""
     base = default_schedule_params()

server/aps_domain/rush.py

@@ -328,8 +328,9 @@ def apply_rush(world: World, next_id, payload: dict[str, Any]) -> dict[str, Any]
     rush = resolve_rush_payload(world, payload.get("payload") or payload)
     strategy = str(payload.get("strategy") or "DELIVERY_FIRST").upper()
     order = inject_rush_trunk(world, next_id, rush)
-    planning_context = world.get("planningContext") or {}
-    if planning_context.get("trialOnly"):
+    from server.aps_domain.params import is_trial_world
+
+    if is_trial_world(world):
         # 试排世界必须走柔性能力池引擎：主轨 RULE 引擎读不到 flex* 主数据，
         # 全量重排只会产出 0 工单的空壳版本，排产推演/查看排产结果都看不到急单。
         # 改为先把主干急单桥接进柔性订单池，再重跑 PoolEngine 试排（与 run_flex_schedule 同轨）。