【Feature】新增mindie-motor_standalone_coordinator部署模式-MindIE-Motor-AtomGit

M.agents/skills/motor-dev/references/coordinator.md+16-0

	@@ -328,6 +328,22 @@ looks up the registered instance first and submits its actual role, job name, mo
328	Duplicate IDs in one request return 400; identity conflicts return 409. Membership write is in-process	328	Duplicate IDs in one request return 400; identity conflicts return 409. Membership write is in-process
329	(`apply_refresh`); there is no separate Scheduler ACK.	329	(`apply_refresh`); there is no separate Scheduler ACK.
330		330
		331	+Standalone External Deployer events are selected by `_is_standalone_instance_refresh()` in Mgmt: top-level
		332	+`model_name` / `dispatch_capabilities` / `engine_type`, or `instances[].endpoints` as an address array without
		333	+`job_name`. Each External endpoint carries an explicit `id` (DP rank, unique in the instance) plus `address`. Controller InsEventMsg uses per-instance `job_name` and dict-shaped endpoints. When all standalone
		334	+top-level fields are omitted, the endpoint-array instance shape is the discriminator. Mixed shapes in one request
		335	+are rejected. With `model_name` absent, Mgmt queries `/v1/models` through `NativeEngineApiClient` in instance/endpoint
		336	+order until it finds exactly one model ID. That call uses the blocking `requests` client (`timeout=2` per endpoint) and
		337	+is offloaded with `asyncio.to_thread` so `/liveness`, `/readiness`, and `GET /instances` stay responsive while engines
		338	+are still starting. An empty or multi-model response requires an explicit `model_name`. The configured AIGW model, when
		339	+present, remains the final case-insensitive consistency check. External `dispatch_capabilities` accepts only
		340	+`prefill_handoff_decode` (default) and `concurrent_engine_sync`; `decode_colocation` is connector-derived Decode-only
		341	+and is rejected at parse time so it cannot be stamped onto every P/D instance.
		342	+
		343	+`ControllerApiClient.report_alarms` skips Controller loading when `has_motor_controller_config()` is false: missing
		344	+`USER_CONFIG_PATH`, missing file, empty/invalid JSON, or a dict without `motor_controller_config`. Do not treat a
		345	+missing config as Controller-present; that would POST defaults to `127.0.0.1:1026` (same as Coordinator mgmt).
		346	+
331	`motor.coordinator.domain` keeps its package-level compatibility exports (for example `InstanceReadiness` and	347	`motor.coordinator.domain` keeps its package-level compatibility exports (for example `InstanceReadiness` and
332	`RequestManager`) behind module `__getattr__` lazy loading. Domain submodules are imported directly by Coordinator models,	348	`RequestManager`) behind module `__getattr__` lazy loading. Domain submodules are imported directly by Coordinator models,
333	so `domain/__init__.py` must not eagerly import modules that depend on `models.request`; doing so creates a	349	so `domain/__init__.py` must not eagerly import modules that depend on `models.request`; doing so creates a

Mdocs/zh/user_guide/api/management_interfaces.md+63-0

	@@ -312,6 +312,69 @@ curl -X POST "http://{IP}:{Port}/instances/refresh" \
312	\| data.event_type \| string \| 事件类型，与请求`event`对应。 \|	312	\| data.event_type \| string \| 事件类型，与请求`event`对应。 \|
313	\| data.instance_count \| integer \| 实例数量。 \|	313	\| data.instance_count \| integer \| 实例数量。 \|
314		314
		315	+### External Deployer 精简实例事件
		316	+
		317	+Coordinator 独立部署复用同一 URL，支持 `event=set/add/del` 的精简协议。原 Controller
		318	+完整协议保持不变。请求满足以下任一条件时按精简协议解析：顶层包含 `model_name`、
		319	+`dispatch_capabilities` 或 `engine_type`；或 `instances[].endpoints` 为
		320	+`[{"address": "host:port"}]` 数组且实例不含 `job_name`。未命中上述形态时按 Controller
		321	+完整协议解析。同一请求混用两种实例形态时返回错误。
		322	+
		323	+精简请求顶层字段：
		324	+
		325	+\| 参数名 \| 类型 \| 必填 \| 说明 \|
		326	+\|---\|---\|---\|---\|
		327	+\| `event` \| string \| 是 \| `set` 全量替换、`add` 加入纳管、`del` 踢出纳管。 \|
		328	+\| `model_name` \| string \| 否（多模型时必填） \| 单模型时可省略，Coordinator 从第一个可达引擎 `/v1/models` 自动发现；引擎返回多个模型时必须显式指定。若 Coordinator 配置了 `motor_coordinator_config.aigw.id`，须与其一致（忽略大小写）。 \|
		329	+\| `dispatch_capabilities` \| string \| 否 \| 仅允许 `prefill_handoff_decode`（默认，handoff）与 `concurrent_engine_sync`（trigger/sync）。`decode_colocation` 等其余值会被拒绝。 \|
		330	+\| `engine_type` \| string \| 否 \| 默认 `vllm`；当前独立精简协议仅支持 `vllm`。 \|
		331	+\| `instances` \| array \| 是 \| 待全量或增量处理的实例。 \|
		332	+\| `instances[].id` \| integer \| 是 \| 全局唯一实例 ID，必须大于 0。 \|
		333	+\| `instances[].role` \| string \| 是 \| 仅支持 `prefill` 或 `decode`。 \|
		334	+\| `instances[].endpoints` \| array \| 是 \| 至少一个业务 Endpoint。 \|
		335	+\| `endpoints[].id` \| integer \| 是 \| 实例内 DP 序号，整数 ≥ 0；同一实例内不可重复。 \|
		336	+\| `endpoints[].address` \| string \| 是 \| 引擎 `host:port`；IPv6 使用 `[addr]:port`。 \|
		337	+
		338	+```JSON
		339	+{
		340	+ "event": "set",
		341	+ "model_name": "qwen3-8b",
		342	+ "instances": [
		343	+ {
		344	+ "id": 1,
		345	+ "role": "prefill",
		346	+ "endpoints": [
		347	+ {"id": 0, "address": "10.0.0.11:8000"}
		348	+ ]
		349	+ },
		350	+ {
		351	+ "id": 2,
		352	+ "role": "prefill",
		353	+ "endpoints": [
		354	+ {"id": 0, "address": "10.0.0.12:8000"}
		355	+ ]
		356	+ },
		357	+ {
		358	+ "id": 3,
		359	+ "role": "decode",
		360	+ "endpoints": [
		361	+ {"id": 0, "address": "10.0.0.21:8000"}
		362	+ ]
		363	+ }
		364	+ ]
		365	+}
		366	+```
		367	+
		368	+Coordinator 会生成 `job_name`、`status` 和 `parallel_config` 等内部字段。省略
		369	+`model_name` 时由引擎 `/v1/models` 自动发现；省略 `dispatch_capabilities` 与
		370	+`engine_type` 时分别默认为 `prefill_handoff_decode` 与 `vllm`。`add/del` 只改变
		371	+Coordinator 纳管状态，不管理 P/D 生命周期；重启后必须由 External Deployer
		372	+重放完整 `set`。为避免同一实例 ID 误删已更新的实例，`del` 必须携带与注册时一致的
		373	+`role` 和 Endpoint 物理身份（外部协议中为 `address`）；不一致时接口返回
		374	+HTTP 409。
		375	+
		376	+---
		377	+
315	## 精度告警状态清理接口	378	## 精度告警状态清理接口
316		379
317	接口功能	380	接口功能

Mdocs/zh/user_guide/deployment/standalone.md+102-0

	@@ -75,6 +75,8 @@ curl http://127.0.0.1:1026/liveness
75	curl http://127.0.0.1:1026/readiness # 此时 instance_count=0 属正常	75	curl http://127.0.0.1:1026/readiness # 此时 instance_count=0 属正常
76	```	76	```
77		77
		78	+若第二节编写了 `coordinator.json`，启动前 `export USER_CONFIG_PATH=...`。
		79	+
78	---	80	---
79		81
80	## 四、注册 P/D 实例	82	## 四、注册 P/D 实例
	@@ -94,6 +96,106 @@ curl http://127.0.0.1:1026/instances # 列出已登记实例
94	curl http://127.0.0.1:1026/readiness # 调度是否就绪（需同时有可用 P 和 D）	96	curl http://127.0.0.1:1026/readiness # 调度是否就绪（需同时有可用 P 和 D）
95	```	97	```
96		98
		99	+External Deployer 也可以直接向 `POST /instances/refresh` 发送精简事件。
		100	+Coordinator standalone 当前只支持单模型。省略 `model_name` 时，Coordinator 会按
		101	+`instances[]` 顺序访问第一个可达 vLLM endpoint 的 `/v1/models`；接口只返回一个模型
		102	+ID 时自动采用该 ID。多模型场景必须在请求顶层显式携带 `model_name`。
		103	+`dispatch_capabilities` 和 `engine_type` 也可以省略并使用默认值，因此 `instances[]`
		104	+中的每个元素只需提供 `id`、`role` 和 `endpoints[]`。
		105	+
		106	+### 请求模板
		107	+
		108	+`set` — 全量替换纳管拓扑（首次部署或 Coordinator 重启后必须重放）：
		109	+
		110	+```json
		111	+{
		112	+ "event": "set",
		113	+ "instances": [
		114	+ {
		115	+ "id": 1,
		116	+ "role": "prefill",
		117	+ "endpoints": [
		118	+ {"id": 0, "address": "10.0.0.11:8000"}
		119	+ ]
		120	+ },
		121	+ {
		122	+ "id": 2,
		123	+ "role": "prefill",
		124	+ "endpoints": [
		125	+ {"id": 0, "address": "10.0.0.12:8000"}
		126	+ ]
		127	+ },
		128	+ {
		129	+ "id": 3,
		130	+ "role": "decode",
		131	+ "endpoints": [
		132	+ {"id": 0, "address": "10.0.0.21:8000"}
		133	+ ]
		134	+ }
		135	+ ]
		136	+}
		137	+```
		138	+
		139	+`add` — 增量加入纳管：
		140	+
		141	+```json
		142	+{
		143	+ "event": "add",
		144	+ "instances": [
		145	+ {
		146	+ "id": 4,
		147	+ "role": "prefill",
		148	+ "endpoints": [
		149	+ {"id": 0, "address": "10.0.0.13:8000"}
		150	+ ]
		151	+ }
		152	+ ]
		153	+}
		154	+```
		155	+
		156	+`del` — 增量踢出纳管（`id` 须与已登记实例一致）：
		157	+
		158	+```json
		159	+{
		160	+ "event": "del",
		161	+ "instances": [
		162	+ {
		163	+ "id": 4,
		164	+ "role": "prefill",
		165	+ "endpoints": [
		166	+ {"id": 0, "address": "10.0.0.13:8000"}
		167	+ ]
		168	+ }
		169	+ ]
		170	+}
		171	+```
		172	+
		173	+```bash
		174	+# 将上方 set 模板保存为 topology_set.json 后执行
		175	+curl -X POST http://127.0.0.1:1026/instances/refresh \
		176	+ -H 'Content-Type: application/json' \
		177	+ -d @topology_set.json
		178	+```
		179	+
		180	+### 字段说明
		181	+
		182	+\| 字段 \| 层级 \| 必填 \| 说明 \|
		183	+\|------\|------\|------\|------\|
		184	+\| `event` \| 顶层 \| 是 \| 事件类型：`set` 全量替换；`add` 增量加入；`del` 增量删除 \|
		185	+\| `model_name` \| 顶层 \| 否（多模型时必填） \| 单模型时默认从第一个可达 endpoint 的 `/v1/models` 自动获取；接口返回多个模型时必须显式指定。若 Coordinator 配置了 `motor_coordinator_config.aigw.id`，须与其一致（大小写不敏感） \|
		186	+\| `dispatch_capabilities` \| 顶层 \| 否 \| 仅允许 `prefill_handoff_decode`（默认，handoff）与 `concurrent_engine_sync`（trigger/sync）；`decode_colocation` 等其余值会被拒绝 \|
		187	+\| `engine_type` \| 顶层 \| 否 \| 默认 `vllm`；External Deployer 当前仅支持 `vllm` \|
		188	+\| `instances` \| 顶层 \| 是 \| 本批次涉及的实例列表；`set` 时为完整拓扑，`add`/`del` 时为增量子集 \|
		189	+\| `instances[].id` \| 实例 \| 是 \| 全局唯一实例 ID，整数 ≥ 1；同一请求内不可重复 \|
		190	+\| `instances[].role` \| 实例 \| 是 \| 实例角色：`prefill` 或 `decode` \|
		191	+\| `instances[].endpoints` \| 实例 \| 是 \| 该实例可达的引擎 HTTP 端点，至少 1 个；多 DP 时在同一数组内列出多个 endpoint \|
		192	+\| `endpoints[].id` \| 端点 \| 是 \| 实例内 DP 序号，整数 ≥ 0；同一实例内不可重复 \|
		193	+\| `endpoints[].address` \| 端点 \| 是 \| 引擎 `host:port`；IPv6 使用 `[addr]:port`（如 `"[2001:db8::1]:8200"`） \|
		194	+
		195	+`set` 全量替换纳管拓扑，`add` 增量加入纳管，`del` 增量踢出纳管。这些事件只改变
		196	+Coordinator 的内存实例池，不会启动、停止或销毁 P/D 进程。Coordinator 重启后，
		197	+External Deployer 必须重放完整 `set`。
		198	+
