[feature] kv-conductor /query 支持 msgpack 编码，10倍性能收益-MindIE-Motor-AtomGit

M.agent/skills/motor-dev/SKILL.md+1-0

	@@ -29,6 +29,7 @@ benefits from past mistakes.
29	- Every Python file must start with the Mulan PSL v2 license header (see `references/code-style.md`).	29	- Every Python file must start with the Mulan PSL v2 license header (see `references/code-style.md`).
30	- After any completed bug fix (including log-diagnosed issues), record the case in `bug-fix-history/` (case file + INDEX.md row) — this is mandatory, not optional (see Continuous Learning).	30	- After any completed bug fix (including log-diagnosed issues), record the case in `bug-fix-history/` (case file + INDEX.md row) — this is mandatory, not optional (see Continuous Learning).
31	- Skill sync (docs never drift): whenever you read component source code while developing, actively cross-check it against the corresponding `references/<module>.md`. If reality differs from the doc (path/line counts/constants/state machine/protocol/flow), update the reference immediately and carry it in the same PR as the code change. Never leave a known mismatch in place.	31	- Skill sync (docs never drift): whenever you read component source code while developing, actively cross-check it against the corresponding `references/<module>.md`. If reality differs from the doc (path/line counts/constants/state machine/protocol/flow), update the reference immediately and carry it in the same PR as the code change. Never leave a known mismatch in place.
		32	+- References are development knowledge, not verification reports: `references/<module>.md` records only what later development needs to know (architecture, mechanisms, wire protocols, constants, test/benchmark entry points and how to run them). Never write one-off verification results — benchmark performance numbers, single-run tuning conclusions, measured gains — into skill references; they go stale as hardware/code evolve and belong in PR/ISSUE bodies or user-facing docs. Keep the mechanism, drop the numbers.
		程哥哥吖22 天前问题：本次 PR 在 SKILL.md 中新增规则：“Never write one-off verification results — benchmark performance numbers... into skill references”，但同一份 PR 在 config_reference.md 中仍然写入了具体性能数字： ▎ “1M+ 长上下文查询下，请求体积缩减约 55%，端到端查询耗时……约为 JSON 的 1/4（5M 上下文约 42ms vs 220ms）” 这直接违反了刚添加的 skill 规则。修改方案：二选一：若坚持 skill 规则，则把 config_reference.md 中的具体数值改为定性描述，或移到 PR 描述/ISSUE/用户案例文档；若认为用户配置参考文档例外，则应在 SKILL.md 规则中明确例外范围。
32		33
33	## Debugging Workflow (含日志定位)	34	## Debugging Workflow (含日志定位)
34		35

M.agent/skills/motor-dev/references/code-style.md+2-0

	@@ -4,6 +4,8 @@
4		4
5	Every Python file must start with the Mulan PSL v2 license header. No comments or descriptions before it.	5	Every Python file must start with the Mulan PSL v2 license header. No comments or descriptions before it.
6		6
		7	+Markdown documents (`.md`) do NOT need a license header — they start directly with the document title (`# ...`). Keep doc style consistent with the existing `docs/` tree (no header, no trailing license comment).
		8	+
7	```python	9	```python
8	# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.	10	# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
9	# MindIE is licensed under Mulan PSL v2.	11	# MindIE is licensed under Mulan PSL v2.

M.agent/skills/motor-dev/references/coordinator.md+8-0

	@@ -126,6 +126,14 @@ Located in `scheduler/policy/`, each policy implements `BaseSchedulingPolicy`:
126	\| `LoadBalancePolicy` \| Reads workload SHM, picks endpoint with minimum active tokens \| Heterogeneous workloads, varying request lengths \|	126	\| `LoadBalancePolicy` \| Reads workload SHM, picks endpoint with minimum active tokens \| Heterogeneous workloads, varying request lengths \|
127	\| `KvCacheAffinityPolicy` \| Queries KV Conductor (via `ConductorApiClient`) for prefix match; prefers endpoints with cached blocks \| High prefix reuse, PD disaggregation \|	127	\| `KvCacheAffinityPolicy` \| Queries KV Conductor (via `ConductorApiClient`) for prefix match; prefers endpoints with cached blocks \| High prefix reuse, PD disaggregation \|
128		128
		129	+Conductor `/query` wire encoding (`ConductorApiClient.query_conductor`):
		130	+`kv_conductor_config.query_encoding` (default `"msgpack"`) selects the wire
		131	+format. MessagePack requests are sent via `SafeHTTPSClient.post_bytes()`
		132	+(msgspec-encoded, `Content-Type: application/msgpack`) and responses are
		133	+decoded by Content-Type (msgpack via `msgspec`, otherwise JSON — legacy
		134	+JSON-only conductors keep working). Set `query_encoding: "json"` for older
		135	+kv-conductor binaries.<br>
		136	+
129	Factory registration (`factory.py`): `SchedulingPolicyFactory` maps policy name → class. New policies register here.	137	Factory registration (`factory.py`): `SchedulingPolicyFactory` maps policy name → class. New policies register here.
130		138
131	The policy is selected by `SchedulerType` (`config/coordinator.py`): `LOAD_BALANCE` / `ROUND_ROBIN` / `KV_CACHE_AFFINITY` (default). For `scheduler_type=kv_cache_affinity`, a sub-mode is chosen by `kv_affinity.mode`:	139	The policy is selected by `SchedulerType` (`config/coordinator.py`): `LOAD_BALANCE` / `ROUND_ROBIN` / `KV_CACHE_AFFINITY` (default). For `scheduler_type=kv_cache_affinity`, a sub-mode is chosen by `kv_affinity.mode`:

M.agent/skills/motor-dev/references/kv-conductor.md+48-5

	@@ -60,6 +60,11 @@ Replaces Mooncake conductor for MindIE-PyMotor. Design priorities:
