Propagate request correlation context

Cargo.toml

@@ -41,6 +41,7 @@ tracing-opentelemetry = "0.33.0"
 opentelemetry = { version = "0.32", default-features = false, features = ["trace", "metrics", "internal-logs"] }
 opentelemetry_sdk = { version = "0.32", default-features = false, features = ["rt-tokio", "trace", "metrics", "internal-logs"] }
 tokio = { version = "1.53.1", features = ["macros", "net", "rt-multi-thread", "signal", "sync", "time"] }
+tower = "0.5.3"
 tower-http = "0.7.1"
 http = "1.5.0"
 redis = { version = "1.7.0", default-features = false, features = [

_context/wiki/architecture.md

@@ -10,6 +10,7 @@ Tower layers execute outside-in:
 ```text
 TCP/TLS listener
   -> HttpMetricsLayer
+  -> request_context_layer      preserves/generates X-Correlation-ID
   -> TraceLayer (extract incoming trace context)
   -> /contextforge-rs nested router
   -> CORS layer
@@ -84,6 +85,7 @@ into successful response hooks. See [Routing](routing.md) and
 | State | Owner | Lifetime |
 | --- | --- | --- |
 | Parsed config and shared upstream HTTP client | Gateway | Process. |
+| Correlation ID | Request context and HTTP request extensions | One request; echoed downstream and propagated to the selected backend. |
 | JWKS keys | JWT authorization service | Five-minute cache; fetched when verification needs them. |
 | User config | Redis store and optional local LRU | Redis is authoritative; local capacity 50,000, default expiry 60 seconds. |
 | Principal, claims, virtual-host ID, config snapshot | HTTP request extensions | One request. |

_context/wiki/config.md

@@ -181,8 +181,8 @@ Integer values are limited to the IEEE 754 safe range.
 
 **Header apply order:** backend Host for HTTPS → configured passthrough →
 automatic `Mcp-Param-*` forwarding → `add_headers` → `remove_headers` → current
-trace-context injection. RMCP generates the outbound method, name, and protocol
-headers for the routed request.
+request-correlation and trace-context injection. RMCP generates the outbound
+method, name, and protocol headers for the routed request.
 
 Passthrough values come from the current HTTP request. A new backend transport
 is constructed per routed operation; no initialization-time header snapshot is
@@ -195,7 +195,7 @@ reused across requests.
 | Body-framing | `Content-Length`, `Content-Type` |
 | Hop-by-hop | `Connection`, `Keep-Alive`, `Proxy-Authenticate`, `Proxy-Authorization`, `Proxy-Connection`, `TE`, `Trailer`, `Trailers`, `Transfer-Encoding`, `Upgrade` |
 | RMCP-reserved | `Mcp-Session-Id`, `Accept`, `Last-Event-Id` |
-| Gateway-managed | `Host` (set from backend URL host + port; never overridden by config) |
+| Gateway-managed | `Host` (set from backend URL host + port) and `X-Correlation-ID` (preserved or generated per request); never overridden by config |
 | MCP standard | `Mcp-Method`, `Mcp-Name`, `Mcp-Protocol-Version`, `Mcp-Param-*` |
 
 `Authorization` and `Cookie` are not protected here because backend
@@ -448,9 +448,11 @@ scraping; the five-minute query window tolerates sparse samples.
 
 Incoming W3C trace context is extracted by `ExtractingMakeSpan` and current
 context is injected into each backend request after configured header changes.
-This propagation is implemented, not a future gap.
+`X-Correlation-ID` is preserved when valid or generated when absent, scoped to
+the request, returned on the response, and injected into the selected backend
+request. This propagation is implemented, not a future gap.
 
-HTTP spans carry method, URI, and version. Authentication, configuration,
+HTTP spans carry method, path, and version. Authentication, configuration,
 routed operations, and CPEX also have instrumentation. A complete MCP semantic
 attribute set and coverage of every operation are still separate work; do not
 interpret HTTP tracing alone as full MCP observability.

_context/wiki/routing.md

@@ -51,9 +51,9 @@ does not implement template listing or dynamic URI matching.
 
 ## Header forwarding
 
-Applied in order per upstream call: Host (from backend URL, HTTPS only) → passthrough (`BackendMCPGateway::passthrough_headers`) → `Mcp-Param-*` auto-forward → add (`add_headers`, overrides passthrough) → remove (`remove_headers`) → current trace-context injection.
+Applied in order per upstream call: Host (from backend URL, HTTPS only) → passthrough (`BackendMCPGateway::passthrough_headers`) → `Mcp-Param-*` auto-forward → add (`add_headers`, overrides passthrough) → remove (`remove_headers`) → current request-correlation and trace-context injection.
 
-Protected headers that config can never touch: `Host`, `Content-Length`, `Content-Type`, all RFC 7230 hop-by-hop headers, `Mcp-Session-Id`, `Accept`, `Last-Event-Id`, and all computed MCP standard headers (`Mcp-Method`, `Mcp-Name`, `Mcp-Protocol-Version`, `Mcp-Param-*`).
+Protected headers that config can never touch: `Host`, `X-Correlation-ID`, `Content-Length`, `Content-Type`, all RFC 7230 hop-by-hop headers, `Mcp-Session-Id`, `Accept`, `Last-Event-Id`, and all computed MCP standard headers (`Mcp-Method`, `Mcp-Name`, `Mcp-Protocol-Version`, `Mcp-Param-*`).
 
 For clients on `≥ 2026-07-28`, `call_tool` validates `Mcp-Param-*` headers against `BackendMCPGateway::tool_schemas` when a schema is published, before
 plugins or backend I/O. Without a schema the headers pass through without local

crates/contextforge-data-plane-lib/Cargo.toml

@@ -24,7 +24,7 @@ tokio-util = "0.7"
 axum.workspace = true
 axum-otel-metrics = "0.14"
 tower-http = { workspace = true, features = ["cors", "trace"] }
-tower = "0.5.3"
+tower.workspace = true
 http.workspace = true
 futures = { version = "0.3", features = ["std", "alloc"] }
 jsonwebtoken.workspace = true

crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs

@@ -41,6 +41,8 @@ pub(super) async fn connect_backend_for_request(
 ) -> Result<RunningService<RoleClient, GatewayBackendClient>, ErrorData> {
     let mut headers = HashMap::new();
     let downstream_headers = cx.extensions.get::<Parts>().map(|parts| &parts.headers);
+    let request_context =
+        cx.extensions.get::<Parts>().and_then(|parts| parts.extensions.get::<crate::RequestObservabilityContext>());
 
     if let Some(host) = backend.url.host_str()
         && backend.url.scheme() == "https"
@@ -54,7 +56,7 @@ pub(super) async fn connect_backend_for_request(
     }
 
     apply_header_config(&mut headers, backend, downstream_headers);
-    contextforge_data_plane_observability::inject_current_context(&mut headers);
+    contextforge_data_plane_observability::inject_current_context(&mut headers, request_context);
 
     let config = StreamableHttpClientTransportConfig::with_uri(backend.url.to_string()).custom_headers(headers);
     let transport = StreamableHttpClientTransport::with_client(mcp_service.http_client.clone(), config);
@@ -131,7 +133,7 @@ fn apply_header_config(
 }
 
 /// Returns `true` for headers that config must never touch:
-/// - Gateway-managed: `Host`
+/// - Gateway-managed: `Host`, `X-Correlation-ID`
 /// - Body-framing: `Content-Length`, `Content-Type` (gateway owns framing; forwarding corrupts body or enables encoding-dispatch bypass)
 /// - Hop-by-hop (RFC 7230 §6.1): `Connection`, `Keep-Alive`, `Proxy-Authenticate`, `Proxy-Authorization`, `TE`, `Trailer`, `Trailers`, `Transfer-Encoding`, `Upgrade`
 /// - Non-standard hop-by-hop: `Proxy-Connection` (must not cross gateway boundary)
@@ -142,6 +144,7 @@ fn apply_header_config(
 fn is_protected_header(name: &http::HeaderName) -> bool {
     const PROTECTED: &[&str] = &[
         "host",
+        "x-correlation-id",
         // body-framing: gateway owns these; forwarding corrupts framing or enables encoding-dispatch bypass
         "content-length",
         "content-type",
@@ -283,6 +286,17 @@ mod tests {
         assert!(headers.is_empty(), "no RMCP-reserved header must reach the upstream config");
     }
 
+    #[test]
+    fn correlation_id_cannot_be_configured_as_a_backend_header() {
+        let mut headers = HashMap::new();
+        let ds = downstream(&[("X-Correlation-ID", "downstream-id")]);
+        let cfg = backend(&["x-correlation-id"], &[("X-Correlation-ID", "configured-id")], &["x-correlation-id"]);
+
+        apply_header_config(&mut headers, &cfg, Some(&ds));
+
+        assert!(headers.is_empty(), "request correlation is owned by the gateway");
+    }
+
     #[test]
     fn mcp_param_headers_are_forwarded_but_cannot_be_changed_by_backend_config() {
         let mut headers = HashMap::new();

crates/contextforge-data-plane-lib/src/lib.rs

@@ -37,6 +37,9 @@ pub use config_stores::{ConfigStore, ConfigStoreError, get_global_config};
 pub use config_stores::set_global_config;
 
 pub use crate::common::*;
+pub use contextforge_data_plane_observability::{
+    CORRELATION_ID_HEADER, RequestObservabilityContext, current_request_context,
+};
 
 pub type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
 pub type Result<T> = std::result::Result<T, Error>;
@@ -167,6 +170,7 @@ impl Gateway {
         let app = axum::Router::new()
             .nest("/contextforge-rs", app)
             .layer(TraceLayer::new_for_http().make_span_with(contextforge_data_plane_observability::ExtractingMakeSpan))
+            .layer(middleware::from_fn(contextforge_data_plane_observability::request_context_layer))
             .layer(HttpMetricsLayerBuilder::new().build());
 
         Ok(app)

crates/contextforge-data-plane-lib/tests/gateway.rs

@@ -7,6 +7,8 @@ mod compatibility;
 mod completions;
 #[path = "gateway/future_contracts/mod.rs"]
 mod future_contracts;
+#[path = "gateway/observability.rs"]
+mod observability;
 #[path = "gateway/plugins.rs"]
 mod plugins;
 #[path = "gateway/prompts.rs"]

crates/contextforge-data-plane-lib/tests/gateway/observability.rs

@@ -0,0 +1,40 @@
+use std::sync::Arc;
+
+use contextforge_data_plane_cpex::CpexRuntimeRegistry;
+use contextforge_data_plane_lib::{CORRELATION_ID_HEADER, Result};
+
+use crate::harness::{TEST_USER_ID, connect_modern_client, modern_client_info, start_gateway, sum_request, token};
+
+const CORRELATION_ID: &str = "gateway-request-123";
+
+fn client_with_correlation_id() -> reqwest::Client {
+    let mut headers = http::HeaderMap::new();
+    headers.insert(
+        http::header::AUTHORIZATION,
+        http::HeaderValue::from_str(&format!("Bearer {}", token(TEST_USER_ID))).expect("valid authorization header"),
+    );
+    headers.insert(CORRELATION_ID_HEADER, http::HeaderValue::from_static(CORRELATION_ID));
+    reqwest::Client::builder().default_headers(headers).build().expect("client should build")
+}
+
+#[tokio::test]
+async fn correlation_id_is_propagated_to_the_backend() -> Result<()> {
+    let gateway = start_gateway(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await;
+    let service =
+        connect_modern_client(gateway.gateway_url(), client_with_correlation_id(), modern_client_info()).await;
+
+    service.call_tool(sum_request("sum", 1, 2)).await?;
+
+    let backend_headers = gateway
+        .backend_state
+        .request_headers
+        .lock()
+        .expect("backend request headers lock poisoned")
+        .last()
+        .cloned()
+        .expect("backend should receive the tool call");
+    assert_eq!(backend_headers[CORRELATION_ID_HEADER], CORRELATION_ID);
+
+    drop(service);
+    gateway.shutdown().await
+}

crates/contextforge-data-plane-observability/Cargo.toml

@@ -10,18 +10,21 @@ readme.workspace = true
 repository.workspace = true
 
 [dependencies]
+axum.workspace = true
 http.workspace = true
 opentelemetry.workspace = true
 opentelemetry-otlp = { version = "0.32", default-features = false, features = ["grpc-tonic", "http-proto", "reqwest-blocking-client", "trace", "metrics", "internal-logs", "tls-aws-lc"] }
 opentelemetry_sdk.workspace = true
 tonic = { version = "0.14", default-features = false }
+tokio.workspace = true
 tower-http = { workspace = true, features = ["trace"] }
 tracing.workspace = true
 tracing-opentelemetry.workspace = true
 tracing-subscriber = { version = "0.3", default-features = false, features = ["env-filter", "fmt", "tracing-log"] }
+uuid.workspace = true
 
 [dev-dependencies]
-tokio.workspace = true
+tower.workspace = true
 
 [lints]
 workspace = true

crates/contextforge-data-plane-observability/src/lib.rs

@@ -9,5 +9,8 @@ mod runtime;
 mod traces;
 
 pub use config::{ObservabilityConfig, OtlpProtocol};
-pub use propagation::{ExtractingMakeSpan, inject_current_context};
+pub use propagation::{
+    CORRELATION_ID_HEADER, ExtractingMakeSpan, RequestObservabilityContext, current_request_context,
+    inject_current_context, request_context_layer,
+};
 pub use runtime::{Guard, init_tracing_logging};