97	---	199	---
98		200
99	## 五、端到端验证	201	## 五、端到端验证

Mmotor/common/resources/http_msg_spec.py+148-2

	@@ -9,14 +9,29 @@
9	# See the Mulan PSL v2 for more details.	9	# See the Mulan PSL v2 for more details.
10		10
11	from enum import Enum	11	from enum import Enum
12	-from pydantic import BaseModel, Field, model_validator	12	+
		13	+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
13		14
14	from motor.common.logger import get_logger	15	from motor.common.logger import get_logger
15	-from motor.common.resources.instance import Instance, ParallelConfig	16	+from motor.common.resources.dispatch import DispatchPlan
		17	+from motor.common.resources.instance import InsStatus, Instance, ParallelConfig, PDRole
16	from motor.common.resources.endpoint import Endpoint, DeviceInfo, EndpointStatus	18	from motor.common.resources.endpoint import Endpoint, DeviceInfo, EndpointStatus
		19	+from motor.common.utils.net import split_address
17		20
18	logger = get_logger(__name__)	21	logger = get_logger(__name__)
19		22
		23	+_EXTERNAL_ALLOWED_DISPATCH_PLANS = (
		24	+ DispatchPlan.PREFILL_HANDOFF_DECODE,
		25	+ DispatchPlan.CONCURRENT_ENGINE_SYNC,
		26	+)
		27	+
		28	+
		29	+def _require_external_dispatch_plan(value: DispatchPlan) -> DispatchPlan:
		30	+ if value not in _EXTERNAL_ALLOWED_DISPATCH_PLANS:
		31	+ allowed = ", ".join(plan.value for plan in _EXTERNAL_ALLOWED_DISPATCH_PLANS)
		32	+ raise ValueError(f"External Deployer dispatch_capabilities must be one of: {allowed}")
		33	+ return value
		34	+
20		35
21	class ServerInfo(BaseModel):	36	class ServerInfo(BaseModel):
22	server_id: str = Field(..., description="Host IP address")	37	server_id: str = Field(..., description="Host IP address")
	@@ -166,3 +181,134 @@ class InsEventMsg(BaseModel):
166	if len(instance_ids) != len(set(instance_ids)):	181	if len(instance_ids) != len(set(instance_ids)):
167	raise ValueError("duplicate instance IDs in one request")	182	raise ValueError("duplicate instance IDs in one request")
168	return self	183	return self
		184	+
		185	+
		186	+class ExternalEndpoint(BaseModel):
		187	+ """Minimal routable endpoint supplied by an External Deployer."""
		188	+
		189	+ model_config = ConfigDict(extra="forbid")
		190	+
		191	+ id: int = Field(..., ge=0, description="Data-parallel endpoint ID")
		192	+ address: str = Field(
		193	+ ..., min_length=1, description="Engine host:port, e.g. 192.168.1.20:8200 or [2001:db8::1]:8200"
		194	+ )
		195	+
		196	+ def parsed_host_port(self) -> tuple[str, str]:
		197	+ host, port = split_address(self.address.strip())
		198	+ if not host:
		199	+ raise ValueError(f"address must contain a host, got {self.address!r}")
		200	+ if not port:
		201	+ raise ValueError(f"address must contain a port, got {self.address!r}")
		202	+ try:
		203	+ port_int = int(port)
		204	+ except (TypeError, ValueError) as exc:
		205	+ raise ValueError(f"address port must be an integer, got {port!r}") from exc
		206	+ if not 1 <= port_int <= 65535:
		207	+ raise ValueError(f"address port must be in range 1-65535, got {port_int}")
		208	+ return host, str(port_int)
		209	+
		210	+
		211	+class ExternalInstance(BaseModel):
		212	+ """Minimal P/D instance supplied by an External Deployer."""
		213	+
		214	+ model_config = ConfigDict(extra="forbid")
		215	+
		216	+ id: int = Field(..., ge=1, description="Globally unique instance ID")
		217	+ role: PDRole = Field(..., description="Only prefill or decode is supported")
		218	+ endpoints: list[ExternalEndpoint] = Field(..., min_length=1, description="Routable engine endpoints")
		219	+
		220	+
		221	+class ExternalInsEventMsg(BaseModel):
		222	+ """Standalone Coordinator instance event containing only routing information."""
		223	+
		224	+ model_config = ConfigDict(extra="forbid")
		225	+
		226	+ event: EventType
		227	+ model_name: str \| None = Field(
		228	+ default=None,
		229	+ min_length=1,
		230	+ description=(
		231	+ "The single served model. Optional in single-model deployments; "
		232	+ "Coordinator resolves it from the first reachable engine /v1/models. "
		233	+ "Required when any engine advertises multiple models."
		234	+ ),
		235	+ )
		236	+ dispatch_capabilities: DispatchPlan = Field(
		liu9月8日 `dispatch_capabilities` 直接使用整个 `DispatchPlan` 枚举，调用方可传入 `decode_colocation`，会给全部 P/D 实例只挂该能力；统一 PD 协同能力交集为空，易导致无可路由 pair / 误降级。建议 External 协议只允许 `prefill_handoff_decode` / `concurrent_engine_sync`，并在 `to_internal` 或字段校验处显式拒绝其余值。灏悦心陈 9月9日评论： 9月9日评论：
		237	+ default=DispatchPlan.PREFILL_HANDOFF_DECODE,
		238	+ description="P/D coordination plan; only prefill_handoff_decode or concurrent_engine_sync",
		239	+ )
		240	+ engine_type: str = Field(
		241	+ default="vllm", min_length=1, description="Inference engine family shared by all instances"
		242	+ )
		243	+ instances: list[ExternalInstance]
		244	+
		245	+ @field_validator("dispatch_capabilities")
		246	+ @classmethod
		247	+ def reject_non_pd_dispatch_plans(cls, value: DispatchPlan) -> DispatchPlan:
		248	+ return _require_external_dispatch_plan(value)
		249	+
		250	+ def to_internal(self, configured_model_name: str = "", resolved_model_name: str = "") -> InsEventMsg:
		251	+ if self.event not in (EventType.SET, EventType.ADD, EventType.DEL):
		252	+ raise ValueError("External Deployer only supports event='set', 'add', or 'del'")
		253	+
		254	+ model_name = (self.model_name or resolved_model_name).strip()
		255	+ if not model_name:
		256	+ raise ValueError("model_name is required when it cannot be resolved from native engine /v1/models")
		257	+ if configured_model_name and model_name.casefold() != configured_model_name.casefold():
		258	+ raise ValueError(
		259	+ f"external model_name {model_name!r} does not match configured single model {configured_model_name!r}"
		260	+ )
		261	+ internal_model_name = configured_model_name or model_name
		262	+
		263	+ dispatch_capability = _require_external_dispatch_plan(self.dispatch_capabilities).value
		264	+ engine_type = self.engine_type.strip().lower()
		265	+ if engine_type != "vllm":
		266	+ raise ValueError("External Deployer currently supports engine_type='vllm' only")
		267	+
		268	+ instance_ids: set[int] = set()
		269	+ internal_instances: list[Instance] = []
		270	+ for external_instance in self.instances:
		271	+ if external_instance.id in instance_ids:
		272	+ raise ValueError(f"duplicate instance id: {external_instance.id}")
		273	+ instance_ids.add(external_instance.id)
		274	+ if external_instance.role not in (PDRole.ROLE_P, PDRole.ROLE_D):
		275	+ raise ValueError(f"unsupported external instance role: {external_instance.role.value}")
		276	+
		277	+ endpoint_ids: set[int] = set()
		278	+ endpoint_addresses: set[tuple[str, str]] = set()
		279	+ endpoints_by_ip: dict[str, dict[int, Endpoint]] = {}
		280	+ for external_endpoint in external_instance.endpoints:
		281	+ if external_endpoint.id in endpoint_ids:
		282	+ raise ValueError(f"duplicate endpoint id {external_endpoint.id} in instance {external_instance.id}")
		283	+ endpoint_ids.add(external_endpoint.id)
		284	+ host, business_port = external_endpoint.parsed_host_port()
		285	+ address_key = (host, business_port)
		286	+ if address_key in endpoint_addresses:
		287	+ raise ValueError(
		288	+ f"duplicate endpoint address {external_endpoint.address!r} in instance {external_instance.id}"
		289	+ )
		290	+ endpoint_addresses.add(address_key)
		291	+ endpoint = Endpoint(
		292	+ id=external_endpoint.id,
		293	+ ip=host,
		294	+ business_port=business_port,
		295	+ status=EndpointStatus.NORMAL,
		296	+ )
		297	+ endpoints_by_ip.setdefault(endpoint.ip, {})[endpoint.id] = endpoint
		298	+
		299	+ internal_instances.append(
		300	+ Instance(
		301	+ job_name=f"external-{external_instance.role.value}-{external_instance.id}",
		302	+ model_name=internal_model_name,
		303	+ engine_type=engine_type,
		304	+ dispatch_capabilities=[dispatch_capability],
		305	+ id=external_instance.id,
		306	+ role=external_instance.role.value,
		307	+ status=InsStatus.ACTIVE,
		308	+ parallel_config=ParallelConfig(dp_size=len(endpoint_ids)),
		309	+ enable_multi_endpoints=True,
		310	+ endpoints=endpoints_by_ip,
		311	+ )
		312	+ )
		313	+
		314	+ return InsEventMsg(event=self.event, instances=internal_instances)