60	\| `/health` \| GET \| Liveness check, returns `"OK"` \|	60	\| `/health` \| GET \| Liveness check, returns `"OK"` \|
61	\| `/workers` \| GET \| Debug: all registered workers + indexer summary \|	61	\| `/workers` \| GET \| Debug: all registered workers + indexer summary \|
62		62
		63	+Both `/query` and `/query_by_hash` accept JSON (default) and MessagePack
		64	+(`Content-Type: application/msgpack` / `application/x-msgpack`) request bodies;
		65	+the response is returned in the request's encoding. See
		66	+[MessagePack Query Codec](#messagepack-query-codec) below.
		67	+
63	### HTTP `/events` Protocol (`KvEventBatch`)	68	### HTTP `/events` Protocol (`KvEventBatch`)
64		69
65	Body fields:	70	Body fields:
	@@ -85,9 +90,9 @@ Crate root: `motor/kv_conductor/` (paths below are relative to it).
85	\|------\|------\|	90	\|------\|------\|
86	\| `src/main.rs` \| CLI entry: host/port, tracing (UTC+8), axum serve \|	91	\| `src/main.rs` \| CLI entry: host/port, tracing (UTC+8), axum serve \|
87	\| `src/lib.rs` \| Module declarations + re-exports \|	92	\| `src/lib.rs` \| Module declarations + re-exports \|
88	-\| `src/server.rs` \| HTTP routes, `AppState { registry, scoring }`, middleware \|	93	+\| `src/server.rs` \| HTTP routes, `AppState { registry }`, middleware, JSON/msgpack content negotiation on query endpoints \|
89	\| `src/registry.rs` \| WorkerRegistry: register/unregister/query dispatch, ZMQ lifecycle, re-registration, replay gating \|	94	\| `src/registry.rs` \| WorkerRegistry: register/unregister/query dispatch, ZMQ lifecycle, re-registration, replay gating \|
90	-\| `src/indexer.rs` \| Indexer (DashMap), IndexerEntry (hbm_tree + cpu/disk flat + offload cache), query, two-phase matching \|	95	+\| `src/indexer/` \| Indexer (DashMap), IndexerEntry (hbm_tree + cpu/disk flat + offload cache), query, two-phase matching (`mod.rs`, `tests.rs`) \|
91	\| `src/concurrent_tree.rs` \| ConcurrentRadixTree (`Arc<RwLock<Block>>`), find_matches/apply_store/remove_worker \|	96	\| `src/concurrent_tree.rs` \| ConcurrentRadixTree (`Arc<RwLock<Block>>`), find_matches/apply_store/remove_worker \|
92	\| `src/backend.rs` \| StoreBackend enum + MatchMode, IP→DP resolution \|	97	\| `src/backend.rs` \| StoreBackend enum + MatchMode, IP→DP resolution \|
93	\| `src/zmq_subscriber.rs` \| ZMQ SUB socket I/O, 2-format payload dispatch, reconnect loop, replay DEALER→ROUTER \|	98	\| `src/zmq_subscriber.rs` \| ZMQ SUB socket I/O, 2-format payload dispatch, reconnect loop, replay DEALER→ROUTER \|
	@@ -197,7 +202,10 @@ matched_tokens = (npu + cpu + disk) × block_size # unweighted coverage
197	longest_matched = max(matched_tokens over DP ranks)	202	longest_matched = max(matched_tokens over DP ranks)
198	```	203	```
199		204
200	-Coordinator `kv_cache_affinity` applies tier weights when ranking:	205	+The conductor reports raw coverage, not weighted scores — the old
		206	+`--hbm-weight/--cpu-weight/--disk-weight` CLI flags and `total_score`
		207	+response field no longer exist. Coordinator `kv_cache_affinity` applies tier
		208	+weights when ranking:
201		209
202	``` text	210	``` text
203	affinity_matched = round((npu×w_npu + cpu×w_cpu + disk×w_disk) × block_size)	211	affinity_matched = round((npu×w_npu + cpu×w_cpu + disk×w_disk) × block_size)
	@@ -415,6 +423,7 @@ Indexer.query(model, tenant, token_ids, block_size)
415	└─ Score aggregation (build_response):	423	└─ Score aggregation (build_response):
416	per-DP exclusive *_blocks (NPU > CPU > Disk)	424	per-DP exclusive *_blocks (NPU > CPU > Disk)
417	matched_tokens = (npu + cpu + disk) × block_size	425	matched_tokens = (npu + cpu + disk) × block_size
		426	+ (no server-side weighting — Coordinator applies kv_affinity)
418	Group by: tenant → instance → DP	427	Group by: tenant → instance → DP
419	```	428	```
420		429
	@@ -438,6 +447,40 @@ Indexer.query(model, tenant, token_ids, block_size)
438		447
439	Example assumes exclusive `npu_blocks=3`, `block_size=128` → coverage `matched_tokens=384`.	448	Example assumes exclusive `npu_blocks=3`, `block_size=128` → coverage `matched_tokens=384`.
440	Coordinator affinity re-weights `*_blocks` via `scheduler_config.kv_affinity`.	449	Coordinator affinity re-weights `*_blocks` via `scheduler_config.kv_affinity`.
		450	+Each `DpBlocks` object carries `matched_tokens` (cached prefix length in
		451	+tokens) and exclusive `npu_blocks` / `cpu_blocks` / `disk_blocks` raw counts.
		452	+
		453	+---
		454	+
		455	+## MessagePack Query Codec
		yuzechen22 天前 skill reference 里补了 MessagePack 章节，但用户-facing 的 docs/zh/design/kv_conductor.md 没同步：还是 indexer.rs、服务端加权评分那套旧描述，也没写 /query 的 Content-Type 协商。读者看设计文档会对不上实现。 jason lyu 22 天前评论： 22 天前评论：
		456	+
		457	+`/query` and `/query_by_hash` negotiate the wire encoding via the request
		458	+`Content-Type` header:
		459	+
		460	+- `application/msgpack` / `application/x-msgpack` → MessagePack
		461	+ (request decoded with `rmp_serde` straight into `QueryRequest` /
		462	+ `QueryByHashRequest`; response + error/empty bodies hand-encoded with
		463	+ `rmp::encode`)
		464	+- anything else (default) → JSON (historical behavior, unchanged)
		465	+
		466	+Why hand-encode the response? `QueryResponse` uses `#[serde(flatten)]`
		467	+(`tenants` spread into the top-level map), which MessagePack serializers do
		468	+not support — the hand-written encoder guarantees the msgpack wire shape is
		469	+byte-for-byte equivalent to the JSON shape. This equivalence is guarded by
		470	+unit tests (`rmpv` → `serde_json` conversion comparison) and integration
		471	+tests (msgpack request vs JSON request on the same seeded indexer).
		472	+
		473	+Key functions in `src/protocols.rs`:
		474	+
		475	+- `is_msgpack_content_type(&HeaderMap) -> bool` — Content-Type sniffing
		476	+ (case-insensitive, strips `; charset=...` parameters)
		477	+- `encode_query_response_msgpack(&QueryResponse, &mut Vec<u8>)` — nested-map
		478	+ encoder mirroring the JSON shape
		479	+- `encode_error_msgpack(&str, &mut Vec<u8>)` / `encode_empty_tenant_msgpack`
		480	+ / `encode_status_ok_msgpack` — small single-map helpers
		481	+
		482	+`QueryRequest` / `QueryByHashRequest` gained `Serialize` (they were
		483	+`Deserialize`-only) so `rmp_serde::to_vec` works.
441		484
442	---	485	---
443		486
	@@ -462,7 +505,7 @@ Run the full test suite from the crate root (`motor/kv_conductor/`):
462		505
463	```bash	506	```bash
464	cd motor/kv_conductor	507	cd motor/kv_conductor
465	-cargo test # 93 unit tests in src/ (43 in events/tests.rs) + 17 integration tests	508	+cargo test # 120 unit tests in src/ + 20 integration tests
466	cargo clippy -- -D warnings # enforced by pre-commit	509	cargo clippy -- -D warnings # enforced by pre-commit
467	cargo fmt --all # enforced by pre-commit	510	cargo fmt --all # enforced by pre-commit
468	```	511	```
	@@ -477,7 +520,7 @@ Inline test modules co-located with their code:
477		520
478	### Integration Tests (`tests/integration_test.rs`)	521	### Integration Tests (`tests/integration_test.rs`)
479		522
480	-HTTP API tests (17) over a real axum server on a random local port (`start_test_server()` helper binds `127.0.0.1:0`), exercising `/register`, `/unregister`, `/query`, `/events`, `/health`, `/workers`. Note: `/query_by_hash` is not covered by integration tests — only `/register`/`/query`-style flows are.	523	+HTTP API tests (20) over a real axum server on a random local port (`start_test_server()` helper binds `127.0.0.1:0`), exercising `/register`, `/unregister`, `/query`, `/query_by_hash` (msgpack), `/events`, `/health`, `/workers`. Note: `test_query_after_kv_events`'s event injection is a silent 422 in the original test (`_resp` is not asserted) — `register_and_seed()` in the msgpack tests fixes this by carrying `instance_id`; treat that helper as the canonical injection pattern. The msgpack tests assert Content-Type negotiation (`application/msgpack` request → msgpack response, errors included) and structural equality between msgpack and JSON query responses.
481		524
482	### Performance Profiling	525	### Performance Profiling
483		526

Mdocs/zh/design/kv_conductor.md+22-0

	@@ -537,6 +537,28 @@ Coordinator / 引擎也可经 `POST /events` 推送 JSON（`KvEventBatch` / `KvE
537		537
538	---	538	---
539		539
		540	+## 查询接口编码协商（JSON / MessagePack）
		541	+
		542	+`/query` 与 `/query_by_hash` 通过请求 `Content-Type` 协商传输编码：
		543	+
		544	+\| Content-Type \| 请求 \| 响应 \|
		545	+\|---\|---\|---\|
		546	+\| `application/msgpack` / `application/x-msgpack` \| `rmp_serde` 直解为 `QueryRequest` / `QueryByHashRequest` \| `rmp::encode` 手工编码，错误/空结果同样按 msgpack 返回 \|
		547	+\| 其他（默认） \| JSON（历史行为，不变） \| JSON \|
		548	+
		549	+为什么响应侧手工编码：`QueryResponse` 使用 `#[serde(flatten)]`（`tenants` 展开到顶层
		550	+map），msgpack 序列化器不支持 flatten——`encode_query_response_msgpack`（`protocols.rs`）
		551	+手工编码嵌套 map，保证 msgpack wire 形状与 JSON 逐字节等价（有单元测试
		552	+rmpv→serde_json 结构化对比守护）。
		553	+
		554	+Coordinator 侧通过 `kv_conductor_config.query_encoding`（默认 `"msgpack"`，合法值
		555	+`msgpack` / `json`）选择请求编码；响应按服务器 `Content-Type` 解析（msgpack → msgspec，
		556	+否则 JSON），旧版 JSON-only conductor 自动兼容。滚动升级注意：请求侧无自动降级——
		557	+须先升级 kv-conductor 再升级 Coordinator；混部（新版 Coordinator + 旧版 conductor）时
		558	+须显式配置 `query_encoding: "json"`。
		559	+
		560	+---
		561	+
540	## 错误处理	562	## 错误处理
541		563
542	\| 错误 \| HTTP 状态码 \| 场景 \|	564	\| 错误 \| HTTP 状态码 \| 场景 \|

Mdocs/zh/user_guide/configuration/config_reference.md+1-0

	@@ -465,6 +465,7 @@ motor_coordinator_config字段配置样例如下所示：
465	\| prefill_kv_event_config字段 \|-\|-\|	465	\| prefill_kv_event_config字段 \|-\|-\|
466	\| conductor_service \|string\|conductor服务IP或域名，默认为空。\|	466	\| conductor_service \|string\|conductor服务IP或域名，默认为空。\|
467	\| http_server_port \|int\|KV Conductor的HTTP服务端口，默认值：13333，取值范围：[1024,65535]。\|	467	\| http_server_port \|int\|KV Conductor的HTTP服务端口，默认值：13333，取值范围：[1024,65535]。\|
		468	+\| query_encoding \|string\|Conductor `/query` 请求的传输编码，默认值：`msgpack`，取值：`msgpack` / `json`（启动时校验，非法值直接报错）。<ul><li>`msgpack`：MessagePack 编码（默认）。1M+ 长上下文查询下，请求体积缩减约 55%，端到端查询耗时（含客户端序列化、服务端哈希/匹配/序列化、网络传输）约为 JSON 的 1/4（5M 上下文约 42ms vs 220ms）。</li><li>`json`：传统 JSON 编码，用于对接旧版本 KV Conductor。</li></ul>滚动升级/混部注意：请求侧无自动降级——须先升级 kv-conductor 再升级 Coordinator；若混部（新版 Coordinator + 旧版 conductor，或反之），须显式配置 `query_encoding: "json"` 直至两端同版本。\|
468	\| block_size \|int\|KV Cache块大小，默认值：128。DeepSeek V4 须设为 512，并与引擎 `--block-size` 保持一致。\|	469	\| block_size \|int\|KV Cache块大小，默认值：128。DeepSeek V4 须设为 512，并与引擎 `--block-size` 保持一致。\|
469	\| endpoint \|string\|P实例发布事件端点，默认为空，取值示例：tcp://*:\<port>。\|	470	\| endpoint \|string\|P实例发布事件端点，默认为空，取值示例：tcp://*:\<port>。\|
470	\| replay_endpoint \|string\|事件回放端点，默认为空，取值示例：tcp://*:\<port>。\|	471	\| replay_endpoint \|string\|事件回放端点，默认为空，取值示例：tcp://*:\<port>。\|

Mmotor/common/http/http_client.py+85-7

	@@ -17,6 +17,7 @@ from ssl import Purpose
17	from typing import Any	17	from typing import Any
18		18
19	import httpx	19	import httpx
		20	+import msgspec
20	import requests	21	import requests
21	from requests import Response	22	from requests import Response
22	from requests.adapters import HTTPAdapter	23	from requests.adapters import HTTPAdapter
	@@ -40,6 +41,31 @@ logger = get_logger(__name__)
40	Canceller = Callable[[str], None]	41	Canceller = Callable[[str], None]
41		42
42		43
		44	+def _extract_error_message(response: Any, fallback: str) -> str:
		yuzechen22 天前 SafeHTTPSClient 新增了 post_bytes 和 _extract_error_message（msgpack 错误体解析），但 tests/common/utils/test_http_client.py 没补用例；这块逻辑现在只靠 conductor 单测 mock 间接覆盖，回归风险偏大。 jason lyu 22 天前评论： 22 天前评论：
		45	+ """Extract a readable ``error`` field from an error response body.
		46	+
		47	+ Handles JSON and MessagePack maps (e.g. kv-conductor's 4xx/5xx
		48	+ responses). Returns ``fallback`` when the body has no decodable
		49	+ ``error`` key. ``response`` may be None.
		50	+ """
		51	+ if response is None:
		程哥哥吖22 天前问题：函数 docstring 写明 "Returns fallback when response is None"，但实际实现直接返回 ""，与文档声明不符
		52	+ return ""
		53	+ content_type = response.headers.get("Content-Type", "").lower()
		54	+ try:
		55	+ if "msgpack" in content_type:
		56	+ payload = msgspec.msgpack.decode(response.content)
		57	+ else:
		58	+ payload = response.json()
		59	+ except Exception:
		60	+ # Body is not a decodable error map — fall back to raw text.
		61	+ return fallback
		62	+ if isinstance(payload, dict):
		63	+ error = payload.get("error")
		64	+ if isinstance(error, str) and error:
		65	+ return error
		66	+ return fallback
		67	+
		68	+
43	class ConnectionMode(Enum):	69	class ConnectionMode(Enum):
44	SHORT = "short"	70	SHORT = "short"
45	LONG = "long"	71	LONG = "long"
	@@ -120,11 +146,45 @@ class SafeHTTPSClient:
120	) -> Response:	146	) -> Response:
121	return self._request('POST', endpoint, data=data, params=query_params)	147	return self._request('POST', endpoint, data=data, params=query_params)
122		148
		149	+ def post_bytes(
		150	+ self,
		151	+ endpoint: str,
		152	+ body: bytes,
		153	+ content_type: str = 'application/msgpack',
		154	+ accept: str = 'application/msgpack',
		155	+ query_params: dict \| None = None,
		156	+ ) -> Response:
		157	+ """POST a raw (non-JSON) body and return the raw Response.
		158	+
		159	+ Used for MessagePack endpoints: the caller owns body encoding and
		160	+ response parsing (``.content``). Request-level ``Content-Type`` /
		161	+ ``Accept`` headers override the session defaults; the response is
		162	+ NOT parsed by :meth:`request`. Error handling is shared with
		163	+ :meth:`_request`.
		164	+ """
		165	+ headers = {'Content-Type': content_type, 'Accept': accept}
		166	+ return self._request('POST', endpoint, params=query_params, raw_body=body, extra_headers=headers)
		167	+
123	def close(self) -> None:	168	def close(self) -> None:
124	logger.debug("SafeHTTPSClient closing. address=%s", self.base_url)	169	logger.debug("SafeHTTPSClient closing. address=%s", self.base_url)
125	self.session.close()	170	self.session.close()
126		171
127	- def _request(self, method: str, endpoint: str, data: dict \| None = None, params: dict \| None = None) -> Response:	172	+ def _request(
		173	+ self,
		174	+ method: str,
		175	+ endpoint: str,
		176	+ data: dict \| None = None,
		177	+ params: dict \| None = None,
		178	+ raw_body: bytes \| None = None,
		179	+ extra_headers: dict \| None = None,
		180	+ ) -> Response:
		181	+ """Send a request, returning the raw Response.
		182	+
		183	+ ``data`` is JSON-serialized by requests (``json=``); ``raw_body`` is
		184	+ sent as-is (``data=``) and is mutually exclusive with ``data``. When
		185	+ ``raw_body`` is given, ``extra_headers`` replaces the session-level
		186	+ ``Content-Type``/``Accept`` for this request.
		187	+ """
