From 8311bab6740a59b888f1a82b598b4e5869a2aa9e Mon Sep 17 00:00:00 2001
From: Francisco Javier Arceo <farceo@redhat.com>
Date: Sat, 1 Aug 2026 23:16:57 -0400
Subject: [PATCH 1/2] fix: preserve Claude Code Messages transport fidelity

Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
---
 .github/workflows/claude-code-e2e.yml         |  66 +++++
 .../agentic-server-core/src/executor/error.rs |  16 +-
 .../src/executor/inference.rs                 |  55 +++-
 .../src/executor/messages_loop.rs             |  82 ++++--
 .../src/executor/messages_stream.rs           | 107 +++++--
 .../agentic-server-core/src/executor/mod.rs   |   2 +-
 crates/agentic-server-core/src/proxy.rs       |  61 +++-
 .../tests/cassettes/README.md                 |  26 ++
 .../claude-code-cache-control-request.json    |  80 ++++++
 .../tests/messages_loop_test.rs               |  91 +++++-
 .../tests/messages_stream_test.rs             | 188 +++++++++++-
 crates/agentic-server/src/handler/common.rs   |  20 +-
 .../src/handler/http/messages.rs              |  92 +++---
 crates/agentic-server/tests/messages_test.rs  | 213 +++++++++++++-
 scripts/claude-code-smoke.sh                  | 142 ++++++++++
 scripts/claude_code_replay_server.py          | 268 ++++++++++++++++++
 scripts/test_claude_code_replay_server.py     | 145 ++++++++++
 17 files changed, 1533 insertions(+), 121 deletions(-)
 create mode 100644 .github/workflows/claude-code-e2e.yml
 create mode 100644 crates/agentic-server-core/tests/fixtures/claude-code-cache-control-request.json
 create mode 100755 scripts/claude-code-smoke.sh
 create mode 100755 scripts/claude_code_replay_server.py
 create mode 100644 scripts/test_claude_code_replay_server.py

