[bugfix] ZJ部署过程遇到的若干问题Bug-MindIE-Motor-AtomGit

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

	@@ -391,7 +391,7 @@ motor_coordinator_config字段配置样例如下所示：
391	\| transport_max_retry \| int/null \| Coordinator 传输失败的最大尝试次数；`null` 时使用 `max_retry`。默认值：`null` \|	391	\| transport_max_retry \| int/null \| Coordinator 传输失败的最大尝试次数；`null` 时使用 `max_retry`。默认值：`null` \|
392	\| retry_delay \| float \| 每次重试前的等待时间（秒）。默认值：`0.2` \|	392	\| retry_delay \| float \| 每次重试前的等待时间（秒）。默认值：`0.2` \|
393	\| first_token_timeout \| int \| 等待首 token 返回的超时时间（秒）。默认值：`600` \|	393	\| first_token_timeout \| int \| 等待首 token 返回的超时时间（秒）。默认值：`600` \|
394	-\| infer_timeout \| int \| 单次推理请求的总超时时间（秒）。默认值：`3600` \|	394	+\| infer_timeout \| int \| 单次推理请求的总超时时间（秒）。非流式场景作用于单次转发；流式场景作为整个流式请求的整体墙钟超时（从请求到达算起，超时后中断并返回 504）。默认：`3600` \|
tobking13 天前已过期 PR需求描述需要优化下，以具体修复问题为主题，不要体现xx局点信息 zhoujing 13 天前评论： 13 天前评论：
395	\| upstream_error_body_max_bytes \| int \| 向客户端透传引擎 HTTP 错误体的最大字节数，避免返回超大错误响应。默认值：`65536` \|	395	\| upstream_error_body_max_bytes \| int \| 向客户端透传引擎 HTTP 错误体的最大字节数，避免返回超大错误响应。默认值：`65536` \|
396	\| reschedule_config字段 \|-\|-\|	396	\| reschedule_config字段 \|-\|-\|
397	\| enable \| bool \| 故障场景重调度功能开关。默认：`false`。<br>模型重计算由引擎侧负责，该配置不控制引擎侧重计算；`recompute_enabled`仅作为`reschedule_enabled`的旧配置兼容别名；`recompute_max_retry`已移除并会被忽略。 \|	397	\| enable \| bool \| 故障场景重调度功能开关。默认：`false`。<br>模型重计算由引擎侧负责，该配置不控制引擎侧重计算；`recompute_enabled`仅作为`reschedule_enabled`的旧配置兼容别名；`recompute_max_retry`已移除并会被忽略。 \|

Mdocs/zh/user_guide/features/max_tokens_adaptation.md+4-0

	@@ -164,6 +164,10 @@ Context budget clamped req_id=<request-id> parameter=max_tokens requested=20000
164		164
165	- 该功能只调整有效的正整数 `max_tokens` 或 `max_completion_tokens`。请求未携带这些参数，	165	- 该功能只调整有效的正整数 `max_tokens` 或 `max_completion_tokens`。请求未携带这些参数，
166	或参数不是正整数时，Coordinator 不做调整。	166	或参数不是正整数时，Coordinator 不做调整。
		167	+- 请求中携带非正整数（如 0、负数、布尔、非 int 类型）的 `max_tokens` 或 `max_completion_tokens`
		168	+ 时，Coordinator 不会返回 400，而是移除该参数并记录 WARNING 日志，请求按未携带该参数处理
		169	+ （即按模型默认输出上限继续执行）。如需感知此类配置错误，请检查 Coordinator 日志中的
		170	+ `Invalid max_tokens=` / `Invalid max_completion_tokens=` 告警。
167	- 当输入 token 数已经达到或超过模型上下文上限时，Coordinator 不修改输出上限，	171	- 当输入 token 数已经达到或超过模型上下文上限时，Coordinator 不修改输出上限，
168	请求仍由后端推理引擎完成上下文校验并返回对应错误。	172	请求仍由后端推理引擎完成上下文校验并返回对应错误。
169	- `TokenizerManager` 无法得到有效 token ID 时，Coordinator 不根据估算值裁剪请求，	173	- `TokenizerManager` 无法得到有效 token ID 时，Coordinator 不根据估算值裁剪请求，

Mmotor/common/utils/error.py+1-0

	@@ -13,6 +13,7 @@ NODE_FAULT = "Node fault"
13	CLIENT_DISCONNECT = "Client disconnected"	13	CLIENT_DISCONNECT = "Client disconnected"
14	DISPATCH_ABORT = "Cancelled by dispatch"	14	DISPATCH_ABORT = "Cancelled by dispatch"
15	SCOPE_ABORT = "Cancelled via cancel scope"	15	SCOPE_ABORT = "Cancelled via cancel scope"
		16	+INFER_TIMEOUT = "Infer timeout"
16		17
17		18
18	class RequestCancelledError(Exception):	19	class RequestCancelledError(Exception):