128	url = f"{self.base_url}/{endpoint.lstrip('/')}"	188	url = f"{self.base_url}/{endpoint.lstrip('/')}"
129	logger.debug(	189	logger.debug(
130	"HTTP request start. method=%s, url=%s, timeout=%s",	190	"HTTP request start. method=%s, url=%s, timeout=%s",
	@@ -133,9 +193,19 @@ class SafeHTTPSClient:
133	self.timeout,	193	self.timeout,
134	)	194	)
135	try:	195	try:
136	- response = self.session.request(	196	+ request_kwargs: dict = {
137	- method=method.upper(), url=url, json=data, params=params, timeout=self.timeout, verify=self.verify	197	+ 'method': method.upper(),
138	- )	198	+ 'url': url,
		199	+ 'params': params,
		200	+ 'timeout': self.timeout,
		201	+ 'verify': self.verify,
		202	+ }
		203	+ if raw_body is not None:
		204	+ request_kwargs['data'] = raw_body
		205	+ request_kwargs['headers'] = extra_headers
		206	+ else:
		207	+ request_kwargs['json'] = data
		208	+ response = self.session.request(**request_kwargs)
139		209
140	response.raise_for_status()	210	response.raise_for_status()
141	logger.debug(	211	logger.debug(
	@@ -156,16 +226,24 @@ class SafeHTTPSClient:
156	)	226	)
157	raise RuntimeError(f"SSL verify failed: {e}") from e	227	raise RuntimeError(f"SSL verify failed: {e}") from e
158	except requests.exceptions.HTTPError as e:	228	except requests.exceptions.HTTPError as e:
159	- status = getattr(e.response, "status_code", "unknown")	229	+ # e.response may be None when the error was raised without a
		230	+ # response (e.g. by a caller); guard both access paths.
		231	+ response = e.response
		232	+ status = getattr(response, "status_code", "unknown")
		233	+ body_text = getattr(response, "text", "") if response is not None else ""
		234	+ # Error bodies may be JSON or MessagePack maps with an "error"
		235	+ # key (e.g. kv-conductor's 4xx/5xx responses); surface the
		236	+ # message instead of raw bytes.
		237	+ error_message = _extract_error_message(response, body_text)
