feat(pd): route Anthropic Messages and Responses API through dual dispatch

crates/protocols/src/responses.rs

@@ -2941,7 +2941,7 @@ pub struct ResponsesRequest {
     pub store: Option<bool>,
 
     /// Whether to stream the response
-    #[serde(default)]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub stream: Option<bool>,
 
     /// Temperature for sampling

e2e_test/router/test_pd_messages.py

@@ -0,0 +1,108 @@
+"""Anthropic Messages API tests for PD (Prefill-Decode) disaggregated routing.
+
+/v1/messages enters PD dual dispatch like chat completions: the HTTP PD
+router forwards the request to both legs with bootstrap fields injected,
+and the gRPC PD router runs the mode-parameterized Messages pipeline. Every
+create below exercises prefill/decode pair selection and dual dispatch.
+
+Backends:
+- "pd_http": HTTP mode (SGLang only - vLLM does not support HTTP)
+- "pd_grpc": gRPC mode (both SGLang and vLLM)
+
+Requirements:
+    - SGLang: sgl_kernel package
+    - vLLM: NIXL or Mooncake KV transfer support
+    - GPUs: num_prefill + num_decode (default: 2 GPUs for 1+1)
+
+Usage:
+    # SGLang (runs both HTTP and gRPC)
+    pytest e2e_test/router/test_pd_messages.py -v
+
+    # vLLM (runs gRPC only, HTTP skipped)
+    E2E_RUNTIME=vllm pytest e2e_test/router/test_pd_messages.py -v
+"""
+
+from __future__ import annotations
+
+import logging
+
+import anthropic
+import pytest
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.fixture
+def anthropic_client(setup_backend):
+    """Anthropic SDK client pointed at the gateway under test."""
+    _, _, _, gateway = setup_backend
+    client = anthropic.Anthropic(base_url=gateway.base_url, api_key="not-used")
+    yield client
+    client.close()
+
+
+class MessagesOverPD:
+    """Shared test bodies; subclasses pin the backend mode via markers."""
+
+    def test_non_streaming_message(self, model, anthropic_client):
+        """Basic message creation through PD dual dispatch."""
+        response = anthropic_client.messages.create(
+            model=model,
+            max_tokens=64,
+            messages=[{"role": "user", "content": "Say hello in one sentence."}],
+        )
+
+        assert response.id is not None
+        assert response.role == "assistant"
+        assert response.content is not None
+        assert len(response.content) > 0
+        assert response.content[0].type == "text"
+        assert len(response.content[0].text) > 0
+        assert response.usage is not None
+        assert response.usage.output_tokens > 0
+
+    def test_streaming_message(self, model, anthropic_client):
+        """Streaming events arrive and deltas concatenate to a full message."""
+        expected_event_types = {
+            "message_start",
+            "content_block_delta",
+            "message_stop",
+        }
+
+        with anthropic_client.messages.stream(
+            model=model,
+            max_tokens=64,
+            messages=[{"role": "user", "content": "Count from 1 to 3."}],
+        ) as stream:
+            event_types = set()
+            for event in stream:
+                event_types.add(event.type)
+            full_text = stream.get_final_text()
+
+        missing = expected_event_types - event_types
+        assert not missing, f"Missing expected event types: {missing}"
+        assert len(full_text) > 0
+
+
+@pytest.mark.skip(
+    reason="SGLang's /v1/messages does not carry PD bootstrap fields through "
+    "to the scheduler yet: the decode leg rejects with 'Disaggregated request "
+    "received without bootstrap room id'. Unskip when the engine forwards them."
+)
+@pytest.mark.engine("sglang")
+@pytest.mark.gpu(2)
+@pytest.mark.model("meta-llama/Llama-3.1-8B-Instruct")
+@pytest.mark.e2e
+@pytest.mark.skip_for_runtime("vllm", reason="vLLM does not support HTTP mode")
+@pytest.mark.parametrize("setup_backend", ["pd_http"], indirect=True)
+class TestPDMessagesHttp(MessagesOverPD):
+    """Messages API through the HTTP PD router's dual dispatch."""
+
+
+@pytest.mark.engine("sglang", "vllm")
+@pytest.mark.gpu(2)
+@pytest.mark.model("meta-llama/Llama-3.1-8B-Instruct")
+@pytest.mark.e2e
+@pytest.mark.parametrize("setup_backend", ["pd_grpc"], indirect=True)
+class TestPDMessagesGrpc(MessagesOverPD):
+    """Messages API through the gRPC PD pipeline."""

e2e_test/router/test_pd_responses.py

@@ -5,6 +5,7 @@
 bootstrap injection, and dual dispatch.
 
 Backends:
+- "pd_http": HTTP mode (SGLang only - vLLM does not support HTTP)
 - "pd_grpc": gRPC mode (both SGLang and vLLM)
 
 Requirements:
@@ -30,6 +31,55 @@
 logger = logging.getLogger(__name__)
 
 
+@pytest.mark.skip(
+    reason="SGLang's /v1/responses does not accept PD-disaggregated requests "
+    "yet (bootstrap fields are not carried through to the scheduler). Unskip "
+    "when the engine forwards them."
+)
+@pytest.mark.engine("sglang")
+@pytest.mark.gpu(2)
+@pytest.mark.model("meta-llama/Llama-3.1-8B-Instruct")
+@pytest.mark.e2e
+@pytest.mark.skip_for_runtime("vllm", reason="vLLM does not support HTTP mode")
+@pytest.mark.parametrize("setup_backend", ["pd_http"], indirect=True)
+class TestPDResponsesHttp:
+    """Responses API through the HTTP PD router's dual dispatch.
+
+    Creation and streaming only: in HTTP mode the gateway proxies to the
+    engine's own /v1/responses, so storage semantics (retrieval, chaining)
+    are the engine's and are not asserted here.
+    """
+
+    def test_basic_response_creation(self, model, api_client):
+        """Test basic response creation."""
+        resp = api_client.responses.create(model=model, input="What is 2+2?")
+
+        assert resp.id is not None
+        assert resp.error is None
+        assert resp.status == "completed"
+        assert len(resp.output_text) > 0
+        assert resp.usage is not None
+
+    def test_streaming_response(self, model, api_client):
+        """Test streaming response."""
+        resp = api_client.responses.create(
+            model=model, input="Count to 5", stream=True, max_output_tokens=50
+        )
+
+        events = list(resp)
+        created_events = [e for e in events if e.type == "response.created"]
+        assert len(created_events) > 0
+
+        delta_events = [e for e in events if e.type == "response.output_text.delta"]
+        assert len(delta_events) > 0
+        streamed_text = "".join(e.delta for e in delta_events)
+        assert len(streamed_text) > 0
+
+        completed_events = [e for e in events if e.type == "response.completed"]
+        assert len(completed_events) == 1
+        assert completed_events[0].response.status == "completed"
+
+
 @pytest.mark.engine("sglang", "vllm")
 @pytest.mark.gpu(2)
 @pytest.mark.model("meta-llama/Llama-3.1-8B-Instruct")
@@ -62,9 +112,12 @@ def test_streaming_response(self, model, api_client):
 
         delta_events = [e for e in events if e.type == "response.output_text.delta"]
         assert len(delta_events) > 0
+        streamed_text = "".join(e.delta for e in delta_events)
+        assert len(streamed_text) > 0
 
         completed_events = [e for e in events if e.type == "response.completed"]
         assert len(completed_events) == 1
+        assert completed_events[0].response.status == "completed"
 
     def test_previous_response_id_chaining(self, model, api_client):
         """Test chaining responses using previous_response_id."""