[sync] vLLM 2026-07-12 — 5f8e73cb8..4c81772e8

.sync/vllm-sha

@@ -1 +1 @@
-83762b77b07e97c77f986e4bf5a9474952e47bb3
+4c81772e8bdf9e37ab2b8adfd80e4308b618fa92

aphrodite/compilation/passes/fusion/allreduce_rms_fusion.py

@@ -567,7 +567,14 @@ def replacement(
             )
             return allreduce[1], allreduce[2]
 
-        pm.register_replacement(pattern, replacement, self.get_inputs(), pm.fwd_only, pm_pass)
+        pm.register_replacement(
+            pattern,
+            replacement,
+            self.get_inputs(),
+            pm.fwd_only,
+            pm_pass,
+            extra_check=_norm_input_weight_dtype_match,
+        )
 
         first_return_only = lambda fn: lambda a, b, c: fn(a, b, c)[0]
 
@@ -722,7 +729,14 @@ def replacement(
             # quant_out, rms_norm_residual
             return allreduce[4], allreduce[2]
 
-        pm.register_replacement(pattern, replacement, self.get_inputs(), pm.fwd_only, pm_pass)
+        pm.register_replacement(
+            pattern,
+            replacement,
+            self.get_inputs(),
+            pm.fwd_only,
+            pm_pass,
+            extra_check=_norm_input_weight_dtype_match,
+        )
 
 
 class AllReduceFusedRMSNormStaticQuantNVFP4Pattern(BasePattern):
@@ -899,7 +913,14 @@ def replacement(
             # quant_out, rms_norm_residual, output_scale
             return allreduce[4], allreduce[2], allreduce[5]
 
-        pm.register_replacement(pattern, replacement, self.get_inputs(), pm.fwd_only, pm_pass)
+        pm.register_replacement(
+            pattern,
+            replacement,
+            self.get_inputs(),
+            pm.fwd_only,
+            pm_pass,
+            extra_check=_norm_input_weight_dtype_match,
+        )
 
 
 class AllReduceFusionPass(AphroditePatternMatcherPass):

aphrodite/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py

@@ -1184,6 +1184,7 @@ def reset_cache(self) -> None:
         for status in self._req_status.values():
             for group_state in status.group_states:
                 group_state.next_stored_block_idx = 0
+            status.transfer_jobs.clear()
 
         # Discard jobs and save job_counter to be able to discard worker responses
         self._stale_job_threshold = self._job_counter

aphrodite/entrypoints/anthropic/protocol.py

@@ -137,6 +137,10 @@ class AnthropicMessagesRequest(BaseModel):
         default=None,
         description="KVTransfer parameters used for disaggregated serving.",
     )
+    ec_transfer_params: dict[str, Any] | None = Field(
+        default=None,
+        description="ECTransfer parameters used for encoder-cache disaggregated serving.",
+    )
     chat_template_kwargs: dict[str, Any] | None = Field(
         default=None,
         description=(
@@ -208,6 +212,7 @@ class AnthropicMessagesResponse(BaseModel):
 
     # Aphrodite-specific fields that are not in Anthropic spec
     kv_transfer_params: dict[str, Any] | None = Field(default=None, description="KVTransfer parameters.")
+    ec_transfer_params: dict[str, Any] | None = Field(default=None, description="ECTransfer parameters.")
 
     def model_post_init(self, __context):
         if not self.id:

aphrodite/entrypoints/anthropic/serving.py

@@ -484,6 +484,7 @@ def _build_base_request(
             top_p=anthropic_request.top_p,
             top_k=anthropic_request.top_k,
             kv_transfer_params=anthropic_request.kv_transfer_params,
+            ec_transfer_params=anthropic_request.ec_transfer_params,
             chat_template_kwargs=anthropic_request.chat_template_kwargs,
         )
 
@@ -621,6 +622,7 @@ def messages_full_converter(
                 generator.usage,
             ),
             kv_transfer_params=generator.kv_transfer_params,
+            ec_transfer_params=generator.ec_transfer_params,
         )
         choice = generator.choices[0]
         if choice.finish_reason == "stop":

aphrodite/entrypoints/openai/chat_completion/protocol.py

@@ -160,6 +160,7 @@ class ChatCompletionResponse(OpenAIBaseModel):
     # ``return_prompt_text=True`` on the request).
     prompt_text: str | None = None
     kv_transfer_params: dict[str, Any] | None = Field(default=None, description="KVTransfer parameters.")
+    ec_transfer_params: dict[str, Any] | None = Field(default=None, description="ECTransfer parameters.")
     metrics: PerRequestTimingMetrics | None = None
 
 
@@ -484,6 +485,11 @@ class ChatCompletionRequest(OpenAIBaseModel):
         description="KVTransfer parameters used for disaggregated serving.",
     )
 
+    ec_transfer_params: dict[str, Any] | None = Field(
+        default=None,
+        description="ECTransfer parameters used for encoder-cache disaggregated serving.",
+    )
+
     aphrodite_xargs: dict[str, str | int | float | list[str | int | float]] | None = Field(
         default=None,
         validation_alias=AliasChoices("aphrodite_xargs", "aphrodite_xargs"),
@@ -701,6 +707,9 @@ def to_sampling_params(
         if self.kv_transfer_params:
             # Pass in kv_transfer_params via extra_args
             extra_args["kv_transfer_params"] = self.kv_transfer_params
+        if self.ec_transfer_params:
+            # Pass in ec_transfer_params via extra_args
+            extra_args["ec_transfer_params"] = self.ec_transfer_params
         return SamplingParams.from_optional(
             n=self.n,
             presence_penalty=self.presence_penalty,

aphrodite/entrypoints/openai/chat_completion/serving.py

@@ -973,6 +973,7 @@ async def chat_completion_full_generator(
             prompt_token_ids=(final_res.prompt_token_ids if request.return_token_ids else None),
             prompt_text=prompt_text,
             kv_transfer_params=final_res.kv_transfer_params,
+            ec_transfer_params=final_res.ec_transfer_params,
             metrics=per_request_metrics,
         )
 

aphrodite/entrypoints/openai/completion/protocol.py

@@ -230,6 +230,11 @@ class CompletionRequest(OpenAIBaseModel):
         description="KVTransfer parameters used for disaggregated serving.",
     )
 
+    ec_transfer_params: dict[str, Any] | None = Field(
+        default=None,
+        description="ECTransfer parameters used for encoder-cache disaggregated serving.",
+    )
+
     aphrodite_xargs: dict[str, str | int | float] | None = Field(
         default=None,
         validation_alias=AliasChoices("aphrodite_xargs", "aphrodite_xargs"),
@@ -375,6 +380,9 @@ def to_sampling_params(
         if self.kv_transfer_params:
             # Pass in kv_transfer_params via extra_args
             extra_args["kv_transfer_params"] = self.kv_transfer_params
+        if self.ec_transfer_params:
+            # Pass in ec_transfer_params via extra_args
+            extra_args["ec_transfer_params"] = self.ec_transfer_params
         return SamplingParams.from_optional(
             n=self.n,
             presence_penalty=self.presence_penalty,
@@ -636,6 +644,7 @@ class CompletionResponse(OpenAIBaseModel):
 
     # Aphrodite-specific fields that are not in OpenAI spec
     kv_transfer_params: dict[str, Any] | None = Field(default=None, description="KVTransfer parameters.")
+    ec_transfer_params: dict[str, Any] | None = Field(default=None, description="ECTransfer parameters.")
     metrics: PerRequestTimingMetrics | None = None
 
 

aphrodite/entrypoints/openai/completion/serving.py

@@ -479,6 +479,7 @@ def request_output_to_completion_response(
         num_prompt_tokens = 0
         num_generated_tokens = 0
         kv_transfer_params = None
+        ec_transfer_params = None
         last_final_res = None
         for final_res in final_res_batch:
             last_final_res = final_res
@@ -584,6 +585,7 @@ def request_output_to_completion_response(
 
         if final_res_batch:
             kv_transfer_params = final_res_batch[0].kv_transfer_params
+            ec_transfer_params = final_res_batch[0].ec_transfer_params
         return CompletionResponse(
             id=request_id,
             created=created_time,
@@ -592,6 +594,7 @@ def request_output_to_completion_response(
             usage=usage,
             system_fingerprint=self.system_fingerprint,
             kv_transfer_params=kv_transfer_params,
+            ec_transfer_params=ec_transfer_params,
             metrics=per_request_metrics,
         )
 

aphrodite/entrypoints/openai/responses/context.py

@@ -195,6 +195,7 @@ def __init__(
 
         self.input_messages: list[ResponseRawMessageAndToken] = []
         self.kv_transfer_params: dict[str, Any] | None = None
+        self.ec_transfer_params: dict[str, Any] | None = None
 
     def append_output(self, output) -> None:
         self.last_output = output
@@ -205,6 +206,8 @@ def append_output(self, output) -> None:
         self.num_output_tokens += len(output.outputs[0].token_ids or [])
         if output.kv_transfer_params is not None:
             self.kv_transfer_params = output.kv_transfer_params
+        if output.ec_transfer_params is not None:
+            self.ec_transfer_params = output.ec_transfer_params
 
         # Accumulate text, token_ids, and logprobs for streaming mode
         delta_output = output.outputs[0]
@@ -323,6 +326,7 @@ def __init__(
         self.output_messages: list[ResponseRawMessageAndToken] = []
         self._accumulated_token_ids: list[int] = []
         self.kv_transfer_params: dict[str, Any] | None = None
+        self.ec_transfer_params: dict[str, Any] | None = None
 
     def append_output(self, output: RequestOutput) -> None:
         self.num_prompt_tokens = len(output.prompt_token_ids or [])
@@ -331,6 +335,9 @@ def append_output(self, output: RequestOutput) -> None:
         if output.kv_transfer_params is not None:
             self.kv_transfer_params = output.kv_transfer_params
 
+        if output.ec_transfer_params is not None:
+            self.ec_transfer_params = output.ec_transfer_params
+
         completion = output.outputs[0]
         self.finish_reason = completion.finish_reason
 
@@ -610,6 +617,7 @@ def __init__(
         self.is_first_turn = True
         self.first_tok_of_message = True
         self.kv_transfer_params: dict[str, Any] | None = None
+        self.ec_transfer_params: dict[str, Any] | None = None
 
     def append_output(self, output: RequestOutput) -> None:
         if self.first_tok_of_message:
@@ -625,6 +633,8 @@ def append_output(self, output: RequestOutput) -> None:
         self._update_decode_token_usage(output)
         if output.kv_transfer_params is not None:
             self.kv_transfer_params = output.kv_transfer_params
+        if output.ec_transfer_params is not None:
+            self.ec_transfer_params = output.ec_transfer_params
 
         if output.finished:
             self.finish_reason = output.outputs[0].finish_reason

aphrodite/entrypoints/openai/responses/protocol.py

@@ -280,6 +280,10 @@ class ResponsesRequest(OpenAIBaseModel):
         default=None,
         description="KVTransfer parameters used for disaggregated serving.",
     )
+    ec_transfer_params: dict[str, Any] | None = Field(
+        default=None,
+        description="ECTransfer parameters used for encoder-cache disaggregated serving.",
+    )
     chat_template_kwargs: dict[str, Any] | None = Field(
         default=None,
         description=(
@@ -394,6 +398,8 @@ def to_sampling_params(
         extra_args: dict[str, Any] = self.aphrodite_xargs if self.aphrodite_xargs else {}
         if self.kv_transfer_params:
             extra_args["kv_transfer_params"] = self.kv_transfer_params
+        if self.ec_transfer_params:
+            extra_args["ec_transfer_params"] = self.ec_transfer_params
 
         return SamplingParams.from_optional(
             temperature=temperature,
@@ -636,6 +642,7 @@ class ResponsesResponse(OpenAIBaseModel):
 
     # Aphrodite-specific fields that are not in OpenAI spec
     kv_transfer_params: dict[str, Any] | None = Field(default=None, description="KVTransfer parameters.")
+    ec_transfer_params: dict[str, Any] | None = Field(default=None, description="ECTransfer parameters.")
 
     # --8<-- [start:responses-response-extra-params]
     # These are populated when enable_response_messages is set to True
@@ -677,6 +684,7 @@ def from_request(
         input_messages: ResponseInputOutputMessage | None = None,
         output_messages: ResponseInputOutputMessage | None = None,
         kv_transfer_params: dict[str, Any] | None = None,
+        ec_transfer_params: dict[str, Any] | None = None,
     ) -> "ResponsesResponse":
         incomplete_details: IncompleteDetails | None = None
         if status == "incomplete":
@@ -715,6 +723,7 @@ def from_request(
             user=request.user,
             usage=usage,
             kv_transfer_params=kv_transfer_params,
+            ec_transfer_params=ec_transfer_params,
         )
 
 

aphrodite/entrypoints/openai/responses/serving.py

@@ -867,6 +867,7 @@ async def responses_full_generator(
             status=status,
             usage=usage,
             kv_transfer_params=context.kv_transfer_params,
+            ec_transfer_params=context.ec_transfer_params,
         )
 
         if request.store:

aphrodite/entrypoints/scale_out/token_in_token_out/protocol.py

@@ -132,6 +132,10 @@ def validate_token_ids(cls, v: list[int]) -> list[int]:
         default=None,
         description="KVTransfer parameters used for disaggregated serving.",
     )
+    ec_transfer_params: dict[str, Any] | None = Field(
+        default=None,
+        description="ECTransfer parameters used for encoder-cache disaggregated serving.",
+    )
 
     # Tracks which keys the caller explicitly set inside ``sampling_params``
     # when the request was parsed from a JSON body. Lets the server tell
@@ -238,6 +242,10 @@ class GenerateResponse(BaseModel):
         default=None,
         description="KVTransfer parameters used for disaggregated serving.",
     )
+    ec_transfer_params: dict[str, Any] | None = Field(
+        default=None,
+        description="ECTransfer parameters used for encoder-cache disaggregated serving.",
+    )
 
 
 ####### Derender (postprocessing) #######

aphrodite/entrypoints/scale_out/token_in_token_out/serving.py

@@ -311,6 +311,7 @@ async def serve_tokens_full_generator(
             usage=usage,
             prompt_logprobs=clamp_prompt_logprobs(final_res.prompt_logprobs),
             kv_transfer_params=final_res.kv_transfer_params,
+            ec_transfer_params=final_res.ec_transfer_params,
         )
 
         # Log complete response if output logging is enabled

aphrodite/envs.py

@@ -207,6 +207,8 @@
     APHRODITE_DISABLE_REQUEST_ID_RANDOMIZATION: bool = False
     APHRODITE_NIXL_SIDE_CHANNEL_HOST: str = "localhost"
     APHRODITE_NIXL_SIDE_CHANNEL_PORT: int = 5600
+    APHRODITE_EC_SIDE_CHANNEL_HOST: str = "localhost"
+    APHRODITE_EC_SIDE_CHANNEL_PORT: int = 5601
     APHRODITE_MOONCAKE_BOOTSTRAP_PORT: int = 8998
     APHRODITE_MOONCAKE_STORE_TIER_LOG: bool = False
     APHRODITE_MOONCAKE_LOAD_RECV_THREADS: int = 1
@@ -1383,6 +1385,12 @@ def _resolve_rust_frontend_path() -> str | None:
     "APHRODITE_NIXL_SIDE_CHANNEL_HOST": lambda: os.getenv("APHRODITE_NIXL_SIDE_CHANNEL_HOST", "localhost"),
     # Port used for NIXL handshake between remote agents.
     "APHRODITE_NIXL_SIDE_CHANNEL_PORT": lambda: int(os.getenv("APHRODITE_NIXL_SIDE_CHANNEL_PORT", "5600")),
+    # IP address used for the EC connector's ZMQ side channel
+    # (producer ROUTER bind, consumer DEALER dial).
+    "APHRODITE_EC_SIDE_CHANNEL_HOST": lambda: os.getenv("APHRODITE_EC_SIDE_CHANNEL_HOST", "localhost"),
+    # Port for the EC connector's ZMQ side channel; advertised to peers
+    # via `ec_transfer_params.peer_port` on the producer's response.
+    "APHRODITE_EC_SIDE_CHANNEL_PORT": lambda: int(os.getenv("APHRODITE_EC_SIDE_CHANNEL_PORT", "5601")),
     # Port used for Mooncake handshake between remote agents.
     "APHRODITE_MOONCAKE_BOOTSTRAP_PORT": lambda: int(os.getenv("APHRODITE_MOONCAKE_BOOTSTRAP_PORT", "8998")),
     # Log per-batch memory/disk tier breakdown on external GETs.

aphrodite/outputs.py

@@ -102,6 +102,7 @@ class RequestOutput:
                                   None if decoder-only.
         num_cached_tokens: The number of tokens with prefix cache hit.
         kv_transfer_params: The params for remote K/V transfer.
+        ec_transfer_params: The params for remote encoder-cache transfer.
     """
 
     def __init__(
@@ -119,6 +120,7 @@ def __init__(
         num_cached_tokens: int | None = None,
         *,
         kv_transfer_params: dict[str, Any] | None = None,
+        ec_transfer_params: dict[str, Any] | None = None,
         # Forward compatibility, code that uses args added in new release can
         # still run with older versions of Aphrodite without breaking.
         **kwargs: Any,
@@ -137,12 +139,14 @@ def __init__(
         self.encoder_prompt_token_ids = encoder_prompt_token_ids
         self.num_cached_tokens = num_cached_tokens
         self.kv_transfer_params = kv_transfer_params
+        self.ec_transfer_params = ec_transfer_params
 
     def add(self, next_output: "RequestOutput", aggregate: bool) -> None:
         """Merge subsequent RequestOutput into this one"""
 
         self.finished |= next_output.finished
         self.kv_transfer_params = next_output.kv_transfer_params
+        self.ec_transfer_params = next_output.ec_transfer_params
 
         for next_completion in next_output.outputs:
             for i, completion in enumerate(self.outputs):

aphrodite/v1/core/sched/interface.py

@@ -9,6 +9,7 @@
 
 if TYPE_CHECKING:
     from aphrodite.config import AphroditeConfig
+    from aphrodite.distributed.ec_transfer.ec_connector.base import ECConnectorBase
     from aphrodite.distributed.kv_transfer.kv_connector.v1 import KVConnectorBase_V1
     from aphrodite.v1.core.sched.output import GrammarOutput, SchedulerOutput
     from aphrodite.v1.engine import EngineCoreOutputs
@@ -244,3 +245,6 @@ def shutdown(self) -> None:
 
     def get_kv_connector(self) -> "KVConnectorBase_V1 | None":
         return None
+
+    def get_ec_connector(self) -> "ECConnectorBase | None":
+        return None