Mmotor/config/config_utils.py+19-0

	@@ -410,6 +410,25 @@ def resolve_config_json_path(json_path: str \| None) -> tuple[str \| None, Path \|
410	return json_path, Path(json_path) if json_path else None	410	return json_path, Path(json_path) if json_path else None
411		411
412		412
		413	+def has_motor_controller_config(json_path: str \| None = None) -> bool:
		414	+ """Return whether the user config file includes motor_controller_config."""
		415	+ _, config_path = resolve_config_json_path(json_path)
		416	+ if not config_path or not config_path.exists():
		417	+ return False
		418	+ try:
		419	+ with open(config_path, "r", encoding="utf-8") as f:
		420	+ content = f.read().strip()
		421	+ if not content:
		422	+ return False
		423	+ raw = json.loads(content)
		424	+ if not isinstance(raw, dict):
		425	+ return False
		426	+ return ConfigKey.MOTOR_CONTROLLER.value in raw
		427	+ except (json.JSONDecodeError, OSError) as e:
		428	+ logger.warning("Failed to inspect motor_controller_config presence: %s", e)
		429	+ return False
		430	+
		431	+
413	def apply_config_path_metadata(config: Any, config_path: Path \| None) -> None:	432	def apply_config_path_metadata(config: Any, config_path: Path \| None) -> None:
414	if config_path:	433	if config_path:
415	config.config_path = str(config_path)	434	config.config_path = str(config_path)

Mmotor/coordinator/api_client/conductor_api_client.py+7-1

	@@ -95,8 +95,11 @@ class ConductorApiClient:
95	@classmethod	95	@classmethod
96	def register_kv_instance(cls, instances: list[Instance]) -> None:	96	def register_kv_instance(cls, instances: list[Instance]) -> None:
97	"""Register all KVA-eligible instance endpoints with the KV conductor."""	97	"""Register all KVA-eligible instance endpoints with the KV conductor."""
98	- logger.info("register_kv_instance started.")
99	reg = cls._kv_reg()	98	reg = cls._kv_reg()
		99	+ if not reg.conductor_service:
		100	+ logger.debug("conductor_service is empty; skip KV conductor instance registration")
		101	+ return
		102	+ logger.info("register_kv_instance started.")
100	mode = cls._resolve_backend_mode()	103	mode = cls._resolve_backend_mode()
101	sb = cls._resolve_store_backend()	104	sb = cls._resolve_store_backend()
102		105
	@@ -117,6 +120,9 @@ class ConductorApiClient:
117	@classmethod	120	@classmethod
118	def unregister_kv_instance(cls, instances: list[Instance]) -> None:	121	def unregister_kv_instance(cls, instances: list[Instance]) -> None:
119	"""Unregister all KVA-eligible instance endpoints from the KV conductor."""	122	"""Unregister all KVA-eligible instance endpoints from the KV conductor."""
		123	+ if not cls._kv_reg().conductor_service:
		124	+ logger.debug("conductor_service is empty; skip KV conductor instance unregistration")
		125	+ return
120	logger.info("unregister_kv_instance started.")	126	logger.info("unregister_kv_instance started.")
121		127
122	for instance in instances:	128	for instance in instances:

Mmotor/coordinator/api_client/controller_api_client.py+27-5

	@@ -11,6 +11,7 @@
11		11
12	from motor.common.http.http_client import SafeHTTPSClient	12	from motor.common.http.http_client import SafeHTTPSClient
13	from motor.common.logger import get_logger	13	from motor.common.logger import get_logger
		14	+from motor.config.config_utils import has_motor_controller_config
14	from motor.config.controller import ControllerConfig	15	from motor.config.controller import ControllerConfig
15	from motor.config.coordinator import CoordinatorConfig	16	from motor.config.coordinator import CoordinatorConfig
16	from motor.coordinator.domain.probe import is_master_from_role_shm	17	from motor.coordinator.domain.probe import is_master_from_role_shm
	@@ -19,8 +20,20 @@ logger = get_logger(__name__)
19		20
20		21
21	class ControllerApiClient:	22	class ControllerApiClient:
22	- controller_config = ControllerConfig.from_json()	23	+ controller_config: ControllerConfig \| None = None
23	- coordinator_config = CoordinatorConfig.from_json()	24	+ coordinator_config: CoordinatorConfig \| None = None
		25	+
		26	+ @classmethod
		27	+ def _get_coordinator_config(cls) -> CoordinatorConfig:
		28	+ if cls.coordinator_config is None:
		29	+ cls.coordinator_config = CoordinatorConfig.from_json()
		30	+ return cls.coordinator_config
		31	+
		32	+ @classmethod
		33	+ def _get_controller_config(cls) -> ControllerConfig:
		34	+ if cls.controller_config is None:
		35	+ cls.controller_config = ControllerConfig.from_json()
		36	+ return cls.controller_config
24		37
25	@classmethod	38	@classmethod
26	def report_alarms(cls, params: dict) -> dict:	39	def report_alarms(cls, params: dict) -> dict:
	@@ -32,7 +45,16 @@ class ControllerApiClient:
32	"precision_alarm_cleared_by_auto_recovery": False,	45	"precision_alarm_cleared_by_auto_recovery": False,
33	}	46	}
34	try:	47	try:
35	- if cls.coordinator_config.standby_config.enable_master_standby:	48	+ coordinator_config = cls._get_coordinator_config()
		49	+ if not has_motor_controller_config(coordinator_config.config_path):
		50	+ logger.debug("No motor_controller_config in user config; skip alarm reporting.")
		51	+ return {
		52	+ "ok": True,
		53	+ "precision_alarm_cleared": False,
		54	+ "precision_alarm_cleared_by_auto_recovery": False,
		55	+ }
		56	+
		57	+ if coordinator_config.standby_config.enable_master_standby:
