diff --git a/crates/headroom-proxy/src/config.rs b/crates/headroom-proxy/src/config.rs
index 37c8efde0..e457d4995 100644
--- a/crates/headroom-proxy/src/config.rs
+++ b/crates/headroom-proxy/src/config.rs
@@ -30,6 +30,47 @@ pub enum CompressionMode {
     LiveZone,
 }
 
+/// Policy for stripping internal `x-headroom-*` headers from upstream-bound
+/// requests (PR-A5, fixes P5-49).
+///
+/// When `enabled` (default), every header whose name starts with
+/// `x-headroom-` is dropped before the upstream call. Stops fingerprinting
+/// of the proxy via subscription-revocation flags (`x-headroom-bypass`,
+/// `x-headroom-mode`, etc.) and prevents leakage of internal user-id /
+/// stack / base-url headers.
+///
+/// When `disabled`, internal headers are forwarded verbatim. This is an
+/// explicit operator opt-in for diagnostic shadow tracing — NOT a fallback.
+/// Document the trade-off in `docs/configuration.md` before flipping this.
+///
+/// Source priority: CLI flag → `HEADROOM_PROXY_STRIP_INTERNAL_HEADERS`
+/// env var → default (`enabled`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
+#[clap(rename_all = "snake_case")]
+pub enum StripInternalHeaders {
+    /// Strip every `x-headroom-*` header from upstream-bound requests.
+    /// Default. Operationally safe.
+    Enabled,
+    /// Forward `x-headroom-*` to upstream verbatim. Diagnostic-only;
+    /// exposes internal flags to the upstream and reveals the proxy.
+    Disabled,
+}
+
+impl StripInternalHeaders {
+    /// Stable snake_case name suitable for log fields.
+    pub fn as_str(self) -> &'static str {
+        match self {
+            StripInternalHeaders::Enabled => "enabled",
+            StripInternalHeaders::Disabled => "disabled",
+        }
+    }
+
+    /// Convenience: is the strip switched on?
+    pub fn is_enabled(self) -> bool {
+        matches!(self, StripInternalHeaders::Enabled)
+    }
+}
+
 /// Policy for automatically deriving `frozen_message_count` from the
 /// customer's `cache_control` markers (PR-A4).
 ///
@@ -195,6 +236,21 @@ pub struct CliArgs {
         default_value_t = CacheControlAutoFrozen::Enabled,
     )]
     pub cache_control_auto_frozen: CacheControlAutoFrozen,
+
+    /// Strip internal `x-headroom-*` headers from upstream-bound
+    /// requests (PR-A5, fixes P5-49). Default `enabled`. The `disabled`
+    /// path is operator opt-in for diagnostic shadow tracing only —
+    /// NOT a fallback per realignment build constraint #4.
+    ///
+    /// Source priority: CLI flag → `HEADROOM_PROXY_STRIP_INTERNAL_HEADERS`
+    /// env var → default (`enabled`).
+    #[arg(
+        long = "strip-internal-headers",
+        env = "HEADROOM_PROXY_STRIP_INTERNAL_HEADERS",
+        value_enum,
+        default_value_t = StripInternalHeaders::Enabled,
+    )]
+    pub strip_internal_headers: StripInternalHeaders,
 }
 
 fn parse_duration(s: &str) -> Result<Duration, String> {
@@ -236,6 +292,10 @@ pub struct Config {
     /// adds the derivation function (`compute_frozen_count`); Phase
     /// B's dispatcher consumes the resolved value here.
     pub cache_control_auto_frozen: CacheControlAutoFrozen,
+    /// Whether to strip internal `x-headroom-*` headers from
+    /// upstream-bound requests. PR-A5 default-on guard against
+    /// fingerprinting / leakage of internal flags.
+    pub strip_internal_headers: StripInternalHeaders,
 }
 
 impl Config {
@@ -261,6 +321,7 @@ impl Config {
             compression_max_body_bytes,
             compression_mode: args.compression_mode,
             cache_control_auto_frozen: args.cache_control_auto_frozen,
+            strip_internal_headers: args.strip_internal_headers,
         }
     }
 
@@ -282,6 +343,10 @@ impl Config {
             // Match production default so the cache-control walker is
             // exercised under test without per-test opt-in.
             cache_control_auto_frozen: CacheControlAutoFrozen::Enabled,
+            // Production default: strip internal `x-headroom-*` headers
+            // from upstream-bound requests. Tests opt out per-case via
+            // `start_proxy_with`.
+            strip_internal_headers: StripInternalHeaders::Enabled,
         }
     }
 }