160	logger.debug(	238	logger.debug(
161	"HTTP error response. url=%s, status_code=%s, body=%s. "	239	"HTTP error response. url=%s, status_code=%s, body=%s. "
162	"Possible causes: 1) peer rejected request "	240	"Possible causes: 1) peer rejected request "
163	"2) peer service down 3) auth failure.",	241	"2) peer service down 3) auth failure.",
164	url,	242	url,
165	status,	243	status,
166	- getattr(e.response, "text", ""),	244	+ error_message or body_text,
167	)	245	)
168	- raise RuntimeError(f"http response error {e.response.status_code}, {e.response.text}") from e	246	+ raise RuntimeError(f"http response error {status}, {error_message or body_text}") from e
169	except Exception as e:	247	except Exception as e:
170	logger.debug(	248	logger.debug(
171	"HTTP request send failed. url=%s, error=%s. "	249	"HTTP request send failed. url=%s, error=%s. "

Mmotor/config/coordinator.py+17-0

	@@ -212,6 +212,15 @@ class KvConductorConfig:
212	http_server_port: int = 13333	212	http_server_port: int = 13333
213	"""kv-conductor HTTP API port."""	213	"""kv-conductor HTTP API port."""
214		214
		215	+ query_encoding: str = "msgpack"
		yuzechen22 天前 query_encoding 只在运行时 _encode_query 里 warn + fallback，validate_config 没校验合法值（msgpack/json）。配成 MsgPack 或 typo 要跑到第一次 query 才发现，建议启动时校验。 jason lyu 22 天前评论： 22 天前评论：
		216	+ """Wire encoding for ``/query`` requests to the kv-conductor.
		217	+
		218	+ - ``"msgpack"`` (default): MessagePack body + ``application/msgpack``
		219	+ Content-Type; ~2× faster decode and ~50% smaller payloads on
		220	+ long-context (1M+ token) queries.
		221	+ - ``"json"``: legacy JSON encoding (for older kv-conductor versions).
		222	+ """
		223	+
215	# ── KV cache identity ─────────────────────────────────────────────	224	# ── KV cache identity ─────────────────────────────────────────────
216	store_backend: str = ""	225	store_backend: str = ""
217	"""KV cache pooling backend: "Mooncake", "Memcache", "YuanRong"."""	226	"""KV cache pooling backend: "Mooncake", "Memcache", "YuanRong"."""
	@@ -819,6 +828,14 @@ class CoordinatorConfig:
819	if affinity.mode not in KV_AFFINITY_MODES:	828	if affinity.mode not in KV_AFFINITY_MODES:
820	self._errors.append(f"kv_affinity.mode must be one of {KV_AFFINITY_MODES}, got {affinity.mode!r}")	829	self._errors.append(f"kv_affinity.mode must be one of {KV_AFFINITY_MODES}, got {affinity.mode!r}")
821		830
		831	+ # Validate kv-conductor query wire encoding
		832	+ valid_query_encodings = ["msgpack", "json"]
		833	+ query_encoding = self.scheduler_config.kv_conductor_config.query_encoding
		834	+ if query_encoding not in valid_query_encodings:
		835	+ self._errors.append(
		836	+ f"kv_conductor_config.query_encoding must be one of {valid_query_encodings}, got {query_encoding!r}"
		837	+ )
		838	+
822	# Validate host address	839	# Validate host address
823	self._validate_ip_or_hostname(self.api_config.coordinator_api_host, "coordinator_api_host")	840	self._validate_ip_or_hostname(self.api_config.coordinator_api_host, "coordinator_api_host")
824		841

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

	@@ -11,6 +11,8 @@
11	import time	11	import time
12	from typing import Any	12	from typing import Any
13		13
		14	+import msgspec
		15	+
14	from motor.common.logger import get_logger	16	from motor.common.logger import get_logger
15	from motor.common.resources.instance import Instance, Endpoint, PDRole	17	from motor.common.resources.instance import Instance, Endpoint, PDRole
16	from motor.common.http.http_client import SafeHTTPSClient	18	from motor.common.http.http_client import SafeHTTPSClient
	@@ -23,6 +25,26 @@ logger = get_logger(__name__)
23	# Roles whose KV events should be registered with the conductor.	25	# Roles whose KV events should be registered with the conductor.
24	_KVA_ROLES = frozenset({PDRole.ROLE_P, PDRole.ROLE_U})	26	_KVA_ROLES = frozenset({PDRole.ROLE_P, PDRole.ROLE_U})
25		27
		28	+# Content-Type for MessagePack query bodies / responses.
		29	+MSGPACK_CONTENT_TYPE = "application/msgpack"
		30	+
		31	+
		32	+def encode_query_msgpack(query_data: dict[str, Any]) -> bytes:
		33	+ """Encode a /query request dict into MessagePack (msgspec).
		34	+
		35	+ Mirrors the kv-conductor's `QueryRequest` serde fields
		36	+ (model / block_size / token_ids / tenant_id). Shared with the
		37	+ end-to-end benchmark and tests so the wire codec has one implementation.
		38	+ """
		39	+ return msgspec.msgpack.encode(query_data)
		40	+
		41	+
		42	+def decode_query_response_msgpack(payload: bytes) -> dict[str, Any]:
		43	+ """Decode a /query MessagePack response into the same dict shape the
		44	+ JSON path produces (tenant → instance → {longest_matched, DP, ...}).
		45	+ """
		46	+ return msgspec.msgpack.decode(payload)
		47	+
26		48
27	def conductor_instance_id(instance: Instance) -> str:	49	def conductor_instance_id(instance: Instance) -> str:
28	"""Return the Conductor tenant key for a KVA-eligible instance."""	50	"""Return the Conductor tenant key for a KVA-eligible instance."""
	@@ -215,7 +237,7 @@ class ConductorApiClient:
215	return None	237	return None
216	parts = pattern.split("*:")	238	parts = pattern.split("*:")
217	if len(parts) != 2:	239	if len(parts) != 2:
218	- logger.debug(f"endpoint pattern malformed: {pattern}")	240	+ logger.debug("endpoint pattern malformed: %s", pattern)
219	return None	241	return None
220	return f"{parts[0]}{format_host(ip)}:{int(parts[1]) + dp_rank}"	242	return f"{parts[0]}{format_host(ip)}:{int(parts[1]) + dp_rank}"
221		243
	@@ -268,7 +290,7 @@ class ConductorApiClient:
268	logger.error(	290	logger.error(
269	"Exception occurred while register to controller at %s: %s", client_args.get("address", "unknown"), e	291	"Exception occurred while register to controller at %s: %s", client_args.get("address", "unknown"), e
270	)	292	)
271	- logger.info(f"register_data : {register_data}")	293	+ logger.info("register_data : %s", register_data)
272		294
273	@classmethod	295	@classmethod
274	def unregister_post(cls, instance: Instance, endpoint: Endpoint) -> None:	296	def unregister_post(cls, instance: Instance, endpoint: Endpoint) -> None:
	@@ -303,7 +325,7 @@ class ConductorApiClient:
303	logger.error(	325	logger.error(
304	"Exception occurred while register to conductor at %s: %s", client_args.get('address', 'unknown'), e	326	"Exception occurred while register to conductor at %s: %s", client_args.get('address', 'unknown'), e
305	)	327	)
306	- logger.info(f"unregister_data : {register_data}")	328	+ logger.info("unregister_data : %s", register_data)
307		329
308	# ── Circuit breaker for /query ──────────────────────────────────	330	# ── Circuit breaker for /query ──────────────────────────────────
309	_query_failures: int = 0	331	_query_failures: int = 0
	@@ -311,10 +333,48 @@ class ConductorApiClient:
311	_QUERY_CB_THRESHOLD: int = 3 # consecutive failures to trip	333	_QUERY_CB_THRESHOLD: int = 3 # consecutive failures to trip
312	_QUERY_CB_COOLDOWN: float = 30.0 # seconds to stay open	334	_QUERY_CB_COOLDOWN: float = 30.0 # seconds to stay open
313		335
		336	+ @classmethod
		337	+ def _encode_query(cls, query_data: dict[str, Any], encoding: str) -> tuple[bytes \| None, str]:
		338	+ """Encode the query body for the requested wire encoding.
		339	+
		340	+ Returns ``(body, content_type)``; JSON bodies are encoded lazily by
		341	+ SafeHTTPSClient (``body=None``).
		342	+ """
		343	+ if encoding == "msgpack":
		344	+ return encode_query_msgpack(query_data), MSGPACK_CONTENT_TYPE
		345	+ if encoding == "json":
		346	+ return None, "application/json"
		347	+ logger.warning("Unknown query_encoding=%s, falling back to msgpack", encoding)
		348	+ return encode_query_msgpack(query_data), MSGPACK_CONTENT_TYPE
		349	+
		350	+ @classmethod
		351	+ def _decode_query_response(cls, response: Any, encoding: str) -> dict[str, Any]:
		352	+ """Decode a /query response honoring the server's Content-Type.
		353	+
		354	+ MessagePack responses are decoded with msgspec; everything else
		355	+ (including legacy JSON servers that ignore the Accept header) is
		356	+ parsed as JSON — so an upgraded client keeps working against an
		357	+ older kv-conductor binary.
		358	+ """
		359	+ content_type = response.headers.get("Content-Type", "").lower()
		360	+ if content_type.startswith(MSGPACK_CONTENT_TYPE):
		361	+ return decode_query_response_msgpack(response.content)
		362	+ if encoding == "msgpack":
		363	+ logger.debug(
		364	+ "conductor replied with Content-Type=%s (expected msgpack); parsing as JSON",
		365	+ content_type or "none",
		366	+ )
		367	+ return response.json()
		368	+
314	@classmethod	369	@classmethod
315	def query_conductor(cls, instances: list[Instance], encoded_ids: list[int]) -> dict[str, Any]:	370	def query_conductor(cls, instances: list[Instance], encoded_ids: list[int]) -> dict[str, Any]:
316	"""Query KV conductor for prefix cache matched blocks.	371	"""Query KV conductor for prefix cache matched blocks.
317		372
		373	+ Wire encoding is selected by ``kv_conductor_config.query_encoding``
		374	+ (default ``"msgpack"``): MessagePack bodies are faster to serialize
		375	+ and smaller on long-context queries. Response parsing follows the
		376	+ server's Content-Type, so legacy JSON conductors still work.
		377	+
318	Circuit breaker: after ``_QUERY_CB_THRESHOLD`` consecutive failures,	378	Circuit breaker: after ``_QUERY_CB_THRESHOLD`` consecutive failures,
319	skip queries for ``_QUERY_CB_COOLDOWN`` seconds.	379	skip queries for ``_QUERY_CB_COOLDOWN`` seconds.
320	"""	380	"""
	@@ -342,16 +402,28 @@ class ConductorApiClient:
342	if TENANT_ID != "default":	402	if TENANT_ID != "default":
		jason lyu22 天前服务端返回msgpack错误映射时，代码会走json()解析失败并触发熔断，与测试预期不符。 jason lyu 22 天前评论： 22 天前评论：
343	query_data["tenant_id"] = TENANT_ID	403	query_data["tenant_id"] = TENANT_ID
344		404
345	- logger.debug(f"query_data : {query_data}")	405	+ logger.debug(
		406	+ "query_data : model=%s block_size=%s tokens=%d",
		407	+ query_data["model"],
		408	+ query_data["block_size"],
		409	+ len(encoded_ids),
		410	+ )
		411	+
		412	+ encoding = getattr(reg, "query_encoding", "msgpack")
		413	+ body, content_type = cls._encode_query(query_data, encoding)
346		414
347	client_args = {"address": format_address(reg.conductor_service, reg.http_server_port)}	415	client_args = {"address": format_address(reg.conductor_service, reg.http_server_port)}
348		416
349	try:	417	try:
350	with SafeHTTPSClient(timeout=3, **client_args) as client:	418	with SafeHTTPSClient(timeout=3, **client_args) as client:
351	- response = client.post("/query", query_data)	419	+ if body is not None:
352	- logger.info("conductor query response: %s", response)	420	+ response = client.post_bytes("/query", body, content_type=content_type)
		421	+ else:
		422	+ response = client.do_post("/query", data=query_data)
		423	+ parsed = cls._decode_query_response(response, encoding)
		424	+ logger.debug("conductor query ok: %s instances, %d bytes", len(parsed), len(response.content))
353	cls._query_failures = 0 # reset on success	425	cls._query_failures = 0 # reset on success
354	- return response	426	+ return parsed
355	except Exception as e:	427	except Exception as e:
356	cls._query_failures += 1	428	cls._query_failures += 1
357	if cls._query_failures >= cls._QUERY_CB_THRESHOLD:	429	if cls._query_failures >= cls._QUERY_CB_THRESHOLD:

Mmotor/kv_conductor/Cargo.toml+1-0

	@@ -35,6 +35,7 @@ thiserror = "2"
35	clap = { version = "4", features = ["derive"] }	35	clap = { version = "4", features = ["derive"] }
36	tower-http = { version = "0.5", features = ["cors", "trace"] }	36	tower-http = { version = "0.5", features = ["cors", "trace"] }
37	zmq = "0.10"	37	zmq = "0.10"
		38	+rmp = "0.8"
38	rmp-serde = "1"	39	rmp-serde = "1"
39	rmpv = { version = "1", features = ["with-serde"] }	40	rmpv = { version = "1", features = ["with-serde"] }
40	tokio-util = "0.7"	41	tokio-util = "0.7"

Mmotor/kv_conductor/src/protocols.rs+256-2

	@@ -207,7 +207,7 @@ pub struct UnregisterRequest {
207	}	207	}
208		208
209	/// POST /query request body (matching Python client).	209	/// POST /query request body (matching Python client).
210	-#[derive(Debug, Clone, Deserialize)]	210	+#[derive(Debug, Clone, Serialize, Deserialize)]
211	pub struct QueryRequest {	211	pub struct QueryRequest {
212	pub model: String,	212	pub model: String,
213	pub block_size: u32,	213	pub block_size: u32,
	@@ -219,7 +219,7 @@ pub struct QueryRequest {
219	/// POST /query_by_hash request body — query using pre-computed block hashes	219	/// POST /query_by_hash request body — query using pre-computed block hashes
220	/// instead of raw token IDs. This avoids redundant XXH3 computation when the	220	/// instead of raw token IDs. This avoids redundant XXH3 computation when the
221	/// caller has already hashed the sequence.	221	/// caller has already hashed the sequence.
222	-#[derive(Debug, Clone, Deserialize)]	222	+#[derive(Debug, Clone, Serialize, Deserialize)]
223	pub struct QueryByHashRequest {	223	pub struct QueryByHashRequest {
224	pub model: String,	224	pub model: String,
225	pub block_size: u32,	225	pub block_size: u32,
	@@ -274,6 +274,102 @@ pub struct QueryResponse {
274	pub tenants: HashMap<String, HashMap<InstanceId, InstanceMatchData>>,	274	pub tenants: HashMap<String, HashMap<InstanceId, InstanceMatchData>>,
275	}	275	}
276		276
		277	+// ---------------------------------------------------------------------------
		278	+// MessagePack query codec
		279	+//
		280	+// `/query` and `/query_by_hash` accept both JSON and MessagePack bodies
		281	+// (selected by `Content-Type: application/msgpack`). The request side is
		282	+// decoded with `rmp_serde` straight into `QueryRequest` / `QueryByHashRequest`
		283	+// (neither uses serde `flatten`, so the generic path works). The response
		284	+// side is hand-encoded below because `QueryResponse` relies on
		285	+// `#[serde(flatten)]`, which MessagePack serializers do not support.
		286	+//
		287	+// The MessagePack response mirrors the JSON wire shape exactly:
		288	+//
		289	+// ```text
		290	+// { tenant_id: { instance_id: { longest_matched, DP:
		291	+// { rank: { matched_tokens, npu_blocks, cpu_blocks, disk_blocks } } } } }
		292	+// ```
		293	+
		294	+/// Content-Type values accepted as MessagePack on the query endpoints.
		295	+pub const MSGPACK_CONTENT_TYPES: [&str; 2] = ["application/msgpack", "application/x-msgpack"];
		296	+
		297	+/// True when the request `Content-Type` header selects MessagePack.
		298	+pub fn is_msgpack_content_type(headers: &axum::http::HeaderMap) -> bool {
		299	+ headers
		300	+ .get(axum::http::header::CONTENT_TYPE)
		301	+ .and_then(\|v\| v.to_str().ok())
		302	+ .map(\|ct\| {
		303	+ // Strip parameters such as "; charset=utf-8".
		304	+ let ct = ct
		305	+ .split(';')
		306	+ .next()
		307	+ .unwrap_or("")
		308	+ .trim()
		309	+ .to_ascii_lowercase();
		310	+ MSGPACK_CONTENT_TYPES.contains(&ct.as_str())
		311	+ })
		312	+ .unwrap_or(false)
		313	+}
		314	+
		315	+/// Encode a full `/query` response into MessagePack.
		316	+///
		317	+/// Written with `rmp::encode` instead of `rmp_serde` to avoid the
		318	+/// `#[serde(flatten)]` map-merge pitfall on `QueryResponse`.
		319	+pub fn encode_query_response_msgpack(response: &QueryResponse, out: &mut Vec<u8>) {
		320	+ use rmp::encode::*;
		321	+ write_map_len(
		322	+ out,
		323	+ u32::try_from(response.tenants.len()).expect("tenants len fits u32"),
		324	+ )
		325	+ .expect("write map len");
		326	+ for (tenant, instances) in &response.tenants {
		327	+ write_str(out, tenant).expect("write tenant");
		328	+ write_map_len(
		329	+ out,
		330	+ u32::try_from(instances.len()).expect("instances len fits u32"),
		331	+ )
		jason lyu22 天前用rmp::encode手写编码DpBlocks时，matched_tokens可能大于u32上限(4294967295)，write_u32会panic，应改用write_u64。 jason lyu 22 天前评论： 22 天前评论：
		332	+ .expect("write map len");
		333	+ for (instance, data) in instances {
		334	+ write_str(out, instance).expect("write instance");
		335	+ write_map_len(out, 2).expect("instance map len");
		336	+ write_str(out, "longest_matched").expect("write key");
		337	+ write_u32(out, data.longest_matched).expect("write longest_matched");
		338	+ write_str(out, "DP").expect("write key");
		339	+ write_map_len(out, u32::try_from(data.dp.len()).expect("dp len fits u32"))
		340	+ .expect("write map len");
		341	+ for (rank, blocks) in &data.dp {
		342	+ write_str(out, rank).expect("write rank");
		343	+ write_map_len(out, 4).expect("blocks map len");
		344	+ write_str(out, "matched_tokens").expect("write key");
		345	+ write_u32(out, blocks.matched_tokens).expect("write matched_tokens");
		346	+ write_str(out, "npu_blocks").expect("write key");
		347	+ write_u32(out, blocks.npu_blocks).expect("write npu_blocks");
		348	+ write_str(out, "cpu_blocks").expect("write key");
		349	+ write_u32(out, blocks.cpu_blocks).expect("write cpu_blocks");
		350	+ write_str(out, "disk_blocks").expect("write key");
		351	+ write_u32(out, blocks.disk_blocks).expect("write disk_blocks");
		352	+ }
		353	+ }
		354	+ }
		355	+}
		356	+
		357	+/// Encode a single-key error map `{ "error": "..." }` into MessagePack.
		358	+pub fn encode_error_msgpack(message: &str, out: &mut Vec<u8>) {
		359	+ use rmp::encode::*;
		360	+ write_map_len(out, 1).expect("map len");
		361	+ write_str(out, "error").expect("write key");
		362	+ write_str(out, message).expect("write error");
		363	+}
		364	+
		365	+/// Encode the empty query result `{ "<tenant_id>": {} }` into MessagePack.
		366	+pub fn encode_empty_tenant_msgpack(tenant_id: &str, out: &mut Vec<u8>) {
		367	+ use rmp::encode::*;
		368	+ write_map_len(out, 1).expect("map len");
		369	+ write_str(out, tenant_id).expect("write tenant");
		370	+ write_map_len(out, 0).expect("empty map");
		371	+}
		372	+
277	// ---------------------------------------------------------------------------	373	// ---------------------------------------------------------------------------
278	// KV event types (for POST /events, push-based KV cache event ingestion)	374	// KV event types (for POST /events, push-based KV cache event ingestion)
279	// ---------------------------------------------------------------------------	375	// ---------------------------------------------------------------------------
	@@ -840,4 +936,162 @@ mod tests {
840	};	936	};
841	assert_eq!(store.parent_hash, Some(999));	937	assert_eq!(store.parent_hash, Some(999));
842	}	938	}
		939	+
		940	+ // -----------------------------------------------------------------------
		941	+ // MessagePack query codec
		942	+ // -----------------------------------------------------------------------
		943	+
		944	+ #[test]
		945	+ fn test_query_request_msgpack_roundtrip() {
		946	+ let req = QueryRequest {
		947	+ model: "llama-7b".into(),
		948	+ block_size: 128,
		949	+ token_ids: (0..100_000).map(\|i\| i % 32000).collect(),
		950	+ tenant_id: "default".into(),
		951	+ };
		952	+ let encoded = rmp_serde::to_vec(&req).unwrap();
		953	+ let decoded: QueryRequest = rmp_serde::from_slice(&encoded).unwrap();
		954	+ assert_eq!(decoded.model, req.model);
		955	+ assert_eq!(decoded.block_size, req.block_size);
		956	+ assert_eq!(decoded.tenant_id, req.tenant_id);
		957	+ assert_eq!(decoded.token_ids, req.token_ids);
		958	+ }
		959	+
		960	+ #[test]
		961	+ fn test_query_by_hash_request_msgpack_roundtrip() {
		962	+ let req = QueryByHashRequest {
		963	+ model: "llama-7b".into(),
		964	+ block_size: 128,
		965	+ block_hashes: (0..10_000).map(\|i\| (i as u64) * 2654435761).collect(),
		966	+ tenant_id: "default".into(),
		967	+ };
		968	+ let encoded = rmp_serde::to_vec(&req).unwrap();
		969	+ let decoded: QueryByHashRequest = rmp_serde::from_slice(&encoded).unwrap();
		970	+ assert_eq!(decoded.block_hashes, req.block_hashes);
		971	+ }
		972	+
		973	+ /// Convert a MessagePack value into its JSON equivalent so the two wire
		974	+ /// shapes can be compared structurally.
		975	+ fn rmpv_to_json(v: &rmpv::Value) -> serde_json::Value {
		976	+ match v {
		977	+ rmpv::Value::Nil => serde_json::Value::Null,
		978	+ rmpv::Value::Boolean(b) => serde_json::Value::Bool(*b),
		979	+ rmpv::Value::Integer(i) => {
		980	+ if let Some(u) = i.as_u64() {
		981	+ serde_json::Value::from(u)
		982	+ } else {
		983	+ serde_json::Value::from(i.as_i64().unwrap_or_default())
		984	+ }
		985	+ }
		986	+ rmpv::Value::F64(f) => serde_json::Value::from(*f),
		987	+ rmpv::Value::F32(f) => serde_json::Value::from(*f),
		988	+ rmpv::Value::String(s) => {
		989	+ serde_json::Value::String(s.as_str().unwrap_or_default().to_string())
		990	+ }
		991	+ rmpv::Value::Binary(b) => serde_json::Value::String(format!("{b:?}")),
		992	+ rmpv::Value::Array(a) => serde_json::Value::Array(a.iter().map(rmpv_to_json).collect()),
		993	+ rmpv::Value::Map(m) => {
		994	+ let mut map = serde_json::Map::new();
		995	+ for (k, v) in m {
		996	+ let key = match k {
		997	+ rmpv::Value::String(s) => s.as_str().unwrap_or_default().to_string(),
		998	+ other => format!("{other:?}"),
		999	+ };
		1000	+ map.insert(key, rmpv_to_json(v));
		1001	+ }
		1002	+ serde_json::Value::Object(map)
		1003	+ }
		1004	+ rmpv::Value::Ext(..) => serde_json::Value::Null,
		1005	+ }
		1006	+ }
		1007	+
		1008	+ fn sample_query_response() -> QueryResponse {
		1009	+ let mut tenants = HashMap::new();
		1010	+ let mut instances = HashMap::new();
		1011	+ let mut dp = HashMap::new();
		1012	+ dp.insert(
		1013	+ "0".to_string(),
		1014	+ DpBlocks {
		1015	+ matched_tokens: 384,
		1016	+ npu_blocks: 3,
		1017	+ cpu_blocks: 0,
		1018	+ disk_blocks: 0,
		1019	+ },
		1020	+ );
		1021	+ dp.insert(
		1022	+ "1".to_string(),
		1023	+ DpBlocks {
		1024	+ matched_tokens: 512,
		1025	+ npu_blocks: 1,
		1026	+ cpu_blocks: 3,
		1027	+ disk_blocks: 0,
		1028	+ },
		1029	+ );
		1030	+ instances.insert(
		1031	+ "prefill-0".to_string(),
		1032	+ InstanceMatchData {
		1033	+ longest_matched: 512,
		1034	+ dp,
		1035	+ },
		1036	+ );
		1037	+ tenants.insert("default".to_string(), instances);
		1038	+ QueryResponse { tenants }
		1039	+ }
		1040	+
		1041	+ #[test]
		1042	+ fn test_query_response_msgpack_matches_json_shape() {
		1043	+ let response = sample_query_response();
		1044	+ let mut buf = Vec::new();
		1045	+ encode_query_response_msgpack(&response, &mut buf);
		1046	+
		1047	+ // Decode the msgpack payload and compare with the JSON wire shape
		1048	+ // field-by-field. This guards the hand-written encoder against
		1049	+ // drifting from the serde_json shape (which the Python client parses).
		1050	+ let msgpack_value = rmpv::decode::read_value(&mut buf.as_slice()).unwrap();
		1051	+ let msgpack_json = rmpv_to_json(&msgpack_value);
		1052	+ let json_value = serde_json::to_value(&response).unwrap();
		1053	+ assert_eq!(
		1054	+ msgpack_json, json_value,
		1055	+ "msgpack response diverges from JSON wire shape"
		1056	+ );
		1057	+ }
		1058	+
		1059	+ #[test]
		1060	+ fn test_query_response_msgpack_empty_tenant() {
		1061	+ let response = QueryResponse::default();
		1062	+ let mut buf = Vec::new();
		1063	+ encode_query_response_msgpack(&response, &mut buf);
		1064	+ let msgpack_value = rmpv::decode::read_value(&mut buf.as_slice()).unwrap();
		1065	+ assert_eq!(rmpv_to_json(&msgpack_value), serde_json::json!({}));
		1066	+ }
		1067	+
		1068	+ #[test]
		1069	+ fn test_error_msgpack_encoding() {
		1070	+ let mut err = Vec::new();
		1071	+ encode_error_msgpack("boom", &mut err);
		1072	+ assert_eq!(
		1073	+ rmpv_to_json(&rmpv::decode::read_value(&mut err.as_slice()).unwrap()),
		1074	+ serde_json::json!({"error": "boom"})
		1075	+ );
		1076	+ }
		1077	+
		1078	+ #[test]
		1079	+ fn test_is_msgpack_content_type() {
		1080	+ let cases: Vec<(&str, bool)> = vec![
		1081	+ ("application/msgpack", true),
		1082	+ ("application/x-msgpack", true),
		1083	+ ("application/msgpack; charset=utf-8", true),
		1084	+ ("Application/MSGPACK", true),
		1085	+ ("application/json", false),
		1086	+ ("", false),
		1087	+ ("text/plain", false),
		1088	+ ];
		1089	+ for (ct, expected) in cases {
		1090	+ let mut headers = axum::http::HeaderMap::new();
		1091	+ if !ct.is_empty() {
		1092	+ headers.insert(axum::http::header::CONTENT_TYPE, ct.parse().unwrap());
		1093	+ }
		1094	+ assert_eq!(is_msgpack_content_type(&headers), expected, "ct={ct}");
		1095	+ }
		1096	+ }
843	}	1097	}