diff --git a/.github/workflows/claude-code-e2e.yml b/.github/workflows/claude-code-e2e.yml
new file mode 100644
index 00000000..2bb4d705
--- /dev/null
+++ b/.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
diff --git a/crates/agentic-server-core/src/executor/error.rs b/crates/agentic-server-core/src/executor/error.rs
index f2eb5307..e520224b 100644
--- a/crates/agentic-server-core/src/executor/error.rs
+++ b/crates/agentic-server-core/src/executor/error.rs
@@ -12,9 +12,17 @@ pub enum ExecutorError {
     #[error("storage error: {0}")]
     Storage(#[from] 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.
     ///
@@ -69,7 +77,7 @@ impl ExecutorError {
     pub fn http_status(&self) -> StatusCode {
         match self {
             Self::Storage(e) if e.is_not_found() => StatusCode::NOT_FOUND,
-            Self::LLMRequest { status, .. } => *status,
+            Self::LLMRequest { status, .. } | Self::LLMTransport { status, .. } => *status,
             Self::Tool(ToolError::Config(_)) | Self::InvalidRequest(_) | Self::JsonError(_) => StatusCode::BAD_REQUEST,
             Self::Tool(ToolError::Execution(_)) | Self::CompactionFailed { .. } => StatusCode::BAD_GATEWAY,
             Self::ParseError(_) => StatusCode::UNPROCESSABLE_ENTITY,
@@ -82,7 +90,7 @@ impl ExecutorError {
     pub fn error_code(&self) -> &'static str {
         match self {
             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::Config(_)) | Self::InvalidRequest(_) | Self::ParseError(_) | Self::JsonError(_) => {
                 "invalid_request_error"
             }
diff --git a/crates/agentic-server-core/src/executor/inference.rs b/crates/agentic-server-core/src/executor/inference.rs
index c4932102..d5a63bc1 100644
--- a/crates/agentic-server-core/src/executor/inference.rs
+++ b/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);
 
diff --git a/crates/agentic-server-core/src/executor/messages_loop.rs b/crates/agentic-server-core/src/executor/messages_loop.rs
index bb79356e..68c2de1e 100644
--- a/crates/agentic-server-core/src/executor/messages_loop.rs
+++ b/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,7 +143,10 @@ 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
@@ -110,7 +154,10 @@ pub async fn run_messages_loop(
             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
diff --git a/crates/agentic-server-core/src/executor/messages_stream.rs b/crates/agentic-server-core/src/executor/messages_stream.rs
index dfaf8f2f..a6580164 100644
--- a/crates/agentic-server-core/src/executor/messages_stream.rs
+++ b/crates/agentic-server-core/src/executor/messages_stream.rs
@@ -22,44 +22,79 @@ use async_stream::stream;
 use futures::StreamExt;
 use serde_json::{Value, json};
 
-use crate::executor::inference::{BoxStream, call_inference};
+use crate::executor::error::{ExecutorError, ExecutorResult};
+use crate::executor::inference::{BoxStream, response_lines, send_request};
 use crate::executor::request::ExecutionContext;
+use crate::proxy::processed_response_headers;
 use crate::tool::ToolRegistry;
 use crate::types::messages::tool_seam;
-use crate::utils::common::serialize_to_string;
+use crate::utils::common::{deserialize_from_str, serialize_to_string};
 
 // Shared with the non-streaming loop so the two Messages loops can't drift.
-use crate::executor::messages_loop::{GATEWAY_TOOL_TIMEOUT, MAX_GATEWAY_TOOL_ROUNDS};
+use crate::executor::messages_loop::{
+    GATEWAY_TOOL_TIMEOUT, MAX_GATEWAY_TOOL_ROUNDS, MessagesResponse, MessagesUpstream,
+};
 /// vLLM streaming chunk timeout (per line). Generous — the loop's own budget is
 /// the round cap, not this.
 const CHUNK_TIMEOUT: Duration = Duration::from_secs(120);
 
 /// Drive the streaming Messages-native loop, yielding Anthropic SSE lines for
 /// the client. Owns the multi-round → single-message accumulation.
-#[must_use]
-pub fn run_messages_stream(
+///
+/// # Errors
+///
+/// Returns an executor error when the initial request cannot be serialized or
+/// when the upstream rejects it before streaming begins.
+pub async fn run_messages_stream(
     mut request: Value,
     registry: Arc<ToolRegistry>,
     exec_ctx: Arc<ExecutionContext>,
-    auth: Option<String>,
-) -> BoxStream {
-    let url = format!("{}/v1/messages", exec_ctx.llm_base_url);
+    upstream: MessagesUpstream,
+) -> ExecutorResult<MessagesResponse<BoxStream>> {
     request["stream"] = Value::Bool(true);
 
-    Box::pin(stream! {
+    // Prime the first upstream request before the handler commits an HTTP 200.
+    // This lets initial vLLM errors retain their original status and body.
+    let first_body = serialize_to_string(&request)?;
+    let first_response = send_request(
+        &exec_ctx.client,
+        upstream.url(),
+        first_body,
+        None,
+        Some(upstream.headers()),
+    )
+    .await?;
+    let response_headers = processed_response_headers(first_response.headers());
+
+    let body: BoxStream = Box::pin(stream! {
         let mut acc = MessagesStreamAccumulator::new(exec_ctx.messages_gateway_tools.clone());
+        let mut prepared_response = Some(first_response);
 
         for _round in 0..MAX_GATEWAY_TOOL_ROUNDS {
-            let body = match serialize_to_string(&request) {
-                Ok(b) => b,
-                Err(e) => { yield error_sse(&e.to_string()); return; }
+            let response = if let Some(response) = prepared_response.take() {
+                response
+            } else {
+                let body = match serialize_to_string(&request) {
+                    Ok(b) => b,
+                    Err(e) => { yield error_sse(&e.to_string()); return; }
+                };
+                match send_request(
+                    &exec_ctx.client,
+                    upstream.url(),
+                    body,
+                    None,
+                    Some(upstream.headers()),
+                )
+                .await
+                {
+                    Ok(response) => response,
+                    Err(e) => { yield executor_error_sse(&e); return; }
+                }
             };
-            let mut upstream = Box::pin(call_inference(
-                body, url.clone(), Arc::clone(&exec_ctx.client), auth.clone(), CHUNK_TIMEOUT,
-            ));
+            let mut response_stream = Box::pin(response_lines(response, CHUNK_TIMEOUT));
 
             acc.begin_round();
-            while let Some(line) = upstream.next().await {
+            while let Some(line) = response_stream.next().await {
                 let line = match line {
                     Ok(l) => l,
                     Err(e) => { yield error_sse(&e.to_string()); return; }
@@ -67,6 +102,9 @@ pub fn run_messages_stream(
                 for out in acc.push(&line) {
                     yield out;
                 }
+                if acc.has_upstream_error() {
+                    return;
+                }
             }
 
             // Round finished. Continue only for a pure gateway-tool round; a
@@ -88,6 +126,10 @@ pub fn run_messages_stream(
 
         // Round budget exhausted.
         yield error_sse(&format!("gateway tool loop exceeded {MAX_GATEWAY_TOOL_ROUNDS} rounds"));
+    });
+    Ok(MessagesResponse {
+        body,
+        headers: response_headers,
     })
 }
 
@@ -111,6 +153,12 @@ struct BufferedBlock {
     is_gateway_tool: bool,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum RoundState {
+    Active,
+    UpstreamError,
+}
+
 impl BufferedBlock {
     fn apply_delta(&mut self, delta: &Value) {
         match delta.get("type").and_then(Value::as_str) {
@@ -174,6 +222,8 @@ struct MessagesStreamAccumulator {
     has_client_tool_use: bool,
     /// Buffered terminal `message_delta` from the final round (emitted by `finish`).
     final_message_delta: Option<Value>,
+    /// Whether this round is still active or terminated with an upstream error.
+    round_state: RoundState,
     /// Operator-configured client-tool → gateway-executor aliases, so a client
     /// tool like Claude Code's `WebSearch` is classified gateway-owned (and
     /// suppressed) the same way the built-in `web_search` is.
@@ -191,6 +241,7 @@ impl MessagesStreamAccumulator {
             ended_on_tool_use: false,
             has_client_tool_use: false,
             final_message_delta: None,
+            round_state: RoundState::Active,
             gateway_map,
         }
     }
@@ -201,6 +252,7 @@ impl MessagesStreamAccumulator {
         self.blocks.clear();
         self.ended_on_tool_use = false;
         self.has_client_tool_use = false;
+        self.round_state = RoundState::Active;
         // F6: clear the previous round's terminal so a clean-EOF round can't
         // re-emit a stale stop_reason.
         self.final_message_delta = None;
@@ -239,6 +291,10 @@ impl MessagesStreamAccumulator {
         self.ended_on_tool_use && self.gateway_call_count() > 0 && !self.has_client_tool_use
     }
 
+    fn has_upstream_error(&self) -> bool {
+        self.round_state == RoundState::UpstreamError
+    }
+
     /// Translate one upstream SSE line into zero or more client SSE lines.
     fn push(&mut self, line: &str) -> Vec<String> {
         let Some(data) = line.strip_prefix("data: ") else {
@@ -262,6 +318,10 @@ impl MessagesStreamAccumulator {
                 self.final_message_delta = Some(event);
                 Vec::new()
             }
+            Some("error") => {
+                self.round_state = RoundState::UpstreamError;
+                vec![sse("error", &event)]
+            }
             // `message_stop` (per-round terminal) is suppressed; `finish` emits
             // the single client-visible terminal. Everything else is dropped.
             _ => Vec::new(),
@@ -365,6 +425,21 @@ fn error_sse(message: &str) -> String {
     format!("event: error\ndata: {json}\n\n")
 }
 
+fn executor_error_sse(error: &ExecutorError) -> String {
+    if let ExecutorError::LLMRequest { body, .. } = error
+        && let Ok(value) = deserialize_from_str::<Value>(body)
+        && value.get("type").and_then(Value::as_str) == Some("error")
+    {
+        let data = if body.contains(['\r', '\n']) {
+            serialize_to_string(&value).unwrap_or_else(|_| body.clone())
+        } else {
+            body.clone()
+        };
+        return format!("event: error\ndata: {data}\n\n");
+    }
+    error_sse(&error.to_string())
+}
+
 /// Execute reconstructed gateway calls (concurrent, per-call timeout). Errors
 /// become error `tool_result`s (E5).
 async fn execute_gateway_calls(
diff --git a/crates/agentic-server-core/src/executor/mod.rs b/crates/agentic-server-core/src/executor/mod.rs
index e45f8633..b74248dc 100644
--- a/crates/agentic-server-core/src/executor/mod.rs
+++ b/crates/agentic-server-core/src/executor/mod.rs
@@ -20,7 +20,7 @@ pub use compaction::compact_response;
 pub use engine::{BoxStream, ExecuteRequest, create_conversation, execute};
 pub use error::{ExecutorError, ExecutorResult};
 pub use inference::call_inference;
-pub use messages_loop::run_messages_loop;
+pub use messages_loop::{MessagesResponse, MessagesUpstream, run_messages_loop};
 pub use messages_stream::run_messages_stream;
 pub use modes::{ConversationHandler, ResponseHandler};
 pub use persist::persist_response;
diff --git a/crates/agentic-server-core/src/proxy.rs b/crates/agentic-server-core/src/proxy.rs
index be8f83d8..da982670 100644
--- a/crates/agentic-server-core/src/proxy.rs
+++ b/crates/agentic-server-core/src/proxy.rs
@@ -23,6 +23,7 @@ const HOP_BY_HOP: &[&str] = &[
 ];
 
 const REQUEST_DROP_EXTRA: &[&str] = &["host", "content-length"];
+const PROCESSED_RESPONSE_DROP_EXTRA: &[&str] = &["content-length", "content-encoding"];
 
 fn is_hop_by_hop(name: &str) -> bool {
     HOP_BY_HOP.iter().any(|h| h.eq_ignore_ascii_case(name))
@@ -92,7 +93,13 @@ impl ProxyState {
     }
 }
 
-fn filter_request_headers(headers: &HeaderMap, config: &Config, auth: ProxyAuth) -> reqwest::header::HeaderMap {
+/// Build the request headers forwarded to an upstream API.
+///
+/// Hop-by-hop and origin-specific headers are removed, all other headers stay
+/// open-ended, and the configured credential is injected only when the client
+/// did not supply one.
+#[must_use]
+pub fn upstream_request_headers(headers: &HeaderMap, config: &Config, auth: ProxyAuth) -> reqwest::header::HeaderMap {
     let mut out = reqwest::header::HeaderMap::new();
     for (name, value) in headers {
         if is_request_drop(name.as_str()) {
@@ -143,6 +150,20 @@ fn filter_response_headers(headers: &reqwest::header::HeaderMap) -> HeaderMap {
     out
 }
 
+/// Build response headers for an upstream body consumed or transformed in-process.
+///
+/// Hop-by-hop headers are removed along with representation metadata that may no
+/// longer describe the emitted body. Request IDs, retry guidance, and rate-limit
+/// metadata remain open-ended.
+#[must_use]
+pub fn processed_response_headers(headers: &reqwest::header::HeaderMap) -> HeaderMap {
+    let mut out = filter_response_headers(headers);
+    for name in PROCESSED_RESPONSE_DROP_EXTRA {
+        out.remove(*name);
+    }
+    out
+}
+
 fn is_sse_content_type(headers: &reqwest::header::HeaderMap) -> bool {
     headers
         .get(reqwest::header::CONTENT_TYPE)
@@ -189,7 +210,7 @@ pub fn error_response_for_auth(status: StatusCode, code: &str, message: &str, au
 /// Uses the non-streaming client; the response body is returned as a full
 /// [`ProxyBody::Full`] payload.
 pub async fn proxy_get(path: &str, request_headers: &HeaderMap, state: &ProxyState) -> ProxyResponse {
-    let llm_headers = filter_request_headers(request_headers, &state.config, ProxyAuth::OpenAiBearer);
+    let llm_headers = upstream_request_headers(request_headers, &state.config, ProxyAuth::OpenAiBearer);
     let base = state.config.llm_api_base.trim_end_matches('/');
     let url = format!("{base}/{}", path.trim_start_matches('/'));
 
@@ -242,7 +263,7 @@ pub async fn proxy_request_with_path(
         .and_then(|v| v.get("stream")?.as_bool())
         .unwrap_or(false);
 
-    let llm_headers = filter_request_headers(&request.headers, &state.config, auth);
+    let llm_headers = upstream_request_headers(&request.headers, &state.config, auth);
 
     let base = state.config.llm_api_base.trim_end_matches('/');
     let mut url = format!("{base}/{}", path.trim_start_matches('/'));
@@ -370,7 +391,7 @@ mod tests {
         headers.insert("x-custom", "value".parse().unwrap());
 
         let config = test_config_no_key();
-        let filtered = filter_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
+        let filtered = upstream_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
 
         assert!(filtered.contains_key("content-type"));
         assert!(filtered.contains_key("x-custom"));
@@ -386,7 +407,7 @@ mod tests {
         headers.insert("accept", "*/*".parse().unwrap());
 
         let config = test_config_no_key();
-        let filtered = filter_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
+        let filtered = upstream_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
 
         assert!(!filtered.contains_key("host"));
         assert!(!filtered.contains_key("content-length"));
@@ -397,7 +418,7 @@ mod tests {
     fn auth_injected_when_no_client_auth() {
         let headers = HeaderMap::new();
         let config = test_config();
-        let filtered = filter_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
+        let filtered = upstream_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
 
         assert_eq!(
             filtered.get("authorization").unwrap().to_str().unwrap(),
@@ -411,7 +432,7 @@ mod tests {
         headers.insert("authorization", "Bearer client-token".parse().unwrap());
 
         let config = test_config();
-        let filtered = filter_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
+        let filtered = upstream_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
 
         assert_eq!(
             filtered.get("authorization").unwrap().to_str().unwrap(),
@@ -424,7 +445,7 @@ mod tests {
         let mut headers = HeaderMap::new();
         headers.insert("x-api-key", "client-anthropic-key".parse().unwrap());
 
-        let filtered = filter_request_headers(&headers, &test_config(), ProxyAuth::Anthropic);
+        let filtered = upstream_request_headers(&headers, &test_config(), ProxyAuth::Anthropic);
 
         assert_eq!(filtered.get("x-api-key").unwrap(), "client-anthropic-key");
         assert!(!filtered.contains_key("authorization"));
@@ -432,7 +453,7 @@ mod tests {
 
     #[test]
     fn anthropic_auth_uses_configured_key_as_api_key_fallback() {
-        let filtered = filter_request_headers(&HeaderMap::new(), &test_config(), ProxyAuth::Anthropic);
+        let filtered = upstream_request_headers(&HeaderMap::new(), &test_config(), ProxyAuth::Anthropic);
 
         assert_eq!(filtered.get("x-api-key").unwrap(), "test-key");
         assert!(!filtered.contains_key("authorization"));
@@ -445,7 +466,7 @@ mod tests {
             openai_api_key: Some("  ".to_owned()),
             ..test_config()
         };
-        let filtered = filter_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
+        let filtered = upstream_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
 
         assert!(!filtered.contains_key("authorization"));
     }
@@ -454,7 +475,7 @@ mod tests {
     fn no_auth_injected_when_key_none() {
         let headers = HeaderMap::new();
         let config = test_config_no_key();
-        let filtered = filter_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
+        let filtered = upstream_request_headers(&headers, &config, ProxyAuth::OpenAiBearer);
 
         assert!(!filtered.contains_key("authorization"));
     }
@@ -473,6 +494,24 @@ mod tests {
         assert!(!filtered.contains_key("connection"));
     }
 
+    #[test]
+    fn processed_response_headers_preserve_metadata_and_strip_representation_headers() {
+        let mut headers = reqwest::header::HeaderMap::new();
+        headers.insert("request-id", "req_123".parse().unwrap());
+        headers.insert("retry-after", "3".parse().unwrap());
+        headers.insert("anthropic-ratelimit-requests-remaining", "7".parse().unwrap());
+        headers.insert("content-length", "99".parse().unwrap());
+        headers.insert("content-encoding", "gzip".parse().unwrap());
+
+        let filtered = processed_response_headers(&headers);
+
+        assert_eq!(filtered["request-id"], "req_123");
+        assert_eq!(filtered["retry-after"], "3");
+        assert_eq!(filtered["anthropic-ratelimit-requests-remaining"], "7");
+        assert!(!filtered.contains_key("content-length"));
+        assert!(!filtered.contains_key("content-encoding"));
+    }
+
     #[test]
     fn sse_content_type_detected() {
         let mut headers = reqwest::header::HeaderMap::new();
diff --git a/crates/agentic-server-core/tests/cassettes/README.md b/crates/agentic-server-core/tests/cassettes/README.md
index b5d31756..af8a2f12 100644
--- a/crates/agentic-server-core/tests/cassettes/README.md
+++ b/crates/agentic-server-core/tests/cassettes/README.md
@@ -29,6 +29,32 @@ python tests/cassettes/record_cassette.py --mode responses --turns 1 --no-stream
 
 The recorder scripts (`record_reasoning_cassettes.sh`, `record_tool_call_cassettes.sh`, etc.) use `printf` to feed fixed prompts per test so no manual input is needed.
 
+## Claude Code cache-control parity
+
+The literal fixture at `../fixtures/claude-code-cache-control-request.json` mirrors the cache-bearing parts of a
+Claude Code Messages request: multi-block `system`, a structured user message, and both `WebSearch` and client-owned
+tool declarations. It includes explicit `5m` and `1h` TTLs. The Messages HTTP and loop integration tests assert that
+the fixture remains unchanged through transparent proxying and every streaming and non-streaming gateway-tool round.
+
+The fixture is hand-checked test data, not a captured cassette. A dedicated GitHub Actions job also runs the real,
+pinned Claude Code CLI through `agentic-server` against the recorded streaming vLLM Messages cassette and a
+deterministic local search backend. It uses a placeholder API key and makes no request to Anthropic or a live model.
+
+To run the same end-to-end check locally, install the pinned dependencies, build the server, and run the harness:
+
+```bash
+npm install --global '@anthropic-ai/claude-code@2.1.218'
+python -m pip install 'PyYAML==6.0.3'
+cargo build -p agentic-server
+bash scripts/claude-code-smoke.sh
+```
+
+The harness starts both local services, opts `WebSearch` into gateway execution with
+`MESSAGES_GATEWAY_TOOL_ALIASES=WebSearch=web_search`, and invokes Claude Code's non-interactive interface in safe
+mode. It disables nonessential traffic and session persistence, maps every Claude model tier to the replayed `qwen3`
+model, then asserts two Messages rounds, one search request, a hidden `tool_result`, and the cache-bearing system and
+user blocks emitted by Claude Code 2.1.218.
+
 ## Modes
 
 | Mode | Description |
diff --git a/crates/agentic-server-core/tests/fixtures/claude-code-cache-control-request.json b/crates/agentic-server-core/tests/fixtures/claude-code-cache-control-request.json
new file mode 100644
index 00000000..41838764
--- /dev/null
+++ b/crates/agentic-server-core/tests/fixtures/claude-code-cache-control-request.json
@@ -0,0 +1,80 @@
+{
+  "model": "qwen3",
+  "max_tokens": 1024,
+  "stream": false,
+  "system": [
+    {
+      "type": "text",
+      "text": "You are Claude Code, Anthropic's official CLI for Claude.",
+      "cache_control": {
+        "type": "ephemeral",
+        "ttl": "1h"
+      }
+    },
+    {
+      "type": "text",
+      "text": "<attribution>claude-code-session</attribution>",
+      "cache_control": {
+        "type": "ephemeral",
+        "ttl": "5m"
+      }
+    }
+  ],
+  "messages": [
+    {
+      "role": "user",
+      "content": [
+        {
+          "type": "text",
+          "text": "Use WebSearch to find the latest stable Rust release.",
+          "cache_control": {
+            "type": "ephemeral",
+            "ttl": "5m"
+          }
+        }
+      ]
+    }
+  ],
+  "tools": [
+    {
+      "name": "WebSearch",
+      "description": "Search the web",
+      "input_schema": {
+        "type": "object",
+        "properties": {
+          "query": {
+            "type": "string"
+          }
+        },
+        "required": [
+          "query"
+        ]
+      }
+    },
+    {
+      "name": "Read",
+      "description": "Read a file from the local workspace",
+      "input_schema": {
+        "type": "object",
+        "properties": {
+          "file_path": {
+            "type": "string"
+          }
+        },
+        "required": [
+          "file_path"
+        ]
+      },
+      "cache_control": {
+        "type": "ephemeral",
+        "ttl": "1h"
+      }
+    }
+  ],
+  "thinking": {
+    "type": "adaptive"
+  },
+  "metadata": {
+    "user_id": "claude-code-smoke"
+  }
+}
diff --git a/crates/agentic-server-core/tests/messages_loop_test.rs b/crates/agentic-server-core/tests/messages_loop_test.rs
index 23283e97..9960ce71 100644
--- a/crates/agentic-server-core/tests/messages_loop_test.rs
+++ b/crates/agentic-server-core/tests/messages_loop_test.rs
@@ -13,7 +13,9 @@
 use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
-use agentic_core::executor::{ConversationHandler, ExecutionContext, ResponseHandler, run_messages_loop};
+use agentic_core::executor::{
+    ConversationHandler, ExecutionContext, ExecutorResult, MessagesUpstream, ResponseHandler, run_messages_loop,
+};
 use agentic_core::storage::{ConversationStore, ResponseStore};
 use agentic_core::tool::{ToolRegistry, WebSearchHandler};
 use agentic_core::types::messages::{GatewayToolMap, ToolParam, registry_tools};
@@ -48,6 +50,7 @@ fn cassette_bodies_at(path: &str) -> Vec<Value> {
 }
 
 const MULTIROUND_DIR: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/cassettes/messages_multiround");
+const CLAUDE_CODE_CACHE_CONTROL_REQUEST: &str = include_str!("fixtures/claude-code-cache-control-request.json");
 
 /// Mock vLLM `/v1/messages` — serves the recorded response bodies in order and
 /// records each request body it received (to assert the loop fed the
@@ -153,13 +156,24 @@ async fn build_exec_ctx(vllm_url: &str, search_url: &str) -> ExecutionContext {
 }
 
 async fn build_tool_registry(tools: &Vec<ToolParam>, exec_ctx: &ExecutionContext) -> ToolRegistry {
-    let mut registry_tool_params = registry_tools(Some(tools), &GatewayToolMap::default());
+    let mut registry_tool_params = registry_tools(Some(tools), &exec_ctx.messages_gateway_tools);
     let mut gateway_executors = exec_ctx.gateway_executors.clone();
     ToolRegistry::build_with_handlers(&mut registry_tool_params, &mut gateway_executors)
         .await
         .unwrap()
 }
 
+async fn run_test_messages_loop(
+    request: Value,
+    registry: &ToolRegistry,
+    exec_ctx: &ExecutionContext,
+) -> ExecutorResult<Value> {
+    let upstream = MessagesUpstream::new(&exec_ctx.llm_base_url, None, reqwest::header::HeaderMap::new());
+    run_messages_loop(request, registry, exec_ctx, &upstream)
+        .await
+        .map(|response| response.body)
+}
+
 fn web_search_request() -> Value {
     serde_json::json!({
         "model": "qwen3",
@@ -181,7 +195,7 @@ async fn messages_loop_hides_gateway_tool_and_surfaces_final_text() {
     let tools: Vec<ToolParam> = serde_json::from_value(request["tools"].clone()).unwrap();
     let registry = build_tool_registry(&tools, &exec_ctx).await;
 
-    let result = run_messages_loop(request, &registry, &exec_ctx, None)
+    let result = run_test_messages_loop(request, &registry, &exec_ctx)
         .await
         .expect("loop runs");
 
@@ -248,7 +262,7 @@ async fn repro_f3_next_round_preserves_thinking_and_text_blocks() {
     let request = web_search_request();
     let tools: Vec<ToolParam> = serde_json::from_value(request["tools"].clone()).unwrap();
     let registry = build_tool_registry(&tools, &exec_ctx).await;
-    run_messages_loop(request, &registry, &exec_ctx, None).await.unwrap();
+    run_test_messages_loop(request, &registry, &exec_ctx).await.unwrap();
 
     // Inspect the assistant turn the loop appended for round 2.
     let reqs = upstream.requests.lock().await;
@@ -317,7 +331,7 @@ async fn run_against(bodies: Vec<Value>, request: Value) -> (Value, usize) {
     let exec_ctx = build_exec_ctx(&vllm_url, &search_url).await;
     let tools: Vec<ToolParam> = serde_json::from_value(request["tools"].clone()).unwrap_or_default();
     let registry = build_tool_registry(&tools, &exec_ctx).await;
-    let result = run_messages_loop(request, &registry, &exec_ctx, None).await.unwrap();
+    let result = run_test_messages_loop(request, &registry, &exec_ctx).await.unwrap();
     (result, upstream.calls.load(Ordering::SeqCst))
 }
 
@@ -376,7 +390,7 @@ async fn messages_loop_multi_round_sequential() {
     let tools: Vec<ToolParam> = serde_json::from_value(request["tools"].clone()).unwrap();
     let registry = build_tool_registry(&tools, &exec_ctx).await;
 
-    let result = run_messages_loop(request, &registry, &exec_ctx, None).await.unwrap();
+    let result = run_test_messages_loop(request, &registry, &exec_ctx).await.unwrap();
 
     assert_eq!(upstream.calls.load(Ordering::SeqCst), 3, "three upstream rounds");
     // Two gateway searches executed (rounds 0 and 1).
@@ -399,7 +413,7 @@ async fn messages_loop_parallel_tool_use() {
     let tools: Vec<ToolParam> = serde_json::from_value(request["tools"].clone()).unwrap();
     let registry = build_tool_registry(&tools, &exec_ctx).await;
 
-    let result = run_messages_loop(request, &registry, &exec_ctx, None).await.unwrap();
+    let result = run_test_messages_loop(request, &registry, &exec_ctx).await.unwrap();
 
     // Two parallel gateway calls both executed against the backend.
     captured.recv().await.expect("first parallel search");
@@ -442,7 +456,7 @@ async fn messages_loop_tool_failure_becomes_error_tool_result() {
     let tools: Vec<ToolParam> = serde_json::from_value(request["tools"].clone()).unwrap();
     let registry = build_tool_registry(&tools, &exec_ctx).await;
 
-    let result = run_messages_loop(request, &registry, &exec_ctx, None).await.unwrap();
+    let result = run_test_messages_loop(request, &registry, &exec_ctx).await.unwrap();
 
     // The request did NOT fail — it looped to a final answer.
     assert_eq!(result["stop_reason"], "end_turn");
@@ -486,7 +500,7 @@ async fn messages_loop_caps_at_max_rounds() {
     let tools: Vec<ToolParam> = serde_json::from_value(request["tools"].clone()).unwrap();
     let registry = build_tool_registry(&tools, &exec_ctx).await;
 
-    let result = run_messages_loop(request, &registry, &exec_ctx, None).await.unwrap();
+    let result = run_test_messages_loop(request, &registry, &exec_ctx).await.unwrap();
 
     let calls = upstream.calls.load(Ordering::SeqCst);
     assert!(calls <= 10, "loop must cap at MAX_GATEWAY_TOOL_ROUNDS (got {calls})");
@@ -547,7 +561,7 @@ async fn messages_loop_preserves_multi_block_system_across_rounds() {
     let tools: Vec<ToolParam> = serde_json::from_value(request["tools"].clone()).unwrap();
     let registry = build_tool_registry(&tools, &exec_ctx).await;
 
-    let result = run_messages_loop(request, &registry, &exec_ctx, None).await.unwrap();
+    let result = run_test_messages_loop(request, &registry, &exec_ctx).await.unwrap();
     assert_eq!(result["stop_reason"], "end_turn");
     assert_eq!(upstream.calls.load(Ordering::SeqCst), 2, "tool round + final round");
 
@@ -559,3 +573,60 @@ async fn messages_loop_preserves_multi_block_system_across_rounds() {
         "round 2 (after append_round_to_history) still carries the system blocks unchanged"
     );
 }
+
+#[tokio::test]
+async fn messages_loop_preserves_claude_code_cache_control_across_rounds() {
+    let round0 = serde_json::json!({
+        "id": "m", "type": "message", "role": "assistant", "model": "qwen3",
+        "content": [{"type": "tool_use", "id": "t1", "name": "WebSearch", "input": {"query": "rust"}}],
+        "stop_reason": "tool_use", "usage": {"input_tokens": 5, "output_tokens": 3}
+    });
+    let round1 = serde_json::json!({
+        "id": "m2", "type": "message", "role": "assistant", "model": "qwen3",
+        "content": [{"type": "text", "text": "Rust is stable."}],
+        "stop_reason": "end_turn", "usage": {"input_tokens": 5, "output_tokens": 3}
+    });
+    let (vllm_url, upstream, _v) = spawn_mock_vllm_messages(vec![round0, round1]).await;
+    let (search_url, mut search_requests, _s) = spawn_mock_search().await;
+    let mut exec_ctx = build_exec_ctx(&vllm_url, &search_url).await;
+    exec_ctx.messages_gateway_tools = GatewayToolMap::from_pairs([("WebSearch", "web_search")]);
+
+    let request: Value = serde_json::from_str(CLAUDE_CODE_CACHE_CONTROL_REQUEST).unwrap();
+    let original_system = request["system"].clone();
+    let original_user_message = request["messages"][0].clone();
+    let original_tools = request["tools"].clone();
+    let tools: Vec<ToolParam> = serde_json::from_value(original_tools.clone()).unwrap();
+    let registry = build_tool_registry(&tools, &exec_ctx).await;
+
+    let result = run_test_messages_loop(request, &registry, &exec_ctx).await.unwrap();
+    assert_eq!(result["stop_reason"], "end_turn");
+    let search_request = search_requests
+        .try_recv()
+        .expect("WebSearch alias dispatches to the gateway search backend");
+    assert_eq!(search_request.body["query"], "rust");
+
+    let requests = upstream.requests.lock().await;
+    assert_eq!(requests.len(), 2, "tool round + final round");
+    for upstream_request in requests.iter() {
+        assert_eq!(upstream_request["system"], original_system);
+        assert_eq!(upstream_request["messages"][0], original_user_message);
+        assert_eq!(upstream_request["tools"], original_tools);
+    }
+    assert_eq!(
+        requests[1]["system"][0]["cache_control"],
+        serde_json::json!({"type": "ephemeral", "ttl": "1h"})
+    );
+    assert_eq!(
+        requests[1]["system"][1]["cache_control"],
+        serde_json::json!({"type": "ephemeral", "ttl": "5m"})
+    );
+    assert_eq!(
+        requests[1]["messages"][0]["content"][0]["cache_control"],
+        serde_json::json!({"type": "ephemeral", "ttl": "5m"})
+    );
+    assert_eq!(
+        requests[1]["tools"][1]["cache_control"],
+        serde_json::json!({"type": "ephemeral", "ttl": "1h"})
+    );
+    assert!(requests[1]["tools"][0].get("cache_control").is_none());
+}
diff --git a/crates/agentic-server-core/tests/messages_stream_test.rs b/crates/agentic-server-core/tests/messages_stream_test.rs
index ecdc80f3..71588f10 100644
--- a/crates/agentic-server-core/tests/messages_stream_test.rs
+++ b/crates/agentic-server-core/tests/messages_stream_test.rs
@@ -11,7 +11,9 @@
 use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
-use agentic_core::executor::{ConversationHandler, ExecutionContext, ResponseHandler, run_messages_stream};
+use agentic_core::executor::{
+    BoxStream, ConversationHandler, ExecutionContext, MessagesUpstream, ResponseHandler, run_messages_stream,
+};
 use agentic_core::storage::{ConversationStore, ResponseStore};
 use agentic_core::tool::{ToolRegistry, WebSearchHandler};
 use agentic_core::types::messages::{GatewayToolMap, ToolParam, registry_tools};
@@ -35,6 +37,7 @@ const MULTIROUND: &str = concat!(
     env!("CARGO_MANIFEST_DIR"),
     "/tests/cassettes/messages_multiround/multiround-web-search-qwen3-streaming.yaml"
 );
+const CLAUDE_CODE_CACHE_CONTROL_REQUEST: &str = include_str!("fixtures/claude-code-cache-control-request.json");
 
 /// Load each streaming turn's SSE body (the raw event-stream text) from the cassette.
 fn cassette_turn_streams() -> Vec<String> {
@@ -102,6 +105,44 @@ async fn spawn_mock_vllm_stream(streams: Vec<String>) -> (String, UpstreamState,
     (format!("http://{addr}"), state, handle)
 }
 
+async fn spawn_mock_vllm_stream_then_error(
+    first_stream: String,
+    error_body: &'static str,
+) -> (String, Arc<AtomicUsize>, tokio::task::JoinHandle<()>) {
+    let calls = Arc::new(AtomicUsize::new(0));
+    let route_calls = Arc::clone(&calls);
+    let app = Router::new().route(
+        "/v1/messages",
+        post(move |_body: axum::body::Bytes| {
+            let n = route_calls.fetch_add(1, Ordering::SeqCst);
+            let first_stream = first_stream.clone();
+            async move {
+                if n == 0 {
+                    Response::builder()
+                        .status(StatusCode::OK)
+                        .header("content-type", "text/event-stream")
+                        .body(axum::body::Body::from(first_stream))
+                        .unwrap()
+                        .into_response()
+                } else {
+                    Response::builder()
+                        .status(StatusCode::BAD_REQUEST)
+                        .header("content-type", "application/json")
+                        .body(axum::body::Body::from(error_body))
+                        .unwrap()
+                        .into_response()
+                }
+            }
+        }),
+    );
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let handle = tokio::spawn(async move {
+        axum::serve(listener, app).await.ok();
+    });
+    (format!("http://{addr}"), calls, handle)
+}
+
 async fn spawn_mock_search() -> (String, tokio::task::JoinHandle<()>) {
     let app = Router::new().route(
         "/v1/search",
@@ -133,6 +174,18 @@ async fn build_exec_ctx(vllm_url: &str, search_url: &str) -> Arc<ExecutionContex
     )
 }
 
+async fn run_test_messages_stream(
+    request: Value,
+    registry: Arc<ToolRegistry>,
+    exec_ctx: Arc<ExecutionContext>,
+) -> BoxStream {
+    let upstream = MessagesUpstream::new(&exec_ctx.llm_base_url, None, reqwest::header::HeaderMap::new());
+    run_messages_stream(request, registry, exec_ctx, upstream)
+        .await
+        .map(|response| response.body)
+        .unwrap()
+}
+
 #[tokio::test]
 async fn messages_stream_presents_one_message_and_hides_gateway_tool() {
     let (vllm_url, upstream, _v) = spawn_mock_vllm_stream(cassette_turn_streams()).await;
@@ -154,7 +207,7 @@ async fn messages_stream_presents_one_message_and_hides_gateway_tool() {
             .unwrap(),
     );
 
-    let stream = run_messages_stream(request, registry, Arc::clone(&exec_ctx), None);
+    let stream = run_test_messages_stream(request, registry, Arc::clone(&exec_ctx)).await;
     let chunks: Vec<String> = stream.collect().await;
     let sse = chunks.join("");
 
@@ -207,6 +260,78 @@ async fn messages_stream_presents_one_message_and_hides_gateway_tool() {
     );
 }
 
+#[tokio::test]
+async fn messages_stream_preserves_error_event_after_gateway_tool_round() {
+    let error_body = r#"{"type":"error","error":{"type":"invalid_request_error","message":"bad second round"}}"#;
+    let first_stream = cassette_turn_streams().remove(0);
+    let (vllm_url, calls, _v) = spawn_mock_vllm_stream_then_error(first_stream, error_body).await;
+    let (search_url, _s) = spawn_mock_search().await;
+    let exec_ctx = build_exec_ctx(&vllm_url, &search_url).await;
+    let request = serde_json::json!({
+        "model": "qwen3", "max_tokens": 1024, "stream": true,
+        "messages": [{"role": "user", "content": "Use web_search."}],
+        "tools": [{"name": "web_search", "description": "s", "input_schema": {"type": "object"}}]
+    });
+    let tools: Vec<ToolParam> = serde_json::from_value(request["tools"].clone()).unwrap();
+    let mut registry_tool_params = registry_tools(Some(&tools), &GatewayToolMap::default());
+    let mut gateway_executors = exec_ctx.gateway_executors.clone();
+    let registry = Arc::new(
+        ToolRegistry::build_with_handlers(&mut registry_tool_params, &mut gateway_executors)
+            .await
+            .unwrap(),
+    );
+
+    let stream = run_test_messages_stream(request, registry, Arc::clone(&exec_ctx)).await;
+    let sse = stream.collect::<Vec<_>>().await.join("");
+
+    assert_eq!(calls.load(Ordering::SeqCst), 2);
+    assert!(
+        sse.contains(&format!("event: error\ndata: {error_body}\n\n")),
+        "the original Anthropic error object should survive as the SSE error event: {sse}"
+    );
+}
+
+#[tokio::test]
+async fn messages_stream_forwards_upstream_sse_error_and_stops() {
+    let error = r#"{"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}"#;
+    let upstream_stream = format!(
+        "event: message_start\ndata: {{\"type\":\"message_start\",\"message\":{{\"id\":\"m\"}}}}\n\n\
+         event: error\ndata: {error}\n\n\
+         event: message_stop\ndata: {{\"type\":\"message_stop\"}}\n\n\
+         data: [DONE]\n\n"
+    );
+    let (vllm_url, upstream, _v) = spawn_mock_vllm_stream(vec![upstream_stream]).await;
+    let (search_url, _s) = spawn_mock_search().await;
+    let exec_ctx = build_exec_ctx(&vllm_url, &search_url).await;
+    let request = serde_json::json!({
+        "model": "qwen3", "max_tokens": 1024, "stream": true,
+        "messages": [{"role": "user", "content": "Use web_search."}],
+        "tools": [{"name": "web_search", "description": "s", "input_schema": {"type": "object"}}]
+    });
+    let tools: Vec<ToolParam> = serde_json::from_value(request["tools"].clone()).unwrap();
+    let mut registry_tool_params = registry_tools(Some(&tools), &GatewayToolMap::default());
+    let mut gateway_executors = exec_ctx.gateway_executors.clone();
+    let registry = Arc::new(
+        ToolRegistry::build_with_handlers(&mut registry_tool_params, &mut gateway_executors)
+            .await
+            .unwrap(),
+    );
+
+    let stream = run_test_messages_stream(request, registry, Arc::clone(&exec_ctx)).await;
+    let sse = stream.collect::<Vec<_>>().await.join("");
+
+    assert_eq!(upstream.calls.load(Ordering::SeqCst), 1);
+    assert!(sse.contains("event: error"), "error event missing: {sse}");
+    assert!(
+        sse.contains(r#""type":"overloaded_error""#),
+        "upstream error payload missing: {sse}"
+    );
+    assert!(
+        !sse.contains("event: message_stop"),
+        "an error must terminate without message_stop: {sse}"
+    );
+}
+
 // Multi-round streaming: replay the live-recorded multi-round streaming cassette
 // and assert the same single-lifecycle / contiguous-index / hidden-tool
 // invariants hold across a tool round + a final round.
@@ -231,7 +356,7 @@ async fn messages_stream_multiround_single_lifecycle() {
             .unwrap(),
     );
 
-    let stream = run_messages_stream(request, registry, Arc::clone(&exec_ctx), None);
+    let stream = run_test_messages_stream(request, registry, Arc::clone(&exec_ctx)).await;
     let sse = stream.collect::<Vec<_>>().await.join("");
 
     assert!(
@@ -302,7 +427,7 @@ async fn messages_stream_preserves_multi_block_system_across_rounds() {
             .unwrap(),
     );
 
-    let stream = run_messages_stream(request, registry, Arc::clone(&exec_ctx), None);
+    let stream = run_test_messages_stream(request, registry, Arc::clone(&exec_ctx)).await;
     let _chunks: Vec<String> = stream.collect().await;
 
     assert_eq!(
@@ -320,3 +445,58 @@ async fn messages_stream_preserves_multi_block_system_across_rounds() {
         "round 2 still carries the system blocks unchanged"
     );
 }
+
+#[tokio::test]
+async fn messages_stream_preserves_claude_code_cache_control_across_rounds() {
+    let streams = cassette_turn_streams()
+        .into_iter()
+        .map(|stream| stream.replace(r#""web_search""#, r#""WebSearch""#))
+        .collect();
+    let (vllm_url, upstream, _v) = spawn_mock_vllm_stream(streams).await;
+    let (search_url, _s) = spawn_mock_search().await;
+    let mut exec_ctx = build_exec_ctx(&vllm_url, &search_url).await;
+    Arc::get_mut(&mut exec_ctx).unwrap().messages_gateway_tools =
+        GatewayToolMap::from_pairs([("WebSearch", "web_search")]);
+
+    let mut request: Value = serde_json::from_str(CLAUDE_CODE_CACHE_CONTROL_REQUEST).unwrap();
+    request["stream"] = Value::Bool(true);
+    let original_system = request["system"].clone();
+    let original_user_message = request["messages"][0].clone();
+    let original_tools = request["tools"].clone();
+    let tools: Vec<ToolParam> = serde_json::from_value(original_tools.clone()).unwrap();
+    let mut registry_tool_params = registry_tools(Some(&tools), &exec_ctx.messages_gateway_tools);
+    let mut gateway_executors = exec_ctx.gateway_executors.clone();
+    let registry = Arc::new(
+        ToolRegistry::build_with_handlers(&mut registry_tool_params, &mut gateway_executors)
+            .await
+            .unwrap(),
+    );
+
+    let stream = run_test_messages_stream(request, registry, Arc::clone(&exec_ctx)).await;
+    let _chunks: Vec<String> = stream.collect().await;
+
+    let requests = upstream.requests.lock().await;
+    assert_eq!(requests.len(), 2, "tool round + final round");
+    for upstream_request in requests.iter() {
+        assert_eq!(upstream_request["system"], original_system);
+        assert_eq!(upstream_request["messages"][0], original_user_message);
+        assert_eq!(upstream_request["tools"], original_tools);
+    }
+    assert_eq!(
+        requests[1]["system"][0]["cache_control"],
+        serde_json::json!({"type": "ephemeral", "ttl": "1h"})
+    );
+    assert_eq!(
+        requests[1]["system"][1]["cache_control"],
+        serde_json::json!({"type": "ephemeral", "ttl": "5m"})
+    );
+    assert_eq!(
+        requests[1]["messages"][0]["content"][0]["cache_control"],
+        serde_json::json!({"type": "ephemeral", "ttl": "5m"})
+    );
+    assert_eq!(
+        requests[1]["tools"][1]["cache_control"],
+        serde_json::json!({"type": "ephemeral", "ttl": "1h"})
+    );
+    assert!(requests[1]["tools"][0].get("cache_control").is_none());
+}
diff --git a/crates/agentic-server/src/handler/common.rs b/crates/agentic-server/src/handler/common.rs
index 7d83efd8..a24ad606 100644
--- a/crates/agentic-server/src/handler/common.rs
+++ b/crates/agentic-server/src/handler/common.rs
@@ -85,12 +85,22 @@ pub(super) fn extract_bearer(headers: &HeaderMap, config_key: Option<&str>) -> O
 }
 
 pub(super) fn sse_response(stream: BoxStream) -> Response {
+    sse_response_with_headers(stream, HeaderMap::new())
+}
+
+pub(super) fn sse_response_with_headers(stream: BoxStream, mut headers: HeaderMap) -> Response {
     let byte_stream = stream.map(|line| Ok::<Bytes, std::convert::Infallible>(Bytes::from(line)));
-    Response::builder()
-        .status(StatusCode::OK)
-        .header("Content-Type", "text/event-stream; charset=utf-8")
-        .header("Cache-Control", "no-cache")
-        .header("X-Accel-Buffering", "no")
+    headers.insert(
+        http::header::CONTENT_TYPE,
+        http::HeaderValue::from_static("text/event-stream; charset=utf-8"),
+    );
+    headers.insert(http::header::CACHE_CONTROL, http::HeaderValue::from_static("no-cache"));
+    headers.insert("x-accel-buffering", http::HeaderValue::from_static("no"));
+    let mut builder = Response::builder().status(StatusCode::OK);
+    for (name, value) in &headers {
+        builder = builder.header(name, value);
+    }
+    builder
         .body(Body::from_stream(byte_stream))
         .expect("valid SSE response")
 }
diff --git a/crates/agentic-server/src/handler/http/messages.rs b/crates/agentic-server/src/handler/http/messages.rs
index 9b92b9fb..416cb330 100644
--- a/crates/agentic-server/src/handler/http/messages.rs
+++ b/crates/agentic-server/src/handler/http/messages.rs
@@ -6,12 +6,15 @@ use bytes::Bytes;
 use http::HeaderMap;
 use tracing::debug;
 
-use agentic_core::executor::{ExecutorError, run_messages_loop, run_messages_stream};
-use agentic_core::proxy::{ProxyAuth, ProxyRequest, error_response_for_auth, proxy_request_with_path};
+use agentic_core::executor::{ExecutorError, MessagesUpstream, run_messages_loop, run_messages_stream};
+use agentic_core::proxy::{
+    ProxyAuth, ProxyBody, ProxyRequest, ProxyResponse, error_response_for_auth, proxy_request_with_path,
+    upstream_request_headers,
+};
 use agentic_core::tool::ToolRegistry;
 use agentic_core::types::messages::{MessagesRequest, has_gateway_tool, registry_tools};
 
-use super::super::common::{convert_response, read_bytes_with_auth, sse_response};
+use super::super::common::{convert_response, read_bytes_with_auth, sse_response_with_headers};
 use crate::app::AppState;
 
 async fn proxy_messages(
@@ -35,30 +38,24 @@ async fn proxy_messages(
     )
 }
 
-/// Extract the client's Anthropic credential — `x-api-key` (Anthropic-native) or
-/// an `Authorization: Bearer` — falling back to the server's configured key.
-/// Consistent with the proxy path forwarding the client's `x-api-key` (E15).
-fn extract_client_key(headers: &HeaderMap, config_key: Option<&str>) -> Option<String> {
-    headers
-        .get("x-api-key")
-        .and_then(|v| v.to_str().ok())
-        .filter(|s| !s.is_empty())
-        .map(str::to_owned)
-        .or_else(|| {
-            headers
-                .get("authorization")
-                .and_then(|v| v.to_str().ok())
-                .and_then(|v| v.strip_prefix("Bearer "))
-                .filter(|s| !s.is_empty())
-                .map(str::to_owned)
-        })
-        .or_else(|| config_key.filter(|s| !s.is_empty()).map(str::to_owned))
-}
-
-/// Render an executor error as the Anthropic error envelope
-/// (`{"type":"error","error":{"type","message"}}`), consistent with the proxy
-/// path (E14).
-fn messages_error_response(err: &ExecutorError) -> Response {
+/// Preserve upstream Messages errors verbatim; render local executor failures
+/// as an Anthropic error envelope, consistent with the proxy path (E14).
+fn messages_error_response(err: ExecutorError) -> Response {
+    if let ExecutorError::LLMRequest {
+        status,
+        body,
+        mut headers,
+    } = err
+    {
+        headers
+            .entry(http::header::CONTENT_TYPE)
+            .or_insert(http::HeaderValue::from_static("application/json"));
+        return convert_response(ProxyResponse {
+            status,
+            headers,
+            body: ProxyBody::Full(Bytes::from(body)),
+        });
+    }
     convert_response(error_response_for_auth(
         err.http_status(),
         err.error_code(),
@@ -69,9 +66,13 @@ fn messages_error_response(err: &ExecutorError) -> Response {
 
 /// Drive the Messages-native gateway tool loop (non-streaming or streaming) for
 /// a request that declares a gateway-owned tool.
-async fn execute_messages(state: &AppState, headers: &HeaderMap, req: &MessagesRequest, body: &Bytes) -> Response {
-    let auth = extract_client_key(headers, state.openai_api_key.as_deref());
-
+async fn execute_messages(
+    state: &AppState,
+    headers: &HeaderMap,
+    query: Option<&str>,
+    req: &MessagesRequest,
+    body: &Bytes,
+) -> Response {
     // Build the request-scoped registry from the declared tools (M6). Gateway
     // ownership (incl. configured aliases like Claude Code's `WebSearch`) is
     // resolved against the operator-configured map.
@@ -80,23 +81,38 @@ async fn execute_messages(state: &AppState, headers: &HeaderMap, req: &MessagesR
     let mut executors = state.exec_ctx.gateway_executors.clone();
     let registry = match ToolRegistry::build_with_handlers(&mut tools, &mut executors).await {
         Ok(r) => r,
-        Err(e) => return messages_error_response(&ExecutorError::from(e)),
+        Err(e) => return messages_error_response(ExecutorError::from(e)),
     };
 
     // Parse the raw body to a JSON Value the loop forwards upstream untouched —
     // preserving every Anthropic field (tool_choice, stop_sequences, …).
     let request_json: serde_json::Value = match serde_json::from_slice(body) {
         Ok(v) => v,
-        Err(e) => return messages_error_response(&ExecutorError::from(e)),
+        Err(e) => return messages_error_response(ExecutorError::from(e)),
     };
 
+    let upstream = MessagesUpstream::new(
+        &state.exec_ctx.llm_base_url,
+        query,
+        upstream_request_headers(headers, &state.proxy_state.config, ProxyAuth::Anthropic),
+    );
     if req.stream {
-        let stream = run_messages_stream(request_json, Arc::new(registry), Arc::clone(&state.exec_ctx), auth);
-        sse_response(stream)
+        match run_messages_stream(request_json, Arc::new(registry), Arc::clone(&state.exec_ctx), upstream).await {
+            Ok(response) => sse_response_with_headers(response.body, response.headers),
+            Err(e) => messages_error_response(e),
+        }
     } else {
-        match run_messages_loop(request_json, &registry, &state.exec_ctx, auth.as_deref()).await {
-            Ok(message) => axum::Json(message).into_response(),
-            Err(e) => messages_error_response(&e),
+        match run_messages_loop(request_json, &registry, &state.exec_ctx, &upstream).await {
+            Ok(message) => {
+                let mut response = axum::Json(message.body).into_response();
+                response.headers_mut().extend(message.headers);
+                response.headers_mut().insert(
+                    http::header::CONTENT_TYPE,
+                    http::HeaderValue::from_static("application/json"),
+                );
+                response
+            }
+            Err(e) => messages_error_response(e),
         }
     }
 }
@@ -119,7 +135,7 @@ pub async fn messages(State(state): State<AppState>, request: Request) -> Respon
             "routing HTTP messages request"
         );
         if route_to_loop {
-            return execute_messages(&state, &parts.headers, &req, &bytes).await;
+            return execute_messages(&state, &parts.headers, parts.uri.query(), &req, &bytes).await;
         }
     }
 
diff --git a/crates/agentic-server/tests/messages_test.rs b/crates/agentic-server/tests/messages_test.rs
index f39f344c..6817b82e 100644
--- a/crates/agentic-server/tests/messages_test.rs
+++ b/crates/agentic-server/tests/messages_test.rs
@@ -14,6 +14,9 @@ use tokio::sync::Mutex;
 
 use common::{spawn_gateway, test_config, test_state};
 
+const CLAUDE_CODE_CACHE_CONTROL_REQUEST: &[u8] =
+    include_bytes!("../../agentic-server-core/tests/fixtures/claude-code-cache-control-request.json");
+
 #[derive(Clone, Debug)]
 struct RecordedRequest {
     uri: String,
@@ -25,27 +28,39 @@ async fn spawn_recording_upstream(
     status: StatusCode,
     content_type: &'static str,
     response_body: &'static str,
+) -> (String, Arc<Mutex<Vec<RecordedRequest>>>, tokio::task::JoinHandle<()>) {
+    spawn_recording_upstream_with_headers(status, content_type, response_body, HeaderMap::new()).await
+}
+
+async fn spawn_recording_upstream_with_headers(
+    status: StatusCode,
+    content_type: &'static str,
+    response_body: &'static str,
+    response_headers: HeaderMap,
 ) -> (String, Arc<Mutex<Vec<RecordedRequest>>>, tokio::task::JoinHandle<()>) {
     let requests = Arc::new(Mutex::new(Vec::new()));
     let route_requests = Arc::clone(&requests);
     let count_tokens_requests = Arc::clone(&requests);
+    let route_response_headers = response_headers.clone();
     let app = Router::new()
         .route(
             "/v1/messages",
             post(move |OriginalUri(uri), headers: HeaderMap, body: Bytes| {
                 let route_requests = Arc::clone(&route_requests);
+                let response_headers = route_response_headers.clone();
                 async move {
                     route_requests.lock().await.push(RecordedRequest {
                         uri: uri.to_string(),
                         headers,
                         body,
                     });
-                    Response::builder()
+                    let mut response = Response::builder()
                         .status(status)
                         .header("content-type", content_type)
                         .body(axum::body::Body::from(response_body))
-                        .unwrap()
-                        .into_response()
+                        .unwrap();
+                    response.headers_mut().extend(response_headers);
+                    response.into_response()
                 }
             }),
         )
@@ -53,18 +68,20 @@ async fn spawn_recording_upstream(
             "/v1/messages/count_tokens",
             post(move |OriginalUri(uri), headers: HeaderMap, body: Bytes| {
                 let route_requests = Arc::clone(&count_tokens_requests);
+                let response_headers = response_headers.clone();
                 async move {
                     route_requests.lock().await.push(RecordedRequest {
                         uri: uri.to_string(),
                         headers,
                         body,
                     });
-                    Response::builder()
+                    let mut response = Response::builder()
                         .status(status)
                         .header("content-type", content_type)
                         .body(axum::body::Body::from(response_body))
-                        .unwrap()
-                        .into_response()
+                        .unwrap();
+                    response.headers_mut().extend(response_headers);
+                    response.into_response()
                 }
             }),
         );
@@ -136,6 +153,25 @@ async fn messages_forwards_system_attribution_blocks_verbatim() {
     assert_eq!(requests[0].body.as_ref(), body);
 }
 
+#[tokio::test]
+async fn messages_proxy_preserves_claude_code_cache_control_body_verbatim() {
+    let (llm_url, requests, _upstream) =
+        spawn_recording_upstream(StatusCode::OK, "application/json", r#"{"id":"msg_cache"}"#).await;
+    let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&llm_url))).await;
+
+    let response = reqwest::Client::new()
+        .post(format!("{gateway_url}/v1/messages"))
+        .body(CLAUDE_CODE_CACHE_CONTROL_REQUEST.to_vec())
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::OK);
+    let requests = requests.lock().await;
+    assert_eq!(requests.len(), 1);
+    assert_eq!(requests[0].body.as_ref(), CLAUDE_CODE_CACHE_CONTROL_REQUEST);
+}
+
 #[tokio::test]
 async fn messages_forwards_sse_bytes_unchanged() {
     let sse = "event: message_start\ndata: {\"type\":\"message_start\"}\n\nevent: message_stop\ndata: {\"type\":\"message_stop\"}\n\n";
@@ -196,6 +232,55 @@ async fn messages_preserves_upstream_error_status_and_body() {
     );
 }
 
+#[tokio::test]
+async fn messages_gateway_loop_preserves_upstream_error_status_and_body() {
+    let upstream_error = r#"{"type":"error","error":{"type":"invalid_request_error","message":"bad thinking field"}}"#;
+    let mut upstream_headers = HeaderMap::new();
+    upstream_headers.insert("request-id", "req_error".parse().unwrap());
+    upstream_headers.insert("retry-after", "7".parse().unwrap());
+    let (llm_url, _requests, _upstream) = spawn_recording_upstream_with_headers(
+        StatusCode::BAD_REQUEST,
+        "application/json",
+        upstream_error,
+        upstream_headers,
+    )
+    .await;
+    let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&llm_url))).await;
+    let body = br#"{"model":"qwen3","max_tokens":256,"stream":false,"messages":[{"role":"user","content":"search"}],"tools":[{"name":"web_search","input_schema":{"type":"object"}}]}"#;
+
+    let response = reqwest::Client::new()
+        .post(format!("{gateway_url}/v1/messages"))
+        .body(body.to_vec())
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(response.headers()["request-id"], "req_error");
+    assert_eq!(response.headers()["retry-after"], "7");
+    assert_eq!(response.text().await.unwrap(), upstream_error);
+}
+
+#[tokio::test]
+async fn messages_gateway_stream_preserves_initial_upstream_error_status_and_body() {
+    let upstream_error =
+        r#"{"type":"error","error":{"type":"invalid_request_error","message":"unsupported adaptive thinking"}}"#;
+    let (llm_url, _requests, _upstream) =
+        spawn_recording_upstream(StatusCode::BAD_REQUEST, "application/json", upstream_error).await;
+    let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&llm_url))).await;
+    let body = br#"{"model":"qwen3","max_tokens":256,"stream":true,"messages":[{"role":"user","content":"search"}],"tools":[{"name":"web_search","input_schema":{"type":"object"}}]}"#;
+
+    let response = reqwest::Client::new()
+        .post(format!("{gateway_url}/v1/messages"))
+        .body(body.to_vec())
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(response.text().await.unwrap(), upstream_error);
+}
+
 #[tokio::test]
 async fn messages_returns_anthropic_error_for_unreachable_upstream() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -224,6 +309,34 @@ async fn messages_returns_anthropic_error_for_unreachable_upstream() {
     );
 }
 
+#[tokio::test]
+async fn messages_gateway_loop_returns_anthropic_error_for_unreachable_upstream() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let dead_addr = listener.local_addr().unwrap();
+    drop(listener);
+    let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&format!("http://{dead_addr}")))).await;
+    let body = br#"{"model":"qwen3","max_tokens":256,"stream":false,"messages":[{"role":"user","content":"search"}],"tools":[{"name":"web_search","input_schema":{"type":"object"}}]}"#;
+
+    let response = reqwest::Client::new()
+        .post(format!("{gateway_url}/v1/messages"))
+        .body(body.to_vec())
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
+    assert_eq!(
+        response.json::<serde_json::Value>().await.unwrap(),
+        serde_json::json!({
+            "type": "error",
+            "error": {
+                "type": "api_error",
+                "message": "LLM unavailable",
+            },
+        })
+    );
+}
+
 /// Mock vLLM `/v1/messages` that returns a canned Anthropic message and records
 /// how many times it was called (to prove routing).
 async fn spawn_mock_vllm_messages(body: &'static str) -> (String, Arc<Mutex<usize>>, tokio::task::JoinHandle<()>) {
@@ -277,6 +390,94 @@ async fn messages_with_web_search_tool_routes_to_native_loop() {
     assert_eq!(json["stop_reason"], "end_turn");
 }
 
+#[tokio::test]
+async fn messages_gateway_loop_forwards_query_and_open_headers() {
+    let final_msg = r#"{"id":"m","type":"message","role":"assistant","model":"qwen3","content":[{"type":"text","text":"done"}],"stop_reason":"end_turn","usage":{"input_tokens":5,"output_tokens":1}}"#;
+    let mut upstream_headers = HeaderMap::new();
+    upstream_headers.insert("request-id", "req_terminal".parse().unwrap());
+    upstream_headers.insert("anthropic-ratelimit-requests-remaining", "41".parse().unwrap());
+    let (llm_url, requests, _upstream) =
+        spawn_recording_upstream_with_headers(StatusCode::OK, "application/json", final_msg, upstream_headers).await;
+    let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&llm_url))).await;
+    let body = br#"{"model":"qwen3","max_tokens":256,"stream":false,"messages":[{"role":"user","content":"search"}],"tools":[{"name":"web_search","input_schema":{"type":"object"}}]}"#;
+
+    let response = reqwest::Client::new()
+        .post(format!("{gateway_url}/v1/messages?beta=true"))
+        .header("anthropic-version", "2023-06-01")
+        .header("anthropic-beta", "future-beta-unknown,interleaved-thinking-2025-05-14")
+        .header("x-claude-code-session-id", "session-loop")
+        .header("x-claude-code-agent-id", "agent-loop")
+        .header("x-api-key", "anthropic-key")
+        .body(body.to_vec())
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::OK);
+    assert_eq!(response.headers()["request-id"], "req_terminal");
+    assert_eq!(response.headers()["anthropic-ratelimit-requests-remaining"], "41");
+    let requests = requests.lock().await;
+    assert_eq!(requests.len(), 1);
+    assert_eq!(requests[0].uri, "/v1/messages?beta=true");
+    assert_eq!(requests[0].headers["anthropic-version"], "2023-06-01");
+    assert_eq!(
+        requests[0].headers["anthropic-beta"],
+        "future-beta-unknown,interleaved-thinking-2025-05-14"
+    );
+    assert_eq!(requests[0].headers["x-claude-code-session-id"], "session-loop");
+    assert_eq!(requests[0].headers["x-claude-code-agent-id"], "agent-loop");
+    assert_eq!(requests[0].headers["x-api-key"], "anthropic-key");
+    assert!(!requests[0].headers.contains_key("authorization"));
+}
+
+#[tokio::test]
+async fn messages_gateway_stream_forwards_query_and_open_headers() {
+    let final_sse = concat!(
+        "event: message_start\n",
+        "data: {\"type\":\"message_start\",\"message\":{\"id\":\"m\",\"type\":\"message\",\"role\":\"assistant\",\"model\":\"qwen3\",\"content\":[],\"stop_reason\":null,\"stop_sequence\":null,\"usage\":{\"input_tokens\":1,\"output_tokens\":0}}}\n\n",
+        "event: content_block_start\n",
+        "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
+        "event: content_block_delta\n",
+        "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"done\"}}\n\n",
+        "event: content_block_stop\n",
+        "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
+        "event: message_delta\n",
+        "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",\"stop_sequence\":null},\"usage\":{\"output_tokens\":1}}\n\n",
+        "event: message_stop\n",
+        "data: {\"type\":\"message_stop\"}\n\n",
+    );
+    let mut upstream_headers = HeaderMap::new();
+    upstream_headers.insert("request-id", "req_stream".parse().unwrap());
+    upstream_headers.insert("anthropic-ratelimit-tokens-remaining", "900".parse().unwrap());
+    let (llm_url, requests, _upstream) =
+        spawn_recording_upstream_with_headers(StatusCode::OK, "text/event-stream", final_sse, upstream_headers).await;
+    let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&llm_url))).await;
+    let body = br#"{"model":"qwen3","max_tokens":256,"stream":true,"messages":[{"role":"user","content":"search"}],"tools":[{"name":"web_search","input_schema":{"type":"object"}}]}"#;
+
+    let response = reqwest::Client::new()
+        .post(format!("{gateway_url}/v1/messages?beta=true"))
+        .header("anthropic-version", "2023-06-01")
+        .header("anthropic-beta", "future-beta-unknown")
+        .header("x-claude-code-session-id", "session-stream")
+        .header("x-api-key", "anthropic-key")
+        .body(body.to_vec())
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::OK);
+    assert_eq!(response.headers()["request-id"], "req_stream");
+    assert_eq!(response.headers()["anthropic-ratelimit-tokens-remaining"], "900");
+    assert!(response.text().await.unwrap().contains("done"));
+    let requests = requests.lock().await;
+    assert_eq!(requests.len(), 1);
+    assert_eq!(requests[0].uri, "/v1/messages?beta=true");
+    assert_eq!(requests[0].headers["anthropic-version"], "2023-06-01");
+    assert_eq!(requests[0].headers["anthropic-beta"], "future-beta-unknown");
+    assert_eq!(requests[0].headers["x-claude-code-session-id"], "session-stream");
+    assert_eq!(requests[0].headers["x-api-key"], "anthropic-key");
+}
+
 // A request with NO gateway-owned tool stays on the transparent proxy — the
 // native loop is never engaged.
 #[tokio::test]
diff --git a/scripts/claude-code-smoke.sh b/scripts/claude-code-smoke.sh
new file mode 100755
index 00000000..31758e67
--- /dev/null
+++ b/scripts/claude-code-smoke.sh
@@ -0,0 +1,142 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+CLAUDE_BIN="${CLAUDE_BIN:-claude}"
+AGENTIC_SERVER_BIN="${AGENTIC_SERVER_BIN:-target/debug/agentic-server}"
+PYTHON_BIN="${PYTHON_BIN:-python3}"
+CASSETTE="${CASSETTE:-crates/agentic-server-core/tests/cassettes/messages/messages-web-search-Qwen-Qwen3-30B-A3B-FP8-streaming.yaml}"
+
+choose_port() {
+  "$PYTHON_BIN" -c 'import socket; sock = socket.socket(); sock.bind(("127.0.0.1", 0)); print(sock.getsockname()[1]); sock.close()'
+}
+
+REPLAY_PORT="${REPLAY_PORT:-$(choose_port)}"
+GATEWAY_PORT="${GATEWAY_PORT:-$(choose_port)}"
+
+if ! command -v "$CLAUDE_BIN" >/dev/null 2>&1; then
+  echo "error: Claude Code is not installed: ${CLAUDE_BIN}" >&2
+  exit 2
+fi
+if [[ ! -x "$AGENTIC_SERVER_BIN" ]]; then
+  echo "error: agentic-server is not executable: ${AGENTIC_SERVER_BIN}; run cargo build -p agentic-server" >&2
+  exit 2
+fi
+if [[ ! -f "$CASSETTE" ]]; then
+  echo "error: Messages cassette not found: ${CASSETTE}" >&2
+  exit 2
+fi
+
+temp_dir="$(mktemp -d)"
+capture_path="${temp_dir}/capture.jsonl"
+replay_log="${temp_dir}/replay.log"
+gateway_log="${temp_dir}/gateway.log"
+claude_output="${temp_dir}/claude.json"
+claude_debug="${temp_dir}/claude-debug.log"
+claude_config="${temp_dir}/claude-config"
+replay_pid=""
+gateway_pid=""
+mkdir -p "$claude_config"
+
+cleanup() {
+  local status=$?
+  trap - EXIT INT TERM
+  if [[ -n "$gateway_pid" ]]; then
+    kill "$gateway_pid" >/dev/null 2>&1 || true
+    wait "$gateway_pid" >/dev/null 2>&1 || true
+  fi
+  if [[ -n "$replay_pid" ]]; then
+    kill "$replay_pid" >/dev/null 2>&1 || true
+    wait "$replay_pid" >/dev/null 2>&1 || true
+  fi
+  if [[ "$status" -ne 0 ]]; then
+    echo "--- replay server log ---" >&2
+    sed -n '1,240p' "$replay_log" >&2 || true
+    echo "--- agentic-server log ---" >&2
+    sed -n '1,240p' "$gateway_log" >&2 || true
+    echo "--- Claude Code output ---" >&2
+    sed -n '1,240p' "$claude_output" >&2 || true
+    echo "--- Claude Code debug log ---" >&2
+    sed -n '1,240p' "$claude_debug" >&2 || true
+    echo "--- replay capture ---" >&2
+    sed -n '1,240p' "$capture_path" >&2 || true
+  fi
+  rm -r "$temp_dir"
+  exit "$status"
+}
+trap cleanup EXIT INT TERM
+
+wait_until_ready() {
+  local label="$1"
+  local url="$2"
+  for attempt in $(seq 1 60); do
+    if curl --connect-timeout 1 --max-time 2 --fail --silent "$url" >/dev/null; then
+      return 0
+    fi
+    echo "${label} not ready (attempt ${attempt}/60)"
+    sleep 1
+  done
+  echo "error: ${label} did not become ready" >&2
+  return 1
+}
+
+"$PYTHON_BIN" scripts/claude_code_replay_server.py serve \
+  --cassette "$CASSETTE" \
+  --capture "$capture_path" \
+  --port "$REPLAY_PORT" \
+  >"$replay_log" 2>&1 &
+replay_pid=$!
+wait_until_ready "replay server" "http://127.0.0.1:${REPLAY_PORT}/health"
+
+env \
+  LLM_API_BASE="http://127.0.0.1:${REPLAY_PORT}" \
+  GATEWAY_HOST=127.0.0.1 \
+  GATEWAY_PORT="$GATEWAY_PORT" \
+  SKIP_LLM_READY_CHECK=true \
+  DATABASE_URL="sqlite://${temp_dir}/agentic.db" \
+  MESSAGES_GATEWAY_TOOL_ALIASES=WebSearch=web_search \
+  YOU_API_KEY=ci-placeholder \
+  YOU_API_BASE_URL="http://127.0.0.1:${REPLAY_PORT}" \
+  "$AGENTIC_SERVER_BIN" \
+  >"$gateway_log" 2>&1 &
+gateway_pid=$!
+wait_until_ready "agentic-server" "http://127.0.0.1:${GATEWAY_PORT}/ready"
+
+prompt="Use WebSearch to find the latest stable Rust release, then answer with its version only."
+env \
+  -u CLAUDE_CODE_USE_VERTEX \
+  -u ANTHROPIC_VERTEX_PROJECT_ID \
+  -u CLAUDE_CODE_USE_BEDROCK \
+  -u CLAUDE_CODE_USE_FOUNDRY \
+  ANTHROPIC_BASE_URL="http://127.0.0.1:${GATEWAY_PORT}" \
+  ANTHROPIC_API_KEY=ci-placeholder \
+  ANTHROPIC_MODEL=qwen3 \
+  ANTHROPIC_DEFAULT_OPUS_MODEL=qwen3 \
+  ANTHROPIC_DEFAULT_SONNET_MODEL=qwen3 \
+  ANTHROPIC_DEFAULT_HAIKU_MODEL=qwen3 \
+  CLAUDE_CONFIG_DIR="$claude_config" \
+  DISABLE_AUTOUPDATER=1 \
+  CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 \
+  "$CLAUDE_BIN" \
+  --safe-mode \
+  --print \
+  "$prompt" \
+  --output-format json \
+  --debug-file "$claude_debug" \
+  --no-session-persistence \
+  --permission-mode dontAsk \
+  --model qwen3 \
+  --tools WebSearch \
+  --allowedTools WebSearch \
+  >"$claude_output"
+
+"$PYTHON_BIN" - "$claude_output" <<'PY'
+import json
+import sys
+
+result = json.load(open(sys.argv[1]))
+assert result.get("is_error") is False, result
+assert "1.89.0" in result.get("result", ""), result
+print(f"Claude Code result: {result['result']}")
+PY
+
+"$PYTHON_BIN" scripts/claude_code_replay_server.py assert-capture --capture "$capture_path"
diff --git a/scripts/claude_code_replay_server.py b/scripts/claude_code_replay_server.py
new file mode 100755
index 00000000..906aafae
--- /dev/null
+++ b/scripts/claude_code_replay_server.py
@@ -0,0 +1,268 @@
+#!/usr/bin/env python3
+"""Replay recorded vLLM Messages turns for the Claude Code CI acceptance test."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import threading
+from dataclasses import dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlsplit
+
+import yaml
+
+
+@dataclass(frozen=True)
+class ReplayTurn:
+    status_code: int
+    content_type: str
+    body: bytes
+
+
+def load_turns(path: Path) -> list[ReplayTurn]:
+    document = yaml.safe_load(path.read_text())
+    return [
+        ReplayTurn(
+            status_code=turn["response"]["status_code"],
+            content_type=turn["response"]["headers"]["content-type"],
+            body="".join(turn["response"]["sse"]).encode(),
+        )
+        for turn in document["turns"]
+    ]
+
+
+def adapt_stream(stream: str, declared_tool_name: str) -> str:
+    return stream.replace('"name":"web_search"', f'"name":"{declared_tool_name}"')
+
+
+def append_capture(path: Path, kind: str, body: dict[str, Any]) -> None:
+    with path.open("a") as capture:
+        capture.write(json.dumps({"kind": kind, "body": body}, separators=(",", ":")) + "\n")
+
+
+def load_capture(path: Path) -> list[dict[str, Any]]:
+    return [json.loads(line) for line in path.read_text().splitlines() if line]
+
+
+def _content_blocks(request: dict[str, Any]) -> list[dict[str, Any]]:
+    blocks: list[dict[str, Any]] = []
+    for message in request.get("messages", []):
+        content = message.get("content", [])
+        if isinstance(content, list):
+            blocks.extend(block for block in content if isinstance(block, dict))
+    return blocks
+
+
+def _cache_breakpoint_ttls(request: dict[str, Any]) -> list[str]:
+    effective_blocks: list[dict[str, Any]] = []
+    effective_blocks.extend(tool for tool in request.get("tools", []) if isinstance(tool, dict))
+    system = request.get("system", [])
+    if isinstance(system, list):
+        effective_blocks.extend(block for block in system if isinstance(block, dict))
+    effective_blocks.extend(_content_blocks(request))
+
+    ttls: list[str] = []
+    for block in effective_blocks:
+        cache_control = block.get("cache_control")
+        if isinstance(cache_control, dict):
+            ttls.append(cache_control.get("ttl", "5m"))
+    return ttls
+
+
+def validate_capture(records: list[dict[str, Any]]) -> None:
+    messages = [record["body"] for record in records if record["kind"] == "messages"]
+    transports = [record["body"] for record in records if record["kind"] == "messages_transport"]
+    searches = [record["body"] for record in records if record["kind"] == "search"]
+    assert len(messages) == 2, f"expected two Messages rounds, got {len(messages)}"
+    assert len(transports) == 2, f"expected transport capture for both Messages rounds, got {len(transports)}"
+    assert len(searches) == 1, f"expected one search request, got {len(searches)}"
+
+    session_ids = set()
+    for transport in transports:
+        assert transport["path"] == "/v1/messages?beta=true", transport
+        headers = transport["headers"]
+        assert headers.get("anthropic-version") == "2023-06-01", headers
+        assert headers.get("anthropic-beta"), headers
+        assert headers.get("x-api-key") == "ci-placeholder", headers
+        session_id = headers.get("x-claude-code-session-id")
+        assert session_id, headers
+        session_ids.add(session_id)
+    assert len(session_ids) == 1, "expected one Claude Code session ID across gateway tool rounds"
+
+    system = messages[0].get("system")
+    assert isinstance(system, list) and len(system) >= 2, "expected a multi-block system prompt"
+    cached_system_blocks = [block for block in system if "cache_control" in block]
+    assert len(cached_system_blocks) >= 2, "expected cache_control on at least two system blocks"
+    assert messages[1].get("system") == system, "expected the system prompt to survive the gateway tool round"
+
+    cache_ttls = _cache_breakpoint_ttls(messages[0])
+    assert len(cache_ttls) <= 4, f"expected at most four cache breakpoints, got {len(cache_ttls)}"
+    assert all(ttl in {"1h", "5m"} for ttl in cache_ttls), f"unexpected cache TTLs: {cache_ttls}"
+    seen_short_ttl = False
+    for ttl in cache_ttls:
+        if ttl == "5m":
+            seen_short_ttl = True
+        assert not (ttl == "1h" and seen_short_ttl), "1h cache breakpoints must precede 5m breakpoints"
+
+    assert any(
+        block.get("type") == "text" and "cache_control" in block for block in _content_blocks(messages[0])
+    ), "expected cache_control on the user prompt"
+
+    tools = messages[0].get("tools", [])
+    web_search = next((tool for tool in tools if tool.get("name") == "WebSearch"), None)
+    assert web_search is not None, "expected Claude Code to declare WebSearch"
+    assert messages[1].get("tools") == tools, "expected tool declarations to survive the gateway tool round"
+
+    assert any(
+        block.get("type") == "tool_result" for block in _content_blocks(messages[1])
+    ), "expected a tool_result in the second Messages round"
+    assert searches[0].get("query"), "expected a non-empty search query"
+
+
+@dataclass
+class ReplayState:
+    turns: list[ReplayTurn]
+    capture_path: Path
+    next_turn: int = 0
+
+    def __post_init__(self) -> None:
+        self.lock = threading.Lock()
+
+    def take_turn(self) -> ReplayTurn | None:
+        with self.lock:
+            if self.next_turn >= len(self.turns):
+                return None
+            turn = self.turns[self.next_turn]
+            self.next_turn += 1
+            return turn
+
+
+def _declared_web_search_name(request: dict[str, Any]) -> str:
+    for tool in request.get("tools", []):
+        if tool.get("name") == "WebSearch":
+            return "WebSearch"
+    return "web_search"
+
+
+def make_handler(state: ReplayState) -> type[BaseHTTPRequestHandler]:
+    class ReplayHandler(BaseHTTPRequestHandler):
+        def do_GET(self) -> None:
+            if urlsplit(self.path).path == "/health":
+                self._send_bytes(200, "text/plain", b"")
+                return
+            self.send_error(404)
+
+        def do_POST(self) -> None:
+            path = urlsplit(self.path).path
+            request = self._read_json()
+            if request is None:
+                return
+
+            if path == "/v1/messages/count_tokens":
+                self._send_json(200, {"input_tokens": 512})
+                return
+            if path == "/v1/search":
+                append_capture(state.capture_path, "search", request)
+                self._send_json(
+                    200,
+                    {
+                        "results": {
+                            "web": [
+                                {
+                                    "url": "https://www.rust-lang.org/",
+                                    "title": "Rust",
+                                    "description": "Rust language release",
+                                    "snippets": ["Rust 1.89.0 is the latest stable release."],
+                                }
+                            ],
+                            "news": [],
+                        },
+                        "metadata": {"query": request.get("query", ""), "search_uuid": "ci-search", "latency": 0.0},
+                    },
+                )
+                return
+            if path != "/v1/messages":
+                self.send_error(404)
+                return
+
+            append_capture(
+                state.capture_path,
+                "messages_transport",
+                {
+                    "path": self.path,
+                    "headers": {name.lower(): value for name, value in self.headers.items()},
+                },
+            )
+            append_capture(state.capture_path, "messages", request)
+            turn = state.take_turn()
+            if turn is None:
+                self._send_json(409, {"error": {"type": "api_error", "message": "cassette exhausted"}})
+                return
+            body = adapt_stream(turn.body.decode(), _declared_web_search_name(request)).encode()
+            self._send_bytes(turn.status_code, turn.content_type, body)
+
+        def _read_json(self) -> dict[str, Any] | None:
+            try:
+                content_length = int(self.headers.get("Content-Length", "0"))
+                body = json.loads(self.rfile.read(content_length))
+            except (ValueError, json.JSONDecodeError):
+                self.send_error(400, "request body must be valid JSON")
+                return None
+            if not isinstance(body, dict):
+                self.send_error(400, "request body must be a JSON object")
+                return None
+            return body
+
+        def _send_json(self, status: int, body: dict[str, Any]) -> None:
+            self._send_bytes(status, "application/json", json.dumps(body, separators=(",", ":")).encode())
+
+        def _send_bytes(self, status: int, content_type: str, body: bytes) -> None:
+            self.send_response(status)
+            self.send_header("Content-Type", content_type)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, format: str, *args: object) -> None:
+            return
+
+    return ReplayHandler
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    serve = subparsers.add_parser("serve")
+    serve.add_argument("--cassette", required=True, type=Path)
+    serve.add_argument("--port", required=True, type=int)
+    serve.add_argument("--capture", required=True, type=Path)
+
+    assert_capture = subparsers.add_parser("assert-capture")
+    assert_capture.add_argument("--capture", required=True, type=Path)
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    if args.command == "assert-capture":
+        records = load_capture(args.capture)
+        validate_capture(records)
+        messages = sum(record["kind"] == "messages" for record in records)
+        transports = sum(record["kind"] == "messages_transport" for record in records)
+        searches = sum(record["kind"] == "search" for record in records)
+        print(f"capture valid: messages={messages} transports={transports} searches={searches}")
+        return
+
+    args.capture.parent.mkdir(parents=True, exist_ok=True)
+    args.capture.write_text("")
+    state = ReplayState(load_turns(args.cassette), args.capture)
+    server = ThreadingHTTPServer(("127.0.0.1", args.port), make_handler(state))
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/test_claude_code_replay_server.py b/scripts/test_claude_code_replay_server.py
new file mode 100644
index 00000000..942b9cab
--- /dev/null
+++ b/scripts/test_claude_code_replay_server.py
@@ -0,0 +1,145 @@
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+import claude_code_replay_server as replay
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parent.parent
+CASSETTE = (
+    REPOSITORY_ROOT
+    / "crates/agentic-server-core/tests/cassettes/messages/"
+    "messages-web-search-Qwen-Qwen3-30B-A3B-FP8-streaming.yaml"
+)
+
+
+def cache_control(ttl: str = "5m") -> dict[str, str]:
+    return {"type": "ephemeral", "ttl": ttl}
+
+
+def messages_request(*, with_tool_result: bool = False) -> dict:
+    messages = [
+        {
+            "role": "user",
+            "content": [{"type": "text", "text": "Find Rust.", "cache_control": cache_control()}],
+        }
+    ]
+    if with_tool_result:
+        messages.extend(
+            [
+                {
+                    "role": "assistant",
+                    "content": [{"type": "tool_use", "id": "tool-1", "name": "WebSearch", "input": {}}],
+                },
+                {
+                    "role": "user",
+                    "content": [{"type": "tool_result", "tool_use_id": "tool-1", "content": "Rust 1.89.0"}],
+                },
+            ]
+        )
+    return {
+        "model": "qwen3",
+        "stream": True,
+        "system": [
+            {"type": "text", "text": "attribution"},
+            {"type": "text", "text": "instructions", "cache_control": cache_control("1h")},
+            {"type": "text", "text": "runtime", "cache_control": cache_control()},
+        ],
+        "messages": messages,
+        "tools": [
+            {
+                "name": "WebSearch",
+                "description": "Search the web",
+                "input_schema": {"type": "object", "properties": {}},
+            }
+        ],
+    }
+
+
+def messages_transport(**header_overrides: str) -> dict:
+    headers = {
+        "anthropic-version": "2023-06-01",
+        "anthropic-beta": "interleaved-thinking-2025-05-14",
+        "x-api-key": "ci-placeholder",
+        "x-claude-code-session-id": "session-1",
+    }
+    headers.update(header_overrides)
+    return {"path": "/v1/messages?beta=true", "headers": headers}
+
+
+def capture_records(first_request: dict, second_request: dict) -> list[dict]:
+    return [
+        {"kind": "messages_transport", "body": messages_transport()},
+        {"kind": "messages", "body": first_request},
+        {"kind": "search", "body": {"query": "latest stable Rust release"}},
+        {"kind": "messages_transport", "body": messages_transport()},
+        {"kind": "messages", "body": second_request},
+    ]
+
+
+class ReplayServerTests(unittest.TestCase):
+    def test_load_turns_reads_recorded_streams(self) -> None:
+        turns = replay.load_turns(CASSETTE)
+
+        self.assertEqual(len(turns), 2)
+        self.assertEqual(turns[0].status_code, 200)
+        self.assertEqual(turns[0].content_type, "text/event-stream; charset=utf-8")
+        self.assertIn(b"event: message_start", turns[0].body)
+
+    def test_adapt_stream_uses_declared_claude_tool_name(self) -> None:
+        recorded = 'data: {"content_block":{"name":"web_search"}}\n\n'
+
+        adapted = replay.adapt_stream(recorded, "WebSearch")
+
+        self.assertIn('"name":"WebSearch"', adapted)
+        self.assertNotIn('"name":"web_search"', adapted)
+
+    def test_validate_capture_accepts_claude_code_wire_shape_and_tool_round(self) -> None:
+        records = capture_records(messages_request(), messages_request(with_tool_result=True))
+
+        replay.validate_capture(records)
+
+    def test_validate_capture_rejects_more_than_four_cache_breakpoints(self) -> None:
+        first = messages_request()
+        second = messages_request(with_tool_result=True)
+        for request in (first, second):
+            request["tools"][0]["cache_control"] = cache_control("1h")
+            request["system"].append(
+                {"type": "text", "text": "extra runtime", "cache_control": cache_control()}
+            )
+
+        with self.assertRaisesRegex(AssertionError, "at most four cache breakpoints"):
+            replay.validate_capture(capture_records(first, second))
+
+    def test_validate_capture_rejects_short_ttl_before_long_ttl(self) -> None:
+        first = messages_request()
+        second = messages_request(with_tool_result=True)
+        for request in (first, second):
+            request["tools"][0]["cache_control"] = cache_control()
+
+        with self.assertRaisesRegex(AssertionError, "1h cache breakpoints must precede 5m"):
+            replay.validate_capture(capture_records(first, second))
+
+    def test_validate_capture_rejects_missing_claude_transport_header(self) -> None:
+        first_transport = messages_transport()
+        del first_transport["headers"]["anthropic-beta"]
+        records = capture_records(messages_request(), messages_request(with_tool_result=True))
+        records[0]["body"] = first_transport
+
+        with self.assertRaises(AssertionError):
+            replay.validate_capture(records)
+
+    def test_validate_capture_requires_two_messages_rounds(self) -> None:
+        records = [
+            {"kind": "messages", "body": messages_request()},
+            {"kind": "search", "body": {"query": "latest stable Rust release"}},
+        ]
+
+        with self.assertRaisesRegex(AssertionError, "two Messages rounds"):
+            replay.validate_capture(records)
+
+
+if __name__ == "__main__":
+    unittest.main()

From f86a76f0e67f94b15eef924bc69b8c60440b3b0a Mon Sep 17 00:00:00 2001
From: Francisco Javier Arceo <farceo@redhat.com>
Date: Mon, 3 Aug 2026 23:22:28 -0400
Subject: [PATCH 2/2] fix: drop accept-encoding for messages gateway loop

Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
---
 crates/agentic-server-core/src/proxy.rs      | 3 ++-
 crates/agentic-server/tests/messages_test.rs | 2 ++
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/crates/agentic-server-core/src/proxy.rs b/crates/agentic-server-core/src/proxy.rs
index da982670..d3a79e9f 100644
--- a/crates/agentic-server-core/src/proxy.rs
+++ b/crates/agentic-server-core/src/proxy.rs
@@ -22,7 +22,7 @@ const HOP_BY_HOP: &[&str] = &[
     "upgrade",
 ];
 
-const REQUEST_DROP_EXTRA: &[&str] = &["host", "content-length"];
+const REQUEST_DROP_EXTRA: &[&str] = &["host", "content-length", "accept-encoding"];
 const PROCESSED_RESPONSE_DROP_EXTRA: &[&str] = &["content-length", "content-encoding"];
 
 fn is_hop_by_hop(name: &str) -> bool {
@@ -369,6 +369,7 @@ mod tests {
     fn request_drop_includes_host_and_content_length() {
         assert!(is_request_drop("host"));
         assert!(is_request_drop("content-length"));
+        assert!(is_request_drop("accept-encoding"));
         assert!(is_request_drop("connection"));
         assert!(!is_request_drop("content-type"));
     }
diff --git a/crates/agentic-server/tests/messages_test.rs b/crates/agentic-server/tests/messages_test.rs
index 6817b82e..3e0b6084 100644
--- a/crates/agentic-server/tests/messages_test.rs
+++ b/crates/agentic-server/tests/messages_test.rs
@@ -408,6 +408,7 @@ async fn messages_gateway_loop_forwards_query_and_open_headers() {
         .header("x-claude-code-session-id", "session-loop")
         .header("x-claude-code-agent-id", "agent-loop")
         .header("x-api-key", "anthropic-key")
+        .header("accept-encoding", "gzip")
         .body(body.to_vec())
         .send()
         .await
@@ -427,6 +428,7 @@ async fn messages_gateway_loop_forwards_query_and_open_headers() {
     assert_eq!(requests[0].headers["x-claude-code-session-id"], "session-loop");
     assert_eq!(requests[0].headers["x-claude-code-agent-id"], "agent-loop");
     assert_eq!(requests[0].headers["x-api-key"], "anthropic-key");
+    assert!(!requests[0].headers.contains_key("accept-encoding"));
     assert!(!requests[0].headers.contains_key("authorization"));
 }