Mmotor/coordinator/api_server/inference_server.py+20-0

	@@ -51,6 +51,24 @@ def get_request_manager(request: Request) -> RequestManager:
51	return request.app.state.request_manager	51	return request.app.state.request_manager
52		52
53		53
		54	+def _validate_positive_int_field(body_json: dict[str, Any], field_name: str) -> None:
		55	+ """Validate an optional positive integer field.
		56	+
		57	+ If the field is present but not a positive integer, remove it from the
		58	+ request body and log a warning with the body content before removal.
		59	+ """
		60	+ value = body_json.get(field_name)
		61	+ if value is None:
		62	+ return
		63	+ if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
		64	+ logger.warning(
		65	+ "Invalid %s=%r in request body, removing it.",
		66	+ field_name,
		67	+ value,
		68	+ )
		69	+ body_json.pop(field_name, None)
		y1lou13 天前无效的 max_tokens/max_completion_tokens 被静默删掉后请求按默认值继续——用户配错参数收不到 400，错误被吞掉，建议返回 400 或至少把该行为写进文档。 zhoujing 11 天前评论： 11 天前评论：
		70	+
		71	+
54	def _validate_anthropic_request(body_json: dict[str, Any], *, require_max_tokens: bool = True) -> None:	72	def _validate_anthropic_request(body_json: dict[str, Any], *, require_max_tokens: bool = True) -> None:
55	"""Validate Anthropic-style request body. Raises HTTPException on invalid."""	73	"""Validate Anthropic-style request body. Raises HTTPException on invalid."""
56	if not body_json.get("model"):	74	if not body_json.get("model"):
	@@ -114,6 +132,8 @@ def _validate_openai_request(body_json: dict[str, Any], request_type: RequestTyp
114	status_code=status.HTTP_400_BAD_REQUEST,	132	status_code=status.HTTP_400_BAD_REQUEST,
115	detail=f"Missing required field: {OpenAIField.MODEL}",	133	detail=f"Missing required field: {OpenAIField.MODEL}",
116	)	134	)
		135	+ _validate_positive_int_field(body_json, OpenAIField.MAX_TOKENS)
		136	+ _validate_positive_int_field(body_json, OpenAIField.MAX_COMPLETION_TOKENS)
117	if request_type != RequestType.OPENAI:	137	if request_type != RequestType.OPENAI:
118	return	138	return
119	if OpenAIField.PROMPT not in body_json and OpenAIField.MESSAGES not in body_json and "input" not in body_json:	139	if OpenAIField.PROMPT not in body_json and OpenAIField.MESSAGES not in body_json and "input" not in body_json:

Mmotor/coordinator/router/dispatch_session.py+1-0

	@@ -35,6 +35,7 @@ class AttemptState(str, Enum):
35	class AttemptStopReason(str, Enum):	35	class AttemptStopReason(str, Enum):
36	CLIENT_DISCONNECT = "client_disconnect"	36	CLIENT_DISCONNECT = "client_disconnect"
37	PEER_FAILED = "peer_failed"	37	PEER_FAILED = "peer_failed"
		38	+ TIMEOUT = "timeout"
38	OTHER = "other"	39	OTHER = "other"
39		40
40		41

Mmotor/coordinator/router/strategies/base.py+12-1

	@@ -88,7 +88,7 @@ def check_cancel_error(error: asyncio.CancelledError) -> (str, bool):
88	reason = "Exception"	88	reason = "Exception"
89	if error.args:	89	if error.args:
90	reason = error.args[0]	90	reason = error.args[0]
91	- if reason in {cancel_error.CLIENT_DISCONNECT, cancel_error.DISPATCH_ABORT}:	91	+ if reason in {cancel_error.CLIENT_DISCONNECT, cancel_error.DISPATCH_ABORT, cancel_error.INFER_TIMEOUT}:
92	return reason, False	92	return reason, False
93	elif reason.startswith(cancel_error.SCOPE_ABORT):	93	elif reason.startswith(cancel_error.SCOPE_ABORT):
94	return cancel_error.SCOPE_ABORT, False	94	return cancel_error.SCOPE_ABORT, False
	@@ -140,6 +140,17 @@ class BaseRouter(ABC):
140	)	140	)
141	self._sampling_manager = sampling_manager	141	self._sampling_manager = sampling_manager
142		142
		143	+ def _stream_overall_timeout(self) -> float:
		144	+ """Remaining infer_timeout budget for the streaming response, counted from request arrival.
		145	+
		146	+ Streaming responses are served by uvicorn after the handler returns, so the
		147	+ ``timeout_handler`` decorator cannot bound them; the budget is passed to
		148	+ CommitAwareStreamingResponse and enforced as an overall wall-clock deadline.
		149	+ """
		150	+ infer_timeout = self.config.exception_config.infer_timeout
		151	+ elapsed = time.time() - self.req_info.status.get(ReqState.ARRIVE, time.time())
		152	+ return max(infer_timeout - elapsed, 0.0)
		153	+