36	if not is_master_from_role_shm():	58	if not is_master_from_role_shm():
37	logger.debug("The standby coordinator does not need to report alarms.")	59	logger.debug("The standby coordinator does not need to report alarms.")
38	return {	60	return {
	@@ -97,8 +119,8 @@ class ControllerApiClient:
97		119
98	@classmethod	120	@classmethod
99	def _generate_client_args(cls) -> dict[str, str]:	121	def _generate_client_args(cls) -> dict[str, str]:
100	- api_config = cls.controller_config.api_config	122	+ api_config = cls._get_controller_config().api_config
101	- tls_config = cls.coordinator_config.mgmt_tls_config	123	+ tls_config = cls._get_coordinator_config().mgmt_tls_config
102	address = f"{api_config.controller_api_dns}:{api_config.controller_api_port}"	124	address = f"{api_config.controller_api_dns}:{api_config.controller_api_port}"
103	client_ars = {"address": f"{address}", "tls_config": tls_config}	125	client_ars = {"address": f"{address}", "tls_config": tls_config}
104	return client_ars	126	return client_ars

Mmotor/coordinator/api_client/native_engine_api_client.py+16-0

	@@ -18,6 +18,22 @@ logger = get_logger(__name__)
18	class NativeEngineApiClient:	18	class NativeEngineApiClient:
19	"""Read native engine operational endpoints over the inference channel."""	19	"""Read native engine operational endpoints over the inference channel."""
20		20
		21	+ @staticmethod
		22	+ def query_model_ids(address: str, tls_config: TLSConfig \| None) -> list[str]:
		23	+ """Return the model IDs advertised by a native engine."""
		24	+ with SafeHTTPSClient(address=address, tls_config=tls_config, timeout=2) as client:
		25	+ response = client.do_get("/v1/models")
		26	+ if response.status_code != 200:
		27	+ raise RuntimeError(f"native engine /v1/models returned status={response.status_code}")
		28	+ payload = response.json()
		29	+ if not isinstance(payload, dict) or not isinstance(payload.get("data"), list):
		30	+ raise ValueError("native engine /v1/models returned an invalid response")
		31	+ return [
		32	+ model["id"]
		33	+ for model in payload["data"]
		34	+ if isinstance(model, dict) and isinstance(model.get("id"), str) and model["id"]
		35	+ ]
		36	+
21	@staticmethod	37	@staticmethod
22	def query_metrics(address: str, tls_config: TLSConfig \| None) -> str:	38	def query_metrics(address: str, tls_config: TLSConfig \| None) -> str:
23	try:	39	try:

Mmotor/coordinator/api_server/management_server.py+112-11

	@@ -15,8 +15,8 @@ Does not create or start inference Workers; those are started by CoordinatorDaem
15	"""	15	"""
16		16
17	import asyncio	17	import asyncio
18	-import secrets
19	import json	18	import json
		19	+import secrets
20	from concurrent.futures import ThreadPoolExecutor	20	from concurrent.futures import ThreadPoolExecutor
21	from contextlib import asynccontextmanager	21	from contextlib import asynccontextmanager
22	from datetime import datetime, timezone	22	from datetime import datetime, timezone
	@@ -25,7 +25,7 @@ from typing import Any
25	import uvicorn	25	import uvicorn
26	from fastapi import FastAPI, HTTPException, Request, status	26	from fastapi import FastAPI, HTTPException, Request, status
27		27
28	-from motor.common.resources.http_msg_spec import InsEventMsg	28	+from motor.common.resources.http_msg_spec import ExternalInsEventMsg, InsEventMsg
29	from motor.common.http.cert_util import CertUtil	29	from motor.common.http.cert_util import CertUtil
30	from motor.common.logger import get_logger	30	from motor.common.logger import get_logger
31	from motor.common.logger.rate_limited_logger import RateLimitedLogger	31	from motor.common.logger.rate_limited_logger import RateLimitedLogger
	@@ -36,6 +36,7 @@ from motor.coordinator.api_server.base_server import BaseCoordinatorServer
36	from motor.coordinator.api_server.app_builder import AppBuilder	36	from motor.coordinator.api_server.app_builder import AppBuilder
37	from motor.coordinator.scheduler.runtime.scheduler_server import AsyncSchedulerServer	37	from motor.coordinator.scheduler.runtime.scheduler_server import AsyncSchedulerServer
38	from motor.coordinator.api_client.conductor_api_client import ConductorApiClient	38	from motor.coordinator.api_client.conductor_api_client import ConductorApiClient
		39	+from motor.coordinator.api_client.native_engine_api_client import NativeEngineApiClient
39	from motor.coordinator.domain.instance_manager import InstanceIdConflictError, InstanceManager, TYPE_MGMT	40	from motor.coordinator.domain.instance_manager import InstanceIdConflictError, InstanceManager, TYPE_MGMT
40	from motor.coordinator.domain.probe import (	41	from motor.coordinator.domain.probe import (
41	DaemonLivenessProvider,	42	DaemonLivenessProvider,
	@@ -61,6 +62,44 @@ _READINESS_503: dict[ReadinessResult, str] = {
61	_MAX_REQUEST_BODY_SIZE = 10 * 1024 * 1024 # 10MB	62	_MAX_REQUEST_BODY_SIZE = 10 * 1024 * 1024 # 10MB
62	_REQUEST_BODY_PREVIEW_LENGTH = 200	63	_REQUEST_BODY_PREVIEW_LENGTH = 200
63		64
		65	+# Top-level fields that only appear in coordinator-standalone External Deployer events.
		66	+_STANDALONE_TOP_LEVEL_MARKERS = frozenset({"model_name", "dispatch_capabilities", "engine_type"})
		67	+
		68	+
		69	+def _is_standalone_instance_refresh(body: Any) -> bool:
		70	+ """Return True when the payload follows the coordinator-standalone External protocol.
		71	+
		72	+ Controller deployments send full InsEventMsg instances (job_name plus dict endpoints).
		73	+ Standalone External Deployer events may omit all top-level optional fields and only
		74	+ provide id/role/endpoints[], where endpoints is a list of {"id": 0, "address": "host:port"}.
		75	+ """
		76	+ if not isinstance(body, dict):
		77	+ return False
		78	+ if any(marker in body for marker in _STANDALONE_TOP_LEVEL_MARKERS):
		79	+ return True
		80	+
		81	+ instances = body.get("instances")
		82	+ if not isinstance(instances, list) or not instances:
		83	+ return False
		84	+
		85	+ saw_standalone_shape = False
		86	+ saw_controller_shape = False
		87	+ for instance in instances:
		88	+ if not isinstance(instance, dict):
		89	+ continue
		90	+ if "job_name" in instance:
		91	+ saw_controller_shape = True
		92	+ continue
		93	+ endpoints = instance.get("endpoints")
		94	+ if isinstance(endpoints, list):
		95	+ saw_standalone_shape = True
		96	+ elif isinstance(endpoints, dict):
		97	+ saw_controller_shape = True
		98	+
		99	+ if saw_standalone_shape and saw_controller_shape:
		100	+ raise ValueError("mixed controller and coordinator-standalone instance payloads in one request")
		101	+ return saw_standalone_shape
		102	+
64		103
65	def _build_ok_response(message: str) -> dict[str, str]:	104	def _build_ok_response(message: str) -> dict[str, str]:
66	return {"status": "ok", "message": message}	105	return {"status": "ok", "message": message}
	@@ -581,15 +620,7 @@ class ManagementServer(BaseCoordinatorServer):
581	detail=f"Failed to parse request body: {str(e)}",	620	detail=f"Failed to parse request body: {str(e)}",
582	) from e	621	) from e
583		622
584	- try:	623	+ event_msg = await self._parse_instance_event(body)
585	- event_msg = InsEventMsg(**body)
586	- except Exception as e:
587	- body_keys = list(body.keys()) if isinstance(body, dict) else "not a dict"
588	- logger.error("Failed to parse InsEventMsg: %s, body keys: %s", e, body_keys)
589	- raise HTTPException(
590	- status_code=status.HTTP_400_BAD_REQUEST,
591	- detail=f"Invalid request format: {str(e)}",
592	- ) from e
593		624
594	async with self._refresh_lock:	625	async with self._refresh_lock:
595	try:	626	try:
	@@ -615,3 +646,73 @@ class ManagementServer(BaseCoordinatorServer):
615	"instance_count": len(event_msg.instances),	646	"instance_count": len(event_msg.instances),
616	},	647	},
617	)	648	)
		649	+
		650	+ async def _parse_instance_event(self, body: Any) -> InsEventMsg:
		651	+ """Select InsEventMsg (Controller) vs ExternalInsEventMsg (coordinator-standalone).
		652	+
		653	+ Standalone model-name discovery uses blocking HTTP (requests, timeout=2s per endpoint).
		654	+ That I/O is offloaded so /liveness, /readiness, and /instances stay responsive.
		655	+ """
		656	+ try:
		657	+ is_standalone_request = _is_standalone_instance_refresh(body)
		658	+ except ValueError as mixed_protocol_error:
		659	+ raise HTTPException(
		660	+ status_code=status.HTTP_400_BAD_REQUEST,
		661	+ detail=str(mixed_protocol_error),
		662	+ ) from mixed_protocol_error
		663	+
		664	+ if not is_standalone_request:
		665	+ try:
		666	+ return InsEventMsg.model_validate(body)
		667	+ except Exception as controller_error:
		668	+ raise HTTPException(
		669	+ status_code=status.HTTP_400_BAD_REQUEST,
		670	+ detail=f"Invalid request format: {str(controller_error)}",
		671	+ ) from controller_error
		672	+
		673	+ try:
		674	+ # Segment 1: schema-only. No engine I/O, so this stays on the event loop.
		675	+ external_msg = ExternalInsEventMsg.model_validate(body)
		676	+ aigw_model = self.coordinator_config.get_aigw_models() or {}
		677	+ # Segment 2: blocking /v1/models (requests, 2s/endpoint) runs in a worker thread.
		678	+ resolved_model_name = await asyncio.to_thread(self._resolve_external_model_name, external_msg)
		679	+ return external_msg.to_internal(str(aigw_model.get("id", "")), resolved_model_name)
		680	+ except Exception as standalone_error:
		681	+ body_keys = list(body.keys()) if isinstance(body, dict) else "not a dict"
		682	+ logger.error(
		683	+ "Failed to parse coordinator-standalone instance refresh request: error=%s body_keys=%s",
		684	+ standalone_error,
		685	+ body_keys,
		686	+ )
		687	+ raise HTTPException(
		688	+ status_code=status.HTTP_400_BAD_REQUEST,
		689	+ detail=f"Invalid request format: {str(standalone_error)}",
		690	+ ) from standalone_error
		691	+
		692	+ def _resolve_external_model_name(self, event_msg: ExternalInsEventMsg) -> str:
		693	+ """Resolve an omitted model name from the first reachable native engine."""
		694	+ if event_msg.model_name:
		695	+ return ""
		696	+
		697	+ last_error = ""
		698	+ for instance in event_msg.instances:
		699	+ for endpoint in instance.endpoints:
		700	+ address = endpoint.address.strip()
		701	+ try:
		702	+ model_ids = NativeEngineApiClient.query_model_ids(
		703	+ address,
		704	+ self.coordinator_config.infer_tls_config,
		705	+ )
		706	+ except (OSError, RuntimeError, ValueError) as exc:
		707	+ last_error = f"{address} ({exc})"
		708	+ continue
		709	+ if len(model_ids) == 1:
		710	+ return model_ids[0]
		711	+ if not model_ids:
		712	+ raise ValueError(f"{address}/v1/models returned no models; provide model_name explicitly")
		713	+ raise ValueError(
		714	+ f"{address}/v1/models serves multiple models {model_ids}; provide model_name explicitly"
		715	+ )
		716	+
		717	+ hint = f"; last error: {last_error}" if last_error else ""
		718	+ raise ValueError(f"no reachable native engine /v1/models{hint}; provide model_name explicitly")

Mtests/config/test_config_utils.py+25-2

	@@ -1,4 +1,3 @@
1	-# -- coding: utf-8 --
2	# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.	1	# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
3	# MindIE is licensed under Mulan PSL v2.	2	# MindIE is licensed under Mulan PSL v2.
4	# You can use this software according to the terms and conditions of the Mulan PSL v2.	3	# You can use this software according to the terms and conditions of the Mulan PSL v2.
	@@ -9,13 +8,16 @@
9	# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.	8	# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
10	# See the Mulan PSL v2 for more details.	9	# See the Mulan PSL v2 for more details.
11		10
12	-"""Tests for config_utils — re_register_interval_sec resolution."""	11	+"""Tests for config_utils — re_register_interval_sec resolution and controller detection."""
		12	+
		13	+import json
13		14
14	from motor.config.config_utils import (	15	from motor.config.config_utils import (
15	_resolve_re_register_interval_sec,	16	_resolve_re_register_interval_sec,
16	DEFAULT_RE_REGISTER_INTERVAL_SEC,	17	DEFAULT_RE_REGISTER_INTERVAL_SEC,
17	RE_REGISTER_INTERVAL_SEC,	18	RE_REGISTER_INTERVAL_SEC,
18	PREFILL_KV_EVENT_CONFIG,	19	PREFILL_KV_EVENT_CONFIG,
		20	+ has_motor_controller_config,
19	)	21	)
20		22
21	# The top-level key for motor_coordinator config used by the resolver.	23	# The top-level key for motor_coordinator config used by the resolver.
	@@ -84,3 +86,24 @@ class TestResolveReRegisterIntervalSec:
84	{_MOTOR_COORDINATOR_KEY: {PREFILL_KV_EVENT_CONFIG: {RE_REGISTER_INTERVAL_SEC: 0}}}	86	{_MOTOR_COORDINATOR_KEY: {PREFILL_KV_EVENT_CONFIG: {RE_REGISTER_INTERVAL_SEC: 0}}}
85	)	87	)
86	assert result == 0	88	assert result == 0
		89	+
		90	+
		91	+class TestHasMotorControllerConfig:
		92	+ def test_returns_false_when_config_path_missing(self):
		93	+ assert has_motor_controller_config(None) is False
		94	+
		95	+ def test_returns_true_when_motor_controller_config_present(self, tmp_path):
		96	+ config_path = tmp_path / "full.json"
		97	+ config_path.write_text(
		98	+ json.dumps({"motor_controller_config": {}, "motor_coordinator_config": {}}),
		99	+ encoding="utf-8",
		100	+ )
		101	+ assert has_motor_controller_config(str(config_path)) is True
		102	+
		103	+ def test_returns_false_for_standalone_coordinator_only_config(self, tmp_path):
		104	+ config_path = tmp_path / "standalone.json"
		105	+ config_path.write_text(
		106	+ json.dumps({"motor_coordinator_config": {"aigw": {"id": "test-model"}}}),
		107	+ encoding="utf-8",
		108	+ )
		109	+ assert has_motor_controller_config(str(config_path)) is False

