fix(adapters): upstream response decode fidelity (FWD-07, FWD-08, FWD-10, FWD-12) — core, tests follow

src/adapters/forward.rs

@@ -16,7 +16,6 @@ use std::sync::Arc;
 use alkcall::protocol::wire::{CallError, ResponseEnvelope};
 use alkcall::registry::context::OperationContext;
 use alkcall::registry::registration::ResponseStream;
-use alkcall::registry::spec::OperationType;
 use futures::stream;
 use futures::StreamExt;
 use percent_encoding::{utf8_percent_encode, AsciiSet, CONTROLS};
@@ -64,7 +63,7 @@ pub(crate) fn build_request(
             if key == "body" {
                 body = Some(value.clone());
             } else {
-                query_params.push((key.clone(), value_to_query(value)));
+                query_params.push((key.clone(), value_to_path_segment(value)));
             }
         }
     }
@@ -80,12 +79,17 @@ pub(crate) fn build_request(
 
     let mut headers = HeaderMap::new();
     for (k, v) in default_headers {
-        if let (Ok(name), Ok(value)) = (
+        let name = HeaderName::try_from(k.as_str()).map_err(|_| {
-            HeaderName::try_from(k.as_str()),
+            CallError::internal(format!(
-            HeaderValue::try_from(v.as_str()),
+                "default header `{k}` is not a valid HTTP header name; refusing to build a request that silently drops it"
-        ) {
+            ))
-            headers.insert(name, value);
+        })?;
-        }
+        let value = HeaderValue::try_from(v.as_str()).map_err(|_| {
+            CallError::internal(format!(
+                "default header `{k}` has an invalid value (control or non-ASCII bytes are not permitted); refusing to build a request that silently drops it"
+            ))
+        })?;
+        headers.insert(name, value);
     }
 
     if body.is_some() {
@@ -97,24 +101,36 @@ pub(crate) fn build_request(
             let credential = secret.expose_secret().clone();
             match scheme {
                 HttpAuthScheme::Bearer => {
-                    let header_value = format!("Bearer {credential}");
+                    let value =
-                    if let Ok(value) = HeaderValue::try_from(header_value) {
+                        HeaderValue::try_from(format!("Bearer {credential}")).map_err(|_| {
-                        headers.insert(AUTHORIZATION, value);
+                            CallError::internal(format!(
-                    }
+                                "credential for namespace `{namespace}` contains invalid HTTP header characters (control or non-ASCII bytes); refusing to send the request unauthenticated"
+                            ))
+                        })?;
+                    headers.insert(AUTHORIZATION, value);
                 }
                 HttpAuthScheme::ApiKey { header_name } => {
-                    if let (Ok(name), Ok(value)) = (
+                    let name =
-                        HeaderName::try_from(header_name.as_str()),
+                        HeaderName::try_from(header_name.as_str()).map_err(|_| {
-                        HeaderValue::try_from(credential.as_str()),
+                            CallError::internal(format!(
-                    ) {
+                                "API-key auth for namespace `{namespace}` declares invalid header name `{header_name}`; refusing to send the request unauthenticated"
-                        headers.insert(name, value);
+                            ))
-                    }
+                        })?;
+                    let value = HeaderValue::try_from(credential.as_str()).map_err(|_| {
+                        CallError::internal(format!(
+                            "credential for namespace `{namespace}` contains invalid HTTP header characters (control or non-ASCII bytes); refusing to send the request unauthenticated"
+                        ))
+                    })?;
+                    headers.insert(name, value);
                 }
                 HttpAuthScheme::Basic => {
-                    let header_value = format!("Basic {credential}");
+                    let value =
-                    if let Ok(value) = HeaderValue::try_from(header_value) {
+                        HeaderValue::try_from(format!("Basic {credential}")).map_err(|_| {
-                        headers.insert(AUTHORIZATION, value);
+                            CallError::internal(format!(
-                    }
+                                "credential for namespace `{namespace}` contains invalid HTTP header characters (control or non-ASCII bytes); refusing to send the request unauthenticated"
+                            ))
+                        })?;
+                    headers.insert(AUTHORIZATION, value);
                 }
             }
         }
@@ -171,6 +187,10 @@ const PATH_AFTER_PERCENT_ENCODE_SET: &AsciiSet = &CONTROLS
 /// encoded so that a rendered value stays one literal path segment
 /// (FWD-01 traversal/ smuggled-segments gate). Path parameters are never
 /// rejected: they are rendered safely instead.
+/// Raw string form of a scalar input value, used for both path-template
+/// rendering and query-pair emission (FWD-12: the two were byte-identical
+/// helpers; they are one function now, and percent-encoding is applied by
+/// the path renderer and `url::query_pairs_mut` respectively).
 pub(crate) fn value_to_path_segment(value: &Value) -> String {
     let raw = match value {
         Value::String(s) => s.clone(),
@@ -311,13 +331,133 @@ fn assemble_request_url(base_url: &str, rendered_path: &str) -> Result<Url, Call
     Ok(url)
 }
 
-pub(crate) fn value_to_query(value: &Value) -> String {
+#[derive(Debug, thiserror::Error)]
-    match value {
+pub(crate) enum BodyReadError {
-        Value::String(s) => s.clone(),
+    #[error("upstream response body exceeds the {RESPONSE_BODY_CAP}-byte response cap")]
-        Value::Number(n) => n.to_string(),
+    TooLarge,
-        Value::Bool(b) => b.to_string(),
+    #[error("transport error reading response body: {0}")]
-        Value::Null => String::new(),
+    Transport(reqwest::Error),
-        other => other.to_string(),
+    #[error("malformed response body: {0}")]
+    Decode(serde_json::Error),
+}
+
+/// True when a response `Content-Type` header is JSON by mime-essence
+/// semantics: type `application`, subtype `json` or a `+json` structured
+/// suffix (`application/vnd.api+json`, `application/problem+json`, …).
+/// Parameters (`; charset=…`) are ignored.
+pub(crate) fn is_json_content_type(content_type: &str) -> bool {
+    let essence = content_type
+        .split(';')
+        .next()
+        .unwrap_or_default()
+        .trim()
+        .to_ascii_lowercase();
+    match essence.split_once('/') {
+        Some(("application", subtype)) => subtype == "json" || subtype.ends_with("+json"),
+        _ => false,
+    }
+}
+
+/// Bounded string form of an upstream error body for the error envelope
+/// (FWD-10): capped at [`ERROR_BODY_ECHO_CAP`] bytes, lossily decoded,
+/// control characters (which could forge log or display framing) elided,
+/// and truncated with a marker. The echo is never logged by this crate.
+fn bounded_error_body(bytes: bytes::Bytes) -> Option<String> {
+    if bytes.is_empty() {
+        return None;
+    }
+    let truncated = bytes.len() >= ERROR_BODY_ECHO_CAP;
+    let text = String::from_utf8_lossy(&bytes);
+    let printable: String = text
+        .chars()
+        .filter(|c| !c.is_control() || *c == '\n' || *c == '\t')
+        .take(ERROR_BODY_ECHO_CAP)
+        .collect();
+    if printable.is_empty() {
+        None
+    } else if truncated {
+        Some(format!("{printable}\n[truncated]"))
+    } else {
+        Some(printable)
+    }
+}
+
+/// Reads the response body, byte-capped: any read that would push the
+/// accumulated bytes past `cap` returns [`BodyReadError::TooLarge`], so
+/// a hostile upstream cannot grow caller memory past the cap (FWD-07).
+async fn read_body_capped(
+    response: reqwest::Response,
+    cap: usize,
+) -> Result<bytes::Bytes, BodyReadError> {
+    let mut stream = response.bytes_stream();
+    let mut buf: Vec<u8> = Vec::new();
+    while let Some(chunk) = stream.next().await {
+        let chunk = chunk.map_err(BodyReadError::Transport)?;
+        if buf.len().saturating_add(chunk.len()) > cap {
+            return Err(BodyReadError::TooLarge);
+        }
+        buf.extend_from_slice(&chunk);
+    }
+    Ok(buf.into())
+}
+
+/// Reads a `200`-class response into a [`ResponseEnvelope`], dispatching
+/// on the mime essence of its `Content-Type` (FWD-07): `application/json`
+/// and `application/*+json` decode as JSON, `text/*` as a string, and
+/// everything else as a byte array — every path byte-capped at
+/// [`RESPONSE_BODY_CAP`].
+async fn success_envelope(response: reqwest::Response, request_id: &str) -> ResponseEnvelope {
+    let content_type = response
+        .headers()
+        .get(reqwest::header::CONTENT_TYPE)
+        .and_then(|v: &reqwest::header::HeaderValue| v.to_str().ok())
+        .unwrap_or_default()
+        .to_ascii_lowercase();
+    let essence = content_type.split(';').next().unwrap_or_default().trim();
+
+    let read = if is_json_content_type(&content_type) {
+        read_body_capped(response, RESPONSE_BODY_CAP)
+            .await
+            .and_then(|bytes| {
+                serde_json::from_slice::<Value>(&bytes)
+                    .map(|v| ResponseEnvelope::ok(request_id, v))
+                    .map_err(BodyReadError::Decode)
+            })
+    } else if essence.starts_with("text/") {
+        read_body_capped(response, RESPONSE_BODY_CAP)
+            .await
+            .map(|bytes| {
+                ResponseEnvelope::ok(
+                    request_id,
+                    Value::String(String::from_utf8_lossy(&bytes).into_owned()),
+                )
+            })
+    } else {
+        read_body_capped(response, RESPONSE_BODY_CAP)
+            .await
+            .map(|bytes| {
+                let arr: Vec<Value> = bytes
+                    .iter()
+                    .map(|byte| Value::Number((*byte).into()))
+                    .collect();
+                ResponseEnvelope::ok(request_id, Value::Array(arr))
+            })
+    };
+
+    match read {
+        Ok(envelope) => envelope,
+        Err(BodyReadError::TooLarge) => ResponseEnvelope::error(
+            request_id,
+            CallError::new(
+                "HTTP_413",
+                format!("upstream response body exceeds the {RESPONSE_BODY_CAP}-byte response cap"),
+                false,
+            ),
+        ),
+        Err(err) => ResponseEnvelope::error(
+            request_id,
+            CallError::internal(format!("failed to decode response body: {err}")),
+        ),
     }
 }
 
@@ -331,7 +471,6 @@ pub(crate) async fn forward(
     default_headers: &HashMap<String, String>,
     namespace: &str,
     error_status_codes: &[(u16, String)],
-    op_type: OperationType,
     input: Value,
     context: OperationContext,
 ) -> ResponseEnvelope {
@@ -355,17 +494,20 @@ pub(crate) async fn forward(
 
     let request_builder = http_client
         .request(http_method, url.as_str())
-        .headers(headers);
+        .headers(headers)
-
+        .header(ACCEPT, "*/*");
-    let request_builder = if op_type == OperationType::Sub {
-        request_builder.header(ACCEPT, "text/event-stream")
-    } else {
-        request_builder.header(ACCEPT, "*/*")
-    };
 
     let request_builder = match body.as_ref() {
         Some(b) => {
-            let serialized = serde_json::to_string(b).unwrap_or_else(|_| String::from("null"));
+            let serialized = match serde_json::to_string(b) {
+                Ok(s) => s,
+                Err(err) => {
+                    return ResponseEnvelope::error(
+                        request_id,
+                        CallError::internal(format!("failed to serialize request body: {err}")),
+                    );
+                }
+            };
             request_builder.body(serialized)
         }
         None => request_builder,
@@ -381,57 +523,49 @@ pub(crate) async fn forward(
         }
     };
 
+    if !response.status().is_success() {
+        return error_envelope(response, &request_id, error_status_codes).await;
+    }
+
+    success_envelope(response, &request_id).await
+}
+
+/// Builds the non-2xx error envelope for an upstream response (FWD-10):
+/// `HTTP_<status>` mapping per ADR-023, plus a bounded, control-stripped
+/// echo of the upstream error body woven into the message. The body is
+/// read exactly once, capped at [`STATUS_BODY_DRAIN`] bytes — larger
+/// error bodies are truncated for the echo and the connection is
+/// dropped rather than drained further, so a firehose upstream cannot
+/// pin a worker to connection cleanup.
+async fn error_envelope(
+    response: reqwest::Response,
+    request_id: &str,
+    error_status_codes: &[(u16, String)],
+) -> ResponseEnvelope {
     let status = response.status();
-
+    let code = error_status_codes
-    if !status.is_success() {
+        .iter()
-        let code = error_status_codes
+        .find(|(s, _)| *s == status.as_u16())
-            .iter()
+        .map(|(_, c)| c.clone())
-            .find(|(s, _)| *s == status.as_u16())
+        .unwrap_or_else(|| format!("HTTP_{}", status.as_u16()));
-            .map(|(_, code)| code.clone())
+    let mut message = format!(
-            .unwrap_or_else(|| format!("HTTP_{}", status.as_u16()));
+        "HTTP {}: {}",
-        let message = format!(
+        status.as_u16(),
-            "HTTP {}: {}",
+        status.canonical_reason().unwrap_or("")
-            status.as_u16(),
+    );
-            status.canonical_reason().unwrap_or("")
+    match read_body_capped(response, STATUS_BODY_DRAIN).await {
-        );
+        Ok(bytes) => {
-        return ResponseEnvelope::error(request_id, CallError::new(code, message, false));
+            if let Some(echo) = bounded_error_body(bytes) {
-    }
+                message.push_str(": ");
-
+                message.push_str(&echo);
-    let content_type = response
-        .headers()
-        .get(reqwest::header::CONTENT_TYPE)
-        .and_then(|v: &reqwest::header::HeaderValue| v.to_str().ok())
-        .unwrap_or("")
-        .to_string();
-
-    if content_type.contains("application/json") {
-        match response.json::<Value>().await {
-            Ok(v) => ResponseEnvelope::ok(request_id, v),
-            Err(err) => ResponseEnvelope::error(
-                request_id,
-                CallError::internal(format!("failed to decode JSON body: {err}")),
-            ),
-        }
-    } else if content_type.starts_with("text/") {
-        match response.text().await {
-            Ok(t) => ResponseEnvelope::ok(request_id, Value::String(t)),
-            Err(err) => ResponseEnvelope::error(
-                request_id,
-                CallError::internal(format!("failed to decode text body: {err}")),
-            ),
-        }
-    } else {
-        match response.bytes().await {
-            Ok(b) => {
-                let arr: Vec<Value> = b.iter().map(|byte| Value::Number((*byte).into())).collect();
-                ResponseEnvelope::ok(request_id, Value::Array(arr))
             }
-            Err(err) => ResponseEnvelope::error(
-                request_id,
-                CallError::internal(format!("failed to read body: {err}")),
-            ),
         }
+        Err(BodyReadError::TooLarge) => {
+            message.push_str(": [error body too large to echo]");
+        }
+        Err(_) => {}
     }
+    ResponseEnvelope::error(request_id, CallError::new(code, message, false))
 }
 
 /// Converts a parsed SSE event into a response envelope, JSON-decoding
@@ -491,13 +625,20 @@ pub(crate) fn forward_stream(
             .headers(headers)
             .header(ACCEPT, "text/event-stream");
         let request_builder = match body.as_ref() {
-            Some(b) => {
+            Some(b) => match serde_json::to_string(b) {
-                let serialized = serde_json::to_string(b).unwrap_or_else(|_| String::from("null"));
+                Ok(serialized) => request_builder.body(serialized),
-                request_builder.body(serialized)
+                Err(err) => {
-            }
+                    return Err(CallError::internal(format!(
+                        "failed to serialize request body: {err}"
+                    )));
+                }
+            },
             None => request_builder,
         };
-        request_builder.send().await
+        request_builder
+            .send()
+            .await
+            .map_err(|err| CallError::internal(format!("HTTP request failed: {err}")))
     };
 
     let sse = stream::once(init).flat_map(move |result| {
@@ -505,28 +646,33 @@ pub(crate) fn forward_stream(
         let error_status_codes = error_status_codes_stream.clone();
         match result {
             Err(err) => Box::pin(stream::once(async move {
-                ResponseEnvelope::error(
+                ResponseEnvelope::error(request_id, err)
-                    request_id,
-                    CallError::internal(format!("HTTP request failed: {err}")),
-                )
             })) as ResponseStream,
             Ok(response) => {
                 let status = response.status();
                 if !status.is_success() {
-                    let code = error_status_codes
+                    let request_id = request_id.clone();
-                        .iter()
-                        .find(|(s, _)| *s == status.as_u16())
-                        .map(|(_, c)| c.clone())
-                        .unwrap_or_else(|| format!("HTTP_{}", status.as_u16()));
-                    let message = format!(
-                        "HTTP {}: {}",
-                        status.as_u16(),
-                        status.canonical_reason().unwrap_or("")
-                    );
                     Box::pin(stream::once(async move {
-                        ResponseEnvelope::error(request_id, CallError::new(code, message, false))
+                        error_envelope(response, &request_id, &error_status_codes).await
                     })) as ResponseStream
                 } else {
+                    let content_type = response
+                        .headers()
+                        .get(reqwest::header::CONTENT_TYPE)
+                        .and_then(|v: &reqwest::header::HeaderValue| v.to_str().ok())
+                        .unwrap_or_default()
+                        .to_ascii_lowercase();
+                    if !is_sse_content_type(&content_type) {
+                        let message = format!(
+                            "upstream returned Content-Type `{content_type}` on a subscription operation; expected `text/event-stream`"
+                        );
+                        return Box::pin(stream::once(async move {
+                            ResponseEnvelope::error(
+                                request_id,
+                                CallError::new("INVALID_RESPONSE_TYPE", message, false),
+                            )
+                        })) as ResponseStream;
+                    }
                     let request_id_inner = request_id.clone();
                     Box::pin(
                         stream::unfold(
@@ -591,6 +737,14 @@ pub(crate) fn forward_stream(
     Box::pin(sse)
 }
 
+/// True when a response `Content-Type` header is `text/event-stream` by
+/// mime-essence semantics; parameters (`; charset=…`) are ignored. A
+/// subscription forwarder that receives anything else surfaces a loud
+/// error rather than an indefinitely empty stream (FWD-07).
+fn is_sse_content_type(content_type: &str) -> bool {
+    content_type.split(';').next().unwrap_or_default().trim() == "text/event-stream"
+}
+
 /// A parsed SSE event: the `data:` lines joined with `\n`.
 pub(crate) struct SseEvent {
     pub(crate) data: String,
@@ -602,6 +756,26 @@ pub(crate) enum SseParseError {
     BufferOverflow,
 }
 
+/// Maximum size, in bytes, of a buffered upstream response body on any
+/// non-streaming read path in [`forward`] (JSON, text, or binary). A
+/// hostile upstream cannot grow caller memory past this budget; a larger
+/// body fails the read with [`BodyReadError::TooLarge`], surfaced as an
+/// `HTTP_413` error envelope.
+pub(crate) const RESPONSE_BODY_CAP: usize = 16 * 1024 * 1024;
+
+/// Maximum size, in bytes, of the error-body echo surfaced on a non-2xx
+/// upstream response ([`forward`] and [`forward_stream`]). The echo is
+/// bounded, not logged, and carries no request credential material — it
+/// is the upstream's diagnostics (validation details, rate-limit info)
+/// that would otherwise be discarded (FWD-10).
+pub(crate) const ERROR_BODY_ECHO_CAP: usize = 4096;
+
+/// Upper bound on how much of a non-2xx upstream body is read, both for
+/// the bounded echo and for connection reuse. A larger error body is
+/// truncated in the surfaced message and the connection is dropped —
+/// a deliberate trade of pool reuse against unbounded drain time (FWD-10).
+const STATUS_BODY_DRAIN: usize = 64 * 1024;
+
 /// Maximum size, in bytes, of the SSE parser's internal reassembly
 /// buffer. A single event (all `data:` lines plus framing) must fit
 /// within this budget; a stream emitting a longer partial line — or an

src/adapters/from_jsonschema.rs

@@ -105,7 +105,6 @@ impl OperationAdapter for FromJsonSchema {
                 let namespace = namespace.clone();
                 let http_client = Arc::clone(&http_client);
                 let error_status_codes = error_status_codes.clone();
-                let op_type = op_type;
                 async move {
                     forward(
                         &http_client,
@@ -116,7 +115,6 @@ impl OperationAdapter for FromJsonSchema {
                         &default_headers,
                         &namespace,
                         &error_status_codes,
-                        op_type,
                         input,
                         context,
                     )

src/adapters/from_openapi.rs

@@ -302,7 +302,6 @@ impl FromOpenAPI {
                 let namespace = namespace.clone();
                 let http_client = Arc::clone(&http_client);
                 let error_status_codes = error_status_codes.clone();
-                let op_type = op_type;
                 async move {
                     forward(
                         &http_client,
@@ -313,7 +312,6 @@ impl FromOpenAPI {
                         &default_headers,
                         &namespace,
                         &error_status_codes,
-                        op_type,
                         input,
                         context,
                     )

src/client/http_client.rs

@@ -175,14 +175,22 @@ pub enum HttpClientBuildError {
 }
 
 pub struct SharedHttpClient {
-    inner: ArcSwap<ClientWithMiddleware>,
+    inner: ArcSwap<SharedHttpInner>,
-    config: ArcSwap<HttpClientConfig>,
+}
+
+/// Joint holder for the client and its config so a reload swaps both in
+/// one atomic `ArcSwap::store` — a reader can never observe the new
+/// config paired with the previous client (FWD-12).
+#[derive(Clone)]
+struct SharedHttpInner {
+    client: Arc<ClientWithMiddleware>,
+    config: Arc<HttpClientConfig>,
 }
 
 impl std::fmt::Debug for SharedHttpClient {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("SharedHttpClient")
-            .field("config", &self.config.load())
+            .field("config", &self.inner.load().config)
             .finish_non_exhaustive()
     }
 }
@@ -196,27 +204,32 @@ impl SharedHttpClient {
     pub fn new(config: HttpClientConfig) -> Result<Self, HttpClientBuildError> {
         let client = build_client_sync(&config)?;
         Ok(Self {
-            inner: ArcSwap::from_pointee(client),
+            inner: ArcSwap::from_pointee(SharedHttpInner {
-            config: ArcSwap::from_pointee(config),
+                client: Arc::new(client),
+                config: Arc::new(config),
+            }),
         })
     }
 
     pub fn client(&self) -> Arc<ClientWithMiddleware> {
-        self.inner.load_full()
+        Arc::clone(&self.inner.load().client)
     }
 
     pub fn config(&self) -> Arc<HttpClientConfig> {
-        self.config.load_full()
+        Arc::clone(&self.inner.load().config)
     }
 
     /// Rebuild the underlying client and swap it in for new callers
     /// (in-flight requests complete on the previous client). PEM reads
     /// use `tokio::fs`, so this is safe to call from async contexts
-    /// without blocking a worker.
+    /// without blocking a worker. Client and config swap together in a
+    /// single atomic store (FWD-12).
     pub async fn reload(&self, config: HttpClientConfig) -> Result<(), HttpClientBuildError> {
         let client = build_client(&config).await?;
-        self.config.store(Arc::new(config));
+        self.inner.store(Arc::new(SharedHttpInner {
-        self.inner.store(Arc::new(client));
+            client: Arc::new(client),
+            config: Arc::new(config),
+        }));
         Ok(())
     }
 }