143	@staticmethod	154	@staticmethod
144	def build_error_response(e: Exception) -> ErrorResponse:	155	def build_error_response(e: Exception) -> ErrorResponse:
145	if isinstance(e, HTTPException):	156	if isinstance(e, HTTPException):

Mmotor/coordinator/router/strategies/pd_hybrid.py+1-0

	@@ -192,6 +192,7 @@ class PDHybridRouter(BaseRouter):
192	self._generate_stream(req_data, manage_request_context=manage_request_context),	192	self._generate_stream(req_data, manage_request_context=manage_request_context),
193	self._stream_commit_controller,	193	self._stream_commit_controller,
194	on_first_body_sent=self._mark_stream_body_sent,	194	on_first_body_sent=self._mark_stream_body_sent,
		195	+ timeout=self._stream_overall_timeout(),
195	)	196	)
196	return await self._generate_post(req_data, manage_request_context=manage_request_context)	197	return await self._generate_post(req_data, manage_request_context=manage_request_context)
197		198

Mmotor/coordinator/router/strategies/unified_pd.py+3-0

	@@ -298,6 +298,7 @@ class UnifiedPDRouter(BaseRouter):
298	self._generate_stream_response(),	298	self._generate_stream_response(),
299	self._stream_commit_controller,	299	self._stream_commit_controller,
300	on_first_body_sent=self._mark_stream_body_sent,	300	on_first_body_sent=self._mark_stream_body_sent,
		301	+ timeout=self._stream_overall_timeout(),
301	)	302	)
302	return await self._generate_response()	303	return await self._generate_response()
303		304
	@@ -640,6 +641,8 @@ class UnifiedPDRouter(BaseRouter):
640	return AttemptStopReason.PEER_FAILED	641	return AttemptStopReason.PEER_FAILED
641	if reason == cancel_error.CLIENT_DISCONNECT:	642	if reason == cancel_error.CLIENT_DISCONNECT:
642	return AttemptStopReason.CLIENT_DISCONNECT	643	return AttemptStopReason.CLIENT_DISCONNECT
		644	+ if reason == cancel_error.INFER_TIMEOUT:
		645	+ return AttemptStopReason.TIMEOUT
643	return AttemptStopReason.OTHER	646	return AttemptStopReason.OTHER
644		647
645	@staticmethod	648	@staticmethod

Mmotor/coordinator/router/stream_response.py+42-1

	@@ -109,6 +109,7 @@ class CommitAwareStreamingResponse(Response):
109	media_type: str \| None = None,	109	media_type: str \| None = None,
110	background: BackgroundTask \| None = None,	110	background: BackgroundTask \| None = None,
111	on_first_body_sent: Callable[[], None] \| None = None,	111	on_first_body_sent: Callable[[], None] \| None = None,
		112	+ timeout: float \| None = None,
