fix: preserve Claude Code Messages transport fidelity (vllm-project#161)

## Summary

- preserve the original Messages query string and open-ended
Anthropic/Claude Code request headers across every server-side gateway
tool round
- preserve upstream Messages status, body, request IDs, retry guidance,
and rate-limit metadata while keeping local connection failures in valid
Anthropic JSON envelopes
- forward documented upstream SSE error events and terminate without a
synthetic successful message stop
- verify multi-block system attribution and a valid four-breakpoint,
correctly ordered cache-control layout survive transparent proxying and
native gateway tool loops
- run the real Claude Code CLI through agentic-server against recorded
vLLM streaming responses and a local web-search replay service
- pin Claude Code 2.1.218, PyYAML 6.0.3, Node 24, and immutable action
SHAs in the dedicated GitHub Actions check

The E2E job uses localhost services and a placeholder token. It requires
no Anthropic credential, GPU, model download, or paid inference. This
change does not add conversation persistence or other server-side client
state; the request-scoped transport context is reused only for the
gateway-owned tool loop.

For non-streaming requests, response metadata comes from the terminal
upstream round. For streaming requests, HTTP response metadata comes
from the initial upstream response because later rounds occur after the
client response has started.

The visible reasoning double-render remains a vLLM-side issue tracked
separately; this PR closes the agentic-api transport and error-fidelity
gaps found by running Claude Code.

Related to vllm-project#116.

## Test Plan

- cargo test
- cargo clippy --all-targets -- -D warnings
- cargo fmt --all -- --check
- pre-commit 4.4.0: pre-commit run --all-files
- pinned replay-server unit tests: 7 passed
- pinned Claude Code smoke test: capture valid: messages=2 transports=2
searches=1
- bash -n scripts/claude-code-smoke.sh

---------

Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>

Author: franciscojavierarceo

.github/workflows/claude-code-e2e.yml

@@ -0,0 +1,66 @@
+name: Claude Code E2E
+
+run-name: Claude Code through Messages gateway
+
+on:
+  pull_request:
+  merge_group:
+  push:
+    branches:
+      - main
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref == 'refs/heads/main' && github.run_id || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  claude-code-e2e:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+
+      - name: Set up Python
+        uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
+        with:
+          python-version: '3.12'
+
+      - name: Set up Node.js
+        uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0
+        with:
+          node-version: '24'
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@e97e2d8cc328f1b50210efc529dca0028893a2d9 # v1
+        with:
+          toolchain: stable
+
+      - name: Cache cargo registry and build
+        uses: actions/cache@668228422ae6a00e4ad889ee87cd7109ec5666a7 # v4
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            target
+          key: cargo-${{ runner.os }}-${{ hashFiles('Cargo.lock') }}
+          restore-keys: |
+            cargo-${{ runner.os }}-
+
+      - name: Install pinned test dependencies
+        run: |
+          python -m pip install 'PyYAML==6.0.3'
+          npm install --global '@anthropic-ai/claude-code@2.1.218'
+          test "$(claude --version)" = '2.1.218 (Claude Code)'
+
+      - name: Test replay server
+        run: python -m unittest scripts/test_claude_code_replay_server.py -v
+
+      - name: Build agentic-server
+        run: cargo build -p agentic-server
+
+      - name: Run Claude Code through agentic-server
+        run: bash scripts/claude-code-smoke.sh

crates/agentic-server-core/src/executor/error.rs

@@ -29,9 +29,17 @@ pub enum ExecutorError {
         source: StorageError,
     },
 
-    /// The LLM backend returned a non-2xx status or was unreachable.
+    /// The LLM backend returned a non-2xx HTTP response.
     #[error("LLM request failed ({status}): {body}")]
-    LLMRequest { status: StatusCode, body: String },
+    LLMRequest {
+        status: StatusCode,
+        body: String,
+        headers: http::HeaderMap,
+    },
+
+    /// The LLM backend could not be reached or timed out before responding.
+    #[error("{message}")]
+    LLMTransport { status: StatusCode, message: &'static str },
 
     /// A network error occurred reading from the LLM response stream.
     ///
@@ -101,7 +109,7 @@ impl ExecutorError {
     pub fn http_status(&self) -> StatusCode {
         match self.client_visible_error() {
             Self::Storage(e) if e.is_not_found() => StatusCode::NOT_FOUND,
-            Self::LLMRequest { status, .. } => *status,
+            Self::LLMRequest { status, .. } | Self::LLMTransport { status, .. } => *status,
             Self::ConversationLocked { .. }
             | Self::Tool(ToolError::Config(_))
             | Self::InvalidRequest(_)
@@ -122,7 +130,7 @@ impl ExecutorError {
             | Self::ParseError(_)
             | Self::JsonError(_) => "invalid_request_error",
             Self::Storage(e) if e.is_not_found() => "not_found",
-            Self::LLMRequest { .. } | Self::CompactionFailed { .. } => "upstream_error",
+            Self::LLMRequest { .. } | Self::LLMTransport { .. } | Self::CompactionFailed { .. } => "upstream_error",
             Self::Tool(ToolError::Execution(_)) => "tool_error",
             _ => "server_error",
         }

crates/agentic-server-core/src/executor/inference.rs

@@ -10,6 +10,7 @@ use async_stream::stream;
 use futures::{Stream, StreamExt};
 
 use crate::executor::error::{ExecutorError, ExecutorResult};
+use crate::proxy::processed_response_headers;
 
 /// SSE stream of raw lines sent to the client (`data: …\n\n` per event).
 pub type BoxStream = std::pin::Pin<Box<dyn Stream<Item = String> + Send>>;
@@ -55,33 +56,40 @@ fn drain_complete_utf8_lines(buffer: &mut Vec<u8>) -> Vec<String> {
 ///
 /// Shared by both the blocking path (caller reads `.text()`) and the streaming
 /// path (caller reads `.bytes_stream()`). Maps connect/timeout failures and
-/// non-2xx status codes to [`ExecutorError::LLMRequest`].
+/// non-2xx status codes to [`ExecutorError::LLMRequest`] and connection
+/// failures to [`ExecutorError::LLMTransport`].
 pub(super) async fn send_request(
     client: &reqwest::Client,
     url: &str,
     body: String,
     auth: Option<&str>,
+    forwarded_headers: Option<&reqwest::header::HeaderMap>,
 ) -> ExecutorResult<reqwest::Response> {
-    let mut req = client.post(url).header("Content-Type", "application/json").body(body);
+    let mut headers = forwarded_headers.cloned().unwrap_or_default();
+    headers
+        .entry(reqwest::header::CONTENT_TYPE)
+        .or_insert(reqwest::header::HeaderValue::from_static("application/json"));
+    let mut req = client.post(url).headers(headers).body(body);
     if let Some(key) = auth {
         req = req.bearer_auth(key);
     }
 
-    let resp = req.send().await.map_err(|e| ExecutorError::LLMRequest {
+    let resp = req.send().await.map_err(|e| ExecutorError::LLMTransport {
         status: if e.is_timeout() {
             http::StatusCode::GATEWAY_TIMEOUT
         } else {
             http::StatusCode::BAD_GATEWAY
         },
-        body: if e.is_timeout() {
-            "upstream timeout".into()
+        message: if e.is_timeout() {
+            "LLM timeout"
         } else {
-            "upstream unavailable".into()
+            "LLM unavailable"
         },
     })?;
 
     if !resp.status().is_success() {
         let status = resp.status().as_u16();
+        let headers = processed_response_headers(resp.headers());
         // Log and discard any error reading the error body — the status code
         // is the primary signal; an empty body is acceptable here.
         let body = resp
@@ -92,6 +100,7 @@ pub(super) async fn send_request(
         return Err(ExecutorError::LLMRequest {
             status: http::StatusCode::from_u16(status).unwrap_or(http::StatusCode::INTERNAL_SERVER_ERROR),
             body,
+            headers,
         });
     }
 
@@ -107,19 +116,32 @@ pub(super) async fn fetch_response_json(
     client: &reqwest::Client,
     auth: Option<&str>,
 ) -> ExecutorResult<String> {
-    let resp = send_request(client, url, upstream_json, auth).await?;
+    let resp = send_request(client, url, upstream_json, auth, None).await?;
     // Preserve the reqwest::Error as the typed source (NetworkError).
     resp.text().await.map_err(ExecutorError::NetworkError)
 }
 
+/// Makes a non-streaming HTTP POST with caller-supplied upstream headers.
+pub(super) async fn fetch_response_json_with_headers(
+    upstream_json: String,
+    url: &str,
+    client: &reqwest::Client,
+    headers: &reqwest::header::HeaderMap,
+) -> ExecutorResult<(String, http::HeaderMap)> {
+    let resp = send_request(client, url, upstream_json, None, Some(headers)).await?;
+    let response_headers = processed_response_headers(resp.headers());
+    let body = resp.text().await.map_err(ExecutorError::NetworkError)?;
+    Ok((body, response_headers))
+}
+
 /// Step 2 — Call the LLM inference backend; yields raw SSE lines (`data: …`).
 ///
 /// Always requests `stream=true` upstream. Stops on `[DONE]`.
 ///
 /// # Errors
 /// Each stream item is `Result<String, ExecutorError>`. The stream yields `Err` on:
-/// - [`ExecutorError::LLMRequest`] — connect timeout (504), connection failure (502),
-///   or non-2xx HTTP status from the backend
+/// - [`ExecutorError::LLMTransport`] — connect timeout (504) or connection failure (502)
+/// - [`ExecutorError::LLMRequest`] — non-2xx HTTP status from the backend
 /// - [`ExecutorError::NetworkError`] — network failure while reading the response body
 pub fn call_inference(
     upstream_json: String,
@@ -129,11 +151,24 @@ pub fn call_inference(
     chunk_timeout: Duration,
 ) -> impl Stream<Item = Result<String, ExecutorError>> + Send + 'static {
     stream! {
-        let resp = match send_request(&client, &url, upstream_json, auth.as_deref()).await {
+        let resp = match send_request(&client, &url, upstream_json, auth.as_deref(), None).await {
             Ok(r) => r,
             Err(e) => { yield Err(e); return; }
         };
 
+        let mut lines = Box::pin(response_lines(resp, chunk_timeout));
+        while let Some(line) = lines.next().await {
+            yield line;
+        }
+    }
+}
+
+/// Convert a successful upstream response body into normalized SSE data lines.
+pub(super) fn response_lines(
+    resp: reqwest::Response,
+    chunk_timeout: Duration,
+) -> impl Stream<Item = Result<String, ExecutorError>> + Send + 'static {
+    stream! {
         let mut bytes = resp.bytes_stream();
         let mut buf = Vec::with_capacity(8192);
 

crates/agentic-server-core/src/executor/messages_loop.rs

@@ -1,9 +1,8 @@
 //! Messages-native gateway tool loop.
 //!
 //! Runs the server-side gateway-tool loop for `/v1/messages` **natively**: the
-//! client's Anthropic request is forwarded to vLLM `/v1/messages` while
-//! preserving Anthropic fields except for native server-tool declarations that
-//! must be normalized to the function-tool shape vLLM accepts. The assistant turn is
+//! client's Anthropic request is forwarded to vLLM `/v1/messages` essentially
+//! untouched (preserving every Anthropic field), the assistant turn is
 //! inspected, any gateway-owned `tool_use` is executed server-side and hidden,
 //! the loop appends the `tool_result` and re-POSTs, until the model stops asking
 //! for a gateway tool. Only the final assistant message reaches the client.
@@ -19,7 +18,7 @@ use futures::future::join_all;
 use serde_json::{Value, json};
 
 use crate::executor::error::{ExecutorError, ExecutorResult};
-use crate::executor::inference::fetch_response_json;
+use crate::executor::inference::fetch_response_json_with_headers;
 use crate::executor::messages_request::{normalize_native_web_search, web_search_budget_exhausted_result};
 use crate::executor::request::ExecutionContext;
 use crate::tool::ToolRegistry;
@@ -37,6 +36,41 @@ pub(super) const MAX_GATEWAY_TOOL_ROUNDS: usize = 10;
 /// the streaming loop; matches the Responses loop's `gateway::GATEWAY_TOOL_TIMEOUT`.
 pub(super) const GATEWAY_TOOL_TIMEOUT: Duration = Duration::from_secs(60);
 
+/// Per-request transport data reused for every upstream Messages round.
+#[derive(Clone, Debug)]
+pub struct MessagesUpstream {
+    url: String,
+    headers: reqwest::header::HeaderMap,
+}
+
+impl MessagesUpstream {
+    #[must_use]
+    pub fn new(base_url: &str, query: Option<&str>, headers: reqwest::header::HeaderMap) -> Self {
+        let mut url = format!("{}/v1/messages", base_url.trim_end_matches('/'));
+        if let Some(query) = query.filter(|query| !query.is_empty()) {
+            url.push('?');
+            url.push_str(query);
+        }
+        Self { url, headers }
+    }
+
+    pub(super) fn url(&self) -> &str {
+        &self.url
+    }
+
+    pub(super) fn headers(&self) -> &reqwest::header::HeaderMap {
+        &self.headers
+    }
+}
+
+/// A Messages loop result paired with safe metadata from the relevant upstream response.
+pub struct MessagesResponse<T> {
+    /// The completed message or client-facing stream.
+    pub body: T,
+    /// Safe metadata retained from the terminal response, or the initial response for streaming.
+    pub headers: http::HeaderMap,
+}
+
 /// The `tool_result` block for one executed gateway call, fed back next round.
 /// (The model's own `tool_use` block is carried forward via the preserved
 /// assistant content, not reconstructed here — see `append_round_to_history`.)
@@ -58,23 +92,26 @@ pub async fn run_messages_loop(
     mut request: Value,
     registry: &ToolRegistry,
     exec_ctx: &ExecutionContext,
-    auth: Option<&str>,
-) -> ExecutorResult<Value> {
-    let url = format!("{}/v1/messages", exec_ctx.llm_base_url);
+    upstream: &MessagesUpstream,
+) -> ExecutorResult<MessagesResponse<Value>> {
     let mut web_search_budget = normalize_native_web_search(&mut request)?;
     // The loop drives turns itself; force non-streaming upstream regardless of
     // what the client asked (the handler routes streaming elsewhere).
     request["stream"] = Value::Bool(false);
 
     for _round in 0..MAX_GATEWAY_TOOL_ROUNDS {
         let body = serialize_to_string(&request).map_err(ExecutorError::JsonError)?;
-        let resp_text = fetch_response_json(body, &url, &exec_ctx.client, auth).await?;
+        let (resp_text, response_headers) =
+            fetch_response_json_with_headers(body, &upstream.url, &exec_ctx.client, &upstream.headers).await?;
         let message: Value = deserialize_from_str(&resp_text).map_err(ExecutorError::JsonError)?;
 
         // Any error body from upstream is surfaced verbatim (handler maps it to
         // the Anthropic error envelope).
         if message.get("type").and_then(Value::as_str) == Some("error") {
-            return Ok(message);
+            return Ok(MessagesResponse {
+                body: message,
+                headers: response_headers,
+            });
         }
 
         let content = message.get("content").and_then(Value::as_array);
@@ -84,7 +121,10 @@ pub async fn run_messages_loop(
         // client should see. A client-owned tool_use means we cannot continue
         // the loop server-side — return the turn to the client (edge E7).
         let Some(content) = content else {
-            return Ok(message);
+            return Ok(MessagesResponse {
+                body: message,
+                headers: response_headers,
+            });
         };
         let gateway_map = &exec_ctx.messages_gateway_tools;
         let mut gateway_calls: Vec<Value> = Vec::new();
@@ -105,15 +145,21 @@ pub async fn run_messages_loop(
         // must run it) — but the gateway tool_use, if any, must still be hidden
         // (F5): strip gateway blocks from the client-facing content.
         if gateway_calls.is_empty() || stop_reason != Some("tool_use") {
-            return Ok(message);
+            return Ok(MessagesResponse {
+                body: message,
+                headers: response_headers,
+            });
         }
         if has_client_tool_use {
             // Strip the gateway tool_use from the client-facing content (compute
             // before mutating to end the immutable borrow of `message`).
             let stripped = tool_seam::strip_gateway_tool_use(content, gateway_map);
             let mut message = message;
             message["content"] = Value::Array(stripped);
-            return Ok(message);
+            return Ok(MessagesResponse {
+                body: message,
+                headers: response_headers,
+            });
         }
 
         // Pure gateway-tool round: execute the calls, then feed the model's FULL
@@ -129,13 +175,16 @@ pub async fn run_messages_loop(
     // last message. (Open Q1: a dedicated pause_turn signal could go here.)
     // Reaching here means every round emitted a gateway tool_use; surface a
     // minimal terminal so the client isn't left hanging.
-    Ok(json!({
-        "type": "error",
-        "error": {
-            "type": "api_error",
-            "message": format!("gateway tool loop exceeded {MAX_GATEWAY_TOOL_ROUNDS} rounds")
-        }
-    }))
+    Ok(MessagesResponse {
+        body: json!({
+            "type": "error",
+            "error": {
+                "type": "api_error",
+                "message": format!("gateway tool loop exceeded {MAX_GATEWAY_TOOL_ROUNDS} rounds")
+            }
+        }),
+        headers: http::HeaderMap::new(),
+    })
 }
 
 /// Execute the gateway-owned `tool_use` blocks concurrently, each bounded by the