Mmotor/kv_conductor/src/server.rs+109-44

	@@ -22,8 +22,10 @@
22	use std::sync::Arc;	22	use std::sync::Arc;
23		23
24	use axum::{	24	use axum::{
		25	+ body::Bytes,
25	extract::State,	26	extract::State,
26	- http::StatusCode,	27	+ http::{header, HeaderMap, StatusCode},
		28	+ response::{IntoResponse, Response},
27	routing::{get, post},	29	routing::{get, post},
28	Json, Router,	30	Json, Router,
29	};	31	};
	@@ -122,6 +124,68 @@ async fn unregister_handler(
122	}	124	}
		jason lyu22 天前 rmp_serde::from_slice会进行完整拷贝，1M token的msgpack请求解码开销较大，可考虑用零拷贝或流式解析，但这在编程模型上较难实现。 jason lyu 22 天前评论： 22 天前评论：
123	}	125	}
124		126
		127	+/// Parse the query request body into `T`, honoring the `Content-Type` header.
		128	+///
		129	+/// MessagePack bodies (`application/msgpack`) are decoded via `rmp_serde`;
		130	+/// everything else is treated as JSON (the historical default). Callers must
		131	+/// answer in the same encoding — sniff `is_msgpack_content_type` once before
		132	+/// calling and reuse the flag for the response.
		133	+fn parse_query_body<T: serde::de::DeserializeOwned>(
		134	+ headers: &HeaderMap,
		135	+ body: &Bytes,
		136	+) -> Result<T, String> {
		137	+ if is_msgpack_content_type(headers) {
		138	+ rmp_serde::from_slice::<T>(body).map_err(\|e\| format!("invalid msgpack request body: {e}"))
		139	+ } else {
		140	+ serde_json::from_slice::<T>(body).map_err(\|e\| format!("invalid JSON request body: {e}"))
		141	+ }
		142	+}
		143	+
		144	+/// Render the query error/empty-result responses in the request's encoding.
		145	+fn query_error_response(status: StatusCode, message: &str, msgpack: bool) -> Response {
		146	+ if msgpack {
		147	+ let mut buf = Vec::with_capacity(32 + message.len());
		148	+ encode_error_msgpack(message, &mut buf);
		149	+ (status, [(header::CONTENT_TYPE, "application/msgpack")], buf).into_response()
		150	+ } else {
		151	+ (status, Json(serde_json::json!({ "error": message }))).into_response()
		152	+ }
		153	+}
		154	+
		155	+fn empty_tenant_response(tenant_id: &str, msgpack: bool) -> Response {
		156	+ if msgpack {
		157	+ let mut buf = Vec::with_capacity(16 + tenant_id.len());
		158	+ encode_empty_tenant_msgpack(tenant_id, &mut buf);
		159	+ (
		160	+ StatusCode::OK,
		161	+ [(header::CONTENT_TYPE, "application/msgpack")],
		162	+ buf,
		163	+ )
		164	+ .into_response()
		165	+ } else {
		166	+ (StatusCode::OK, Json(serde_json::json!({ tenant_id: {} }))).into_response()
		167	+ }
		168	+}
		169	+
		170	+fn ok_query_response(response: QueryResponse, msgpack: bool) -> Response {
		171	+ if msgpack {
		172	+ let mut buf = Vec::with_capacity(256);
		173	+ encode_query_response_msgpack(&response, &mut buf);
		174	+ (
		175	+ StatusCode::OK,
		176	+ [(header::CONTENT_TYPE, "application/msgpack")],
		177	+ buf,
		178	+ )
		179	+ .into_response()
		180	+ } else {
		181	+ (
		182	+ StatusCode::OK,
		183	+ Json(serde_json::to_value(response).unwrap_or_default()),
		184	+ )
		185	+ .into_response()
		186	+ }
		187	+}
		188	+