diff --git a/crates/headroom-proxy/src/headers.rs b/crates/headroom-proxy/src/headers.rs
index 8ac6a436e..9c360d6f8 100644
--- a/crates/headroom-proxy/src/headers.rs
+++ b/crates/headroom-proxy/src/headers.rs
@@ -24,6 +24,16 @@ const HOP_BY_HOP: &[&str] = &[
 /// must not be copied across (reqwest/hyper sets them itself).
 const CLIENT_MANAGED: &[&str] = &["host", "content-length"];
 
+/// Internal-header prefix dropped from upstream-bound requests when
+/// `Config::strip_internal_headers == StripInternalHeaders::Enabled` (PR-A5,
+/// fixes P5-49). Case-insensitive prefix match. The Rust path mirrors the
+/// Python `_strip_internal_headers` helper.
+///
+/// Response-side `X-Headroom-*` injection (e.g. `x-headroom-tokens-saved`)
+/// is intentionally untouched — that direction is the proxy describing its
+/// own work to the client and never crosses an upstream boundary.
+pub const INTERNAL_HEADER_PREFIX: &str = "x-headroom-";
+
 /// Returns true if `name` is hop-by-hop and must be stripped.
 pub fn is_hop_by_hop(name: &HeaderName) -> bool {
     let n = name.as_str();
@@ -40,6 +50,14 @@ pub fn is_request_drop(name: &HeaderName) -> bool {
     CLIENT_MANAGED.iter().any(|h| h.eq_ignore_ascii_case(n))
 }
 
+/// Returns true when `name` matches the internal `x-headroom-*` prefix
+/// (case-insensitive). Pure function, no regex.
+pub fn is_internal_header(name: &HeaderName) -> bool {
+    name.as_str()
+        .to_ascii_lowercase()
+        .starts_with(INTERNAL_HEADER_PREFIX)
+}
+
 /// Headers we drop on the response side. Same hop-by-hop set; we don't touch
 /// content-length since the response body length is known and we want clients
 /// to see it.
@@ -81,18 +99,43 @@ pub fn set_single(headers: &mut HeaderMap, name: HeaderName, value: &str) {
     }
 }
 
+/// Remove every header whose name starts with `INTERNAL_HEADER_PREFIX`
+/// (case-insensitive). Mutates in place. Returns the number of header
+/// entries removed for structured logging.
+pub fn strip_internal_headers(headers: &mut HeaderMap) -> usize {
+    let to_remove: Vec<HeaderName> = headers
+        .keys()
+        .filter(|n| is_internal_header(n))
+        .cloned()
+        .collect();
+    let mut removed = 0usize;
+    for name in to_remove {
+        // `remove` returns the first value; multi-valued internal headers are
+        // not expected in practice but we drain them all for safety.
+        while headers.remove(&name).is_some() {
+            removed += 1;
+        }
+    }
+    removed
+}
+
 /// Build a fresh HeaderMap suitable for forwarding to the upstream:
 ///   - hop-by-hop and connection-listed headers stripped
 ///   - Host/Content-Length removed (rebuilt by client)
 ///   - X-Forwarded-For appended
 ///   - X-Forwarded-Proto, X-Forwarded-Host set
 ///   - X-Request-Id ensured
+///   - When `strip_internal == true`, `x-headroom-*` headers stripped
+///     (PR-A5, fixes P5-49). Operators can disable via
+///     `HEADROOM_PROXY_STRIP_INTERNAL_HEADERS=disabled` for diagnostic
+///     shadow tracing.
 pub fn build_forward_request_headers(
     incoming: &HeaderMap,
     client_addr: IpAddr,
     forwarded_proto: &str,
     forwarded_host: Option<&str>,
     request_id: &str,
+    strip_internal: bool,
 ) -> HeaderMap {
     let connection_listed = connection_listed_headers(incoming);
     let mut out = HeaderMap::new();
@@ -103,6 +146,9 @@ pub fn build_forward_request_headers(
         if connection_listed.iter().any(|h| h == name.as_str()) {
             continue;
         }
+        if strip_internal && is_internal_header(name) {
+            continue;
+        }
         out.append(name.clone(), value.clone());
     }
     append_xff(&mut out, client_addr);
@@ -169,4 +215,74 @@ mod tests {
         assert!(listed.contains(&"x-foo".to_string()));
         assert!(listed.contains(&"close".to_string()));
     }
+
+    #[test]
+    fn internal_header_detected_case_insensitive() {
+        assert!(is_internal_header(&HeaderName::from_static(
+            "x-headroom-bypass"
+        )));
+        // HeaderName normalizes to lowercase internally, so any cased input
+        // ends up matching the lowercase prefix.
+        assert!(is_internal_header(
+            &HeaderName::from_bytes(b"X-Headroom-Mode").unwrap()
+        ));
+        assert!(is_internal_header(
+            &HeaderName::from_bytes(b"X-HEADROOM-FOO").unwrap()
+        ));
+        assert!(!is_internal_header(&HeaderName::from_static(
+            "x-request-id"
+        )));
+        assert!(!is_internal_header(&HeaderName::from_static(
+            "authorization"
+        )));
+    }
+
+    #[test]
+    fn strip_internal_headers_removes_only_internal_prefix() {
+        let mut h = HeaderMap::new();
+        h.insert("authorization", HeaderValue::from_static("Bearer x"));
+        h.insert("x-headroom-bypass", HeaderValue::from_static("true"));
+        h.insert("x-headroom-mode", HeaderValue::from_static("passthrough"));
+        h.insert("x-request-id", HeaderValue::from_static("req-1"));
+        let removed = strip_internal_headers(&mut h);
+        assert_eq!(removed, 2);
+        assert!(h.get("x-headroom-bypass").is_none());
+        assert!(h.get("x-headroom-mode").is_none());
+        assert!(h.get("authorization").is_some());
+        assert!(h.get("x-request-id").is_some());
+    }
+
+    #[test]
+    fn build_forward_strips_internal_when_enabled() {
+        let mut incoming = HeaderMap::new();
+        incoming.insert("authorization", HeaderValue::from_static("Bearer x"));
+        incoming.insert("x-headroom-bypass", HeaderValue::from_static("true"));
+        let out = build_forward_request_headers(
+            &incoming,
+            "127.0.0.1".parse().unwrap(),
+            "http",
+            Some("h"),
+            "req-1",
+            true,
+        );
+        assert!(out.get("authorization").is_some());
+        assert!(out.get("x-headroom-bypass").is_none());
+    }
+
+    #[test]
+    fn build_forward_keeps_internal_when_disabled() {
+        let mut incoming = HeaderMap::new();
+        incoming.insert("authorization", HeaderValue::from_static("Bearer x"));
+        incoming.insert("x-headroom-bypass", HeaderValue::from_static("true"));
+        let out = build_forward_request_headers(
+            &incoming,
+            "127.0.0.1".parse().unwrap(),
+            "http",
+            Some("h"),
+            "req-1",
+            false,
+        );
+        assert!(out.get("authorization").is_some());
+        assert!(out.get("x-headroom-bypass").is_some());
+    }
 }
diff --git a/crates/headroom-proxy/src/proxy.rs b/crates/headroom-proxy/src/proxy.rs
index 3d80ba15e..628e9a779 100644
--- a/crates/headroom-proxy/src/proxy.rs
+++ b/crates/headroom-proxy/src/proxy.rs
@@ -198,13 +198,41 @@ async fn forward_http(
 
     // Build the outgoing headers off the incoming ones, then optionally drop
     // Host (rewrite_host=true => let reqwest set its own Host for the upstream).
+    // PR-A5 (P5-49): strip internal `x-headroom-*` from upstream-bound
+    // requests when `Config::strip_internal_headers == Enabled` (default).
+    let strip_internal = state.config.strip_internal_headers.is_enabled();
+    let pre_strip_internal_count = req
+        .headers()
+        .iter()
+        .filter(|(name, _)| crate::headers::is_internal_header(name))
+        .count();
     let mut outgoing_headers = build_forward_request_headers(
         req.headers(),
         client_addr.ip(),
         "http",
         forwarded_host.as_deref(),
         &request_id,
+        strip_internal,
     );
+    if strip_internal && pre_strip_internal_count > 0 {
+        tracing::info!(
+            event = "outbound_headers",
+            forwarder = "rust_proxy",
+            stripped_count = pre_strip_internal_count,
+            request_id = %request_id,
+            "stripped internal x-headroom-* headers from upstream-bound request"
+        );
+    } else if !strip_internal && pre_strip_internal_count > 0 {
+        tracing::warn!(
+            event = "outbound_headers",
+            forwarder = "rust_proxy",
+            mode = "disabled",
+            internal_count = pre_strip_internal_count,
+            request_id = %request_id,
+            "HEADROOM_PROXY_STRIP_INTERNAL_HEADERS=disabled; \
+             internal x-headroom-* headers forwarded to upstream"
+        );
+    }
     if !state.config.rewrite_host {
         if let Some(h) = req.headers().get(http::header::HOST) {
             outgoing_headers.insert(http::header::HOST, h.clone());
diff --git a/crates/headroom-proxy/src/websocket.rs b/crates/headroom-proxy/src/websocket.rs
index b976e5675..c3a6358c6 100644
--- a/crates/headroom-proxy/src/websocket.rs
+++ b/crates/headroom-proxy/src/websocket.rs
@@ -49,12 +49,16 @@ pub async fn ws_handler(
         .and_then(|v| v.to_str().ok())
         .map(|s| s.to_string());
 
+    // PR-A5: same strip policy as the HTTP path — operators flip both
+    // simultaneously via the single `Config::strip_internal_headers` knob.
+    let strip_internal_ws = state.config.strip_internal_headers.is_enabled();
     let forward_headers = build_forward_request_headers(
         req.headers(),
         client_addr.ip(),
         "http",
         forwarded_host.as_deref(),
         &request_id,
+        strip_internal_ws,
     );
     // Sec-WebSocket-Protocol must be propagated for subprotocol negotiation.
     let subprotocols: Option<String> = req
diff --git a/crates/headroom-proxy/tests/integration_headers.rs b/crates/headroom-proxy/tests/integration_headers.rs
index d80385a0c..080e93028 100644
--- a/crates/headroom-proxy/tests/integration_headers.rs
+++ b/crates/headroom-proxy/tests/integration_headers.rs
@@ -1,8 +1,10 @@
-//! Header passthrough + hop-by-hop filtering + X-Forwarded-* injection.
+//! Header passthrough + hop-by-hop filtering + X-Forwarded-* injection +
+//! internal `x-headroom-*` strip (PR-A5, fixes P5-49).
 
 mod common;
 
-use common::start_proxy;
+use common::{start_proxy, start_proxy_with};
+use headroom_proxy::config::StripInternalHeaders;
 use wiremock::matchers::{method, path};
 use wiremock::{Mock, MockServer, ResponseTemplate};
 
@@ -61,6 +63,151 @@ async fn custom_headers_pass_through_both_ways() {
     proxy.shutdown().await;
 }
 
+#[tokio::test]
+async fn x_headroom_request_headers_stripped() {
+    let upstream = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/v1/messages"))
+        .respond_with(move |req: &wiremock::Request| {
+            // PR-A5: internal x-headroom-* must NOT reach upstream.
+            assert!(
+                req.headers.get("x-headroom-bypass").is_none(),
+                "x-headroom-bypass leaked upstream"
+            );
+            assert!(
+                req.headers.get("x-headroom-mode").is_none(),
+                "x-headroom-mode leaked upstream"
+            );
+            assert!(
+                req.headers.get("x-headroom-user-id").is_none(),
+                "x-headroom-user-id leaked upstream"
+            );
+            // Legitimate headers must still arrive.
+            assert_eq!(req.headers.get("authorization").unwrap(), "Bearer sk-x");
+            assert_eq!(req.headers.get("anthropic-version").unwrap(), "2023-06-01");
+            ResponseTemplate::new(200).set_body_string("{}")
+        })
+        .mount(&upstream)
+        .await;
+
+    let proxy = start_proxy(&upstream.uri()).await;
+    let resp = reqwest::Client::new()
+        .post(format!("{}/v1/messages", proxy.url()))
+        .header("authorization", "Bearer sk-x")
+        .header("anthropic-version", "2023-06-01")
+        .header("x-headroom-bypass", "true")
+        .header("x-headroom-mode", "passthrough")
+        .header("x-headroom-user-id", "alice")
+        .body("{}")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    proxy.shutdown().await;
+}
+
+#[tokio::test]
+async fn x_headroom_case_insensitive_stripped() {
+    let upstream = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/v1/messages"))
+        .respond_with(move |req: &wiremock::Request| {
+            // Mixed-case variants — all should be stripped.
+            for hdr in [
+                "x-headroom-foo",
+                "x-headroom-bar",
+                "x-headroom-baz",
+                "X-Headroom-Foo",
+                "X-HEADROOM-BAR",
+            ] {
+                assert!(
+                    req.headers.get(hdr).is_none(),
+                    "internal header {hdr} leaked upstream"
+                );
+            }
+            ResponseTemplate::new(200).set_body_string("{}")
+        })
+        .mount(&upstream)
+        .await;
+
+    let proxy = start_proxy(&upstream.uri()).await;
+    let resp = reqwest::Client::new()
+        .post(format!("{}/v1/messages", proxy.url()))
+        .header("X-Headroom-Foo", "1")
+        .header("x-Headroom-Bar", "2")
+        .header("X-HEADROOM-BAZ", "3")
+        .body("{}")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    proxy.shutdown().await;
+}
+
+#[tokio::test]
+async fn legitimate_headers_passthrough_with_strip_enabled() {
+    let upstream = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/echo"))
+        .respond_with(move |req: &wiremock::Request| {
+            // Non-internal x-* headers must NOT be stripped.
+            assert_eq!(req.headers.get("x-api-key").unwrap(), "k1");
+            assert_eq!(req.headers.get("x-trace-id").unwrap(), "trace-1");
+            assert_eq!(req.headers.get("authorization").unwrap(), "Bearer x");
+            assert_eq!(req.headers.get("anthropic-version").unwrap(), "2023-06-01");
+            // Strip happened — internal flag absent.
+            assert!(req.headers.get("x-headroom-bypass").is_none());
+            ResponseTemplate::new(200)
+        })
+        .mount(&upstream)
+        .await;
+
+    let proxy = start_proxy(&upstream.uri()).await;
+    let resp = reqwest::Client::new()
+        .post(format!("{}/echo", proxy.url()))
+        .header("x-api-key", "k1")
+        .header("x-trace-id", "trace-1")
+        .header("authorization", "Bearer x")
+        .header("anthropic-version", "2023-06-01")
+        .header("x-headroom-bypass", "true")
+        .body("{}")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    proxy.shutdown().await;
+}
+
+#[tokio::test]
+async fn disabled_mode_passes_internal_headers_through() {
+    let upstream = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/v1/messages"))
+        .respond_with(move |req: &wiremock::Request| {
+            // Operator opt-in: internal header IS forwarded.
+            assert_eq!(req.headers.get("x-headroom-bypass").unwrap(), "true");
+            assert_eq!(req.headers.get("x-headroom-mode").unwrap(), "passthrough");
+            ResponseTemplate::new(200).set_body_string("{}")
+        })
+        .mount(&upstream)
+        .await;
+
+    let proxy = start_proxy_with(&upstream.uri(), |cfg| {
+        cfg.strip_internal_headers = StripInternalHeaders::Disabled;
+    })
+    .await;
+    let resp = reqwest::Client::new()
+        .post(format!("{}/v1/messages", proxy.url()))
+        .header("x-headroom-bypass", "true")
+        .header("x-headroom-mode", "passthrough")
+        .body("{}")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    proxy.shutdown().await;
+}
+
 #[tokio::test]
 async fn xff_appends_existing_value() {
     let upstream = MockServer::start().await;
diff --git a/docs/content/docs/configuration.mdx b/docs/content/docs/configuration.mdx
index c38d6c0a0..fa0f92abf 100644
--- a/docs/content/docs/configuration.mdx
+++ b/docs/content/docs/configuration.mdx
@@ -257,6 +257,8 @@ headroom proxy --llmlingua --llmlingua-device cuda --llmlingua-rate 0.4
 | `HEADROOM_TELEMETRY` | Set to `off` to disable anonymous telemetry | `on` |
 | `HEADROOM_MEMORY_INJECTION_MODE` | Memory-context routing mode: `live_zone_tail` (default) or `disabled`. The legacy `system_prompt` mode was retired by PR-A2; supplying it raises. | `live_zone_tail` |
 | `HEADROOM_PROXY_PYTHON_FORWARDER_MODE` | Python forwarder serialization mode. `byte_faithful` (default) forwards original request bytes verbatim when no transform mutated the body and re-serializes canonically only when needed — keeps Anthropic prompt-cache hit-rate intact. `legacy_json_kwarg` is an explicit operator opt-in for emergency rollback to the historical `httpx ... json=body` behavior. NOT a fallback — only flip on explicit operator decision. | `byte_faithful` |
+| `HEADROOM_STRIP_INTERNAL_HEADERS` | Python proxy: whether to strip internal `x-headroom-*` request headers (e.g. `x-headroom-bypass`, `x-headroom-mode`, `x-headroom-user-id`, `x-headroom-stack`, `x-headroom-base-url`) before every upstream forwarder call (PR-A5, fixes P5-49). `enabled` (default) stops fingerprinting / leakage. `disabled` is an explicit operator opt-in for diagnostic shadow tracing — NOT a fallback. Inbound reads of these headers (bypass gating, memory user-id resolution) are unaffected because they read `request.headers` directly. | `enabled` |
+| `HEADROOM_PROXY_STRIP_INTERNAL_HEADERS` | Rust proxy: same policy as `HEADROOM_STRIP_INTERNAL_HEADERS` but for the Rust transparent proxy. Stripping happens inside `build_forward_request_headers` so both HTTP and WebSocket upstream calls are gated by one flag. `enabled` default; `disabled` operator opt-in for diagnostic shadow tracing. Response-side `X-Headroom-*` injection (e.g. `x-headroom-tokens-saved`) is unrelated and stays. | `enabled` |
 
 ### Filesystem Contract
 
diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py
index a7c91e6b8..ea96872a8 100644
--- a/headroom/proxy/handlers/anthropic.py
+++ b/headroom/proxy/handlers/anthropic.py
@@ -566,6 +566,21 @@ class AnthropicHandlerMixin:
             # body is undecipherable → 502.
             headers.pop("accept-encoding", None)
             tags = self._extract_tags(headers)
+            # PR-A5 (P5-49): strip internal x-headroom-* from upstream-bound
+            # headers AFTER `_extract_tags` reads them. Inbound bypass gating
+            # uses `request.headers.get(...)` directly above; memory user-id
+            # is read from `request.headers` below if needed. From this
+            # point on, `headers` is the upstream-bound copy.
+            from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+            _pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+            headers = _strip_internal_headers(headers)
+            log_outbound_headers(
+                forwarder="anthropic_messages",
+                stripped_count=_pre_strip_count
+                - sum(1 for k in headers if k.lower().startswith("x-headroom-")),
+                request_id=request_id,
+            )
 
             # Subscription tracker: notify on OAuth requests (not API-key requests)
             _auth_header = headers.get("authorization", "")
@@ -614,10 +629,12 @@ class AnthropicHandlerMixin:
                         detail=f"Budget exceeded for {self.config.budget_period} period",
                     )
 
-            # Memory: Get user ID when memory is enabled (fallback to "default" for simple DevEx)
+            # Memory: Get user ID when memory is enabled (fallback to "default" for simple DevEx).
+            # Reads `request.headers` directly because the local `headers` dict was
+            # stripped of `x-headroom-*` above for the upstream-bound copy (PR-A5).
             memory_user_id: str | None = None
             if self.memory_handler:
-                memory_user_id = headers.get(
+                memory_user_id = request.headers.get(
                     "x-headroom-user-id",
                     os.environ.get("USER", os.environ.get("USERNAME", "default")),
                 )
@@ -2143,6 +2160,16 @@ class AnthropicHandlerMixin:
         headers = dict(request.headers.items())
         headers.pop("host", None)
         headers.pop("content-length", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="anthropic_batch",
+            stripped_count=_pre_strip_count,
+            request_id=request_id,
+        )
 
         # Track compression stats across all batch requests
         total_original_tokens = 0
@@ -2369,6 +2396,16 @@ class AnthropicHandlerMixin:
 
         headers = dict(request.headers.items())
         headers.pop("host", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="anthropic_batch_passthrough",
+            stripped_count=_pre_strip_count,
+            request_id=None,
+        )
 
         body = await request.body()
 
@@ -2470,6 +2507,16 @@ class AnthropicHandlerMixin:
 
         headers = dict(request.headers.items())
         headers.pop("host", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="anthropic_batch_results",
+            stripped_count=_pre_strip_count,
+            request_id=None,
+        )
 
         response = await self.http_client.get(url, headers=headers)  # type: ignore[union-attr]
 
diff --git a/headroom/proxy/handlers/batch.py b/headroom/proxy/handlers/batch.py
index de1d9847d..5ea15a6b4 100644
--- a/headroom/proxy/handlers/batch.py
+++ b/headroom/proxy/handlers/batch.py
@@ -99,6 +99,16 @@ class BatchHandlerMixin:
         headers = dict(request.headers.items())
         headers.pop("host", None)
         headers.pop("content-length", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count_gb = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="google_batch",
+            stripped_count=_pre_strip_count_gb,
+            request_id=request_id,
+        )
 
         # Track compression stats
         total_original_tokens = 0
@@ -330,6 +340,16 @@ class BatchHandlerMixin:
         headers = dict(request.headers.items())
         headers.pop("host", None)
         headers.pop("content-length", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count_gpt = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="google_batch_passthrough",
+            stripped_count=_pre_strip_count_gpt,
+            request_id=None,
+        )
 
         url = f"{self.GEMINI_API_URL}/v1beta/models/{model}:batchGenerateContent"
 
@@ -428,6 +448,16 @@ class BatchHandlerMixin:
 
         headers = dict(request.headers.items())
         headers.pop("host", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count_gp = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="gemini_passthrough",
+            stripped_count=_pre_strip_count_gp,
+            request_id=None,
+        )
 
         # Handle API key
         api_key = headers.pop("x-goog-api-key", None)
@@ -549,6 +579,16 @@ class BatchHandlerMixin:
 
         headers = dict(request.headers.items())
         headers.pop("host", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count_gbr = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="google_batch_results",
+            stripped_count=_pre_strip_count_gbr,
+            request_id=None,
+        )
 
         # Handle API key
         api_key = headers.pop("x-goog-api-key", None)
@@ -729,6 +769,16 @@ class BatchHandlerMixin:
         headers = dict(request.headers.items())
         headers.pop("host", None)
         headers.pop("content-length", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count_oacc = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="openai_batch_chat_completions",
+            stripped_count=_pre_strip_count_oacc,
+            request_id=request_id,
+        )
 
         try:
             # Step 1: Download the input file from OpenAI
@@ -1048,6 +1098,8 @@ class BatchHandlerMixin:
 
         from headroom.proxy.helpers import (
             _read_request_body_bytes,
+            _strip_internal_headers,
+            log_outbound_headers,
             log_outbound_request,
             prepare_outbound_body_bytes,
         )
@@ -1055,6 +1107,14 @@ class BatchHandlerMixin:
         headers = dict(request.headers.items())
         headers.pop("host", None)
         headers.pop("content-length", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        _pre_strip_count_obp = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="openai_batch_passthrough",
+            stripped_count=_pre_strip_count_obp,
+            request_id=None,
+        )
 
         url = f"{self.OPENAI_API_URL}/v1/batches"
 
diff --git a/headroom/proxy/handlers/gemini.py b/headroom/proxy/handlers/gemini.py
index b95594489..427314f3f 100644
--- a/headroom/proxy/handlers/gemini.py
+++ b/headroom/proxy/handlers/gemini.py
@@ -191,11 +191,24 @@ class GeminiHandlerMixin:
         headers.pop("host", None)
         headers.pop("content-length", None)
         tags = self._extract_tags(headers)
+        # PR-A5 (P5-49): strip internal x-headroom-* from upstream-bound
+        # headers AFTER `_extract_tags` reads them. Memory user-id reads
+        # `request.headers` below.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
 
-        # Memory: Get user ID when memory is enabled
+        _pre_strip_count_gem = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="gemini_generate_content",
+            stripped_count=_pre_strip_count_gem,
+            request_id=request_id,
+        )
+
+        # Memory: Get user ID when memory is enabled. Reads `request.headers`
+        # directly because `headers` was stripped of `x-headroom-*` (PR-A5).
         memory_user_id: str | None = None
         if self.memory_handler:
-            memory_user_id = headers.get(
+            memory_user_id = request.headers.get(
                 "x-headroom-user-id",
                 os.environ.get("USER", os.environ.get("USERNAME", "default")),
             )
@@ -513,6 +526,17 @@ class GeminiHandlerMixin:
         headers.pop("accept-encoding", None)
         tags = self._extract_tags(headers)
         is_antigravity = self._is_cloudcode_antigravity_request(body, headers)
+        # PR-A5 (P5-49): strip internal x-headroom-* from upstream-bound headers
+        # AFTER `_extract_tags` and `is_cloudcode_antigravity` reads.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count_cca = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="gemini_cloudcode_assist",
+            stripped_count=_pre_strip_count_cca,
+            request_id=request_id,
+        )
 
         system_instruction = request_payload.get("systemInstruction")
         optimization_system_instruction = None if is_antigravity else system_instruction
@@ -628,6 +652,16 @@ class GeminiHandlerMixin:
         headers.pop("host", None)
         headers.pop("content-length", None)
         tags = self._extract_tags(headers)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count_gem_stream = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="gemini_stream_generate_content",
+            stripped_count=_pre_strip_count_gem_stream,
+            request_id=request_id,
+        )
 
         # Token counting
         tokenizer = get_tokenizer(model)
@@ -702,6 +736,16 @@ class GeminiHandlerMixin:
         headers = dict(request.headers.items())
         headers.pop("host", None)
         headers.pop("content-length", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count_gem_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="gemini_count_tokens",
+            stripped_count=_pre_strip_count_gem_count,
+            request_id=request_id,
+        )
 
         # Convert Gemini format to messages for optimization
         system_instruction = body.get("systemInstruction")
diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py
index 4b6c1e3ca..aede4323f 100644
--- a/headroom/proxy/handlers/openai.py
+++ b/headroom/proxy/handlers/openai.py
@@ -263,11 +263,27 @@ class OpenAIHandlerMixin:
         # if httpx lacks brotli support the response body is undecipherable → 502.
         headers.pop("accept-encoding", None)
         tags = self._extract_tags(headers)
+        # PR-A5 (P5-49): strip internal x-headroom-* from upstream-bound
+        # headers AFTER `_extract_tags` reads them. Inbound bypass gating
+        # uses `request.headers.get(...)` above; memory user-id reads
+        # `request.headers` below. From this point on, `headers` is the
+        # upstream-bound copy.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
 
-        # Memory: Get user ID when memory is enabled
+        _pre_strip_count_chat = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="openai_chat_completions",
+            stripped_count=_pre_strip_count_chat,
+            request_id=request_id,
+        )
+
+        # Memory: Get user ID when memory is enabled. Reads `request.headers`
+        # directly because `headers` was stripped of `x-headroom-*` for the
+        # upstream-bound copy (PR-A5).
         memory_user_id: str | None = None
         if self.memory_handler:
-            memory_user_id = headers.get(
+            memory_user_id = request.headers.get(
                 "x-headroom-user-id",
                 os.environ.get("USER", os.environ.get("USERNAME", "default")),
             )
@@ -1137,11 +1153,24 @@ class OpenAIHandlerMixin:
         # if httpx lacks brotli support the response body is undecipherable → 502.
         headers.pop("accept-encoding", None)
         tags = self._extract_tags(headers)
+        # PR-A5 (P5-49): strip internal x-headroom-* from upstream-bound
+        # headers AFTER `_extract_tags` reads them. Memory user-id reads
+        # `request.headers` below.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
 
-        # Memory: Get user ID when memory is enabled
+        _pre_strip_count_resp = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="openai_responses",
+            stripped_count=_pre_strip_count_resp,
+            request_id=request_id,
+        )
+
+        # Memory: Get user ID when memory is enabled. Reads `request.headers`
+        # directly because `headers` was stripped of `x-headroom-*` (PR-A5).
         memory_user_id: str | None = None
         if self.memory_handler:
-            memory_user_id = headers.get(
+            memory_user_id = request.headers.get(
                 "x-headroom-user-id",
                 os.environ.get("USER", os.environ.get("USERNAME", "default")),
             )
@@ -1617,10 +1646,30 @@ class OpenAIHandlerMixin:
                 "transfer-encoding",  # hop-by-hop
             }
         )
-        upstream_headers: dict[str, str] = {}
+        # PR-A5 (P5-49): also drop internal x-headroom-* from the upstream
+        # WebSocket handshake. Inbound reads on `ws_headers` (memory user-id
+        # below) keep working because we filter only when building
+        # `upstream_headers`, not when reading from `ws_headers`.
+        from headroom.proxy.helpers import (
+            _strip_internal_headers as _strip_internal,
+        )
+        from headroom.proxy.helpers import (
+            log_outbound_headers as _log_outbound_headers,
+        )
+
+        _ws_pre_strip_filtered: dict[str, str] = {}
         for k, v in ws_headers.items():
             if k.lower() not in _skip_headers:
-                upstream_headers[k] = v
+                _ws_pre_strip_filtered[k] = v
+        _ws_pre_strip_count = sum(
+            1 for k in _ws_pre_strip_filtered if k.lower().startswith("x-headroom-")
+        )
+        upstream_headers = _strip_internal(_ws_pre_strip_filtered)
+        _log_outbound_headers(
+            forwarder="openai_responses_ws",
+            stripped_count=_ws_pre_strip_count,
+            request_id=request_id,
+        )
 
         upstream_headers, is_chatgpt_auth = _resolve_codex_routing_headers(upstream_headers)
         _lower_headers = {k.lower(): v for k, v in upstream_headers.items()}
@@ -2771,6 +2820,16 @@ class OpenAIHandlerMixin:
         headers = dict(request.headers.items())
         headers.pop("host", None)
         headers.pop("accept-encoding", None)
+        # PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
+        from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
+
+        _pre_strip_count_pt = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
+        headers = _strip_internal_headers(headers)
+        log_outbound_headers(
+            forwarder="openai_passthrough",
+            stripped_count=_pre_strip_count_pt,
+            request_id=None,
+        )
 
         body = await request.body()
 
diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py
index d84188253..01c26da34 100644
--- a/headroom/proxy/helpers.py
+++ b/headroom/proxy/helpers.py
@@ -563,6 +563,98 @@ def is_anthropic_auth(headers: dict[str, str]) -> bool:
     return False
 
 
+# ---------------------------------------------------------------------------
+# Internal-header stripping (PR-A5 — fixes P5-49).
+# ---------------------------------------------------------------------------
+#
+# `x-headroom-*` request headers (e.g. ``x-headroom-bypass``,
+# ``x-headroom-mode``, ``x-headroom-user-id``, ``x-headroom-stack``,
+# ``x-headroom-base-url``) are internal control flags consumed by the
+# proxy itself. They MUST NOT leak upstream — leaking them would (a)
+# fingerprint the proxy to subscription enforcers and (b) expose the
+# user-id/stack/base-url internals to whichever vendor terminates the
+# request.
+#
+# Inbound read paths (bypass gating, ``_extract_tags`` reading
+# ``x-headroom-*``, memory ``x-headroom-user-id`` lookup) keep using
+# the original dict / ``request.headers``. The stripped copy is what
+# every upstream-bound forwarder receives.
+#
+# Note: response-side ``X-Headroom-*`` injection (e.g.
+# ``x-headroom-tokens-saved``) is unrelated — the proxy is allowed to
+# tell its client about its own work. This helper only filters
+# request-side headers.
+
+_INTERNAL_HEADER_PREFIX = "x-headroom-"
+
+# Operator opt-in env var. ``enabled`` (default) strips internal
+# ``x-headroom-*`` headers from every upstream-bound forwarder.
+# ``disabled`` is an explicit operator opt-in for diagnostic shadow
+# tracing — NOT a fallback. Per realignment build constraint #4 the
+# behaviour is loud, configurable, and never silent.
+_STRIP_INTERNAL_HEADERS_ENV = "HEADROOM_STRIP_INTERNAL_HEADERS"
+StripInternalHeadersMode = Literal["enabled", "disabled"]
+_STRIP_INTERNAL_HEADERS_DEFAULT: StripInternalHeadersMode = "enabled"
+
+
+def get_strip_internal_headers_mode() -> StripInternalHeadersMode:
+    """Return the active internal-header strip mode.
+
+    Read at request time so operators can flip behaviour without a
+    restart. Unknown values raise loudly per the no-silent-fallback
+    build constraint.
+    """
+    raw = os.environ.get(_STRIP_INTERNAL_HEADERS_ENV, "").strip().lower()
+    if not raw:
+        return _STRIP_INTERNAL_HEADERS_DEFAULT
+    if raw in ("enabled", "disabled"):
+        return cast(StripInternalHeadersMode, raw)
+    raise ValueError(
+        f"Invalid {_STRIP_INTERNAL_HEADERS_ENV}={raw!r}; expected 'enabled' or 'disabled'"
+    )
+
+
+def _strip_internal_headers(headers: dict[str, str]) -> dict[str, str]:
+    """Return a copy of ``headers`` with internal ``x-headroom-*`` keys stripped.
+
+    Used at every upstream call site to prevent fingerprinting / leakage of
+    internal flags like ``x-headroom-bypass``, ``x-headroom-mode``,
+    ``x-headroom-user-id``, ``x-headroom-stack``, ``x-headroom-base-url``.
+    Case-insensitive on the prefix. Returns a NEW dict; never mutates the
+    caller's mapping. Pure function. No regex.
+
+    When the operator opt-in ``HEADROOM_STRIP_INTERNAL_HEADERS=disabled``
+    is set, returns a shallow copy unchanged. That mode is for diagnostic
+    shadow tracing only and is documented as a per-deploy choice.
+    """
+    mode = get_strip_internal_headers_mode()
+    if mode == "disabled":
+        # Always return a copy so callers can mutate without surprise.
+        return dict(headers)
+    return {k: v for k, v in headers.items() if not k.lower().startswith(_INTERNAL_HEADER_PREFIX)}
+
+
+def log_outbound_headers(
+    *,
+    forwarder: str,
+    stripped_count: int,
+    request_id: str | None,
+) -> None:
+    """Structured log line for every upstream forwarder header strip.
+
+    Emitted once per outbound request (paired with ``log_outbound_request``).
+    Per realignment build constraint #8 we log every cache-affecting
+    decision; per #8/#11 we never log header values, only the count of
+    stripped internal headers.
+    """
+    logger.info(
+        "event=outbound_headers forwarder=%s stripped_count=%d request_id=%s",
+        forwarder,
+        stripped_count,
+        request_id or "",
+    )
+
+
 async def _read_request_body_bytes(request: Request) -> bytes:
     """Read and (if needed) decompress the request body, returning raw UTF-8 bytes.
 
diff --git a/tests/test_header_isolation.py b/tests/test_header_isolation.py
new file mode 100644
index 000000000..6ce36bbed
--- /dev/null
+++ b/tests/test_header_isolation.py
@@ -0,0 +1,571 @@
+"""Header-isolation tests for PR-A5 (P5-49 fix).
+
+`x-headroom-*` request headers are internal control flags consumed by the
+proxy itself (bypass gating, mode selection, user-id, stack/base-url
+fingerprints). Forwarding them upstream:
+
+  1. Fingerprints the proxy to subscription-revocation enforcers.
+  2. Leaks user-id / stack / base-url internals to whichever vendor
+     terminates the request.
+
+PR-A5 wraps every handler-entry capture of the request headers with
+`_strip_internal_headers`. Inbound read paths (`request.headers.get(...)`
+for bypass gating, `_extract_tags` reading `x-headroom-*`) keep working
+because they never depended on the local outbound-bound dict.
+
+Operator opt-in `HEADROOM_STRIP_INTERNAL_HEADERS=disabled` keeps the
+internal headers in the upstream-bound dict for diagnostic shadow tracing.
+That mode is loud and explicit per realignment build constraint #4 — NOT
+a silent fallback.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+
+pytest.importorskip("fastapi")
+
+from fastapi.testclient import TestClient
+
+from headroom.proxy.helpers import (
+    _strip_internal_headers,
+    get_strip_internal_headers_mode,
+)
+from headroom.proxy.server import ProxyConfig, create_app
+
+# ---------------------------------------------------------------------------
+# Pure helper unit tests
+# ---------------------------------------------------------------------------
+
+
+def test_strip_returns_new_dict_does_not_mutate_caller() -> None:
+    """`_strip_internal_headers` is pure — caller's dict is untouched."""
+    original = {
+        "authorization": "Bearer x",
+        "x-headroom-bypass": "true",
+    }
+    out = _strip_internal_headers(original)
+    assert out is not original
+    assert "x-headroom-bypass" in original
+    assert "x-headroom-bypass" not in out
+    assert out["authorization"] == "Bearer x"
+
+
+def test_strip_x_headroom_bypass_removed() -> None:
+    out = _strip_internal_headers({"x-headroom-bypass": "true", "k": "v"})
+    assert "x-headroom-bypass" not in out
+    assert out["k"] == "v"
+
+
+def test_strip_x_headroom_mode_removed() -> None:
+    out = _strip_internal_headers({"x-headroom-mode": "passthrough", "k": "v"})
+    assert "x-headroom-mode" not in out
+    assert out["k"] == "v"
+
+
+def test_strip_x_headroom_user_id_removed() -> None:
+    out = _strip_internal_headers({"x-headroom-user-id": "u1", "k": "v"})
+    assert "x-headroom-user-id" not in out
+    assert out["k"] == "v"
+
+
+def test_strip_x_headroom_stack_removed() -> None:
+    out = _strip_internal_headers({"x-headroom-stack": "engineer", "k": "v"})
+    assert "x-headroom-stack" not in out
+    assert out["k"] == "v"
+
+
+def test_strip_x_headroom_base_url_removed() -> None:
+    out = _strip_internal_headers({"x-headroom-base-url": "http://x", "k": "v"})
+    assert "x-headroom-base-url" not in out
+    assert out["k"] == "v"
+
+
+def test_strip_case_insensitive_prefix_match() -> None:
+    """Mixed-case `X-Headroom-Foo`, `x-Headroom-Bar`, `X-HEADROOM-BAZ` all stripped."""
+    out = _strip_internal_headers(
+        {
+            "X-Headroom-Foo": "1",
+            "x-Headroom-Bar": "2",
+            "X-HEADROOM-BAZ": "3",
+            "Authorization": "Bearer x",
+        }
+    )
+    assert "X-Headroom-Foo" not in out
+    assert "x-Headroom-Bar" not in out
+    assert "X-HEADROOM-BAZ" not in out
+    assert out["Authorization"] == "Bearer x"
+
+
+def test_strip_legitimate_headers_passthrough() -> None:
+    """Headers without the internal prefix must NOT be stripped."""
+    out = _strip_internal_headers(
+        {
+            "Authorization": "Bearer x",
+            "x-api-key": "k",
+            "x-request-id": "rid-1",
+            "x-trace-id": "tid-1",
+            "Content-Type": "application/json",
+            "anthropic-version": "2023-06-01",
+            "User-Agent": "claude-code/1.0",
+        }
+    )
+    assert out["Authorization"] == "Bearer x"
+    assert out["x-api-key"] == "k"
+    assert out["x-request-id"] == "rid-1"
+    assert out["x-trace-id"] == "tid-1"
+    assert out["Content-Type"] == "application/json"
+    assert out["anthropic-version"] == "2023-06-01"
+    assert out["User-Agent"] == "claude-code/1.0"
+
+
+def test_strip_disabled_mode_passes_internal_headers_through(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """`HEADROOM_STRIP_INTERNAL_HEADERS=disabled` is operator opt-in for diag."""
+    monkeypatch.setenv("HEADROOM_STRIP_INTERNAL_HEADERS", "disabled")
+    out = _strip_internal_headers({"x-headroom-bypass": "true", "authorization": "Bearer x"})
+    assert out["x-headroom-bypass"] == "true"
+    assert out["authorization"] == "Bearer x"
+
+
+def test_strip_disabled_mode_returns_copy_not_alias(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Even in disabled mode the helper returns a NEW dict, never an alias."""
+    monkeypatch.setenv("HEADROOM_STRIP_INTERNAL_HEADERS", "disabled")
+    src = {"x-headroom-bypass": "true"}
+    out = _strip_internal_headers(src)
+    assert out is not src
+
+
+def test_strip_mode_default_is_enabled(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("HEADROOM_STRIP_INTERNAL_HEADERS", raising=False)
+    assert get_strip_internal_headers_mode() == "enabled"
+
+
+def test_strip_mode_invalid_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("HEADROOM_STRIP_INTERNAL_HEADERS", "garbage")
+    with pytest.raises(ValueError, match="HEADROOM_STRIP_INTERNAL_HEADERS"):
+        get_strip_internal_headers_mode()
+
+
+def test_strip_empty_dict_returns_empty_dict() -> None:
+    assert _strip_internal_headers({}) == {}
+
+
+def test_strip_preserves_value_semantics() -> None:
+    """Values are forwarded unchanged for kept headers."""
+    out = _strip_internal_headers(
+        {
+            "Authorization": "Bearer sk-ant-...",
+            "anthropic-version": "2023-06-01",
+        }
+    )
+    assert out["Authorization"] == "Bearer sk-ant-..."
+    assert out["anthropic-version"] == "2023-06-01"
+
+
+# ---------------------------------------------------------------------------
+# End-to-end: x-headroom-* never reaches the upstream
+# ---------------------------------------------------------------------------
+
+
+class _CapturingTransport(httpx.AsyncBaseTransport):
+    def __init__(self) -> None:
+        self.captured_headers: dict[str, str] | None = None
+        self.captured_body: bytes | None = None
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        body = b""
+        async for chunk in request.stream:
+            body += chunk
+        self.captured_body = body
+        self.captured_headers = dict(request.headers.items())
+        return httpx.Response(
+            200,
+            json={
+                "id": "msg_1",
+                "type": "message",
+                "role": "assistant",
+                "content": [{"type": "text", "text": "ok"}],
+                "usage": {
+                    "input_tokens": 10,
+                    "output_tokens": 3,
+                    "cache_read_input_tokens": 0,
+                    "cache_creation_input_tokens": 0,
+                },
+            },
+        )
+
+
+class _FakePrefixTracker:
+    def __init__(self, frozen_count: int = 0):
+        self._frozen_count = frozen_count
+        self._cached_token_count = 0
+        self._last_original_messages: list = []
+        self._last_forwarded_messages: list = []
+
+    def get_frozen_message_count(self) -> int:
+        return self._frozen_count
+
+    def get_last_original_messages(self):  # noqa: ANN201
+        return list(self._last_original_messages)
+
+    def get_last_forwarded_messages(self):  # noqa: ANN201
+        return list(self._last_forwarded_messages)
+
+    def update_from_response(self, **kwargs):  # noqa: ANN003
+        self._last_original_messages = kwargs.get("original_messages", kwargs.get("messages", []))
+        self._last_forwarded_messages = kwargs.get("messages", [])
+        return None
+
+
+def _make_anthropic_app() -> tuple[TestClient, _CapturingTransport]:
+    config = ProxyConfig(
+        optimize=False,
+        cache_enabled=False,
+        rate_limit_enabled=False,
+        cost_tracking_enabled=False,
+        log_requests=False,
+        ccr_inject_tool=False,
+        ccr_handle_responses=False,
+        ccr_context_tracking=False,
+        image_optimize=False,
+    )
+    app = create_app(config)
+    proxy = app.state.proxy
+    transport = _CapturingTransport()
+    proxy.http_client = httpx.AsyncClient(transport=transport)
+
+    fake_tracker = _FakePrefixTracker(frozen_count=0)
+    proxy.session_tracker_store.compute_session_id = lambda request, model, messages: "s1"
+    proxy.session_tracker_store.get_or_create = lambda session_id, provider: fake_tracker
+
+    return TestClient(app), transport
+
+
+def test_x_headroom_bypass_not_forwarded() -> None:
+    client, transport = _make_anthropic_app()
+    resp = client.post(
+        "/v1/messages",
+        headers={
+            "x-api-key": "test-key",
+            "anthropic-version": "2023-06-01",
+            "content-type": "application/json",
+            "x-headroom-bypass": "true",
+        },
+        json={
+            "model": "claude-sonnet-4-6",
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert transport.captured_headers is not None
+    upstream = {k.lower(): v for k, v in transport.captured_headers.items()}
+    assert "x-headroom-bypass" not in upstream
+    # Legitimate headers must reach upstream.
+    assert upstream.get("x-api-key") == "test-key"
+    assert upstream.get("anthropic-version") == "2023-06-01"
+
+
+def test_x_headroom_mode_not_forwarded() -> None:
+    client, transport = _make_anthropic_app()
+    resp = client.post(
+        "/v1/messages",
+        headers={
+            "x-api-key": "test-key",
+            "anthropic-version": "2023-06-01",
+            "content-type": "application/json",
+            "x-headroom-mode": "passthrough",
+        },
+        json={
+            "model": "claude-sonnet-4-6",
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert transport.captured_headers is not None
+    upstream = {k.lower(): v for k, v in transport.captured_headers.items()}
+    assert "x-headroom-mode" not in upstream
+
+
+def test_x_headroom_user_id_not_forwarded() -> None:
+    client, transport = _make_anthropic_app()
+    resp = client.post(
+        "/v1/messages",
+        headers={
+            "x-api-key": "test-key",
+            "anthropic-version": "2023-06-01",
+            "content-type": "application/json",
+            "x-headroom-user-id": "alice",
+        },
+        json={
+            "model": "claude-sonnet-4-6",
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert transport.captured_headers is not None
+    upstream = {k.lower(): v for k, v in transport.captured_headers.items()}
+    assert "x-headroom-user-id" not in upstream
+
+
+def test_x_headroom_stack_not_forwarded() -> None:
+    client, transport = _make_anthropic_app()
+    resp = client.post(
+        "/v1/messages",
+        headers={
+            "x-api-key": "test-key",
+            "anthropic-version": "2023-06-01",
+            "content-type": "application/json",
+            "x-headroom-stack": "engineer",
+        },
+        json={
+            "model": "claude-sonnet-4-6",
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert transport.captured_headers is not None
+    upstream = {k.lower(): v for k, v in transport.captured_headers.items()}
+    assert "x-headroom-stack" not in upstream
+
+
+def test_x_headroom_base_url_not_forwarded() -> None:
+    client, transport = _make_anthropic_app()
+    resp = client.post(
+        "/v1/messages",
+        headers={
+            "x-api-key": "test-key",
+            "anthropic-version": "2023-06-01",
+            "content-type": "application/json",
+            "x-headroom-base-url": "https://override.example",
+        },
+        json={
+            "model": "claude-sonnet-4-6",
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert transport.captured_headers is not None
+    upstream = {k.lower(): v for k, v in transport.captured_headers.items()}
+    assert "x-headroom-base-url" not in upstream
+
+
+def test_case_insensitive_prefix_match_e2e() -> None:
+    """Mixed-case headers are stripped end-to-end."""
+    client, transport = _make_anthropic_app()
+    resp = client.post(
+        "/v1/messages",
+        headers={
+            "x-api-key": "test-key",
+            "anthropic-version": "2023-06-01",
+            "content-type": "application/json",
+            "X-Headroom-Foo": "1",
+            "x-Headroom-Bar": "2",
+            "X-HEADROOM-BAZ": "3",
+        },
+        json={
+            "model": "claude-sonnet-4-6",
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert transport.captured_headers is not None
+    upstream_lower = {k.lower(): v for k, v in transport.captured_headers.items()}
+    assert "x-headroom-foo" not in upstream_lower
+    assert "x-headroom-bar" not in upstream_lower
+    assert "x-headroom-baz" not in upstream_lower
+
+
+def test_legitimate_headers_passthrough_e2e() -> None:
+    """Authorization / x-api-key / x-request-id / Content-Type all preserved."""
+    client, transport = _make_anthropic_app()
+    resp = client.post(
+        "/v1/messages",
+        headers={
+            "Authorization": "Bearer sk-ant-test",
+            "x-api-key": "test-key",
+            "anthropic-version": "2023-06-01",
+            "content-type": "application/json",
+            "x-request-id": "rid-abc",
+            "User-Agent": "claude-code/1.2.3",
+        },
+        json={
+            "model": "claude-sonnet-4-6",
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert transport.captured_headers is not None
+    upstream = {k.lower(): v for k, v in transport.captured_headers.items()}
+    assert upstream.get("x-api-key") == "test-key"
+    assert upstream.get("anthropic-version") == "2023-06-01"
+    assert upstream.get("user-agent") == "claude-code/1.2.3"
+    # Authorization is delivered uppercased; httpx normalizes to lowercase
+    # when reading via dict()-of-Headers, so check via lowercase key.
+    assert upstream.get("authorization") == "Bearer sk-ant-test"
+
+
+def test_inbound_read_path_still_reads_x_headroom_bypass() -> None:
+    """Bypass header still gates compression even though it's stripped from upstream.
+
+    The handler reads `request.headers.get('x-headroom-bypass')` directly.
+    Stripping the local outbound-bound `headers` dict does NOT affect that
+    inbound read path.
+    """
+    config = ProxyConfig(
+        # Compression is the only thing bypass affects observably; turn it on.
+        optimize=True,
+        cache_enabled=False,
+        rate_limit_enabled=False,
+        cost_tracking_enabled=False,
+        log_requests=False,
+        ccr_inject_tool=False,
+        ccr_handle_responses=False,
+        ccr_context_tracking=False,
+        image_optimize=False,
+    )
+    app = create_app(config)
+    proxy = app.state.proxy
+    transport = _CapturingTransport()
+    proxy.http_client = httpx.AsyncClient(transport=transport)
+
+    fake_tracker = _FakePrefixTracker(frozen_count=0)
+    proxy.session_tracker_store.compute_session_id = lambda request, model, messages: "s_bypass"
+    proxy.session_tracker_store.get_or_create = lambda session_id, provider: fake_tracker
+
+    client = TestClient(app)
+
+    # Force a bypass via header. The handler logs "Bypass: skipping compression"
+    # if the inbound read worked; we can't easily intercept the log, so we
+    # primarily assert that:
+    #   1. The request still succeeds.
+    #   2. The upstream did NOT receive the `x-headroom-bypass` header.
+    resp = client.post(
+        "/v1/messages",
+        headers={
+            "x-api-key": "test-key",
+            "anthropic-version": "2023-06-01",
+            "content-type": "application/json",
+            "x-headroom-bypass": "true",
+        },
+        json={
+            "model": "claude-sonnet-4-6",
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+    assert resp.status_code == 200
+    upstream = {k.lower(): v for k, v in (transport.captured_headers or {}).items()}
+    assert "x-headroom-bypass" not in upstream
+
+
+def test_disabled_mode_passes_through_e2e(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """`HEADROOM_STRIP_INTERNAL_HEADERS=disabled` lets internal headers through."""
+    monkeypatch.setenv("HEADROOM_STRIP_INTERNAL_HEADERS", "disabled")
+    client, transport = _make_anthropic_app()
+    resp = client.post(
+        "/v1/messages",
+        headers={
+            "x-api-key": "test-key",
+            "anthropic-version": "2023-06-01",
+            "content-type": "application/json",
+            "x-headroom-mode": "passthrough",
+        },
+        json={
+            "model": "claude-sonnet-4-6",
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert transport.captured_headers is not None
+    upstream = {k.lower(): v for k, v in transport.captured_headers.items()}
+    # Operator opt-in: internal header IS forwarded (diagnostic mode).
+    assert upstream.get("x-headroom-mode") == "passthrough"
+
+
+# ---------------------------------------------------------------------------
+# OpenAI Chat Completions parity check
+# ---------------------------------------------------------------------------
+
+
+def test_openai_chat_x_headroom_bypass_not_forwarded() -> None:
+    """OpenAI handler also strips x-headroom-* before upstream call."""
+    config = ProxyConfig(
+        optimize=False,
+        cache_enabled=False,
+        rate_limit_enabled=False,
+        cost_tracking_enabled=False,
+        log_requests=False,
+        ccr_inject_tool=False,
+        ccr_handle_responses=False,
+        ccr_context_tracking=False,
+        image_optimize=False,
+    )
+    app = create_app(config)
+    proxy = app.state.proxy
+
+    captured: dict[str, object] = {}
+
+    async def _fake_retry(method, url, headers, body, stream=False, **kwargs):  # noqa: ANN001
+        captured["headers"] = dict(headers)
+        return httpx.Response(
+            200,
+            json={
+                "id": "chatcmpl_1",
+                "object": "chat.completion",
+                "model": "gpt-4o",
+                "choices": [
+                    {
+                        "index": 0,
+                        "message": {"role": "assistant", "content": "ok"},
+                        "finish_reason": "stop",
+                    }
+                ],
+                "usage": {"prompt_tokens": 10, "completion_tokens": 1, "total_tokens": 11},
+            },
+        )
+
+    proxy._retry_request = _fake_retry  # type: ignore[attr-defined]
+    proxy.memory_handler = SimpleNamespace(
+        config=SimpleNamespace(inject_context=False, inject_tools=False),
+        search_and_format_context=AsyncMock(return_value=""),
+        has_memory_tool_calls=lambda resp, provider: False,
+    )
+
+    client = TestClient(app)
+    resp = client.post(
+        "/v1/chat/completions",
+        headers={
+            "authorization": "Bearer sk-test",
+            "x-headroom-bypass": "true",
+            "x-headroom-user-id": "u1",
+        },
+        json={
+            "model": "gpt-4o",
+            "messages": [{"role": "user", "content": "hi"}],
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    sent_headers_raw = captured.get("headers")
+    assert isinstance(sent_headers_raw, dict)
+    sent_headers = {k.lower(): v for k, v in sent_headers_raw.items()}
+    assert "x-headroom-bypass" not in sent_headers
+    assert "x-headroom-user-id" not in sent_headers
+    assert sent_headers.get("authorization") == "Bearer sk-test"