112	) -> None:	113	) -> None:
113	super().__init__(	114	super().__init__(
114	content=None,	115	content=None,
	@@ -121,6 +122,7 @@ class CommitAwareStreamingResponse(Response):
121	self._raw_iterator = content	122	self._raw_iterator = content
122	self.controller = controller	123	self.controller = controller
123	self._on_first_body_sent = on_first_body_sent	124	self._on_first_body_sent = on_first_body_sent
		125	+ self._timeout = timeout
124	self._first_body_sent = False	126	self._first_body_sent = False
125	self._finished = False	127	self._finished = False
126	self._consumer_mode: str \| None = None	128	self._consumer_mode: str \| None = None
	@@ -155,6 +157,16 @@ class CommitAwareStreamingResponse(Response):
155	pump_task = asyncio.create_task(self._pump_stream(send, terminal))	157	pump_task = asyncio.create_task(self._pump_stream(send, terminal))
156	ready_task = asyncio.create_task(self.controller.wait_ready())	158	ready_task = asyncio.create_task(self.controller.wait_ready())
157	cancel_reason = None	159	cancel_reason = None
		160	+ timeout_handle = None
		161	+ if self._timeout is not None:
		162	+ # Overall wall-clock deadline for the whole streaming request (infer_timeout).
		163	+ # Same mechanism as client disconnect (stream_task.cancel(msg=...)): cancel the
		164	+ # current task with an explicit reason so Task.cancel() cascades that message
		165	+ # into pump_task and the upstream generator reports "Infer timeout" instead of
		166	+ # a bare "Exception" (asyncio.timeout() cancels without a message).
		167	+ timeout_handle = asyncio.get_running_loop().call_later(
		168	+ self._timeout, asyncio.current_task().cancel, cancel_error.INFER_TIMEOUT
		169	+ )
158	try:	170	try:
159	done, _ = await asyncio.wait(	171	done, _ = await asyncio.wait(
160	(terminal, ready_task),	172	(terminal, ready_task),
	@@ -188,13 +200,25 @@ class CommitAwareStreamingResponse(Response):
188	raise	200	raise
189	except asyncio.CancelledError as error:	201	except asyncio.CancelledError as error:
190	cancel_reason = error.args[0] if error.args else None	202	cancel_reason = error.args[0] if error.args else None
191	- raise	203	+ if cancel_reason == cancel_error.INFER_TIMEOUT:
		204	+ # Overall streaming timeout (infer_timeout) exceeded: stop the upstream
		205	+ # legs first, then surface 504 (JSON pre-commit / SSE post-commit).
		206	+ await self._cancel_and_wait(pump_task, ready_task, reason=cancel_reason)
		207	+ timeout_error = TimeoutError(f"Streaming request timed out after {self._timeout} seconds")
		208	+ if self.controller.committed:
		209	+ await self._send_committed_error(send, timeout_error)
		210	+ else:
		211	+ await self._send_precommit_error(scope, receive, send, timeout_error)
		212	+ else:
		213	+ raise
192	except Exception as error:	214	except Exception as error:
193	if self.controller.committed:	215	if self.controller.committed:
194	await self._send_committed_error(send, error)	216	await self._send_committed_error(send, error)
195	else:	217	else:
196	await self._send_precommit_error(scope, receive, send, error)	218	await self._send_precommit_error(scope, receive, send, error)
197	finally:	219	finally:
		220	+ if timeout_handle is not None:
		221	+ timeout_handle.cancel()
198	await self._cancel_and_wait(pump_task, ready_task, reason=cancel_reason)	222	await self._cancel_and_wait(pump_task, ready_task, reason=cancel_reason)
199		223
200	async def _pump_stream(	224	async def _pump_stream(
	@@ -248,6 +272,19 @@ class CommitAwareStreamingResponse(Response):
248	content={"detail": error.detail},	272	content={"detail": error.detail},
249	headers=error.headers,	273	headers=error.headers,
250	)	274	)
		275	+ elif isinstance(error, TimeoutError) or (
		276	+ isinstance(error, cancel_error.RequestCancelledError) and error.reason == cancel_error.INFER_TIMEOUT
		277	+ ):
		278	+ response = JSONResponse(
		279	+ status_code=status.HTTP_504_GATEWAY_TIMEOUT,
		280	+ content={
		281	+ "error": {
		282	+ "message": sanitize_error_message(str(error)),
		283	+ "type": "TimeoutError",
		284	+ "code": status.HTTP_504_GATEWAY_TIMEOUT,
		285	+ }
		286	+ },
		287	+ )
251	else:	288	else:
252	response = JSONResponse(	289	response = JSONResponse(
253	status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,	290	status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
	@@ -289,6 +326,10 @@ class CommitAwareStreamingResponse(Response):
289	code = error.status_code	326	code = error.status_code
290	elif isinstance(error, httpx.TimeoutException):	327	elif isinstance(error, httpx.TimeoutException):
291	code = status.HTTP_504_GATEWAY_TIMEOUT	328	code = status.HTTP_504_GATEWAY_TIMEOUT
		329	+ elif isinstance(error, TimeoutError) or (
		330	+ isinstance(error, cancel_error.RequestCancelledError) and error.reason == cancel_error.INFER_TIMEOUT
		331	+ ):
		332	+ code = status.HTTP_504_GATEWAY_TIMEOUT
292	elif isinstance(error, httpx.RequestError):	333	elif isinstance(error, httpx.RequestError):
293	code = status.HTTP_502_BAD_GATEWAY	334	code = status.HTTP_502_BAD_GATEWAY
294	elif isinstance(error, HTTPException):	335	elif isinstance(error, HTTPException):

Mmotor/coordinator/scheduler/runtime/scheduler_server.py+1-1

	@@ -384,7 +384,7 @@ class _SchedulerRequestDispatcher:
384	previously_open_ids: list[int] = []	384	previously_open_ids: list[int] = []
385	async with self._workload_commit_lock:	385	async with self._workload_commit_lock:
386	changed = await self._instance_manager.refresh_instances(event_type, instances)	386	changed = await self._instance_manager.refresh_instances(event_type, instances)
387	- if event_type == EventType.SET:	387	+ if event_type == EventType.SET and changed:
388	# Snapshot open instances before clearing so workers can be notified.	388	# Snapshot open instances before clearing so workers can be notified.
389	previously_open_ids = self._cb_manager.get_open_instance_ids()	389	previously_open_ids = self._cb_manager.get_open_instance_ids()
390	self._cb_manager.clear_all()	390	self._cb_manager.clear_all()

Mtests/coordinator/router/test_base_router_request_timeout.py+41-1

 """Tests for BaseRouter._build_request_timeout (bounded TCP connect phase)."""
++import time
 from unittest.mock import MagicMock
++import pytest
++
 from motor.config.coordinator import CoordinatorConfig
 from motor.coordinator.domain import (
     ScheduledResource,  # noqa: F401 -- domain must import first (models.request <-> domain cycle)
 )
++from motor.coordinator.models.request import RequestInfo, ReqState
 from motor.coordinator.router.strategies.base import BaseRouter
         assert timeout.connect == 30.0
         assert timeout.read == 30.0
++
++
++class TestStreamOverallTimeout:
++    """_stream_overall_timeout: remaining infer_timeout budget for streaming responses."""
++
++    def test_full_budget_when_just_arrived(self):
++        config = CoordinatorConfig()
++        config.exception_config.infer_timeout = 3600
++        router = _make_router(config)
++
++        budget = router._stream_overall_timeout()
++
++        assert 3599 < budget <= 3600
++
++    def test_elapsed_time_deducted_from_budget(self):
++        config = CoordinatorConfig()
++        config.exception_config.infer_timeout = 100
++        router = _make_router(config)
++        router.req_info.status[ReqState.ARRIVE] = time.time() - 40
++
++        assert router._stream_overall_timeout() == pytest.approx(60, abs=1)
++
++    def test_budget_floors_at_zero_after_deadline(self):
++        config = CoordinatorConfig()
++        config.exception_config.infer_timeout = 100
++        router = _make_router(config)
++        router.req_info.status[ReqState.ARRIVE] = time.time() - 200
++
++        assert router._stream_overall_timeout() == 0.0
++
++    def test_missing_arrive_time_uses_now(self):
++        config = CoordinatorConfig()
++        config.exception_config.infer_timeout = 100
++        router = _make_router(config)
++        router.req_info.status.pop(ReqState.ARRIVE, None)
++
++        assert router._stream_overall_timeout() == pytest.approx(100, abs=1)

Mtests/coordinator/router/test_cancel_error.py+7-0

	@@ -28,6 +28,7 @@ def _cancelled(*args: str) -> asyncio.CancelledError:
28	(_cancelled(), "Exception", True),	28	(_cancelled(), "Exception", True),
29	(_cancelled(cancel_error.CLIENT_DISCONNECT), cancel_error.CLIENT_DISCONNECT, False),	29	(_cancelled(cancel_error.CLIENT_DISCONNECT), cancel_error.CLIENT_DISCONNECT, False),
30	(_cancelled(cancel_error.DISPATCH_ABORT), cancel_error.DISPATCH_ABORT, False),	30	(_cancelled(cancel_error.DISPATCH_ABORT), cancel_error.DISPATCH_ABORT, False),
		31	+ (_cancelled(cancel_error.INFER_TIMEOUT), cancel_error.INFER_TIMEOUT, False),
31	(_cancelled(cancel_error.SCOPE_ABORT), cancel_error.SCOPE_ABORT, False),	32	(_cancelled(cancel_error.SCOPE_ABORT), cancel_error.SCOPE_ABORT, False),
32	(	33	(
33	_cancelled(	34	_cancelled(
	@@ -59,3 +60,9 @@ def test_request_cancelled_error_carries_reason() -> None:
59	error = RequestCancelledError(cancel_error.CLIENT_DISCONNECT)	60	error = RequestCancelledError(cancel_error.CLIENT_DISCONNECT)
60	assert error.reason == cancel_error.CLIENT_DISCONNECT	61	assert error.reason == cancel_error.CLIENT_DISCONNECT
61	assert str(error) == f"Request cancelled because of {cancel_error.CLIENT_DISCONNECT}"	62	assert str(error) == f"Request cancelled because of {cancel_error.CLIENT_DISCONNECT}"
		63	+
		64	+
		65	+def test_request_cancelled_error_with_infer_timeout_reason() -> None:
		66	+ error = RequestCancelledError(cancel_error.INFER_TIMEOUT)
		67	+ assert error.reason == cancel_error.INFER_TIMEOUT
		68	+ assert str(error) == f"Request cancelled because of {cancel_error.INFER_TIMEOUT}"

Mtests/coordinator/router/test_stream_response.py+110-0

	@@ -1,3 +1,13 @@
		1	+# Copyright (c) Huawei Technologies Co., Ltd. 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	+
1	import asyncio	11	import asyncio
2	import contextvars	12	import contextvars
3	import json	13	import json
	@@ -256,3 +266,103 @@ def test_controller_ignores_stale_attempt_readiness():
256		266
257	controller.mark_ready("prefill", 2)	267	controller.mark_ready("prefill", 2)
258	assert controller.ready_to_commit is True	268	assert controller.ready_to_commit is True
		269	+
		270	+
		271	+@pytest.mark.asyncio
		272	+async def test_response_timeout_cancels_upstream_with_infer_timeout_reason():
		273	+ messages = []
		274	+ source_started = asyncio.Event()
		275	+ cancellation_reasons = []
		276	+ controller = StreamCommitController.requiring({"engine"})
		277	+ controller.begin_attempt(1)
		278	+
		279	+ async def source():
		280	+ source_started.set()
		281	+ try:
		282	+ await asyncio.Event().wait()
		283	+ yield b"never"
		284	+ except asyncio.CancelledError as error:
		285	+ cancellation_reasons.extend(error.args)
		286	+ raise
		287	+
		288	+ async def send(message):
		289	+ messages.append(message)
		290	+
		291	+ response = CommitAwareStreamingResponse(source(), controller, timeout=0.05)
		292	+ await response(_scope(), _never_disconnect, send)
		293	+
		294	+ assert messages[0]["status"] == 504
		295	+ body = json.loads(messages[1]["body"])
		296	+ assert body["error"]["code"] == 504
		297	+ assert controller.committed is False
		298	+ assert cancellation_reasons == [cancel_error.INFER_TIMEOUT]
		299	+
		300	+
		301	+@pytest.mark.asyncio
		302	+async def test_response_timeout_after_commit_sends_504_sse():
		303	+ messages = []
		304	+ controller = StreamCommitController.requiring({"engine"})
		305	+ controller.begin_attempt(1)
		306	+
		307	+ async def source():
		308	+ controller.mark_ready("engine", 1)
		309	+ yield b'data: {"choices":[{"delta":{"content":"A"}}]}\n\n'
		310	+ await asyncio.Event().wait()
		311	+ yield b"never"
		312	+
		313	+ async def send(message):
		314	+ messages.append(message)
		315	+
		316	+ response = CommitAwareStreamingResponse(source(), controller, timeout=0.05)
		317	+ await response(_scope(), _never_disconnect, send)
		318	+
		319	+ assert messages[0]["status"] == 200
		320	+ bodies = [m["body"] for m in messages if m["type"] == "http.response.body"]
		321	+ last_data = [b for b in bodies if b.startswith(b"data:")][-1]
		322	+ payload = json.loads(last_data.decode().removeprefix("data: ").strip())
		323	+ assert payload["error"]["code"] == 504
		324	+ assert messages[-1]["more_body"] is False
		325	+
		326	+
		327	+@pytest.mark.asyncio
		328	+async def test_response_precommit_request_cancelled_infer_timeout_returns_504():
		329	+ messages = []
		330	+ controller = StreamCommitController.requiring({"engine"})
		331	+ controller.begin_attempt(1)
		332	+
		333	+ async def source():
		334	+ raise cancel_error.RequestCancelledError(cancel_error.INFER_TIMEOUT)
		335	+ yield b"" # pylint: disable=unreachable
		336	+
		337	+ async def send(message):
		338	+ messages.append(message)
		339	+
		340	+ response = CommitAwareStreamingResponse(source(), controller)
		341	+ await response(_scope(), _never_disconnect, send)
		342	+
		343	+ assert messages[0]["status"] == 504
		344	+ body = json.loads(messages[1]["body"])
		345	+ assert body["error"]["code"] == 504
		346	+
		347	+
		348	+@pytest.mark.asyncio
		349	+async def test_response_committed_request_cancelled_infer_timeout_returns_504_sse():
		350	+ messages = []
		351	+ controller = StreamCommitController.requiring({"engine"})
		352	+ controller.begin_attempt(1)
		353	+
		354	+ async def source():
		355	+ controller.mark_ready("engine", 1)
		356	+ yield b'data: {"choices":[{"delta":{"content":"A"}}]}\n\n'
		357	+ raise cancel_error.RequestCancelledError(cancel_error.INFER_TIMEOUT)
		358	+
		359	+ async def send(message):
		360	+ messages.append(message)
		361	+
		362	+ response = CommitAwareStreamingResponse(source(), controller)
		363	+ await response(_scope(), _never_disconnect, send)
		364	+
		365	+ bodies = [m["body"] for m in messages if m["type"] == "http.response.body"]
		366	+ last_data = [b for b in bodies if b.startswith(b"data:")][-1]
		367	+ payload = json.loads(last_data.decode().removeprefix("data: ").strip())
		368	+ assert payload["error"]["code"] == 504

Mtests/coordinator/router/test_unified_pd_router.py+1-0

	@@ -429,6 +429,7 @@ async def _invoke_asgi_response(response) -> list[dict]:
429	AttemptStopReason.PEER_FAILED,	429	AttemptStopReason.PEER_FAILED,
430	),	430	),
431	(cancel_error.CLIENT_DISCONNECT, AttemptStopReason.CLIENT_DISCONNECT),	431	(cancel_error.CLIENT_DISCONNECT, AttemptStopReason.CLIENT_DISCONNECT),
		432	+ (cancel_error.INFER_TIMEOUT, AttemptStopReason.TIMEOUT),
432	(cancel_error.DISPATCH_ABORT, AttemptStopReason.OTHER),	433	(cancel_error.DISPATCH_ABORT, AttemptStopReason.OTHER),
433	(cancel_error.SCOPE_ABORT, AttemptStopReason.OTHER),	434	(cancel_error.SCOPE_ABORT, AttemptStopReason.OTHER),
434	],	435	],

Mtests/coordinator/scheduler/test_scheduler_server_main.py+38-0

	@@ -667,6 +667,44 @@ class TestHandleRefreshInstances:
667	response = await dispatcher.dispatch(request)	667	response = await dispatcher.dispatch(request)
668	assert response.response_type == SchedulerResponseType.SUCCESS	668	assert response.response_type == SchedulerResponseType.SUCCESS
669		669
		670	+ @pytest.mark.asyncio
		671	+ async def test_set_event_without_change_does_not_clear_cb(self):
		672	+ """Repeated SET events with no instance change must not clear circuit breakers."""
		673	+ dispatcher, instance_manager, *_ = _make_dispatcher()
		674	+ dispatcher._cb_manager = MagicMock()
		675	+ dispatcher._cb_manager.get_open_instance_ids.return_value = []
		676	+ instance_manager.refresh_instances = AsyncMock(return_value=False)
		677	+
		678	+ request = SchedulerRequest(
		679	+ request_type=SchedulerRequestType.REFRESH_INSTANCES,
		680	+ request_id="req-set-1",
		681	+ data={"event_type": EventType.SET.value, "instances": []},
		682	+ )
		683	+ response = await dispatcher.dispatch(request)
		684	+
		685	+ assert response.response_type == SchedulerResponseType.SUCCESS
		686	+ dispatcher._cb_manager.clear_all.assert_not_called()
		687	+ dispatcher._cb_manager.get_open_instance_ids.assert_not_called()
		688	+
		689	+ @pytest.mark.asyncio
		690	+ async def test_set_event_with_change_clears_cb(self):
		691	+ """A real SET change still snapshots and clears circuit breakers."""
		692	+ dispatcher, instance_manager, *_ = _make_dispatcher()
		693	+ dispatcher._cb_manager = MagicMock()
		694	+ dispatcher._cb_manager.get_open_instance_ids.return_value = [1]
		695	+ instance_manager.refresh_instances = AsyncMock(return_value=True)
		696	+
		697	+ request = SchedulerRequest(
		698	+ request_type=SchedulerRequestType.REFRESH_INSTANCES,
		699	+ request_id="req-set-2",
		700	+ data={"event_type": EventType.SET.value, "instances": []},
		701	+ )
		702	+ response = await dispatcher.dispatch(request)
		703	+
		704	+ assert response.response_type == SchedulerResponseType.SUCCESS
		705	+ dispatcher._cb_manager.get_open_instance_ids.assert_called_once()
		706	+ dispatcher._cb_manager.clear_all.assert_called_once()
		707	+
670		708
671	class TestHandleAllocateOnlyEdgeCases:	709	class TestHandleAllocateOnlyEdgeCases:
672	@pytest.mark.asyncio	710	@pytest.mark.asyncio

Mtests/coordinator/test_http_server.py+54-1

	@@ -30,13 +30,19 @@ from motor.common.standby.standby_manager import StandbyRole, StandbyManager
30	from motor.coordinator.api_server.management_server import ManagementServer	30	from motor.coordinator.api_server.management_server import ManagementServer
31	from motor.coordinator.domain.instance_manager import InstanceIdConflictError	31	from motor.coordinator.domain.instance_manager import InstanceIdConflictError
32	from motor.coordinator.domain.probe import RoleHeartbeatResult	32	from motor.coordinator.domain.probe import RoleHeartbeatResult
33	-from motor.coordinator.api_server.inference_server import InferenceServer, _validate_anthropic_request	33	+from motor.coordinator.api_server.inference_server import (
		34	+ InferenceServer,
		35	+ _validate_anthropic_request,
		36	+ _validate_openai_request,
		37	+ _validate_positive_int_field,
		38	+)
34	from motor.coordinator.domain.request_manager import RequestManager	39	from motor.coordinator.domain.request_manager import RequestManager
35	from motor.config.coordinator import CoordinatorConfig, RateLimitConfig	40	from motor.config.coordinator import CoordinatorConfig, RateLimitConfig
36	from motor.coordinator.domain import InstanceReadiness	41	from motor.coordinator.domain import InstanceReadiness
37	from motor.common.http.key_encryption import encrypt_api_key, set_default_key_encryption_by_name	42	from motor.common.http.key_encryption import encrypt_api_key, set_default_key_encryption_by_name
38	from motor.common.resources import Endpoint, Instance, InsStatus, PDRole	43	from motor.common.resources import Endpoint, Instance, InsStatus, PDRole
39	from motor.coordinator.models.constants import OpenAIField	44	from motor.coordinator.models.constants import OpenAIField
		45	+from motor.coordinator.models.request import RequestType
40	from motor.coordinator.middleware.fastapi_middleware import (	46	from motor.coordinator.middleware.fastapi_middleware import (
41	SimpleRateLimitMiddleware,	47	SimpleRateLimitMiddleware,
42	RateLimitConfigHolder,	48	RateLimitConfigHolder,
	@@ -2192,6 +2198,53 @@ class TestValidateAnthropicRequest:
2192	)	2198	)
2193		2199
2194		2200
		2201	+class TestValidateOpenaiPositiveIntField:
		2202	+ """Unit tests for _validate_positive_int_field (max_tokens / max_completion_tokens)."""
		2203	+
		2204	+ def test_missing_field_is_untouched(self):
		2205	+ body = {"model": "m"}
		2206	+ _validate_positive_int_field(body, OpenAIField.MAX_TOKENS)
		2207	+ assert body == {"model": "m"}
		2208	+
		2209	+ def test_valid_positive_int_is_kept(self):
		2210	+ body = {"model": "m", "max_tokens": 128}
		2211	+ _validate_positive_int_field(body, OpenAIField.MAX_TOKENS)
		2212	+ assert body == {"model": "m", "max_tokens": 128}
		2213	+
		2214	+ @pytest.mark.parametrize("value", [0, -1, 1.5, "128", True, False])
		2215	+ def test_invalid_values_are_removed(self, value):
		2216	+ body = {"model": "m", "max_tokens": value}
		2217	+ _validate_positive_int_field(body, OpenAIField.MAX_TOKENS)
		2218	+ assert "max_tokens" not in body
		2219	+
		2220	+ def test_invalid_value_logs_warning(self, caplog):
		2221	+ body = {"model": "m", "max_tokens": 0}
		2222	+ _validate_positive_int_field(body, OpenAIField.MAX_TOKENS)
		2223	+ assert "Invalid max_tokens=0" in caplog.text
		2224	+
		2225	+ def test_openai_request_removes_invalid_max_tokens(self):
		2226	+ body = {
		2227	+ "model": "m",
		2228	+ "messages": [{"role": "user", "content": "hi"}],
		2229	+ "max_tokens": 0,
		2230	+ "max_completion_tokens": -1,
		2231	+ }
		2232	+ _validate_openai_request(body, RequestType.OPENAI)
		2233	+ assert "max_tokens" not in body
		2234	+ assert "max_completion_tokens" not in body
		2235	+
		2236	+ def test_openai_request_keeps_valid_max_tokens(self):
		2237	+ body = {
		2238	+ "model": "m",
		2239	+ "messages": [{"role": "user", "content": "hi"}],
		2240	+ "max_tokens": 128,
		2241	+ "max_completion_tokens": 64,
		2242	+ }
		2243	+ _validate_openai_request(body, RequestType.OPENAI)
		2244	+ assert body["max_tokens"] == 128
		2245	+ assert body["max_completion_tokens"] == 64
		2246	+
		2247	+
2195	class TestAnthropicEndpoints:	2248	class TestAnthropicEndpoints:
2196	"""Integration tests for Anthropic API endpoints."""	2249	"""Integration tests for Anthropic API endpoints."""
2197		2250

Mtests/examples/observability/test_ccae_motor_backend.py+2-0

	@@ -83,6 +83,7 @@ def test_motor_backend_falls_back_to_default_controller_port(ccae_mods, monkeypa
83		83
84	monkeypatch.setattr(base_mod, "Collector", MagicMock)	84	monkeypatch.setattr(base_mod, "Collector", MagicMock)
85	monkeypatch.setattr(backend_mod, "SafeHTTPSClient", FakeClient)	85	monkeypatch.setattr(backend_mod, "SafeHTTPSClient", FakeClient)
		86	+ monkeypatch.setattr(base_mod, "Log", MagicMock())
86	monkeypatch.setattr(backend_mod, "Log", MagicMock())	87	monkeypatch.setattr(backend_mod, "Log", MagicMock())
87	monkeypatch.setenv("POD_IP", "10.0.0.1")	88	monkeypatch.setenv("POD_IP", "10.0.0.1")
88		89
	@@ -110,6 +111,7 @@ def test_motor_backend_uses_explicit_controller_port(ccae_mods, monkeypatch: pyt
110		111
111	monkeypatch.setattr(base_mod, "Collector", MagicMock)	112	monkeypatch.setattr(base_mod, "Collector", MagicMock)
112	monkeypatch.setattr(backend_mod, "SafeHTTPSClient", FakeClient)	113	monkeypatch.setattr(backend_mod, "SafeHTTPSClient", FakeClient)
		114	+ monkeypatch.setattr(base_mod, "Log", MagicMock())
113	monkeypatch.setattr(backend_mod, "Log", MagicMock())	115	monkeypatch.setattr(backend_mod, "Log", MagicMock())
114	monkeypatch.setenv("POD_IP", "10.0.0.1")	116	monkeypatch.setenv("POD_IP", "10.0.0.1")
115		117