125	/// POST /query	189	/// POST /query
126	///	190	///
127	/// Request body: `{ "model": "...", "block_size": 128, "token_ids": [...], "tenant_id": "default" }`	191	/// Request body: `{ "model": "...", "block_size": 128, "token_ids": [...], "tenant_id": "default" }`
	@@ -129,90 +193,91 @@ async fn unregister_handler(
129	/// Response: `{ "<tenant_id>": { "<instance_id>": { "longest_matched": N,	193	/// Response: `{ "<tenant_id>": { "<instance_id>": { "longest_matched": N,
130	/// "DP": { "<rank>": { "matched_tokens": N, "npu_blocks": N,	194	/// "DP": { "<rank>": { "matched_tokens": N, "npu_blocks": N,
131	/// "cpu_blocks": N, "disk_blocks": N } } } } }`	195	/// "cpu_blocks": N, "disk_blocks": N } } } } }`
132	-async fn query_handler(	196	+///
133	- State(state): State<AppState>,	197	+/// Both JSON (default) and MessagePack (`Content-Type: application/msgpack`)
134	- Json(req): Json<QueryRequest>,	198	+/// encodings are accepted; the response is returned in the request's encoding.
135	-) -> (StatusCode, Json<serde_json::Value>) {	199	+async fn query_handler(State(state): State<AppState>, headers: HeaderMap, body: Bytes) -> Response {
		200	+ let msgpack = is_msgpack_content_type(&headers);
		201	+ let req = match parse_query_body::<QueryRequest>(&headers, &body) {
		202	+ Ok(parsed) => parsed,
		203	+ Err(message) => {
		204	+ return query_error_response(StatusCode::BAD_REQUEST, &message, msgpack);
		205	+ }
		206	+ };
		207	+
136	tracing::debug!(	208	tracing::debug!(
137	model = %req.model,	209	model = %req.model,
138	tenant = %req.tenant_id,	210	tenant = %req.tenant_id,
139	num_tokens = req.token_ids.len(),	211	num_tokens = req.token_ids.len(),
		212	+ msgpack,
140	"query request"	213	"query request"
141	);	214	);
142		215
143	match state.registry.query(&req).await {	216	match state.registry.query(&req).await {
144	- Ok(response) => (	217	+ Ok(response) => ok_query_response(response, msgpack),
145	- StatusCode::OK,
146	- Json(serde_json::to_value(response).unwrap_or_default()),
147	- ),
148	Err(KvConductorError::NoIndexer {	218	Err(KvConductorError::NoIndexer {
149	model_name,	219	model_name,
150	tenant_id,	220	tenant_id,
151	- }) => (	221	+ }) => query_error_response(
152	StatusCode::NOT_FOUND,	222	StatusCode::NOT_FOUND,
153	- Json(serde_json::json!({	223	+ &format!("no indexer for model={model_name} tenant={tenant_id}"),
154	- "error": format!("no indexer for model={} tenant={}", model_name, tenant_id)	224	+ msgpack,
155	- })),
156	),	225	),
157	Err(KvConductorError::NoWorkers {	226	Err(KvConductorError::NoWorkers {
158	model_name: _,	227	model_name: _,
159	tenant_id,	228	tenant_id,
160	- }) => (	229	+ }) => {
161	- StatusCode::OK,
162	// Return empty response structure matching expected format	230	// Return empty response structure matching expected format
163	- Json(serde_json::json!({	231	+ empty_tenant_response(&tenant_id, msgpack)
164	- tenant_id: {}	232	+ }
165	- })),	233	+ Err(e) => query_error_response(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string(), msgpack),
166	- ),
167	- Err(e) => (
168	- StatusCode::INTERNAL_SERVER_ERROR,
169	- Json(serde_json::json!({"error": e.to_string()})),
170	- ),
171	}	234	}
172	}	235	}
173		236
174	/// POST /query_by_hash	237	/// POST /query_by_hash
175	///	238	///
176	/// Same semantics as `/query` but accepts pre-computed block hashes instead	239	/// Same semantics as `/query` but accepts pre-computed block hashes instead
177	-/// of raw token IDs, avoiding redundant XXH3 computation.	240	+/// of raw token IDs, avoiding redundant XXH3 computation. Supports the same
		241	+/// JSON / MessagePack content negotiation as `/query`.
178	async fn query_by_hash_handler(	242	async fn query_by_hash_handler(
179	State(state): State<AppState>,	243	State(state): State<AppState>,
180	- Json(req): Json<QueryByHashRequest>,	244	+ headers: HeaderMap,
181	-) -> (StatusCode, Json<serde_json::Value>) {	245	+ body: Bytes,
		246	+) -> Response {
		247	+ let msgpack = is_msgpack_content_type(&headers);
		248	+ let req = match parse_query_body::<QueryByHashRequest>(&headers, &body) {
		249	+ Ok(parsed) => parsed,
		250	+ Err(message) => {
		251	+ return query_error_response(StatusCode::BAD_REQUEST, &message, msgpack);
		252	+ }
		253	+ };
		254	+
182	tracing::debug!(	255	tracing::debug!(
183	model = %req.model,	256	model = %req.model,
184	tenant = %req.tenant_id,	257	tenant = %req.tenant_id,
185	num_hashes = req.block_hashes.len(),	258	num_hashes = req.block_hashes.len(),
		259	+ msgpack,
186	"query_by_hash request"	260	"query_by_hash request"
187	);	261	);
188		262
189	match state.registry.query_by_hash(&req).await {	263	match state.registry.query_by_hash(&req).await {
190	- Ok(response) => (	264	+ Ok(response) => ok_query_response(response, msgpack),
191	- StatusCode::OK,
192	- Json(serde_json::to_value(response).unwrap_or_default()),
193	- ),
194	Err(KvConductorError::NoIndexer {	265	Err(KvConductorError::NoIndexer {
195	model_name,	266	model_name,
196	tenant_id,	267	tenant_id,
197	- }) => (	268	+ }) => query_error_response(
198	StatusCode::NOT_FOUND,	269	StatusCode::NOT_FOUND,
199	- Json(serde_json::json!({	270	+ &format!("no indexer for model={model_name} tenant={tenant_id}"),
200	- "error": format!("no indexer for model={} tenant={}", model_name, tenant_id)	271	+ msgpack,
201	- })),
202	),	272	),
203	Err(KvConductorError::NoWorkers {	273	Err(KvConductorError::NoWorkers {
204	model_name: _,	274	model_name: _,
205	tenant_id,	275	tenant_id,
206	- }) => (	276	+ }) => {
207	- StatusCode::OK,	277	+ // Return empty response structure matching expected format
208	- Json(serde_json::json!({	278	+ empty_tenant_response(&tenant_id, msgpack)
209	- tenant_id: {}	279	+ }
210	- })),	280	+ Err(e) => query_error_response(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string(), msgpack),
211	- ),
212	- Err(e) => (
213	- StatusCode::INTERNAL_SERVER_ERROR,
214	- Json(serde_json::json!({"error": e.to_string()})),
215	- ),
216	}	281	}
217	}	282	}
218		283

Amotor/kv_conductor/tests/common/mod.rs+48-0

@@ -0,0 +1,48 @@
	1	+// Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
	2	+// MindIE is licensed under Mulan PSL v2.
	3	+// You can use this software according to the terms and conditions of the Mulan PSL v2.
	4	+// You may obtain a copy of Mulan PSL v2 at:
	5	+// http://license.coscl.org.cn/MulanPSL2
	6	+// THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	7	+// EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	8	+// MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	9	+// See the Mulan PSL v2 for more details.
	10	+
	11	+//! Shared helpers for integration tests.
	12	+//!
	13	+//! Note: unit tests inside `src/` cannot reference this directory (separate
	14	+//! crates), so `src/protocols.rs` keeps its own copy of `rmpv_to_json`.
	15	+
	16	+/// Convert a MessagePack value into its JSON equivalent for structural
	17	+/// comparison with the JSON wire shape.
	18	+pub fn rmpv_to_json(v: &rmpv::Value) -> serde_json::Value {
	19	+ use serde_json::Value;
	20	+ match v {
	21	+ rmpv::Value::Nil => Value::Null,
	22	+ rmpv::Value::Boolean(b) => Value::Bool(*b),
	23	+ rmpv::Value::Integer(i) => {
	24	+ if let Some(u) = i.as_u64() {
	25	+ Value::from(u)
	26	+ } else {
	27	+ Value::from(i.as_i64().unwrap_or_default())
	28	+ }
	29	+ }
	30	+ rmpv::Value::F64(f) => Value::from(*f),
	31	+ rmpv::Value::F32(f) => Value::from(*f),
	32	+ rmpv::Value::String(s) => Value::String(s.as_str().unwrap_or_default().to_string()),
	33	+ rmpv::Value::Binary(b) => Value::String(format!("{b:?}")),
	34	+ rmpv::Value::Array(a) => Value::Array(a.iter().map(rmpv_to_json).collect()),
	35	+ rmpv::Value::Map(m) => {
	36	+ let mut map = serde_json::Map::new();
	37	+ for (k, val) in m {
	38	+ let key = match k {
	39	+ rmpv::Value::String(s) => s.as_str().unwrap_or_default().to_string(),
	40	+ other => format!("{other:?}"),
	41	+ };
	42	+ map.insert(key, rmpv_to_json(val));
	43	+ }
	44	+ Value::Object(map)
	45	+ }
	46	+ rmpv::Value::Ext(..) => Value::Null,
	47	+ }
	48	+}

Mmotor/kv_conductor/tests/integration_test.rs+218-0

	@@ -19,6 +19,8 @@ use tokio::net::TcpListener;
19	use kv_conductor::registry::WorkerRegistry;	19	use kv_conductor::registry::WorkerRegistry;
20	use kv_conductor::server::{create_router, AppState};	20	use kv_conductor::server::{create_router, AppState};
21		21
		22	+mod common;
		23	+
22	/// Start a test server on a random port, returning the base URL.	24	/// Start a test server on a random port, returning the base URL.
23	async fn start_test_server() -> (String, tokio::task::JoinHandle<()>) {	25	async fn start_test_server() -> (String, tokio::task::JoinHandle<()>) {
24	let registry = Arc::new(WorkerRegistry::new());	26	let registry = Arc::new(WorkerRegistry::new());
	@@ -855,3 +857,219 @@ async fn test_reregister_different_backend_drops_tree() {
855	"tree should be dropped on backend-change re-registration"	857	"tree should be dropped on backend-change re-registration"
856	);	858	);
857	}	859	}
		860	+
		861	+// ── MessagePack /query content negotiation ──────────────────────────────
		862	+
		863	+/// Register one vLLM-style worker and inject an engine-style stored event so
		864	+/// the indexer has data to answer queries with.
		865	+async fn register_and_seed(client: &Client, base_url: &str) {
		866	+ let reg = json!({
		867	+ "instance_id": "vllm-prefill-42",
		868	+ "medium_endpoints": {
		869	+ "npu": "tcp://10.0.9.1:50090",
		870	+ "cpu": "tcp://10.0.9.1:50090",
		871	+ "disk": "tcp://10.0.9.1:50090"
		872	+ },
		873	+ "type": "vllm",
		874	+ "modelname": "msgpack-model",
		875	+ "block_size": 4,
		876	+ "dp_rank": 0,
		877	+ "tenant_id": "default"
		878	+ });
		879	+ let resp = client
		880	+ .post(format!("{base_url}/register"))
		881	+ .json(&reg)
		882	+ .send()
		883	+ .await
		884	+ .unwrap();
		885	+ assert_eq!(resp.status(), 201);
		886	+
		887	+ let events = json!({
		888	+ "instance_id": "vllm-prefill-42",
		889	+ "events": [
		890	+ {
		891	+ "event_id": 1,
		892	+ "data": {
		893	+ "type": "stored",
		894	+ "parent_hash": null,
		895	+ "blocks": [
		896	+ {"block_hash": 100, "tokens_hash": 12345678901234567890_u64}
		897	+ ]
		898	+ },
		899	+ "dp_rank": 0
		900	+ }
		901	+ ],
		902	+ "shutdown": false
		903	+ });
		904	+ let resp = client
		905	+ .post(format!("{base_url}/events"))
		906	+ .json(&events)
		907	+ .send()
		908	+ .await
		909	+ .unwrap();
		910	+ assert_eq!(resp.status(), 200);
		911	+}
		912	+
		913	+#[tokio::test]
		914	+async fn test_query_msgpack_endpoint() {
		915	+ let (base_url, _handle) = start_test_server().await;
		916	+ let client = Client::new();
		917	+ register_and_seed(&client, &base_url).await;
		918	+
		919	+ let req = kv_conductor::QueryRequest {
		920	+ model: "msgpack-model".into(),
		921	+ block_size: 4,
		922	+ token_ids: (1..=8).collect(),
		923	+ tenant_id: "default".into(),
		924	+ };
		925	+ let resp = client
		926	+ .post(format!("{base_url}/query"))
		927	+ .header(reqwest::header::CONTENT_TYPE, "application/msgpack")
		928	+ .body(rmp_serde::to_vec(&req).unwrap())
		929	+ .send()
		930	+ .await
		931	+ .unwrap();
		932	+
		933	+ assert_eq!(resp.status(), 200);
		934	+ let content_type = resp
		935	+ .headers()
		936	+ .get(reqwest::header::CONTENT_TYPE)
		937	+ .and_then(\|v\| v.to_str().ok())
		938	+ .unwrap_or("")
		939	+ .to_string();
		940	+ assert!(
		941	+ content_type.starts_with("application/msgpack"),
		942	+ "expected msgpack response, got {content_type}"
		943	+ );
		944	+
		945	+ let bytes = resp.bytes().await.unwrap();
		946	+ let msgpack_value = rmpv::decode::read_value(&mut bytes.as_ref()).unwrap();
		947	+ let msgpack_json = common::rmpv_to_json(&msgpack_value);
		948	+
		949	+ // The JSON query must produce the exact same wire shape.
		950	+ let json_resp = client
		951	+ .post(format!("{base_url}/query"))
		952	+ .json(&json!({
		953	+ "model": "msgpack-model",
		954	+ "block_size": 4,
		955	+ "token_ids": [1, 2, 3, 4, 5, 6, 7, 8],
		956	+ "tenant_id": "default"
		957	+ }))
		958	+ .send()
		959	+ .await
		960	+ .unwrap();
		961	+ assert_eq!(json_resp.status(), 200);
		962	+ let json_body: Value = json_resp.json().await.unwrap();
		963	+
		964	+ assert_eq!(
		965	+ msgpack_json, json_body,
		966	+ "msgpack and JSON query responses diverge"
		967	+ );
		968	+ assert!(
		969	+ json_body.get("default").is_some(),
		970	+ "expected a seeded tenant entry"
		971	+ );
		972	+}
		973	+
		974	+#[tokio::test]
		975	+async fn test_query_by_hash_msgpack_endpoint() {
		976	+ let (base_url, _handle) = start_test_server().await;
		977	+ let client = Client::new();
		978	+ register_and_seed(&client, &base_url).await;
		979	+
		980	+ let req = kv_conductor::QueryByHashRequest {
		981	+ model: "msgpack-model".into(),
		982	+ block_size: 4,
		983	+ block_hashes: vec![
		984	+ kv_conductor::hashing::compute_block_hash_for_seq(&[1, 2, 3, 4], 4)[0].0,
		985	+ ],
		986	+ tenant_id: "default".into(),
		987	+ };
		988	+ let resp = client
		989	+ .post(format!("{base_url}/query_by_hash"))
		990	+ .header(reqwest::header::CONTENT_TYPE, "application/x-msgpack")
		991	+ .body(rmp_serde::to_vec(&req).unwrap())
		992	+ .send()
		993	+ .await
		994	+ .unwrap();
		995	+
		996	+ assert_eq!(resp.status(), 200);
		997	+ let content_type = resp
		998	+ .headers()
		999	+ .get(reqwest::header::CONTENT_TYPE)
		1000	+ .and_then(\|v\| v.to_str().ok())
		1001	+ .unwrap_or("")
		1002	+ .to_string();
		1003	+ assert!(content_type.starts_with("application/msgpack"));
		1004	+
		1005	+ let bytes = resp.bytes().await.unwrap();
		1006	+ let msgpack_value = rmpv::decode::read_value(&mut bytes.as_ref()).unwrap();
		1007	+ let msgpack_json = common::rmpv_to_json(&msgpack_value);
		1008	+ assert!(
		1009	+ msgpack_json.get("default").is_some(),
		1010	+ "expected a tenant entry for query_by_hash msgpack"
		1011	+ );
		1012	+}
		1013	+
		1014	+#[tokio::test]
		1015	+async fn test_query_msgpack_error_paths() {
		1016	+ let (base_url, _handle) = start_test_server().await;
		1017	+ let client = Client::new();
		1018	+
		1019	+ // 404: unregistered model, error must come back as msgpack.
		1020	+ let req = kv_conductor::QueryRequest {
		1021	+ model: "no-such-model".into(),
		1022	+ block_size: 4,
		1023	+ token_ids: vec![1, 2, 3, 4],
		1024	+ tenant_id: "default".into(),
		1025	+ };
		1026	+ let resp = client
		1027	+ .post(format!("{base_url}/query"))
		1028	+ .header(reqwest::header::CONTENT_TYPE, "application/msgpack")
		1029	+ .body(rmp_serde::to_vec(&req).unwrap())
		1030	+ .send()
		1031	+ .await
		1032	+ .unwrap();
		1033	+ assert_eq!(resp.status(), 404);
		1034	+ let content_type = resp
		1035	+ .headers()
		1036	+ .get(reqwest::header::CONTENT_TYPE)
		1037	+ .and_then(\|v\| v.to_str().ok())
		1038	+ .unwrap_or("")
		1039	+ .to_string();
		1040	+ assert!(content_type.starts_with("application/msgpack"));
		1041	+ let bytes = resp.bytes().await.unwrap();
		1042	+ let err = common::rmpv_to_json(&rmpv::decode::read_value(&mut bytes.as_ref()).unwrap());
		1043	+ assert!(
		1044	+ err.get("error").is_some(),
		1045	+ "expected msgpack error map, got {err}"
		1046	+ );
		1047	+
		1048	+ // 400: malformed msgpack body.
		1049	+ let resp = client
		1050	+ .post(format!("{base_url}/query"))
		1051	+ .header(reqwest::header::CONTENT_TYPE, "application/msgpack")
		1052	+ .body(vec![0xc1u8, 0xff, 0x00]) // invalid msgpack bytes
		1053	+ .send()
		1054	+ .await
		1055	+ .unwrap();
		1056	+ assert_eq!(resp.status(), 400);
		1057	+ let bytes = resp.bytes().await.unwrap();
		1058	+ let err = common::rmpv_to_json(&rmpv::decode::read_value(&mut bytes.as_ref()).unwrap());
		1059	+ assert!(
		1060	+ err.get("error").is_some(),
		1061	+ "expected msgpack error map for malformed body, got {err}"
		1062	+ );
		1063	+
		1064	+ // The same malformed body without a msgpack Content-Type yields a JSON error.
		1065	+ let resp = client
		1066	+ .post(format!("{base_url}/query"))
		1067	+ .header(reqwest::header::CONTENT_TYPE, "application/json")
		1068	+ .body(vec![0xc1u8, 0xff, 0x00])
		1069	+ .send()
		1070	+ .await
		1071	+ .unwrap();
		1072	+ assert_eq!(resp.status(), 400);
		1073	+ let body: Value = resp.json().await.unwrap();
		1074	+ assert!(body.get("error").is_some());
		1075	+}