Mtests/coordinator/api_client/test_conductor_api_client.py+1-0

	@@ -692,6 +692,7 @@ def _setup_reg_config(
692	from motor.config.coordinator import KvConductorConfig, SchedulerConfig	692	from motor.config.coordinator import KvConductorConfig, SchedulerConfig
693		693
694	reg = KvConductorConfig(	694	reg = KvConductorConfig(
		695	+ conductor_service="kv-conductor",
695	store_backend=store_backend,	696	store_backend=store_backend,
696	pool_endpoint=pool_endpoint,	697	pool_endpoint=pool_endpoint,
697	npu_endpoint=npu_endpoint,	698	npu_endpoint=npu_endpoint,

Atests/coordinator/api_client/test_controller_api_client.py+81-0

@@ -0,0 +1,81 @@
	1	+# Copyright (c) Huawei Technologies Co., Ltd. 2026. All rights reserved.
	2	+# MindIE is licensed under Mulan PSL v2.
	3	+
	4	+import json
	5	+from unittest.mock import MagicMock, patch
	6	+
	7	+from motor.config.coordinator import CoordinatorConfig
	8	+from motor.coordinator.api_client.controller_api_client import ControllerApiClient
	9	+
	10	+
	11	+def setup_function():
	12	+ ControllerApiClient.controller_config = None
	13	+ ControllerApiClient.coordinator_config = None
	14	+
	15	+
	16	+def teardown_function():
	17	+ ControllerApiClient.controller_config = None
	18	+ ControllerApiClient.coordinator_config = None
	19	+
	20	+
	21	+def test_report_alarms_skips_controller_loading_in_standalone_mode(tmp_path):
	22	+ config_path = tmp_path / "standalone.json"
	23	+ config_path.write_text(
	24	+ json.dumps(
	25	+ {
	26	+ "motor_coordinator_config": {
	27	+ "aigw": {"id": "test-model", "p_max_seqlen": 4096, "d_max_seqlen": 4096},
	28	+ }
	29	+ }
	30	+ ),
	31	+ encoding="utf-8",
	32	+ )
	33	+ config = CoordinatorConfig.from_json(str(config_path))
	34	+ ControllerApiClient.coordinator_config = config
	35	+
	36	+ with patch(
	37	+ "motor.coordinator.api_client.controller_api_client.ControllerConfig.from_json",
	38	+ side_effect=AssertionError("Controller config must not be loaded"),
	39	+ ):
	40	+ outcome = ControllerApiClient.report_alarms({"alarm_id": "standalone"})
	41	+
	42	+ assert outcome["ok"] is True
	43	+ assert ControllerApiClient.controller_config is None
	44	+
	45	+
	46	+def test_report_alarms_keeps_controller_mode_compatible(tmp_path):
	47	+ config_path = tmp_path / "full.json"
	48	+ config_path.write_text(
	49	+ json.dumps(
	50	+ {
	51	+ "motor_controller_config": {},
	52	+ "motor_coordinator_config": {},
	53	+ }
	54	+ ),
	55	+ encoding="utf-8",
	56	+ )
	57	+ coordinator_config = CoordinatorConfig.from_json(str(config_path))
	58	+ ControllerApiClient.coordinator_config = coordinator_config
	59	+
	60	+ controller_config = MagicMock()
	61	+ controller_config.api_config.controller_api_dns = "controller.example"
	62	+ controller_config.api_config.controller_api_port = 1024
	63	+ ControllerApiClient.controller_config = controller_config
	64	+
	65	+ response = MagicMock()
	66	+ response.status_code = 200
	67	+ response.json.return_value = {"data": {}}
	68	+ client = MagicMock()
	69	+ client.do_post.return_value = response
	70	+ context = MagicMock()
	71	+ context.__enter__.return_value = client
	72	+
	73	+ with patch(
	74	+ "motor.coordinator.api_client.controller_api_client.SafeHTTPSClient",
	75	+ return_value=context,
	76	+ ) as client_class:
	77	+ outcome = ControllerApiClient.report_alarms({"alarm_id": "controller"})
	78	+
	79	+ assert outcome["ok"] is True
	80	+ client_class.assert_called_once()
	81	+ client.do_post.assert_called_once_with("/observability/add_alarm", {"alarm_id": "controller"})

Mtests/coordinator/api_client/test_native_engine_api_client.py+25-0

 from unittest.mock import MagicMock, patch
++import pytest
++
 from motor.config.tls_config import TLSConfig
 from motor.coordinator.api_client.native_engine_api_client import NativeEngineApiClient
++@patch("motor.coordinator.api_client.native_engine_api_client.SafeHTTPSClient")
++def test_query_model_ids_returns_advertised_models(mock_client):
++    tls_config = TLSConfig(enable_tls=True)
++    response = MagicMock(status_code=200)
++    response.json.return_value = {"data": [{"id": "Qwen3-8B"}, {"id": ""}, {"object": "model"}]}
++    mock_client.return_value.__enter__.return_value.do_get.return_value = response
++
++    result = NativeEngineApiClient.query_model_ids("10.0.0.8:8000", tls_config)
++
++    assert result == ["Qwen3-8B"]
++    mock_client.assert_called_once_with(address="10.0.0.8:8000", tls_config=tls_config, timeout=2)
++    mock_client.return_value.__enter__.return_value.do_get.assert_called_once_with("/v1/models")
++
++
++@patch("motor.coordinator.api_client.native_engine_api_client.SafeHTTPSClient")
++def test_query_model_ids_rejects_non_success_status(mock_client):
++    response = MagicMock(status_code=503)
++    mock_client.return_value.__enter__.return_value.do_get.return_value = response
++
++    with pytest.raises(RuntimeError, match="status=503"):
++        NativeEngineApiClient.query_model_ids("10.0.0.8:8000", None)
++
++
 @patch("motor.coordinator.api_client.native_engine_api_client.SafeHTTPSClient")
 def test_query_metrics_uses_inference_tls(mock_client):
     tls_config = TLSConfig(enable_tls=True)

Mtests/coordinator/core/test_config.py+33-0

	@@ -18,6 +18,39 @@ import pytest
18	from motor.config.coordinator import CoordinatorConfig	18	from motor.config.coordinator import CoordinatorConfig
19		19
20		20
		21	+def test_standalone_coordinator_config_needs_no_engine_or_controller_sections(tmp_path):
		22	+ """A single-model standalone file loads without control-plane sections."""
		23	+ config_path = tmp_path / "standalone.json"
		24	+ config_path.write_text(
		25	+ json.dumps(
		26	+ {
		27	+ "motor_coordinator_config": {
		28	+ "aigw": {"id": "test-model", "p_max_seqlen": 4096, "d_max_seqlen": 4096},
		29	+ "scheduler_config": {"scheduler_type": "load_balance"},
		30	+ }
		31	+ }
		32	+ ),
		33	+ encoding="utf-8",
		34	+ )
		35	+
		36	+ config = CoordinatorConfig.from_json(str(config_path))
		37	+
		38	+ assert config.get_aigw_models()["id"] == "test-model"
		39	+
		40	+
		41	+def test_controller_integration_enabled_is_ignored_when_present(tmp_path):
		42	+ """Legacy standalone configs may still carry the removed key; it is ignored."""
		43	+ config_path = tmp_path / "legacy-standalone.json"
		44	+ config_path.write_text(
		45	+ json.dumps({"motor_coordinator_config": {"controller_integration_enabled": False}}),
		46	+ encoding="utf-8",
		47	+ )
		48	+
		49	+ config = CoordinatorConfig.from_json(str(config_path))
		50	+
		51	+ assert not hasattr(config, "controller_integration_enabled")
		52	+
		53	+
21	@pytest.fixture	54	@pytest.fixture
22	def _temp_json_file():	55	def _temp_json_file():
23	"""Fixture for temporary JSON file that gets cleaned up."""	56	"""Fixture for temporary JSON file that gets cleaned up."""

Atests/coordinator/core/test_external_topology.py+330-0

@@ -0,0 +1,330 @@
	1	+# Copyright (c) Huawei Technologies Co., Ltd. 2026. All rights reserved.
	2	+# MindIE is licensed under Mulan PSL v2.
	3	+
	4	+import pytest
	5	+from pydantic import ValidationError
	6	+
	7	+from motor.common.resources.http_msg_spec import ExternalInsEventMsg
	8	+from motor.coordinator.api_server.management_server import _is_standalone_instance_refresh
	9	+
	10	+
	11	+def _external_instance(**overrides):
	12	+ instance = {
	13	+ "id": 1,
	14	+ "role": "prefill",
	15	+ "endpoints": [{"id": 0, "address": "127.0.0.1:8100"}],
	16	+ }
	17	+ instance.update(overrides)
	18	+ return instance
	19	+
	20	+
	21	+def _external_message(**overrides):
	22	+ message = {
	23	+ "event": "set",
	24	+ "model_name": "test-model",
	25	+ "dispatch_capabilities": "concurrent_engine_sync",
	26	+ "engine_type": "vllm",
	27	+ "instances": [_external_instance()],
	28	+ }
	29	+ message.update(overrides)
	30	+ return message
	31	+
	32	+
	33	+def test_external_set_builds_internal_instance_defaults():
	34	+ message = ExternalInsEventMsg.model_validate(_external_message(model_name="TEST-MODEL"))
	35	+
	36	+ internal = message.to_internal("test-model")
	37	+
	38	+ instance = internal.instances[0]
	39	+ endpoint = instance.endpoints["127.0.0.1"][0]
	40	+ assert instance.job_name == "external-prefill-1"
	41	+ assert instance.model_name == "test-model"
	42	+ assert instance.dispatch_capabilities == ["concurrent_engine_sync"]
	43	+ assert instance.parallel_config.dp_size == 1
	44	+ assert endpoint.business_port == "8100"
	45	+ assert endpoint.status.value == "normal"
	46	+
	47	+
	48	+def test_external_omitted_fields_default_to_handoff_and_vllm():
	49	+ body = {
	50	+ "event": "set",
	51	+ "model_name": "test-model",
	52	+ "instances": [_external_instance()],
	53	+ }
	54	+ message = ExternalInsEventMsg.model_validate(body)
	55	+
	56	+ internal = message.to_internal("test-model")
	57	+
	58	+ instance = internal.instances[0]
	59	+ assert message.dispatch_capabilities.value == "prefill_handoff_decode"
	60	+ assert message.engine_type == "vllm"
	61	+ assert instance.dispatch_capabilities == ["prefill_handoff_decode"]
	62	+ assert instance.engine_type == "vllm"
	63	+
	64	+
	65	+def test_external_omitted_model_name_uses_resolved_engine_model():
	66	+ body = {
	67	+ "event": "set",
	68	+ "instances": [_external_instance()],
	69	+ }
	70	+ message = ExternalInsEventMsg.model_validate(body)
	71	+
	72	+ internal = message.to_internal(resolved_model_name="Qwen3-8B")
	73	+
	74	+ assert internal.instances[0].model_name == "Qwen3-8B"
	75	+
	76	+
	77	+def test_external_set_assigns_endpoint_ids_by_array_order():
	78	+ message = ExternalInsEventMsg.model_validate(
	79	+ _external_message(
	80	+ instances=[
	81	+ _external_instance(
	82	+ endpoints=[
	83	+ {"id": 0, "address": "127.0.0.1:8100"},
	84	+ {"id": 1, "address": "127.0.0.2:8101"},
	85	+ ]
	86	+ )
	87	+ ]
	88	+ )
	89	+ )
	90	+
	91	+ internal = message.to_internal("test-model")
	92	+
	93	+ instance = internal.instances[0]
	94	+ assert instance.parallel_config.dp_size == 2
	95	+ assert instance.endpoints["127.0.0.1"][0].id == 0
	96	+ assert instance.endpoints["127.0.0.2"][1].id == 1
	97	+
	98	+
	99	+def test_external_set_parses_ipv6_address():
	100	+ message = ExternalInsEventMsg.model_validate(
	101	+ _external_message(
	102	+ instances=[
	103	+ _external_instance(
	104	+ endpoints=[{"id": 0, "address": "[2001:db8::1]:8200"}],
	105	+ )
	106	+ ]
	107	+ )
	108	+ )
	109	+
	110	+ internal = message.to_internal("test-model")
	111	+
	112	+ endpoint = internal.instances[0].endpoints["2001:db8::1"][0]
	113	+ assert endpoint.ip == "2001:db8::1"
	114	+ assert endpoint.business_port == "8200"
	115	+
	116	+
	117	+@pytest.mark.parametrize("event", ["add", "del"])
	118	+def test_external_incremental_event_is_preserved(event):
	119	+ message = ExternalInsEventMsg.model_validate(_external_message(event=event))
	120	+
	121	+ internal = message.to_internal("test-model")
	122	+
	123	+ assert internal.event.value == event
	124	+ assert internal.instances[0].id == 1
	125	+
	126	+
	127	+@pytest.mark.parametrize(
	128	+ ("body", "message"),
	129	+ [
	130	+ (_external_message(event="pause"), "event='set', 'add', or 'del'"),
	131	+ (_external_message(instances=[_external_instance(role="union")]), "unsupported external instance role"),
	132	+ (_external_message(engine_type="sglang"), "supports engine_type='vllm'"),
	133	+ (
	134	+ _external_message(
	135	+ instances=[
	136	+ _external_instance(),
	137	+ _external_instance(),
	138	+ ]
	139	+ ),
	140	+ "duplicate instance id",
	141	+ ),
	142	+ (
	143	+ _external_message(
	144	+ instances=[
	145	+ _external_instance(
	146	+ endpoints=[
	147	+ {"id": 0, "address": "127.0.0.1:8100"},
	148	+ {"id": 1, "address": "127.0.0.1:8100"},
	149	+ ]
	150	+ )
	151	+ ]
	152	+ ),
	153	+ "duplicate endpoint address",
	154	+ ),
	155	+ (
	156	+ _external_message(
	157	+ instances=[
	158	+ _external_instance(
	159	+ endpoints=[
	160	+ {"id": 0, "address": "127.0.0.1:8100"},
	161	+ {"id": 0, "address": "127.0.0.2:8101"},
	162	+ ]
	163	+ )
	164	+ ]
	165	+ ),
	166	+ "duplicate endpoint id",
	167	+ ),
	168	+ (
	169	+ _external_message(
	170	+ instances=[
	171	+ _external_instance(
	172	+ endpoints=[{"id": 0, "address": "127.0.0.1:70000"}],
	173	+ )
	174	+ ]
	175	+ ),
	176	+ "range 1-65535",
	177	+ ),
	178	+ (
	179	+ _external_message(
	180	+ instances=[
	181	+ _external_instance(
	182	+ endpoints=[{"id": 0, "address": "127.0.0.1"}],
	183	+ )
	184	+ ]
	185	+ ),
	186	+ "must contain a port",
	187	+ ),
	188	+ ],
	189	+)
	190	+def test_external_event_rejects_invalid_topology(body, message):
	191	+ external = ExternalInsEventMsg.model_validate(body)
	192	+
	193	+ with pytest.raises(ValueError, match=message):
	194	+ external.to_internal("test-model")
	195	+
	196	+
	197	+def test_external_schema_forbids_control_plane_fields():
	198	+ body = _external_message(instances=[_external_instance(job_name="controller-owned")])
	199	+
	200	+ with pytest.raises(ValidationError, match="job_name"):
	201	+ ExternalInsEventMsg.model_validate(body)
	202	+
	203	+
	204	+def test_external_schema_forbids_non_routing_endpoint_fields():
	205	+ body = _external_message(
	206	+ instances=[
	207	+ _external_instance(
	208	+ endpoints=[
	209	+ {
	210	+ "id": 0,
	211	+ "address": "127.0.0.1:8100",
	212	+ "bootstrap_port": 9000,
	213	+ }
	214	+ ]
	215	+ )
	216	+ ]
	217	+ )
	218	+
	219	+ with pytest.raises(ValidationError, match="bootstrap_port"):
	220	+ ExternalInsEventMsg.model_validate(body)
	221	+
	222	+
	223	+def test_external_schema_rejects_legacy_endpoint_fields():
	224	+ body = _external_message(
	225	+ instances=[
	226	+ _external_instance(
	227	+ endpoints=[
	228	+ {
	229	+ "id": 0,
	230	+ "ip": "127.0.0.1",
	231	+ "business_port": "8100",
	232	+ }
	233	+ ]
	234	+ )
	235	+ ]
	236	+ )
	237	+
	238	+ with pytest.raises(ValidationError, match="address"):
	239	+ ExternalInsEventMsg.model_validate(body)
	240	+
	241	+
	242	+def test_external_schema_rejects_unknown_dispatch_capability():
	243	+ body = _external_message(dispatch_capabilities="unknown_dispatch")
	244	+
	245	+ with pytest.raises(ValidationError, match="dispatch_capabilities"):
	246	+ ExternalInsEventMsg.model_validate(body)
	247	+
	248	+
	249	+def test_external_set_rejects_model_mismatch():
	250	+ external = ExternalInsEventMsg.model_validate(_external_message(model_name="another-model"))
	251	+
	252	+ with pytest.raises(ValueError, match="does not match configured single model"):
	253	+ external.to_internal("test-model")
	254	+
	255	+
	256	+def test_is_standalone_detects_minimal_payload_without_top_level_markers():
	257	+ body = {
	258	+ "event": "set",
	259	+ "instances": [
	260	+ {
	261	+ "id": 1,
	262	+ "role": "prefill",
	263	+ "endpoints": [{"id": 0, "address": "127.0.0.1:8100"}],
	264	+ }
	265	+ ],
	266	+ }
	267	+
	268	+ assert _is_standalone_instance_refresh(body) is True
	269	+
	270	+
	271	+def test_is_standalone_detects_top_level_markers():
	272	+ body = _external_message(model_name="test-model")
	273	+
	274	+ assert _is_standalone_instance_refresh(body) is True
	275	+
	276	+
	277	+def test_is_standalone_rejects_controller_endpoint_dict():
	278	+ body = {
	279	+ "event": "add",
	280	+ "instances": [
	281	+ {
	282	+ "job_name": "test-job",
	283	+ "model_name": "test-model",
	284	+ "id": 1,
	285	+ "role": "prefill",
	286	+ "endpoints": {
	287	+ "192.168.1.1": {
	288	+ "0": {
	289	+ "id": 0,
	290	+ "ip": "192.168.1.1",
	291	+ "business_port": "8080",
	292	+ }
	293	+ }
	294	+ },
	295	+ }
	296	+ ],
	297	+ }
	298	+
	299	+ assert _is_standalone_instance_refresh(body) is False
	300	+
	301	+
	302	+def test_is_standalone_rejects_mixed_controller_and_standalone_shapes():
	303	+ body = {
	304	+ "event": "set",
	305	+ "instances": [
	306	+ {
	307	+ "id": 1,
	308	+ "role": "prefill",
	309	+ "endpoints": [{"id": 0, "address": "127.0.0.1:8100"}],
	310	+ },
	311	+ {
	312	+ "job_name": "test-job",
	313	+ "model_name": "test-model",
	314	+ "id": 2,
	315	+ "role": "decode",
	316	+ "endpoints": {
	317	+ "192.168.1.2": {
	318	+ "0": {
	319	+ "id": 0,
	320	+ "ip": "192.168.1.2",
	321	+ "business_port": "8200",
	322	+ }
	323	+ }
	324	+ },
	325	+ },
	326	+ ],
	327	+ }
	328	+
	329	+ with pytest.raises(ValueError, match="mixed controller and coordinator-standalone"):
	330	+ _is_standalone_instance_refresh(body)

Mtests/coordinator/scheduler/test_kva_role_u_support.py+13-1

	@@ -126,6 +126,14 @@ def test_register_post_formats_ipv6_endpoint_and_conductor_address() -> None:
126	assert register_payload["replay_endpoint"] == "tcp://[2001:db8::1]:6669"	126	assert register_payload["replay_endpoint"] == "tcp://[2001:db8::1]:6669"
127		127
128		128
		129	+def _mock_enabled_kv_reg():
		130	+ """Return a KV conductor config with registration enabled for unit tests."""
		131	+ mock_reg = Mock()
		132	+ mock_reg.conductor_service = "kv-conductor"
		133	+ mock_reg.store_backend = "YuanRong"
		134	+ return mock_reg
		135	+
		136	+
129	def test_register_kv_instance_supports_role_u() -> None:	137	def test_register_kv_instance_supports_role_u() -> None:
130	instances = [	138	instances = [
131	_build_instance(PDRole.ROLE_P),	139	_build_instance(PDRole.ROLE_P),
	@@ -133,6 +141,7 @@ def test_register_kv_instance_supports_role_u() -> None:
133	_build_instance(PDRole.ROLE_D),	141	_build_instance(PDRole.ROLE_D),
134	]	142	]
135	with (	143	with (
		144	+ patch.object(ConductorApiClient, "_kv_reg", return_value=_mock_enabled_kv_reg()),
136	patch.object(ConductorApiClient, "_register_hbm_dp") as mock_hbm_dp,	145	patch.object(ConductorApiClient, "_register_hbm_dp") as mock_hbm_dp,
137	patch.object(ConductorApiClient, "_register_yuanrong_dp") as mock_yuanrong_dp,	146	patch.object(ConductorApiClient, "_register_yuanrong_dp") as mock_yuanrong_dp,
138	patch.object(ConductorApiClient, "_register_pool"),	147	patch.object(ConductorApiClient, "_register_pool"),
	@@ -155,7 +164,10 @@ def test_unregister_kv_instance_supports_role_u() -> None:
155	_build_instance(PDRole.ROLE_U),	164	_build_instance(PDRole.ROLE_U),
156	_build_instance(PDRole.ROLE_D),	165	_build_instance(PDRole.ROLE_D),
157	]	166	]
158	- with patch.object(ConductorApiClient, "unregister_post") as mock_unregister_post:	167	+ with (
		168	+ patch.object(ConductorApiClient, "_kv_reg", return_value=_mock_enabled_kv_reg()),
		169	+ patch.object(ConductorApiClient, "unregister_post") as mock_unregister_post,
		170	+ ):
159	ConductorApiClient.unregister_kv_instance(instances)	171	ConductorApiClient.unregister_kv_instance(instances)
160		172
161	assert mock_unregister_post.call_count == 2	173	assert mock_unregister_post.call_count == 2

Mtests/coordinator/test_http_server.py+203-1

	@@ -936,6 +936,14 @@ class TestCoordinatorServerAdvanced:
936	im_instance.validate_refresh_instances = AsyncMock(return_value=None)	936	im_instance.validate_refresh_instances = AsyncMock(return_value=None)
937	im_instance.snapshot_instances = AsyncMock(return_value=[])	937	im_instance.snapshot_instances = AsyncMock(return_value=[])
938	im_mock_cls.return_value = im_instance	938	im_mock_cls.return_value = im_instance
		939	+ self.im_mock = im_instance
		940	+ self.refresh_calls: list[tuple] = []
		941	+
		942	+ async def _capture_refresh(event, instances):
		943	+ self.refresh_calls.append((event, instances))
		944	+ return True
		945	+
		946	+ self._capture_refresh = _capture_refresh
939		947
940	# Mock handle_request to return appropriate JSON response	948	# Mock handle_request to return appropriate JSON response
941	async def mock_handle_request(request, config, scheduler=None, request_manager=None, request_json=None):	949	async def mock_handle_request(request, config, scheduler=None, request_manager=None, request_json=None):
	@@ -1031,14 +1039,16 @@ class TestCoordinatorServerAdvanced:
1031	coordinator_config = CoordinatorConfig()	1039	coordinator_config = CoordinatorConfig()
1032	coordinator_config.api_key_config.enable_api_key = True	1040	coordinator_config.api_key_config.enable_api_key = True
1033	coordinator_config.api_key_config.valid_keys = {"sk-test123456789", "sk-coordinator2024"}	1041	coordinator_config.api_key_config.valid_keys = {"sk-test123456789", "sk-coordinator2024"}
		1042	+ self.coordinator_config = coordinator_config
1034		1043
1035	# Create test server shell (ManagementServer + InferenceServer)	1044	# Create test server shell (ManagementServer + InferenceServer)
1036	self.coordinator_server = _TestServerShell(config=coordinator_config)	1045	self.coordinator_server = _TestServerShell(config=coordinator_config)
		1046	+ self.coordinator_server.instance_manager = self.im_mock
1037	# Skip real ROUTER/PUB/SHM bind in TestClient lifespan	1047	# Skip real ROUTER/PUB/SHM bind in TestClient lifespan
1038	mgmt = self.coordinator_server._mgmt	1048	mgmt = self.coordinator_server._mgmt
1039	mgmt._start_control_plane = AsyncMock()	1049	mgmt._start_control_plane = AsyncMock()
1040	mgmt._stop_control_plane = AsyncMock()	1050	mgmt._stop_control_plane = AsyncMock()
1041	- mgmt._control_plane.apply_refresh = AsyncMock(return_value=True)	1051	+ mgmt._control_plane.apply_refresh = AsyncMock(side_effect=self._capture_refresh)
1042	mgmt._control_plane.scheduler = MagicMock()	1052	mgmt._control_plane.scheduler = MagicMock()
1043	mgmt._control_plane.scheduler.dismiss_precision_alarm_state = AsyncMock(return_value=True)	1053	mgmt._control_plane.scheduler.dismiss_precision_alarm_state = AsyncMock(return_value=True)
1044	self.coordinator_server.setup_rate_limiting()	1054	self.coordinator_server.setup_rate_limiting()
	@@ -1097,6 +1107,198 @@ class TestCoordinatorServerAdvanced:
1097	assert "request_id" in data, "Response missing request_id"	1107	assert "request_id" in data, "Response missing request_id"
1098	assert "data" in data, "Response missing data field"	1108	assert "data" in data, "Response missing data field"
1099		1109
		1110	+ def test_refresh_instances_external_set_converts_minimal_topology(self):
		1111	+ """External SET resolves the omitted model name from a native engine."""
		1112	+ self.coordinator_config.aigw_model = {"id": "test-model"}
		1113	+ body = {
		1114	+ "event": "set",
		1115	+ "dispatch_capabilities": "concurrent_engine_sync",
		1116	+ "engine_type": "vllm",
		1117	+ "instances": [
		1118	+ {
		1119	+ "id": 1,
		1120	+ "role": "prefill",
		1121	+ "endpoints": [{"id": 0, "address": "192.168.1.10:8100"}],
		1122	+ },
		1123	+ {
		1124	+ "id": 2,
		1125	+ "role": "decode",
		1126	+ "endpoints": [{"id": 0, "address": "192.168.1.20:8200"}],
		1127	+ },
		1128	+ ],
		1129	+ }
		1130	+
		1131	+ with patch(
		1132	+ "motor.coordinator.api_server.management_server.NativeEngineApiClient.query_model_ids",
		1133	+ return_value=["TEST-MODEL"],
		1134	+ ) as query_model_ids:
		1135	+ response = self.mgmt_client.post("/instances/refresh", json=body)
		1136	+
		1137	+ assert response.status_code == 200
		1138	+ query_model_ids.assert_called_once_with("192.168.1.10:8100", self.coordinator_config.infer_tls_config)
		1139	+ assert self.refresh_calls
		1140	+ event, instances = self.refresh_calls[-1]
		1141	+ assert event.value == "set"
		1142	+ assert [instance.model_name for instance in instances] == ["test-model", "test-model"]
		1143	+ assert instances[0].dispatch_capabilities == ["concurrent_engine_sync"]
		1144	+ assert [instance.job_name for instance in instances] == ["external-prefill-1", "external-decode-2"]
		1145	+ assert instances[0].parallel_config.dp_size == 1
		1146	+ assert instances[0].endpoints["192.168.1.10"][0].business_port == "8100"
		1147	+
		1148	+ def test_refresh_instances_external_without_markers_resolves_model(self):
		1149	+ """Endpoint array shape keeps a fully minimal External request distinguishable."""
		1150	+ body = {
		1151	+ "event": "set",
		1152	+ "instances": [
		1153	+ {
		1154	+ "id": 1,
		1155	+ "role": "prefill",
		1156	+ "endpoints": [{"id": 0, "address": "127.0.0.1:8100"}],
		1157	+ }
		1158	+ ],
		1159	+ }
		1160	+
		1161	+ with patch(
		1162	+ "motor.coordinator.api_server.management_server.NativeEngineApiClient.query_model_ids",
		1163	+ return_value=["test-model"],
		1164	+ ):
		1165	+ response = self.mgmt_client.post("/instances/refresh", json=body)
		1166	+
		1167	+ assert response.status_code == 200
		1168	+ assert self.refresh_calls
		1169	+ _, instances = self.refresh_calls[-1]
		1170	+ assert instances[0].model_name == "test-model"
		1171	+ assert instances[0].engine_type == "vllm"
		1172	+ assert instances[0].dispatch_capabilities == ["prefill_handoff_decode"]
		1173	+
		1174	+ def test_refresh_instances_external_requires_model_name_for_multi_model_engine(self):
		1175	+ body = {
		1176	+ "event": "set",
		1177	+ "instances": [
		1178	+ {
		1179	+ "id": 1,
		1180	+ "role": "prefill",
		1181	+ "endpoints": [{"id": 0, "address": "127.0.0.1:8100"}],
		1182	+ }
		1183	+ ],
		1184	+ }
		1185	+
		1186	+ with patch(
		1187	+ "motor.coordinator.api_server.management_server.NativeEngineApiClient.query_model_ids",
		1188	+ return_value=["model-a", "model-b"],
		1189	+ ):
		1190	+ response = self.mgmt_client.post("/instances/refresh", json=body)
		1191	+
		1192	+ assert response.status_code == 400
		1193	+ assert "provide model_name explicitly" in response.json()["detail"]
		1194	+
		1195	+ def test_refresh_instances_external_model_resolution_skips_unreachable_endpoint(self):
		1196	+ body = {
		1197	+ "event": "set",
		1198	+ "instances": [
		1199	+ {
		1200	+ "id": 1,
		1201	+ "role": "prefill",
		1202	+ "endpoints": [
		1203	+ {"id": 0, "address": "127.0.0.1:8100"},
		1204	+ {"id": 1, "address": "127.0.0.2:8100"},
		1205	+ ],
		1206	+ }
		1207	+ ],
		1208	+ }
		1209	+
		1210	+ with patch(
		1211	+ "motor.coordinator.api_server.management_server.NativeEngineApiClient.query_model_ids",
		1212	+ side_effect=[OSError("connection refused"), ["test-model"]],
		1213	+ ) as query_model_ids:
		1214	+ response = self.mgmt_client.post("/instances/refresh", json=body)
		1215	+
		1216	+ assert response.status_code == 200
		1217	+ assert query_model_ids.call_count == 2
		1218	+ assert self.refresh_calls
		1219	+ _, instances = self.refresh_calls[-1]
		1220	+ assert instances[0].model_name == "test-model"
		1221	+
		1222	+ def test_refresh_instances_rejects_mixed_controller_and_standalone_payload(self):
		1223	+ body = {
		1224	+ "event": "set",
		1225	+ "instances": [
		1226	+ {
		1227	+ "id": 1,
		1228	+ "role": "prefill",
		1229	+ "endpoints": [{"id": 0, "address": "127.0.0.1:8100"}],
		1230	+ },
		1231	+ {
		1232	+ "job_name": "test-job",
		1233	+ "model_name": "test-model",
		1234	+ "id": 2,
		1235	+ "role": "decode",
		1236	+ "endpoints": {
		1237	+ "192.168.1.2": {
		1238	+ "0": {
		1239	+ "id": 0,
		1240	+ "ip": "192.168.1.2",
		1241	+ "business_port": "8200",
		1242	+ }
		1243	+ }
		1244	+ },
		1245	+ },
		1246	+ ],
		1247	+ }
		1248	+
		1249	+ response = self.mgmt_client.post("/instances/refresh", json=body)
		1250	+
		1251	+ assert response.status_code == 400
		1252	+ assert "mixed controller and coordinator-standalone" in response.json()["detail"]
		1253	+
		1254	+ @pytest.mark.parametrize("event_name", ["add", "del"])
		1255	+ def test_refresh_instances_external_incremental_event_is_forwarded(self, event_name):
		1256	+ """External ADD/DEL only forwards Coordinator membership changes."""
		1257	+ self.coordinator_config.aigw_model = {"id": "test-model"}
		1258	+ body = {
		1259	+ "event": event_name,
		1260	+ "model_name": "test-model",
		1261	+ "dispatch_capabilities": "prefill_handoff_decode",
		1262	+ "engine_type": "vllm",
		1263	+ "instances": [
		1264	+ {
		1265	+ "id": 1,
		1266	+ "role": "prefill",
		1267	+ "endpoints": [{"id": 0, "address": "127.0.0.1:8100"}],
		1268	+ }
		1269	+ ],
		1270	+ }
		1271	+
		1272	+ response = self.mgmt_client.post("/instances/refresh", json=body)
		1273	+
		1274	+ assert response.status_code == 200
		1275	+ assert self.refresh_calls
		1276	+ event, instances = self.refresh_calls[-1]
		1277	+ assert event.value == event_name
		1278	+ assert instances[0].id == 1
		1279	+
		1280	+ def test_refresh_instances_external_rejects_unsupported_event(self):
		1281	+ """External protocol rejects Controller-only lifecycle events."""
		1282	+ self.coordinator_config.aigw_model = {"id": "test-model"}
		1283	+ response = self.mgmt_client.post(
		1284	+ "/instances/refresh",
		1285	+ json={
		1286	+ "event": "pause",
		1287	+ "model_name": "test-model",
		1288	+ "dispatch_capabilities": "concurrent_engine_sync",
		1289	+ "engine_type": "vllm",
		1290	+ "instances": [
		1291	+ {
		1292	+ "id": 1,
		1293	+ "role": "prefill",
		1294	+ "endpoints": [{"id": 0, "address": "127.0.0.1:8100"}],
		1295	+ }
		1296	+ ],
		1297	+ },
		1298	+ )
		1299	+
		1300	+ assert response.status_code == 400
		1301	+
1100	def test_refresh_instances_rejects_duplicate_ids(self):	1302	def test_refresh_instances_rejects_duplicate_ids(self):
1101	"""Duplicate IDs must be rejected before list-to-dict conversion can drop an instance."""	1303	"""Duplicate IDs must be rejected before list-to-dict conversion can drop an instance."""
1102	self.coordinator_server._mgmt._control_plane.apply_refresh.side_effect = ValueError(	1304	self.coordinator_server._mgmt._control_plane.apply_refresh.side_effect = ValueError(

Atests/e2e/test_standalone_coordinator_smoke.py+211-0

@@ -0,0 +1,211 @@
	1	+# Copyright (c) Huawei Technologies Co., Ltd. 2026. All rights reserved.
	2	+# MindIE is licensed under Mulan PSL v2.
	3	+
	4	+"""Opt-in smoke for standalone Coordinator membership events and restart replay.
	5	+
	6	+Required environment:
	7	+* MOTOR_RUN_STANDALONE_COORDINATOR_SMOKE=1
	8	+* MOTOR_STANDALONE_PREFILL_URL=http://host:port
	9	+* MOTOR_STANDALONE_DECODE_URL=http://host:port
	10	+* MOTOR_STANDALONE_MODEL=<served model>
	11	+"""
	12	+
	13	+import json
	14	+import os
	15	+import socket
	16	+import subprocess
	17	+import sys
	18	+import time
	19	+import urllib.error
	20	+import urllib.request
	21	+from pathlib import Path
	22	+from urllib.parse import urlparse
	23	+
	24	+import pytest
	25	+
	26	+
	27	+pytestmark = pytest.mark.integration
	28	+
	29	+
	30	+def _free_port() -> int:
	31	+ with socket.socket() as sock:
	32	+ sock.bind(("127.0.0.1", 0))
	33	+ return int(sock.getsockname()[1])
	34	+
	35	+
	36	+def _required_url(name: str) -> tuple[str, int]:
	37	+ value = os.getenv(name, "")
	38	+ parsed = urlparse(value)
	39	+ if parsed.scheme != "http" or not parsed.hostname or parsed.port is None:
	40	+ pytest.fail(f"{name} must be an http://host:port URL")
	41	+ return parsed.hostname, parsed.port
	42	+
	43	+
	44	+def _json_request(url: str, payload: dict \| None = None) -> tuple[int, dict]:
	45	+ data = None if payload is None else json.dumps(payload).encode()
	46	+ request = urllib.request.Request(
	47	+ url,
	48	+ data=data,
	49	+ headers={"Content-Type": "application/json"},
	50	+ method="GET" if payload is None else "POST",
	51	+ )
	52	+ try:
	53	+ with urllib.request.urlopen(request, timeout=10) as response:
	54	+ return response.status, json.loads(response.read())
	55	+ except urllib.error.HTTPError as error:
	56	+ try:
	57	+ # 服务端返回 JSON 错误体时，本场景保留结构化诊断信息。
	58	+ return error.code, json.loads(error.read())
	59	+ except (json.JSONDecodeError, ValueError):
	60	+ # 网关等组件返回纯文本错误体时，本场景降级为空字典，避免掩盖真实 HTTP 状态码。
	61	+ return error.code, {}
	62	+
	63	+
	64	+def _wait_for_readiness(url: str, expected: bool, timeout: float = 30) -> None:
	65	+ deadline = time.monotonic() + timeout
	66	+ while time.monotonic() < deadline:
	67	+ try:
	68	+ status, body = _json_request(url)
	69	+ if status == 200 and body.get("ready") is expected:
	70	+ return
	71	+ except (OSError, ValueError):
	72	+ pass
	73	+ time.sleep(0.5)
	74	+ pytest.fail(f"readiness did not become {expected}")
	75	+
	76	+
	77	+def _start_coordinator(config_path: Path) -> subprocess.Popen:
	78	+ env = dict(os.environ)
	79	+ env["USER_CONFIG_PATH"] = str(config_path)
	80	+ log_path = config_path.with_name("coordinator-smoke.log")
	81	+ # 日志持续写入文件，本场景不会因无人消费 stdout PIPE 而填满管道并卡住子进程。
	82	+ with log_path.open("a", encoding="utf-8") as log_stream:
	83	+ return subprocess.Popen(
	84	+ [sys.executable, "-m", "motor.coordinator.main"],
	85	+ env=env,
	86	+ stdout=log_stream,
	87	+ stderr=subprocess.STDOUT,
	88	+ text=True,
	89	+ )
	90	+
	91	+
	92	+def _stop_coordinator(process: subprocess.Popen) -> None:
	93	+ process.terminate()
	94	+ try:
	95	+ process.wait(timeout=15)
	96	+ except subprocess.TimeoutExpired:
	97	+ process.kill()
	98	+ process.wait(timeout=5)
	99	+
	100	+
	101	+@pytest.mark.skipif(
	102	+ os.getenv("MOTOR_RUN_STANDALONE_COORDINATOR_SMOKE") != "1",
	103	+ reason="set MOTOR_RUN_STANDALONE_COORDINATOR_SMOKE=1 to run",
	104	+)
	105	+def test_standalone_membership_events_inference_and_restart_replay(tmp_path):
	106	+ prefill_host, prefill_port = _required_url("MOTOR_STANDALONE_PREFILL_URL")
	107	+ decode_host, decode_port = _required_url("MOTOR_STANDALONE_DECODE_URL")
	108	+ model = os.getenv("MOTOR_STANDALONE_MODEL", "").strip()
	109	+ if not model:
	110	+ pytest.fail("MOTOR_STANDALONE_MODEL is required")
	111	+
	112	+ infer_port, mgmt_port, obs_port = _free_port(), _free_port(), _free_port()
	113	+ config = {
	114	+ "motor_coordinator_config": {
	115	+ "aigw": {
	116	+ "id": model,
	117	+ "p_max_seqlen": 32768,
	118	+ "d_max_seqlen": 32768,
	119	+ },
	120	+ "api_config": {
	121	+ "coordinator_api_host": "127.0.0.1",
	122	+ "coordinator_api_infer_port": infer_port,
	123	+ "coordinator_api_mgmt_port": mgmt_port,
	124	+ "coordinator_obs_port": obs_port,
	125	+ },
	126	+ "inference_workers_config": {
	127	+ "num_workers": 1,
	128	+ "worker_metaserver_base_port": _free_port(),
	129	+ },
	130	+ }
	131	+ }
	132	+ config_path = tmp_path / "coordinator.json"
	133	+ config_path.write_text(json.dumps(config), encoding="utf-8")
	134	+ topology = {
	135	+ "event": "set",
	136	+ "model_name": model,
	137	+ "dispatch_capabilities": "concurrent_engine_sync",
	138	+ "engine_type": "vllm",
	139	+ "instances": [
	140	+ {
	141	+ "id": 1,
	142	+ "role": "prefill",
	143	+ "endpoints": [{"id": 0, "address": f"{prefill_host}:{prefill_port}"}],
	144	+ },
	145	+ {
	146	+ "id": 2,
	147	+ "role": "decode",
	148	+ "endpoints": [{"id": 0, "address": f"{decode_host}:{decode_port}"}],
	149	+ },
	150	+ ],
	151	+ }
	152	+ readiness_url = f"http://127.0.0.1:{mgmt_port}/readiness"
	153	+ refresh_url = f"http://127.0.0.1:{mgmt_port}/instances/refresh"
	154	+
	155	+ process = _start_coordinator(config_path)
	156	+ try:
	157	+ # Coordinator 刚启动且尚未纳管完整 P+D 拓扑，本场景会进入未就绪分支。
	158	+ _wait_for_readiness(readiness_url, False)
	159	+
	160	+ # SET 只纳管 Prefill 实例，本场景仍缺少 Decode 实例，因此保持未就绪。
	161	+ prefill_only = {**topology, "instances": [topology["instances"][0]]}
	162	+ assert _json_request(refresh_url, prefill_only)[0] == 200
	163	+ _wait_for_readiness(readiness_url, False)
	164	+
	165	+ # ADD 仅把已在线的 Decode 实例加入 Coordinator 纳管，本场景不会启动 Decode 进程。
	166	+ add_decode = {**topology, "event": "add", "instances": [topology["instances"][1]]}
	167	+ assert _json_request(refresh_url, add_decode)[0] == 200
	168	+
	169	+ # P+D 均已纳管，本场景会进入就绪分支。
	170	+ _wait_for_readiness(readiness_url, True)
	171	+ infer_status, _ = _json_request(
	172	+ f"http://127.0.0.1:{infer_port}/v1/chat/completions",
	173	+ {
	174	+ "model": model,
	175	+ "messages": [{"role": "user", "content": "Reply with OK."}],
	176	+ "max_tokens": 8,
	177	+ },
	178	+ )
	179	+ assert infer_status == 200
	180	+
	181	+ # DEL 仅将 Decode 实例踢出 Coordinator 纳管，本场景不会停止或销毁 Decode 进程。
	182	+ delete_decode = {**topology, "event": "del", "instances": [topology["instances"][1]]}
	183	+ assert _json_request(refresh_url, delete_decode)[0] == 200
	184	+
	185	+ # 缺少受纳管的 Decode 实例，本场景会重新进入未就绪分支。
	186	+ _wait_for_readiness(readiness_url, False)
	187	+
	188	+ # 再次 ADD 同一个 Decode 实例，本场景复用仍在线的外部进程并恢复调度。
	189	+ assert _json_request(refresh_url, add_decode)[0] == 200
	190	+ _wait_for_readiness(readiness_url, True)
	191	+
	192	+ # 删除后重新加入仍可完成推理，证明 DEL 没有真实下线外部 Decode 实例。
	193	+ infer_status, _ = _json_request(
	194	+ f"http://127.0.0.1:{infer_port}/v1/chat/completions",
	195	+ {
	196	+ "model": model,
	197	+ "messages": [{"role": "user", "content": "Reply with OK."}],
	198	+ "max_tokens": 8,
	199	+ },
	200	+ )
	201	+ assert infer_status == 200
	202	+ finally:
	203	+ _stop_coordinator(process)
	204	+
	205	+ process = _start_coordinator(config_path)
	206	+ try:
	207	+ _wait_for_readiness(readiness_url, False)
	208	+ assert _json_request(refresh_url, topology)[0] == 200
	209	+ _wait_for_readiness(readiness_url, True)
	210	+ finally:
	211	+ _stop_coordinator(process)