fix: preserve Claude Code Messages transport fidelity

.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

@@ -18,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::request::ExecutionContext;
 use crate::tool::ToolRegistry;
 use crate::types::messages::tool_seam;
@@ -35,6 +35,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`.)
@@ -56,22 +91,25 @@ 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>> {
     // 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);
@@ -81,7 +119,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();
@@ -102,15 +143,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
@@ -125,13 +172,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