Mtests/common/utils/test_http_client.py+86-23

	@@ -1,5 +1,3 @@
1	-#!/usr/bin/env python3
2	-# -- coding: utf-8 --
3	# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.	1	# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
4	# MindIE is licensed under Mulan PSL v2.	2	# MindIE is licensed under Mulan PSL v2.
5	# 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.
	@@ -47,18 +45,9 @@ def test_init_with_valid_parameters(base_url, cert_files):
47	"""test init with valid parameters"""	45	"""test init with valid parameters"""
48	cert_file, key_file, ca_file = cert_files	46	cert_file, key_file, ca_file = cert_files
49		47
50	- tls_config = TLSConfig(	48	+ tls_config = TLSConfig(enable_tls=True, cert_file=cert_file, key_file=key_file, ca_file=ca_file)
51	- enable_tls=True,
52	- cert_file=cert_file,
53	- key_file=key_file,
54	- ca_file=ca_file
55	- )
56		49
57	- client = SafeHTTPSClient(	50	+ client = SafeHTTPSClient(address=base_url, tls_config=tls_config, timeout=10)
58	- address=base_url,
59	- tls_config=tls_config,
60	- timeout=10
61	- )
62		51
63	assert client.base_url == f"https://{base_url}"	52	assert client.base_url == f"https://{base_url}"
64	assert client.timeout == 10	53	assert client.timeout == 10
	@@ -68,17 +57,10 @@ def test_init_with_valid_parameters(base_url, cert_files):
68		57
69	def test_init_with_missing_cert_files(base_url):	58	def test_init_with_missing_cert_files(base_url):
70	"""test init with missing cert files"""	59	"""test init with missing cert files"""
71	- tls_config = TLSConfig(	60	+ tls_config = TLSConfig(enable_tls=True, cert_file="nonexistent.crt", key_file="nonexistent.key")
72	- enable_tls=True,	61	+ # CertUtil.create_ssl_context returns None if cert files don't exist,
73	- cert_file="nonexistent.crt",
74	- key_file="nonexistent.key"
75	- )
76	- # CertUtil.create_ssl_context returns None if cert files don't exist,
77	# but client can still be initialized (SSL will fail at runtime)	62	# but client can still be initialized (SSL will fail at runtime)
78	- client = SafeHTTPSClient(	63	+ client = SafeHTTPSClient(address=base_url, tls_config=tls_config)
79	- address=base_url,
80	- tls_config=tls_config
81	- )
82	# Client should still initialize, but SSL context creation may have failed	64	# Client should still initialize, but SSL context creation may have failed
83	assert client.base_url == f"https://{base_url}"	65	assert client.base_url == f"https://{base_url}"
84	assert client.protocol == 'https://'	66	assert client.protocol == 'https://'
	@@ -207,3 +189,84 @@ def test_request_timeout(base_url):
207		189
208	call_kwargs = mock_request.call_args[1]	190	call_kwargs = mock_request.call_args[1]
209	assert call_kwargs['timeout'] == 3.5	191	assert call_kwargs['timeout'] == 3.5
		192	+
		193	+
		194	+# ── post_bytes / msgpack error extraction ───────────────────────────────
		195	+
		196	+
		197	+def test_post_bytes_sends_raw_body(base_url):
		198	+ """post_bytes sends the raw body with msgpack Content-Type headers."""
		199	+ client = SafeHTTPSClient(address=base_url)
		200	+ body = b"\x81\xa5hello\xa5world"
		201	+
		202	+ with patch.object(client.session, 'request') as mock_request:
		203	+ mock_response = Mock()
		204	+ mock_response.status_code = 200
		205	+ mock_response.headers = {"Content-Type": "application/msgpack"}
		206	+ mock_request.return_value = mock_response
		207	+
		208	+ resp = client.post_bytes("/query", body)
		209	+
		210	+ assert resp is mock_response
		211	+ kwargs = mock_request.call_args[1]
		212	+ assert kwargs["data"] == body
		213	+ assert kwargs["headers"]["Content-Type"] == "application/msgpack"
		214	+ assert kwargs["headers"]["Accept"] == "application/msgpack"
		215	+
		216	+
		217	+def test_post_bytes_json_headers_override(base_url):
		218	+ """post_bytes honors explicit content_type/accept overrides."""
		219	+ client = SafeHTTPSClient(address=base_url)
		220	+
		221	+ with patch.object(client.session, 'request') as mock_request:
		222	+ mock_request.return_value = Mock(status_code=200, headers={})
		223	+ client.post_bytes("/query", b"x", content_type="application/json", accept="application/json")
		224	+
		225	+ headers = mock_request.call_args[1]["headers"]
		226	+ assert headers["Content-Type"] == "application/json"
		227	+
		228	+
		229	+def _error_response(content_type, content, text=""):
		230	+ mock_response = Mock()
		231	+ mock_response.status_code = 404
		232	+ mock_response.headers = {"Content-Type": content_type}
		233	+ mock_response.content = content
		234	+ mock_response.text = text
		235	+ return mock_response
		236	+
		237	+
		238	+def test_http_error_msgpack_body_extracts_error(base_url):
		239	+ """msgpack error bodies surface the readable error field."""
		240	+ import msgspec
		241	+
		242	+ client = SafeHTTPSClient(address=base_url)
		243	+ err_body = msgspec.msgpack.encode({"error": "no indexer for model=x"})
		244	+ mock_response = _error_response("application/msgpack", err_body)
		245	+
		246	+ with patch.object(client.session, 'request') as mock_request:
		247	+ mock_request.side_effect = requests.exceptions.HTTPError(response=mock_response)
		248	+ with pytest.raises(Exception, match="no indexer for model=x"):
		249	+ client.get("/query")
		250	+
		251	+
		252	+def test_http_error_json_body_extracts_error(base_url):
		253	+ """JSON error bodies surface the readable error field."""
		254	+ client = SafeHTTPSClient(address=base_url)
		255	+ mock_response = _error_response("application/json", b'{"error": "boom-json"}', text="ignored")
		256	+ mock_response.json.return_value = {"error": "boom-json"}
		257	+
		258	+ with patch.object(client.session, 'request') as mock_request:
		259	+ mock_request.side_effect = requests.exceptions.HTTPError(response=mock_response)
		260	+ with pytest.raises(Exception, match="boom-json"):
		261	+ client.get("/query")
		262	+
		263	+
		264	+def test_http_error_non_error_body_falls_back_to_text(base_url):
		265	+ """Non-map / undecodable error bodies fall back to the raw text."""
		266	+ client = SafeHTTPSClient(address=base_url)
		267	+ mock_response = _error_response("text/plain", b"gateway down", text="gateway down")
		268	+
		269	+ with patch.object(client.session, 'request') as mock_request:
		270	+ mock_request.side_effect = requests.exceptions.HTTPError(response=mock_response)
		271	+ with pytest.raises(Exception, match="gateway down"):
		272	+ client.get("/query")

Mtests/coordinator/api_client/test_conductor_api_client.py+100-4

	@@ -12,9 +12,11 @@
12		12
13	from unittest.mock import Mock, patch	13	from unittest.mock import Mock, patch
14		14
		15	+import msgspec
15		16
16	from motor.common.resources.instance import Instance, Endpoint, PDRole	17	from motor.common.resources.instance import Instance, Endpoint, PDRole
17	from motor.coordinator.api_client.conductor_api_client import (	18	from motor.coordinator.api_client.conductor_api_client import (
		19	+ MSGPACK_CONTENT_TYPE,
18	TENANT_ID,	20	TENANT_ID,
19	ConductorApiClient,	21	ConductorApiClient,
20	conductor_instance_id,	22	conductor_instance_id,
	@@ -556,11 +558,37 @@ def _make_mock_instance(instance_id: int):
556	return instance	558	return instance
557		559
558		560
559	-def _mock_successful_query(mock_http, response=None):	561	+def _mock_successful_query(
		562	+ mock_http,
		563	+ response=None,
		564	+ *,
		565	+ encoding="msgpack",
		566	+ response_content_type=None,
		567	+):
		568	+ """Mock a successful /query round trip.
		569	+
		570	+ ``encoding`` selects which client-side path the mock wires up
		571	+ (``post_bytes`` for msgpack, ``do_post`` for JSON). The fake response
		572	+ carries the given ``Content-Type`` (defaults to the request encoding) so
		573	+ the client-side response parsing is exercised end to end.
		574	+ """
560	if response is None:	575	if response is None:
561	response = {TENANT_ID: {}}	576	response = {TENANT_ID: {}}
		577	+ if response_content_type is None:
		578	+ response_content_type = MSGPACK_CONTENT_TYPE if encoding == "msgpack" else "application/json"
		579	+
		580	+ fake_resp = Mock()
		581	+ fake_resp.headers.get.side_effect = lambda key, default=None: (
		582	+ response_content_type if key.lower() == "content-type" else default
		583	+ )
		584	+ fake_resp.content = (
		585	+ msgspec.msgpack.encode(response) if response_content_type.startswith(MSGPACK_CONTENT_TYPE) else b""
		586	+ )
		587	+ fake_resp.json.return_value = response
		588	+
562	mock_client = Mock()	589	mock_client = Mock()
563	- mock_client.post.return_value = response	590	+ mock_client.post_bytes.return_value = fake_resp
		591	+ mock_client.do_post.return_value = fake_resp
564	mock_http.return_value.__enter__.return_value = mock_client	592	mock_http.return_value.__enter__.return_value = mock_client
565		593
566		594
	@@ -570,7 +598,7 @@ def _mock_failed_query(mock_http):
570		598
571	@patch("motor.coordinator.api_client.conductor_api_client.SafeHTTPSClient")	599	@patch("motor.coordinator.api_client.conductor_api_client.SafeHTTPSClient")
572	def test_return_value_on_success(mock_http):	600	def test_return_value_on_success(mock_http):
573	- """On success, query_conductor returns the response dict."""	601	+ """On success (msgpack default), query_conductor returns the response dict."""
574	expected = {TENANT_ID: {"vllm-prefill-1": {"longest_matched": 100, "DP": {"0": 50}}}}	602	expected = {TENANT_ID: {"vllm-prefill-1": {"longest_matched": 100, "DP": {"0": 50}}}}
575	_mock_successful_query(mock_http, response=expected)	603	_mock_successful_query(mock_http, response=expected)
576	instances = [_make_mock_instance(1)]	604	instances = [_make_mock_instance(1)]
	@@ -579,6 +607,67 @@ def test_return_value_on_success(mock_http):
579	assert result == expected	607	assert result == expected
580		608
581		609
		610	+@patch("motor.coordinator.api_client.conductor_api_client.SafeHTTPSClient")
		611	+def test_query_msgpack_wire_format(mock_http):
		612	+ """The msgpack path sends a MessagePack body with the right Content-Type
		613	+ and decodes the MessagePack response.
		614	+ """
		615	+ expected = {TENANT_ID: {"vllm-prefill-1": {"longest_matched": 384, "DP": {"0": 3}}}}
		616	+ _mock_successful_query(mock_http, response=expected)
		617	+ instances = [_make_mock_instance(1)]
		618	+ token_ids = list(range(1000))
		619	+
		620	+ result = ConductorApiClient.query_conductor(instances, token_ids)
		621	+ assert result == expected
		622	+
		623	+ mock_client = mock_http.return_value.__enter__.return_value
		624	+ body = mock_client.post_bytes.call_args[0][1]
		625	+ content_type = mock_client.post_bytes.call_args[1]["content_type"]
		626	+ assert content_type == MSGPACK_CONTENT_TYPE
		627	+ # The request body must decode back to the exact query data.
		628	+ decoded = msgspec.msgpack.decode(body)
		629	+ assert decoded["model"] == "test-model"
		630	+ assert decoded["block_size"] == 128
		631	+ assert decoded["token_ids"] == token_ids
		632	+ # tenant_id is omitted on the wire when it equals the default.
		633	+ assert decoded.get("tenant_id", TENANT_ID) == TENANT_ID
		634	+
		635	+
		636	+@patch("motor.coordinator.api_client.conductor_api_client.SafeHTTPSClient")
		637	+def test_query_json_encoding_config(mock_http):
		638	+ """query_encoding='json' keeps the legacy JSON wire path."""
		639	+ expected = {TENANT_ID: {"vllm-prefill-1": {"longest_matched": 100}}}
		640	+ _mock_successful_query(mock_http, response=expected, encoding="json")
		641	+ instances = [_make_mock_instance(1)]
		642	+
		643	+ with _setup_reg_config("Mooncake", query_encoding="json"):
		644	+ result = ConductorApiClient.query_conductor(instances, [1, 2, 3])
		645	+ assert result == expected
		646	+
		647	+ mock_client = mock_http.return_value.__enter__.return_value
		648	+ assert mock_client.post_bytes.call_count == 0
		649	+ json_data = mock_client.do_post.call_args[1]["data"]
		650	+ assert json_data["token_ids"] == [1, 2, 3]
		651	+
		652	+
		653	+@patch("motor.coordinator.api_client.conductor_api_client.SafeHTTPSClient")
		654	+def test_query_legacy_json_response_fallback(mock_http):
		655	+ """A msgpack request answered by a legacy JSON-only conductor still
		656	+ parses correctly (response Content-Type fallback).
		657	+ """
		658	+ expected = {TENANT_ID: {"vllm-prefill-1": {"longest_matched": 100}}}
		659	+ _mock_successful_query(
		660	+ mock_http,
		661	+ response=expected,
		662	+ encoding="msgpack",
		663	+ response_content_type="application/json",
		664	+ )
		665	+ instances = [_make_mock_instance(1)]
		666	+
		667	+ result = ConductorApiClient.query_conductor(instances, [1, 2, 3])
		668	+ assert result == expected
		669	+
		670	+
582	@patch("motor.coordinator.api_client.conductor_api_client.SafeHTTPSClient")	671	@patch("motor.coordinator.api_client.conductor_api_client.SafeHTTPSClient")
583	def test_return_value_on_failure(mock_http):	672	def test_return_value_on_failure(mock_http):
584	"""On failure, query_conductor returns an empty dict."""	673	"""On failure, query_conductor returns an empty dict."""
	@@ -593,7 +682,13 @@ def test_return_value_on_failure(mock_http):
593		682
594		683
595	def _setup_reg_config(	684	def _setup_reg_config(
596	- store_backend, pool_endpoint="", npu_endpoint="", cpu_endpoint="", disk_endpoint="", replay_endpoint=""	685	+ store_backend,
		686	+ pool_endpoint="",
		687	+ npu_endpoint="",
		688	+ cpu_endpoint="",
		689	+ disk_endpoint="",
		690	+ replay_endpoint="",
		691	+ query_encoding="msgpack",
597	):	692	):
598	"""Patch ConductorApiClient's config for registration testing."""	693	"""Patch ConductorApiClient's config for registration testing."""
599	from motor.config.coordinator import KvConductorConfig, SchedulerConfig	694	from motor.config.coordinator import KvConductorConfig, SchedulerConfig
	@@ -605,6 +700,7 @@ def _setup_reg_config(
605	cpu_endpoint=cpu_endpoint,	700	cpu_endpoint=cpu_endpoint,
606	disk_endpoint=disk_endpoint,	701	disk_endpoint=disk_endpoint,
607	replay_endpoint=replay_endpoint,	702	replay_endpoint=replay_endpoint,
		703	+ query_encoding=query_encoding,
608	)	704	)
609	sched = SchedulerConfig(kv_conductor_config=reg)	705	sched = SchedulerConfig(kv_conductor_config=reg)
610	return patch.object(	706	return patch.object(

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

             setattr(config.standby_config, param, value)
         elif param in ["etcd_port", "etcd_timeout"]:
             setattr(config.etcd_config, param, value)
++        elif param in ["query_encoding"]:
++            setattr(config.scheduler_config.kv_conductor_config, param, value)
++        config.validate_config()
++
++
++def test_config_validation_query_encoding_defaults_ok():
++    """Default query_encoding (msgpack) and json both validate."""
++    config = CoordinatorConfig()
++    config.validate_config()  # default msgpack
++    config.scheduler_config.kv_conductor_config.query_encoding = "json"
++    config.validate_config()
++
++
++def test_config_validation_query_encoding_invalid():
++    """Invalid query_encoding fails at startup validation."""
++    with pytest.raises(ValueError, match="query_encoding must be one of"):
++        config = CoordinatorConfig()
++        config.scheduler_config.kv_conductor_config.query_encoding = "MsgPack"
         config.validate_config()