//! Shared HTTP forwarding core for the HTTP-backed adapters
//! (`from_openapi`, `from_jsonschema`): request construction
//! (path templates, query split, body, credential injection), the
//! reqwest forwarding handlers (`forward` / `forward_stream`), and the
//! SSE frame parser for streaming (SSE → `ResponseEnvelope`) handlers
//! (ADR-049).
//!
//! # Timeout / byte-bound policy for streaming forwards (FWD-15, FWD-14)
//!
//! The two send paths impose deliberately different bounds, mirroring
//! the gateway's dispatch contract (alkcall ADR-021: Once-op invokes
//! carry a 30 s deadline; `HandlerKind::Stream` invokes set
//! `deadline: None` — subscriptions are unbounded in *time* by design):
//!
//! - **`forward`** (request/response) sends through the shared client's
//!   total request timeout (30 s default), so a forwarded call fails
//!   before its caller does.
//! - **`forward_stream`** (subscriptions) sends through
//!   [`SharedHttpClient::stream_client`] — the same config minus the
//!   total request timeout, connect + read timeouts retained. A healthy
//!   subscription longer than 30 s survives; the read timeout remains
//!   the stall guard on upstream staleness (it resets on every body
//!   byte, so a keep-alive-emitting source lives as long as it keeps
//!   the connection warm — and quiet silence past the read timeout
//!   still terminates it). reqwest 0.13's per-request timeout override
//!   can lengthen but never clear a client-level total timeout, hence
//!   the derived client rather than an extension override.
//!
//! Unbounded time without a byte bound would let a hostile upstream
//! stream well-formed 1-MiB-line events forever (the 1 MiB SSE line cap
//! bounds one line, not the stream), so the streaming branch also
//! enforces a total streamed-bytes cap per subscription —
//! [`crate::client::HttpClientConfig::stream_total_byte_cap`], 1 GiB by
//! default, accumulated across every chunk fed to the SSE parser.
//! Exceeding it terminates the stream with a single terminal error
//! envelope (the stream-ends semantics: one error frame, then end —
//! matching the other terminal arms). The line-cap check runs before
//! the buffer grows, so the reassembly buffer can never exceed the
//! line cap.
//!
//! # Streaming payload contract (FWD-17)
//!
//! Each parsed SSE frame becomes one success envelope:
//!
//! - A `data:` payload that is valid JSON surfaces as the decoded value
//!   itself — `123` as a number, `"123"` as a string (the raw-text
//!   fallback this replaces erased that distinction). This holds even
//!   when the frame carries an `event:` name.
//! - Any other payload surfaces as the
//!   `{"data": <raw payload>, "event": <name|null>}` wrapper: the raw
//!   text stays field-addressable and an upstream's named-event
//!   conventions (`event: error`, `event: ping`, …) are visible on
//!   non-JSON frames. The name is the frame's `event:` field (WHATWG
//!   last-wins); `null` when the upstream sent none — the spec's
//!   implicit `message` default is *not* substituted, so "upstream
//!   named it" and "default name" remain distinguishable.
//! - An empty payload is `Null`.
//!
//! Known limitation: a JSON body under a named event surfaces as the
//! decoded value only — the event name is not carried on JSON frames;
//! a non-JSON body under a named event is the only combination where
//! both are visible.
//!
//! # Credentials and input routing
//!
//! The forwarding handler is the no-env-vars credential injection point
//! (ADR-014): it reads `OperationContext.capabilities`, never
//! `std::env::var`. Imported error codes are `HTTP_<status>` to avoid
//! collision with the protocol-level codes (ADR-023).
//!
//! Input-schema enforcement (review 001 OAI-02): every input key must be
//! declared by the operation's `input_schema` or consumed by a path
//! template placeholder; undeclared keys are rejected with `INVALID_INPUT`
//! at call time. There is no pass-through knob — a request the forwarder
//! sends must match what `/schema` advertises, so peer-supplied input
//! cannot add upstream query parameters, headers, or craft a request body
//! the contract does not declare. Declared keys route by designation: a
//! property marked with the `HEADER_PARAM_IN_MARKER` marker key set to "header" is sent as a
//! request header, the declared `GATEWAY_BODY_KEY` property becomes the
//! request body, and every other declared key becomes an upstream query
//! parameter.
//!
//! # Input routing and path placeholders (FWD-18)
//!
//! A key that matches a `{placeholder}` in the path template is consumed
//! by the path and never also emits as a query parameter, regardless of
//! its value's shape — the placeholder check precedes query routing in
//! the request builder. A placeholder renders exactly one literal path
//! segment, so its value must be a scalar: object/array values fail with
//! `INVALID_INPUT` (structural values have no faithful single-segment
//! rendering; splicing the minified JSON into the path was the
//! pre-decision behavior and is rejected now).
//!
//! # Percent handling in the rendered path (FWD-19)
//!
//! Two different contracts apply, by design:
//!
//! - A `%` arriving inside a *value* is always encoded (`%` → `%25`),
//!   so values carrying `%2F` cannot be mistaken for this crate's own
//!   escapes and a value can never inject URL structure.
//! - A `%` in *template/base text* survives verbatim. Templates and
//!   base URLs are assembly-supplied (ADR-066 trust boundary), so an
//!   assembly that writes `/s3%2Fkeys` is presumed to mean a
//!   pre-encoded segment for upstreams that route `%2F` differently
//!   from `/` — that upstream-semantics choice belongs to the assembly,
//!   not this crate. No injection results: the surviving `%2F` still
//!   forms a single segment (the origin check plus the two-pass
//!   percent-encoding over template text see to that), and the
//!   value-side rule above means every bare `%` in a rendered path
//!   traces to template text the assembly author wrote.
//!

use std::collections::HashMap;
use std::sync::Arc;

use alkcall::protocol::wire::{CallError, ResponseEnvelope};
use alkcall::registry::context::OperationContext;
use alkcall::registry::registration::ResponseStream;
use futures::stream;
use futures::StreamExt;
use percent_encoding::{percent_decode_str, utf8_percent_encode, AsciiSet, CONTROLS};
use reqwest::header::{HeaderMap, HeaderName, HeaderValue, ACCEPT, AUTHORIZATION, CONTENT_TYPE};
use reqwest::Method;
use serde_json::Value;
use url::Url;

use crate::client::SharedHttpClient;

/// 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;

// Fractional-path helpers and body echo caps follow; the gateway body key
// and the header-param wire marker used by the input routing live with the
// other request-construction defaults here.

/// 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;

/// The input key that carries the request body on the gateway shape
/// (ADR-047). Declared by `from_openapi`'s generated input schemas
/// (OAI-07 rejects a spec parameter with the same name at import) and
/// consumed by [`build_request`] ahead of the query-parameter routing.
pub(crate) const GATEWAY_BODY_KEY: &str = "body";

/// Property-level marker inside an operation's `input_schema` properties
/// that routes the declared value into an HTTP request header instead of
/// the query string (review 001 OAI-03). `from_openapi` stamps
/// `"wire": "header"` on properties generated from `in: header`
/// parameters; `from_jsonschema` callers can mark their declared
/// properties the same way. Unmarked properties default to query
/// placement, preserving the pre-OAI-03 wire behavior for
/// `in: query`/`in: path` parameters.
pub(crate) const HEADER_PARAM_IN_MARKER: &str = "wire";
pub(crate) const HEADER_PARAM_MARKER_VALUE: &str = "header";

/// The credential scheme forwarded handlers apply to outbound requests
/// (ADR-014). The credential value itself flows through
/// `OperationContext.capabilities` at call time — never through this
/// config.
#[derive(Clone)]
pub enum HttpAuthScheme {
    /// `Authorization: Bearer <token>` from the caller's `Bearer`
    /// capability.
    Bearer,
    /// A named API-key header (e.g. `x-api-key`) carrying the caller's
    /// `ApiKey` capability value.
    ApiKey {
        /// The upstream header the key is sent in.
        header_name: String,
    },
    /// HTTP Basic auth from the caller's `username`/`password`
    /// capability pair.
    Basic,
}

/// Assembly-time configuration for one imported HTTP service: the
/// registry namespace its operations land under, where traffic goes,
/// and how credentials attach.
pub struct HttpServiceConfig {
    /// Registry namespace for the imported operations (the
    /// `<namespace>/<operationId>` op names).
    pub namespace: String,
    /// Outbound base URL every path template is resolved against.
    pub base_url: String,
    /// Credential scheme for outbound requests; `None` sends an
    /// unauthenticated request.
    pub auth: Option<HttpAuthScheme>,
    /// Static headers attached to every outbound request (e.g. a
    /// required `User-Agent`).
    pub default_headers: HashMap<String, String>,
}

#[allow(clippy::too_many_arguments)]
pub(crate) fn build_request(
    base_url: &str,
    path_template: &str,
    method: &str,
    auth_scheme: &Option<HttpAuthScheme>,
    default_headers: &HashMap<String, String>,
    namespace: &str,
    input_schema: &Value,
    input: &Value,
    context: &OperationContext,
) -> Result<(Method, Url, Option<Value>, HeaderMap), CallError> {
    let inputs = input.as_object().ok_or_else(|| {
        CallError::invalid_input(format!(
            "input must be a JSON object (got {}); adapter operations take a named-key input",
            type_name_of(input)
        ))
    })?;

    enforce_input_schema(input_schema, inputs)?;

    let mut query_params: Vec<(String, String)> = Vec::new();
    let mut header_params: Vec<(String, String)> = Vec::new();
    let param_locations = param_locations(input_schema);
    let mut body: Option<Value> = None;
    for (key, value) in inputs {
        if is_path_placeholder(key, path_template) {
            continue;
        }
        if key == GATEWAY_BODY_KEY {
            body = Some(value.clone());
            continue;
        }
        if param_locations.get(key.as_str()) == Some(&ParamLocation::Header) {
            header_params.push((key.clone(), value_to_query(value)));
        } else {
            query_params.push((key.clone(), value_to_query(value)));
        }
    }

    let rendered_path = render_path_template(path_template, Some(inputs))?;
    let mut url = assemble_request_url(base_url, &rendered_path)?;
    if !query_params.is_empty() {
        let mut pairs = url.query_pairs_mut();
        for (k, v) in &query_params {
            pairs.append_pair(k, v);
        }
    }

    let mut headers = HeaderMap::new();
    for (k, v) in &header_params {
        let name = HeaderName::try_from(k.as_str()).map_err(|_| {
            CallError::internal(format!(
                "declared header parameter `{k}` is not a valid HTTP header name; refusing to build a request that silently drops it"
            ))
        })?;
        let value = HeaderValue::try_from(v.as_str()).map_err(|_| {
            CallError::internal(format!(
                "declared header parameter `{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);
    }
    for (k, v) in default_headers {
        let name = HeaderName::try_from(k.as_str()).map_err(|_| {
            CallError::internal(format!(
                "default header `{k}` is not a valid HTTP header name; refusing to build a request that silently drops it"
            ))
        })?;
        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() {
        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
    }

    if let Some(scheme) = auth_scheme {
        if let Some(secret) = context.capabilities.get(namespace) {
            let credential = secret.expose_secret().clone();
            match scheme {
                HttpAuthScheme::Bearer => {
                    let value =
                        HeaderValue::try_from(format!("Bearer {credential}")).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(AUTHORIZATION, value);
                }
                HttpAuthScheme::ApiKey { header_name } => {
                    let name =
                        HeaderName::try_from(header_name.as_str()).map_err(|_| {
                            CallError::internal(format!(
                                "API-key auth for namespace `{namespace}` declares invalid header name `{header_name}`; refusing to send the request unauthenticated"
                            ))
                        })?;
                    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 value =
                        HeaderValue::try_from(format!("Basic {credential}")).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(AUTHORIZATION, value);
                }
            }
        }
    }

    let http_method = Method::from_bytes(method.as_bytes())
        .map_err(|_| CallError::internal(format!("invalid HTTP method `{method}`")))?;
    Ok((http_method, url, body, headers))
}

/// Upstream placement of a declared input property (review 001 OAI-03).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum ParamLocation {
    Query,
    Header,
}

/// Map of declared property name → upstream placement, read from the
/// `input_schema`'s `properties` entries via [`HEADER_PARAM_IN_MARKER`]
/// (query placement is the default).
fn param_locations(input_schema: &Value) -> HashMap<&str, ParamLocation> {
    let mut out = HashMap::new();
    let Some(properties) = input_schema.get("properties").and_then(|p| p.as_object()) else {
        return out;
    };
    for (name, schema) in properties {
        let is_header = schema
            .get(HEADER_PARAM_IN_MARKER)
            .and_then(|w| w.as_str())
            .is_some_and(|w| w == HEADER_PARAM_MARKER_VALUE);
        out.insert(
            name.as_str(),
            if is_header {
                ParamLocation::Header
            } else {
                ParamLocation::Query
            },
        );
    }
    out
}

fn type_name_of(value: &Value) -> &'static str {
    match value {
        Value::Null => "null",
        Value::Bool(_) => "a boolean",
        Value::Number(_) => "a number",
        Value::String(_) => "a string",
        Value::Array(_) => "an array",
        Value::Object(_) => "an object",
    }
}

/// Input-schema enforcement at call time (review 001 OAI-02).
///
/// Every input key must be declared by the input schema's `properties` —
/// including the gateway body property, which `from_openapi` declares as
/// `body` whenever the operation has a requestBody, and including the
/// path-consumed placeholders, which `from_openapi` also declares. An
/// explicit `"additionalProperties": true` opts the operation into
/// catch-all input (JSON Schema semantics: the schema advertises that
/// extra properties are valid), which `from_jsonschema` callers can use
/// for open-shaped endpoints. Anything else undeclared is a rejected
/// `INVALID_INPUT` rather than a silently-added upstream query parameter,
/// so peer input like `{debug: true}` or `{impersonate_id: …}` cannot
/// decorate an upstream request the contract does not advertise.
fn enforce_input_schema(
    input_schema: &Value,
    inputs: &serde_json::Map<String, Value>,
) -> Result<(), CallError> {
    let declared = input_schema
        .get("properties")
        .and_then(|p| p.as_object())
        .map(|p| p.keys().cloned().collect::<Vec<_>>())
        .unwrap_or_default();
    let catch_all = input_schema.get("additionalProperties") == Some(&Value::Bool(true));
    let unknown: Vec<String> = inputs
        .keys()
        .filter(|key| !catch_all && !declared.iter().any(|d| d == *key))
        .cloned()
        .collect();
    if let Some(first) = unknown.first() {
        let declared_list = if declared.is_empty() {
            "none".to_string()
        } else {
            declared.join(", ")
        };
        return Err(CallError::invalid_input(format!(
            "input key `{first}` is not declared by the operation's input schema \
             (declared: {declared_list}); undeclared keys are rejected so peer \
             input cannot shape the upstream request beyond the advertised contract"
        )));
    }
    Ok(())
}

/// Percent-encode set for a spliced path-parameter value: the WHATWG
/// path set (controls, space, `"`, `<`, `>`, `` ` ``, `#`, `?`, `{`, `}`)
/// plus `/` so a value stays one literal segment, plus `%` `?` `#`
/// belt-and-braces, plus `\` so a Windows-style separator cannot smuggle
/// a backslash segment on special-scheme URLs.
///
/// Because `%` is in this set, percent-encoding with it is idempotent:
/// every `%` in a value becomes `%25`, so the result contains `%` only
/// as the lead byte of an escape this crate itself produced.
const PATH_VALUE_ENCODE_SET: &AsciiSet = &CONTROLS
    .add(b' ')
    .add(b'"')
    .add(b'<')
    .add(b'>')
    .add(b'`')
    .add(b'#')
    .add(b'?')
    .add(b'{')
    .add(b'}')
    .add(b'/')
    .add(b'%')
    .add(b'\\');

/// Encode set for the second pass over a rendered segment in
/// [`request_path`]: applied to the text *between* `%` characters, so
/// anything that still looks like a separator is encoded, while the `%`
/// itself is left untouched to preserve this crate's own `%2F`-style
/// escapes.
const PATH_AFTER_PERCENT_ENCODE_SET: &AsciiSet = &CONTROLS
    .add(b' ')
    .add(b'"')
    .add(b'<')
    .add(b'>')
    .add(b'`')
    .add(b'#')
    .add(b'?')
    .add(b'{')
    .add(b'}')
    .add(b'/')
    .add(b'\\');

/// Raw string form of a scalar input value (FWD-12: `value_to_path_segment`
/// and `value_to_query` were byte-identical helpers; the shared scalar
/// extraction is this one function, and the encoding layers mount on it).
fn scalar_value_to_string(value: &Value) -> String {
    match value {
        Value::String(s) => s.clone(),
        Value::Number(n) => n.to_string(),
        Value::Bool(b) => b.to_string(),
        Value::Null => String::new(),
        other => other.to_string(),
    }
}

/// Percent-encoding scheme used by [`render_path_template`]: a value
/// containing `/`, `?`, `#`, or `\` — and any raw `%` it carries — is
/// encoded so that a rendered value stays one literal path segment
/// (FWD-01 traversal/ smuggled-segments gate). Scalars are never
/// rejected: they are rendered safely instead — with the lone-dot
/// exception below (FWD-13). Object/array values under a placeholder
/// key are rejected with `INVALID_INPUT` (FWD-18): a path placeholder
/// designates exactly one literal segment, so a structural value has no
/// faithful rendering — the pre-decision behavior spliced the minified
/// JSON into the path *and* emitted the key as a query parameter
/// (double-routed, neither faithful).
///
/// # Lone-dot values (FWD-13)
///
/// A scalar whose *decoded* form is exactly `.` or `..` — including the
/// `%2e`/`%2E` spellings, which `Url::set_path` normalizes away after
/// decoding regardless of what the encode set preserves — is rejected
/// with `INVALID_INPUT`. `Url::set_path` removes lone dot segments
/// case-insensitively (`/../x` → `/x`, `/./x` → `/x`, `/%2e%2e/x` →
/// `/x`), so no encode-set fix can keep such a value faithful: the
/// upstream would receive a *different* endpoint than the template
/// describes, with the namespace's credentials attached. The rejection
/// is exact-match on the full decoded segment; values like `v1.2.3` or
/// `.hidden-file` render normally. The parameter name is quoted but the
/// value is never echoed.
pub(crate) fn value_to_path_segment(value: &Value) -> Result<String, CallError> {
    let raw = match value {
        Value::String(_) | Value::Number(_) | Value::Bool(_) | Value::Null => {
            scalar_value_to_string(value)
        }
        other => {
            return Err(CallError::invalid_input(format!(
                "path placeholder value must be a scalar (string, number, boolean, or null); got {}: structural values have no faithful single-segment rendering",
                type_name_of(other)
            )))
        }
    };
    reject_lone_dot_value(value, &raw)?;
    Ok(utf8_percent_encode(&raw, PATH_VALUE_ENCODE_SET).to_string())
}

/// FWD-13 gate: a decoded path-value of exactly `.` or `..` cannot be
/// rendered faithfully — `Url::set_path` silently normalizes lone dot
/// segments away, so the upstream would receive a different endpoint
/// than the template describes. The parameter is named but the value is
/// not echoed. Called with both the original [`Value`] (for type-naming)
/// and its raw scalar string.
fn reject_lone_dot_value(value: &Value, raw: &str) -> Result<(), CallError> {
    let type_label = type_name_of(value);
    let is_lone_dot = matches!(raw, "." | "..")
        || raw.eq_ignore_ascii_case("%2e")
        || raw.eq_ignore_ascii_case("%2e%2e");
    if is_lone_dot {
        return Err(CallError::invalid_input(format!(
            "path placeholder value of type {type_label} decodes to a lone dot segment, which cannot appear in a rendered path (Url::set_path would silently normalize it away); pass a concrete non-dot value"
        )));
    }
    Ok(())
}

/// Raw form of a scalar input value for query emission: `&`/`=`
/// separation, escaping, and space encoding are handled by
/// `url::query_pairs_mut` downstream, so the value itself must be raw —
/// a path-rendered value (`%20` for space) would be double-encoded into
/// `%2520` by `append_pair`.
fn value_to_query(value: &Value) -> String {
    scalar_value_to_string(value)
}

/// Single-pass template renderer. Two invariants instead of the old
/// iterative `replace` (FWD-01):
///
/// 1. A rendered value is never re-scanned — each template placeholder
///    is replaced exactly once, so a `{repo}` value that itself contains
///    a later placeholder string cannot trigger a second substitution.
/// 2. The whole template and every input key are walked in one pass, so
///    `?` and `#` in a value split into percent-encoded octets (`%3F`,
///    `%23`) rather than injecting URL structure.
pub(crate) fn render_path_template(
    template: &str,
    inputs: Option<&serde_json::Map<String, Value>>,
) -> Result<String, CallError> {
    let mut out = String::with_capacity(template.len());
    let mut rest = template;
    let mut unresolved: Vec<String> = Vec::new();
    while let Some(start) = rest.find('{') {
        let (head, tail) = rest.split_at(start);
        out.push_str(head);
        let Some(end) = tail.find('}') else {
            return Err(CallError::internal(format!(
                "invalid path template `{template}`: unterminated placeholder"
            )));
        };
        let name = &tail[1..end];
        let raw_value = match inputs.and_then(|map| map.get(name)) {
            Some(v) => v,
            None => {
                unresolved.push(format!("{{{name}}}"));
                rest = &tail[end + 1..];
                continue;
            }
        };
        out.push_str(&value_to_path_segment(raw_value)?);
        rest = &tail[end + 1..];
    }
    out.push_str(rest);
    if !unresolved.is_empty() {
        return Err(CallError::internal(format!(
            "path template `{template}` references unbound placeholder(s): {}",
            unresolved.join(", ")
        )));
    }
    Ok(out)
}

/// True when `input_name` is consumed by a `{input_name}` placeholder in
/// the path template. Drives input routing in [`build_request`]; the
/// value itself is rendered by [`render_path_template`].
pub(crate) fn is_path_placeholder(input_name: &str, template: &str) -> bool {
    let placeholder = format!("{{{input_name}}}");
    template.contains(&placeholder)
}

fn parse_base_url(base_url: &str) -> Result<Url, CallError> {
    let parsed = Url::parse(base_url)
        .map_err(|e| CallError::internal(format!("invalid base_url `{base_url}`: {e}")))?;
    let scheme = parsed.scheme();
    if scheme != "https" && scheme != "http" {
        return Err(CallError::internal(format!(
            "base_url `{base_url}` must use https (or http for plain non-TLS origins); `{scheme}` is not an HTTP scheme"
        )));
    }
    let host = parsed.host_str().unwrap_or_default();
    if host.is_empty() {
        return Err(CallError::internal(format!(
            "base_url `{base_url}` must include an explicit host"
        )));
    }
    if !parsed.username().is_empty() || parsed.password().is_some() {
        return Err(CallError::internal(format!(
            "base_url `{base_url}` must not embed userinfo; credentials are injected per-operation from Capabilities"
        )));
    }
    Ok(parsed)
}

fn request_path(rendered_path: &str) -> Result<String, CallError> {
    let trimmed = rendered_path.trim_matches('/');
    if trimmed.is_empty() {
        return Err(CallError::internal(
            "path template resolves to an empty request path; at least one segment is required",
        ));
    }
    let mut path = String::new();
    for segment in trimmed.split('/') {
        path.push('/');
        let mut pieces = segment.split('%');
        utf8_percent_encode_into(&mut path, pieces.next().unwrap_or_default());
        for piece in pieces {
            path.push('%');
            utf8_percent_encode_into(&mut path, piece);
        }
    }
    Ok(path)
}

fn utf8_percent_encode_into(out: &mut String, text: &str) {
    for piece in utf8_percent_encode(text, PATH_AFTER_PERCENT_ENCODE_SET) {
        out.push_str(piece);
    }
}

fn assemble_request_url(base_url: &str, rendered_path: &str) -> Result<Url, CallError> {
    let base = parse_base_url(base_url)?;
    let request_path = request_path(rendered_path)?;
    let base_path = base.path();
    let base_dir = match base_path.strip_suffix('/') {
        Some(stripped) => stripped,
        None => base_path,
    };
    let mut full_path = String::with_capacity(base_dir.len() + request_path.len() + 1);
    full_path.push_str(base_dir);
    full_path.push_str(&request_path);
    let mut url = base.clone();
    url.set_path(&full_path);
    assert_untouched_by_normalization(&url, base_dir, rendered_path, base_url)?;
    let same_origin = url.scheme() == base.scheme()
        && url.host() == base.host()
        && url.port_or_known_default() == base.port_or_known_default();
    if !same_origin {
        return Err(CallError::internal(format!(
            "request path `{rendered_path}` resolved against `{base_url}` changed the target origin: {} != {}",
            url.origin().ascii_serialization(),
            base.origin().ascii_serialization()
        )));
    }
    Ok(url)
}

/// Post-`set_path` invariant (FWD-13): the decoded segments of the
/// final URL path must equal `base_dir`'s segments plus the decoded
/// rendered segments, byte-identical. `set_path` is the one step this
/// crate does not control, and its parser normalizes lone dot segments
/// away (`/../x` → `/x`); this check turns any future normalizer
/// surprise into a loud `INTERNAL` error instead of a silently
/// re-routed authenticated request.
fn assert_untouched_by_normalization(
    url: &Url,
    base_dir: &str,
    rendered_path: &str,
    base_url: &str,
) -> Result<(), CallError> {
    let rendered_segments = rendered_path
        .trim_matches('/')
        .split('/')
        .filter(|s| !s.is_empty())
        .map(|s| percent_decode_str(s).decode_utf8_lossy().into_owned());
    let base_segments: Vec<&str> = base_dir.split('/').filter(|s| !s.is_empty()).collect();
    let expected: Vec<String> = base_segments
        .iter()
        .copied()
        .map(str::to_string)
        .chain(rendered_segments)
        .collect();
    let actual: Vec<String> = url
        .path()
        .split('/')
        .filter(|s| !s.is_empty())
        .map(|s| percent_decode_str(s).decode_utf8_lossy().into_owned())
        .collect();
    let matches = actual.len() == expected.len()
        && std::iter::Iterator::zip(actual.iter(), expected.iter()).all(|(a, e)| a == e);
    if matches {
        Ok(())
    } else {
        Err(CallError::internal(format!(
            "request path `{rendered_path}` resolved against `{base_url}` was rewritten by URL normalization: expected segments {expected:?}, got {actual:?}"
        )))
    }
}

#[derive(Debug, thiserror::Error)]
pub(crate) enum BodyReadError {
    #[error("upstream response body exceeds the {RESPONSE_BODY_CAP}-byte response cap")]
    TooLarge,
    #[error("transport error reading response body: {0}")]
    Transport(reqwest::Error),
    #[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: axum::body::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<axum::body::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}")),
        ),
    }
}

#[allow(clippy::too_many_arguments)]
pub(crate) async fn forward(
    http_client: &Arc<SharedHttpClient>,
    base_url: &str,
    path_template: &str,
    method: &str,
    auth_scheme: &Option<HttpAuthScheme>,
    default_headers: &HashMap<String, String>,
    namespace: &str,
    input_schema: &Value,
    error_status_codes: &[(u16, String)],
    input: Value,
    context: OperationContext,
) -> ResponseEnvelope {
    let request_id = context.request_id.clone();

    let (http_method, url, body, headers) = match build_request(
        base_url,
        path_template,
        method,
        auth_scheme,
        default_headers,
        namespace,
        input_schema,
        &input,
        &context,
    ) {
        Ok(parts) => parts,
        Err(err) => return ResponseEnvelope::error(request_id, err),
    };

    let http_client = http_client.client();

    let request_builder = http_client
        .request(http_method, url.as_str())
        .headers(headers)
        .header(ACCEPT, "*/*");

    let request_builder = match body.as_ref() {
        Some(b) => {
            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,
    };

    let response: reqwest::Response = match request_builder.send().await {
        Ok(r) => r,
        Err(err) => {
            return ResponseEnvelope::error(
                request_id,
                CallError::internal(format!("HTTP request failed: {err}")),
            );
        }
    };

    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
        .iter()
        .find(|(s, _)| *s == status.as_u16())
        .map(|(_, c)| c.clone())
        .unwrap_or_else(|| format!("HTTP_{}", status.as_u16()));
    let mut message = format!(
        "HTTP {}: {}",
        status.as_u16(),
        status.canonical_reason().unwrap_or("")
    );
    match read_body_capped(response, STATUS_BODY_DRAIN).await {
        Ok(bytes) => {
            if let Some(echo) = bounded_error_body(bytes) {
                message.push_str(": ");
                message.push_str(&echo);
            }
        }
        Err(BodyReadError::TooLarge) => {
            message.push_str(": [error body too large to echo]");
        }
        Err(_) => {}
    }
    ResponseEnvelope::error(request_id, CallError::new(code, message, false))
}

/// Maps a parsed SSE frame to a response envelope under the FWD-17
/// payload contract: a payload that is valid JSON surfaces as the
/// decoded value itself (a number payload stays a number, a
/// quoted-string payload stays a string — the distinction a raw-text
/// fallback would erase); any other payload surfaces as the
/// `{"data": <raw payload>, "event": <event-name|null>}` wrapper, so
/// a non-JSON stream is field-addressable and an upstream's
/// named-event conventions are visible on it. An empty payload is
/// `Null`. JSON payloads intentionally carry no `event` binding: a
/// JSON frame surfaces as itself (shape stability for JSON-first
/// upstreams, and the spec's implicit `event: message` default never
/// wraps a payload).
fn sse_event_envelope(event: SseEvent, request_id: &str) -> ResponseEnvelope {
    let parsed = if event.data.trim().is_empty() {
        Value::Null
    } else {
        match serde_json::from_str::<Value>(&event.data) {
            Ok(value) => value,
            Err(_) => serde_json::json!({
                "data": event.data,
                "event": event.event,
            }),
        }
    };
    ResponseEnvelope::ok(request_id, parsed)
}

#[allow(clippy::too_many_arguments)]
pub(crate) fn forward_stream(
    http_client: &Arc<SharedHttpClient>,
    base_url: &str,
    path_template: &str,
    method: &str,
    auth_scheme: &Option<HttpAuthScheme>,
    default_headers: &HashMap<String, String>,
    namespace: &str,
    input_schema: &Value,
    error_status_codes: &[(u16, String)],
    input: Value,
    context: OperationContext,
) -> ResponseStream {
    let request_id = context.request_id.clone();

    let (http_method, url, body, headers) = match build_request(
        base_url,
        path_template,
        method,
        auth_scheme,
        default_headers,
        namespace,
        input_schema,
        &input,
        &context,
    ) {
        Ok(parts) => parts,
        Err(err) => {
            return Box::pin(stream::once(async move {
                ResponseEnvelope::error(request_id, err)
            }));
        }
    };

    let http_client = Arc::clone(http_client);
    let error_status_codes = error_status_codes.to_vec();

    let request_id_stream = request_id.clone();
    let error_status_codes_stream = error_status_codes.clone();
    let stream_byte_cap = http_client.config().stream_total_byte_cap;

    let init = async move {
        let request_builder = http_client
            .stream_client()
            .request(http_method, url.as_str())
            .headers(headers)
            .header(ACCEPT, "text/event-stream");
        let request_builder = match body.as_ref() {
            Some(b) => match serde_json::to_string(b) {
                Ok(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
            .map_err(|err| CallError::internal(format!("HTTP request failed: {err}")))
    };

    let sse = stream::once(init).flat_map(move |result| {
        let request_id = request_id_stream.clone();
        let error_status_codes = error_status_codes_stream.clone();
        match result {
            Err(err) => Box::pin(stream::once(async move {
                ResponseEnvelope::error(request_id, err)
            })) as ResponseStream,
            Ok(response) => {
                let status = response.status();
                if !status.is_success() {
                    let request_id = request_id.clone();
                    Box::pin(stream::once(async move {
                        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(
                            (
                                response.bytes_stream(),
                                SseParser::new(),
                                false,
                                0u64,
                            ),
                            move |(mut bytes, mut parser, broken, mut total_bytes)| {
                                let request_id = request_id_inner.clone();
                                async move {
                                    if broken {
                                        return None;
                                    }
                                    match bytes.next().await {
                                        Some(Ok(chunk)) => {
                                            let chunk_len = chunk.len() as u64;
                                            if stream_byte_cap > 0
                                                && total_bytes.saturating_add(chunk_len)
                                                    > stream_byte_cap
                                            {
                                                let error = CallError::new(
                                                    "HTTP_413",
                                                    format!(
                                                        "upstream SSE stream exceeded the {stream_byte_cap}-byte total streamed-bytes cap on a subscription operation"
                                                    ),
                                                    false,
                                                );
                                                return Some((
                                                    vec![ResponseEnvelope::error(
                                                        request_id, error,
                                                    )],
                                                    (bytes, parser, true, total_bytes),
                                                ));
                                            }
                                            total_bytes = total_bytes.saturating_add(chunk_len);
                                            match parser.feed(&chunk, false) {
                                                Ok(events) => {
                                                    let envelopes: Vec<ResponseEnvelope> =
                                                        events
                                                            .into_iter()
                                                            .map(|e| {
                                                                sse_event_envelope(
                                                                    e, &request_id,
                                                                )
                                                            })
                                                            .collect();
                                                    Some((
                                                        envelopes,
                                                        (bytes, parser, false, total_bytes),
                                                    ))
                                                }
                                                Err(err) => {
                                                    let error = CallError::internal(format!(
                                                        "SSE parse error: {err}"
                                                    ));
                                                    Some((
                                                        vec![ResponseEnvelope::error(
                                                            request_id, error,
                                                        )],
                                                        (bytes, parser, true, total_bytes),
                                                    ))
                                                }
                                            }
                                        },
                                        Some(Err(err)) => {
                                            let error = CallError::internal(format!(
                                                "SSE stream error: {err}"
                                            ));
                                            Some((
                                                vec![ResponseEnvelope::error(request_id, error)],
                                                (bytes, parser, true, total_bytes),
                                            ))
                                        }
                                        None => match parser.feed(&[], true) {
                                            Ok(events) if !events.is_empty() => {
                                                let envelopes: Vec<ResponseEnvelope> = events
                                                    .into_iter()
                                                    .map(|e| sse_event_envelope(e, &request_id))
                                                    .collect();
                                                Some((
                                                    envelopes,
                                                    (bytes, parser, true, total_bytes),
                                                ))
                                            }
                                            _ => None,
                                        },
                                    }
                                }
                            },
                        )
                        .flat_map(stream::iter),
                    ) as ResponseStream
                }
            }
        }
    });

    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`, plus the
/// frame's `event:` field name when the upstream sent one (WHATWG
/// event-stream semantics: the last `event:` line before the blank
/// line wins; absent → `None`, never the implicit `message` default —
/// consumers distinguish "upstream named it" from "default name").
pub(crate) struct SseEvent {
    pub(crate) data: String,
    pub(crate) event: Option<String>,
}

#[derive(Debug, thiserror::Error)]
pub(crate) enum SseParseError {
    #[error("SSE event buffer exceeded {SSE_EVENT_BUFFER_CAP} bytes without a complete event")]
    BufferOverflow,
}

/// 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
/// unterminated event — trips `SseParseError::BufferOverflow` instead
/// of buffering without bound. The check runs *before* the buffer
/// takes a chunk's bytes, so the reassembly buffer can never exceed
/// the cap (FWD-14). This bounds one *line*, not the
/// stream; the per-subscription total is
/// [`HttpClientConfig::stream_total_byte_cap`], enforced by the
/// `forward_stream` unfold across every `feed`.
pub(crate) const SSE_EVENT_BUFFER_CAP: usize = 1024 * 1024;

/// Incremental byte-level SSE frame parser.
///
/// Holds raw bytes across `feed` calls so a frame split across TCP
/// chunks is reassembled, and only decodes UTF-8 once a complete line
/// (or EOF) bounds the decode window — a multi-byte character split at
/// a chunk boundary is therefore not corrupted. Framing follows the
/// WHATWG `text/event-stream` draft semantics for what the call
/// protocol needs: lines split on `\n` with optional trailing `\r`;
/// `data:` fields accumulate and join with `\n`; `event:` is captured
/// per frame (last one before the blank line wins) and surfaced on
/// [`SseEvent::event`] for non-JSON payloads (FWD-17); `id:` and
/// `retry:` fields are accepted and ignored; a blank line dispatches
/// the pending event; a pending event with data is dispatched at EOF.
pub(crate) struct SseParser {
    buf: Vec<u8>,
    data_lines: Vec<String>,
    data_seen: bool,
    event_name: Option<String>,
    bom_stripped: bool,
}

impl SseParser {
    pub(crate) fn new() -> Self {
        Self {
            buf: Vec::new(),
            data_lines: Vec::new(),
            data_seen: false,
            event_name: None,
            bom_stripped: false,
        }
    }

    /// Feeds one chunk and drains every complete event (a blank line
    /// dispatches; the last line stays buffered unless `eof`). With
    /// `eof`, also dispatches a pending event if it carries data
    /// lines, and flushes the buffer.
    pub(crate) fn feed(&mut self, chunk: &[u8], eof: bool) -> Result<Vec<SseEvent>, SseParseError> {
        if self.buf.len().saturating_add(chunk.len()) > SSE_EVENT_BUFFER_CAP {
            return Err(SseParseError::BufferOverflow);
        }
        self.buf.extend_from_slice(chunk);
        let mut events = Vec::new();
        let mut start = 0usize;
        while let Some(nl) = self.buf[start..].iter().position(|&b| b == b'\n') {
            let end = start + nl;
            let line_end = if end > start && self.buf[end - 1] == b'\r' {
                end - 1
            } else {
                end
            };
            let line = self.buf[start..line_end].to_vec();
            if let Some(event) = self.parse_line(&line) {
                events.push(event);
            }
            start = end + 1;
        }
        if eof {
            if start < self.buf.len() {
                let line = self.buf[start..].to_vec();
                if let Some(event) = self.parse_line(&line) {
                    events.push(event);
                }
            }
            if self.data_seen {
                if let Some(event) = self.complete_event() {
                    events.push(event);
                }
            }
            self.buf.clear();
        } else {
            self.buf.drain(..start);
        }
        Ok(events)
    }

    fn parse_line(&mut self, line: &[u8]) -> Option<SseEvent> {
        if !self.bom_stripped {
            self.bom_stripped = true;
            let bom = b"\xef\xbb\xbf";
            let line = if line.starts_with(bom) {
                &line[bom.len()..]
            } else {
                line
            };
            return self.parse_line(line);
        }
        let text = match std::str::from_utf8(line) {
            Ok(t) => t,
            Err(_) => return None,
        };
        if text.is_empty() {
            return if self.data_seen {
                self.complete_event()
            } else {
                self.discard_event();
                None
            };
        }
        if text.starts_with(':') {
            return None;
        }
        let (field, value) = match text.split_once(':') {
            Some((f, v)) => (f, v.strip_prefix(' ').unwrap_or(v)),
            None => (text, ""),
        };
        if field == "data" {
            self.data_lines.push(value.to_string());
            self.data_seen = true;
        } else if field == "event" && !value.is_empty() {
            self.event_name = Some(value.to_string());
        }
        None
    }

    fn complete_event(&mut self) -> Option<SseEvent> {
        self.data_seen = false;
        Some(SseEvent {
            data: std::mem::take(&mut self.data_lines).join("\n"),
            event: self.event_name.take(),
        })
    }

    fn discard_event(&mut self) {
        self.data_seen = false;
        self.event_name = None;
        self.data_lines.clear();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use alkcall::core::types::Capabilities;
    use alkcall::registry::context::{AbortPolicy, ScopedPeerEnv};
    use serde_json::json;
    use std::collections::HashMap as TestHashMap;
    use std::sync::Arc as TestArc;
    use std::time::Duration;

    struct CapturedRequest {
        #[allow(dead_code)]
        method: String,
        #[allow(dead_code)]
        target: String,
        #[allow(dead_code)]
        headers: TestHashMap<String, String>,
    }

    fn noop_context() -> OperationContext {
        struct NoopEnv;
        #[async_trait::async_trait]
        impl alkcall::registry::env::OperationEnv for NoopEnv {
            async fn invoke_with_policy(
                &self,
                _ns: &str,
                _op: &str,
                _input: Value,
                parent: &OperationContext,
                _policy: AbortPolicy,
            ) -> ResponseEnvelope {
                ResponseEnvelope::ok(parent.request_id.clone(), Value::Null)
            }
            fn contains(&self, _name: &str) -> bool {
                false
            }
        }
        OperationContext {
            request_id: "req-fwd".to_string(),
            parent_request_id: None,
            identity: None,
            handler_identity: None,
            forwarded_for: None,
            capabilities: Capabilities::new(),
            metadata: TestHashMap::new(),
            scoped_env: ScopedPeerEnv::empty(),
            env: TestArc::new(NoopEnv),
            abort_policy: AbortPolicy::default(),
            deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
            internal: true,
            ownership: None,
        }
    }

    fn request_url(base_url: &str, template: &str, input: Value) -> Result<url::Url, CallError> {
        let ctx = noop_context();
        let (_, url, _, _) = build_request(
            base_url,
            template,
            "GET",
            &None,
            &TestHashMap::new(),
            "svc",
            &serde_json::json!({"type": "object", "additionalProperties": true}),
            &input,
            &ctx,
        )?;
        Ok(url)
    }

    #[test]
    fn traversal_value_cannot_escape_template_path() {
        let url = request_url(
            "https://api.example.com",
            "/repos/{owner}/{repo}/issues",
            json!({"owner": "../../admin", "repo": "x"}),
        )
        .expect("request builds");
        assert_eq!(url.host_str(), Some("api.example.com"));
        assert_eq!(url.path(), "/repos/..%2F..%2Fadmin/x/issues");
        assert!(!url.path().contains("/admin"));
    }

    #[test]
    fn structural_characters_cannot_split_or_inject_url_parts() {
        let url = request_url(
            "https://api.example.com",
            "/files/{name}",
            json!({"name": "a?via=query#frag"}),
        )
        .expect("request builds");
        assert_eq!(url.query(), None, "`?` in a value must be encoded");
        assert_eq!(url.fragment(), None, "`#` in a value must be encoded");
        assert_eq!(url.path(), "/files/a%3Fvia=query%23frag");

        let url = request_url(
            "https://api.example.com",
            "/files/{name}",
            json!({"name": "a b/c\\d"}),
        )
        .expect("request builds");
        assert_eq!(url.path_segments().map(|s| s.count()), Some(2));
        assert_eq!(url.path(), "/files/a%20b%2Fc%5Cd");

        let url = request_url(
            "https://api.example.com",
            "/files/{name}",
            json!({"name": "héllo→世界"}),
        )
        .expect("request builds");
        assert_eq!(url.path(), "/files/h%C3%A9llo%E2%86%92%E4%B8%96%E7%95%8C");
    }

    #[test]
    fn rendering_is_single_pass_and_never_re_substitutes() {
        let url = request_url(
            "https://api.example.com",
            "/x/{a}/{b}",
            json!({"a": "{b}", "b": "second"}),
        )
        .expect("request builds");
        assert_eq!(url.path(), "/x/%7Bb%7D/second");

        let rendered = render_path_template(
            "/x/{a}",
            Some(&json!({"a": "../../{b}"}).as_object().unwrap().clone()),
        )
        .expect("renders");
        assert_eq!(rendered, "/x/..%2F..%2F%7Bb%7D");
    }

    #[test]
    fn base_path_prefix_is_preserved() {
        let url = request_url("https://api.openai.com/v1", "/chat/completions", json!({}))
            .expect("request builds");
        assert_eq!(url.path(), "/v1/chat/completions");

        let url =
            request_url("https://api.example.com", "/data", json!({})).expect("request builds");
        assert_eq!(url.path(), "/data");
    }

    #[test]
    fn absolute_url_in_input_cannot_change_origin_or_hop_paths() {
        let url = request_url(
            "https://api.example.com",
            "/fetch/{url}",
            json!({"url": "http://169.254.169.254/latest/meta-data"}),
        )
        .expect("absolute URL in a path value stays an encoded segment");
        assert_eq!(url.host_str(), Some("api.example.com"));
        assert_eq!(
            url.path(),
            "/fetch/http:%2F%2F169.254.169.254%2Flatest%2Fmeta-data"
        );

        let url = request_url(
            "https://api.example.com",
            "/fetch/{target}",
            json!({"target": "https://evil.example.com/x"}),
        )
        .expect("https absolute URL also stays an encoded segment");
        assert_eq!(url.host_str(), Some("api.example.com"));
        assert_eq!(url.path(), "/fetch/https:%2F%2Fevil.example.com%2Fx");
    }

    #[test]
    fn unbound_and_malformed_templates_error_loudly() {
        let err = request_url("https://api.example.com", "/x/{missing}", json!({}))
            .expect_err("unbound placeholder must error");
        assert!(err.message.contains("unbound placeholder"));

        let err = request_url("https://api.example.com", "/x/{open", json!({}))
            .expect_err("unterminated placeholder must error");
        assert!(err.message.contains("unterminated"));

        let err = request_url("https://api.example.com", "/x{missing}/a/b", json!({}))
            .expect_err("partial render without the placeholder is still loud");
        assert!(err.message.contains("unbound placeholder"));
    }

    #[test]
    fn base_url_validation_rejects_bad_inputs() {
        let err = request_url("ftp://api.example.com", "/x", json!({}))
            .expect_err("non-http scheme must be rejected");
        assert!(err.message.contains("not an HTTP scheme"));

        let err = request_url("https://u:p@api.example.com", "/x", json!({}))
            .expect_err("userinfo must be rejected");
        assert!(err.message.contains("userinfo"));

        let err = request_url("not a url at all", "/x", json!({}))
            .expect_err("unparseable base must be rejected");
        assert!(err.message.contains("invalid base_url"));
    }

    #[test]
    fn query_values_remain_encoded_via_query_pairs_mut() {
        let url = request_url(
            "https://api.example.com",
            "/search",
            json!({"q": "a&b=c d", "lang": "en"}),
        )
        .expect("request builds");
        assert_eq!(url.query(), Some("lang=en&q=a%26b%3Dc+d"));
    }

    #[test]
    fn undeclared_input_keys_are_rejected_not_sent_upstream() {
        let ctx = noop_context();
        let schema = json!({
            "type": "object",
            "properties": {"owner": {"type": "string"}, "body": {"type": "object"}},
        });
        for input in [
            json!({"owner": "a", "debug": "true"}),
            json!({"owner": "a", "impersonate_id": "x"}),
            json!({"debug": "true"}),
        ] {
            let err = build_request(
                "https://api.example.com",
                "/x/{owner}",
                "GET",
                &None,
                &TestHashMap::new(),
                "svc",
                &schema,
                &input,
                &ctx,
            )
            .expect_err("undeclared peer input must be rejected");
            assert_eq!(err.code, "INVALID_INPUT", "input was: {input}");
            assert!(err.message.contains("not declared"), "input was: {input}");
        }
    }

    #[test]
    fn declared_body_and_header_params_route_off_the_query_string() {
        let ctx = noop_context();
        let schema = json!({
            "type": "object",
            "properties": {
                "q": {"type": "string"},
                "X-Trace-Id": {"type": "string", "wire": "header"},
                "body": {"type": "object"},
            },
        });
        let (_, url, body, headers) = build_request(
            "https://api.example.com",
            "/search",
            "POST",
            &None,
            &TestHashMap::new(),
            "svc",
            &schema,
            &json!({"q": "rust", "X-Trace-Id": "t-1", "body": {"page": 2}}),
            &ctx,
        )
        .expect("declared input builds");
        assert_eq!(url.query(), Some("q=rust"));
        assert_eq!(
            headers
                .get("x-trace-id")
                .expect("header param sent as header")
                .to_str()
                .expect("ascii header"),
            "t-1"
        );
        assert_eq!(body, Some(json!({"page": 2})));
    }

    #[test]
    fn non_object_input_is_rejected() {
        let ctx = noop_context();
        let schema = json!({"type": "object", "properties": {"q": {"type": "string"}}});
        for input in [json!(null), json!([1]), json!("str"), json!(42)] {
            let err = build_request(
                "https://api.example.com",
                "/x",
                "GET",
                &None,
                &TestHashMap::new(),
                "svc",
                &schema,
                &input,
                &ctx,
            )
            .expect_err("non-object input must be rejected");
            assert_eq!(err.code, "INVALID_INPUT");
        }
    }

    #[test]
    fn additional_properties_true_opts_into_catch_all_input() {
        let ctx = noop_context();
        let schema = json!({
            "type": "object",
            "properties": {"q": {"type": "string"}},
            "additionalProperties": true,
        });
        let (_, url, _, _) = build_request(
            "https://api.example.com",
            "/search",
            "GET",
            &None,
            &TestHashMap::new(),
            "svc",
            &schema,
            &json!({"q": "rust", "debug": "true"}),
            &ctx,
        )
        .expect("catch-all input builds");
        let q = url.query().expect("query present");
        assert!(q.contains("q=rust") && q.contains("debug=true"), "{q}");
    }

    /// FWD-18, routing half pin: a key that matched a placeholder never
    /// also emits as a query parameter — the skip in `build_request`
    /// precedes query routing and is value-shape-agnostic. The scalar
    /// case renders into the path only; the object-value case is
    /// rejected (see the structural-value test below), so here the
    /// companion non-placeholder key proves query routing stays intact
    /// around the placeholder skip.
    #[test]
    fn placeholder_keys_never_double_route_as_query_params() {
        let url = request_url(
            "https://api.example.com",
            "/repos/{owner}",
            json!({"owner": "octocat", "extra": "q-val"}),
        )
        .expect("scalar placeholder builds");
        assert_eq!(url.path(), "/repos/octocat");
        assert_eq!(
            url.query(),
            Some("extra=q-val"),
            "only the non-placeholder key routes to query"
        );

        let ctx = noop_context();
        let schema = json!({
            "type": "object",
            "properties": {"id": {"type": "string"}},
        });
        let err = build_request(
            "https://api.example.com",
            "/items/{id}",
            "GET",
            &None,
            &TestHashMap::new(),
            "svc",
            &schema,
            &json!({"id": "x", "undeclared": "v"}),
            &ctx,
        )
        .expect_err("undeclared key still rejected with a placeholder present");
        assert_eq!(err.code, "INVALID_INPUT");
    }

    /// FWD-18, structural-value decision: an object/array value under a
    /// placeholder key is an `INVALID_INPUT` error, not a spliced JSON
    /// segment. A path placeholder designates exactly one literal
    /// segment; the pre-decision behavior embedded the minified JSON
    /// into the path (and, under an additionalProperties schema, also
    /// emitted the key as a query param — the double-route).
    #[test]
    fn object_or_array_path_values_error_instead_of_splicing_json_into_the_path() {
        for value in [json!({"a": 1}), json!([1, 2])] {
            let err = request_url(
                "https://api.example.com",
                "/things/{id}",
                json!({"id": value}),
            )
            .expect_err("structural path values must be rejected");
            assert_eq!(err.code, "INVALID_INPUT", "value was: {value}");
            assert!(
                err.message.contains("must be a scalar"),
                "value was: {value}"
            );
            assert!(
                !err.message.contains("{\"a\":1}") && !err.message.contains("[1,2]"),
                "error must not echo the structural value: {value}"
            );
        }

        for value in [json!("text"), json!(42), json!(true)] {
            let url = request_url(
                "https://api.example.com",
                "/things/{id}",
                json!({ "id": value }),
            )
            .expect("scalar path values still render");
            assert!(
                url.path().starts_with("/things/"),
                "scalar {value} renders into the path"
            );
        }
    }

    /// FWD-19 pin (documented trade-off, not a rejection): literal `%`
    /// in template text is preserved as-is, so an assembly-supplied
    /// pre-encoded template like `/s3%2Fkeys` reaches the upstream
    /// verbatim (most stacks route `%2F` differently from `/` — the
    /// assembly layer owns that choice, ADR-066). A `%` arriving in a
    /// *value* is always encoded to `%25`, so the only bare `%` in a
    /// rendered path is one the template author placed.
    #[test]
    fn percent_in_template_text_survives_and_percent_in_values_are_always_encoded() {
        let url = request_url(
            "https://api.example.com",
            "/s3%2Fkeys/{name}",
            json!({"name": "x"}),
        )
        .expect("template text may carry literal percent escapes");
        assert_eq!(url.path(), "/s3%2Fkeys/x");

        let url = request_url(
            "https://api.example.com",
            "/files/{name}",
            json!({"name": "a%2Fb"}),
        )
        .expect("value percent is encoded, not preserved");
        assert_eq!(url.path(), "/files/a%252Fb");
    }

    /// Empirical pin of the failure mode FWD-13 describes, against the
    /// locked `url` crate: lone `.`/`..` (and their percent-escaped
    /// spellings) are silently *normalized away* by `Url::set_path`, so
    /// the encode set alone cannot keep the rendered path faithful.
    #[test]
    fn url_set_path_normalizes_lone_dot_and_dot_dot_path_values() {
        let mut url = Url::parse("https://api.example.com").expect("parses");
        for (spliced, normalized) in [
            ("/tenants/../resources", "/resources"),
            ("/tenants/./resources", "/tenants/resources"),
            ("/files/..", "/"),
            ("/repos/%2E%2E/x", "/x"),
        ] {
            url.set_path(spliced);
            assert_eq!(
                url.path(),
                normalized,
                "set_path silently normalized `{spliced}`"
            );
        }
    }

    #[test]
    fn lone_dot_dot_path_value_is_rejected() {
        let err = request_url(
            "https://api.example.com",
            "/tenants/{tenant}/resources",
            json!({"tenant": ".."}),
        )
        .expect_err("lone `..` value must be rejected");
        assert_eq!(err.code, "INVALID_INPUT");
        assert!(
            err.message.contains("lone dot segment"),
            "message must explain the lone-dot rejection: {}",
            err.message
        );
        assert!(
            !err.message.contains(".."),
            "error must not echo the raw value: {}",
            err.message
        );
    }

    #[test]
    fn lone_dot_path_value_is_rejected() {
        let err = request_url(
            "https://api.example.com",
            "/tenants/{tenant}/resources",
            json!({"tenant": "."}),
        )
        .expect_err("lone `.` value must be rejected");
        assert_eq!(err.code, "INVALID_INPUT");
    }

    #[test]
    fn percent_escapes_spellings_of_lone_dots_are_rejected() {
        for value in ["\u{2e}\u{2e}", "%2e%2e", "%2E%2e", "%2e%2E"] {
            let err = request_url(
                "https://api.example.com",
                "/tenants/{tenant}/resources",
                json!({ "tenant": value }),
            )
            .expect_err("escaped lone-dot spellings must be rejected");
            assert_eq!(err.code, "INVALID_INPUT", "value was: {value}");
        }

        for value in ["%2e", "%2E"] {
            let err = request_url(
                "https://api.example.com",
                "/tenants/{tenant}/resources",
                json!({ "tenant": value }),
            )
            .expect_err("escaped lone-dot spellings must be rejected");
            assert_eq!(err.code, "INVALID_INPUT", "value was: {value}");
        }
    }

    #[test]
    fn dotted_but_not_lone_dot_values_still_render() {
        for value in [
            "v1.2.3",
            ".hidden-file",
            "..hidden",
            "hidden..",
            "a..b",
            ".a.b.",
            "...",
        ] {
            let url = request_url(
                "https://api.example.com",
                "/files/{name}",
                json!({ "name": value }),
            )
            .unwrap_or_else(|e| panic!("value `{value}` must render: {e:?}"));
            assert!(
                url.path().starts_with("/files/"),
                "value `{value}` rendered into the path"
            );
        }
    }

    /// FWD-13 belt-and-braces: across a dot/percent/binary corpus the
    /// post-`set_path` invariant holds — the decoded URL path segments
    /// equal the base dir plus the decoded rendered segments,
    /// byte-identical — so a normalizer rewrite (today's lone-dot
    /// removal, tomorrow's regression in the `url` crate) can never
    /// silently re-route an authenticated request. Lone-dot values are
    /// rejected outright; values containing `/` or `\` are rejected as
    /// smuggled separators only if the encode-set layer ever regressed.
    #[test]
    fn post_set_path_invariant_holds_across_dot_percent_binary_corpus() {
        let corpus = [
            "v1.2.3",
            ".hidden-file",
            "..",
            ".",
            "%2e",
            "%2E",
            "%2e%2e",
            "%2E%2E",
            "%252e",
            "hidden..",
            "a..b",
            "...",
            "a%2Fb",
            "a b",
            "h\\éllo→世界",
            "line\nbreak",
            "tab\tchar",
            "\u{7f}\u{1f600}",
        ];
        for value in corpus {
            let outcome = request_url(
                "https://api.example.com/v1",
                "/files/{name}",
                json!({ "name": value }),
            );
            match outcome {
                Ok(url) => {
                    assert_eq!(url.host_str(), Some("api.example.com"));
                    let segments: Vec<String> = url
                        .path_segments()
                        .map(|s| {
                            s.map(|seg| percent_decode_str(seg).decode_utf8_lossy().into_owned())
                                .collect()
                        })
                        .unwrap_or_default();
                    assert_eq!(
                        segments.len(),
                        3,
                        "value `{value:?}` must render as one literal segment under /v1/files/"
                    );
                    assert_eq!(
                        segments.get(2).map(String::as_str),
                        Some(value),
                        "value `{value:?}` must survive byte-identical as the final segment"
                    );
                    assert!(
                        !segments.iter().any(|s| s == "." || s == ".."),
                        "value `{value:?}` must not leave lone dot segments: {:?}",
                        segments
                    );
                }
                Err(err) => {
                    assert_eq!(
                        err.code, "INVALID_INPUT",
                        "value `{value:?}` may only fail as INVALID_INPUT"
                    );
                }
            }
        }
    }

    fn ctx_with_capability(namespace: &str, value: String) -> OperationContext {
        let mut ctx = noop_context();
        ctx.capabilities = Capabilities::new().with_http_token(namespace, value);
        ctx
    }

    fn minimal_client() -> TestArc<SharedHttpClient> {
        TestArc::new(
            SharedHttpClient::new(crate::client::HttpClientConfig::default())
                .expect("client builds"),
        )
    }

    type ServerResponder = Arc<dyn Fn(&CapturedRequest) -> http::Response<Vec<u8>> + Send + Sync>;

    async fn spawn_responder(responder: ServerResponder) -> String {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("local addr");
        tokio::spawn(async move {
            loop {
                let Ok((mut sock, _)) = listener.accept().await else {
                    break;
                };
                let mut buf = vec![0u8; 8192];
                let mut n = 0;
                loop {
                    let read = sock.read(&mut buf[n..]).await.unwrap_or(0);
                    if read == 0 {
                        break;
                    }
                    n += read;
                    if String::from_utf8_lossy(&buf[..n]).contains("\r\n\r\n") {
                        break;
                    }
                }
                let request = String::from_utf8_lossy(&buf[..n]);
                let mut lines = request.lines();
                let request_line = lines.next().unwrap_or_default();
                let mut parts_iter = request_line.split_whitespace();
                let method = parts_iter.next().unwrap_or("GET").to_string();
                let target = parts_iter.next().unwrap_or("/").to_string();
                let mut headers = TestHashMap::new();
                for line in lines {
                    if line.is_empty() {
                        break;
                    }
                    if let Some((k, v)) = line.split_once(':') {
                        headers.insert(k.trim().to_lowercase(), v.trim().to_string());
                    }
                }
                let response = (responder)(&CapturedRequest {
                    method,
                    target,
                    headers,
                });
                let mut head = format!("HTTP/1.1 {}\r\n", response.status());
                for (name, value) in response.headers() {
                    head.push_str(&format!(
                        "{}: {}\r\n",
                        name,
                        value.to_str().unwrap_or_default()
                    ));
                }
                head.push_str(&format!(
                    "content-length: {}\r\n\r\n",
                    response.body().len()
                ));
                let _ = sock.write_all(head.as_bytes()).await;
                let _ = sock.write_all(response.body()).await;
                let _ = sock.flush().await;
                let _ = sock.shutdown().await;
            }
        });
        format!("http://{addr}")
    }

    async fn collect_stream(mut stream: ResponseStream) -> Vec<ResponseEnvelope> {
        let mut out = Vec::new();
        while let Some(envelope) = stream.next().await {
            out.push(envelope);
        }
        out
    }

    fn http_response(status: u16, content_type: &str, body: Vec<u8>) -> http::Response<Vec<u8>> {
        let mut builder = http::Response::builder().status(status);
        if !content_type.is_empty() {
            builder = builder.header("content-type", content_type);
        }
        builder.body(body).expect("static response builds")
    }

    /// Spawns a raw-TCP responder whose response head is written from a
    /// status/content-type pair, then hands the socket to an async
    /// writer so a test can trickle SSE frames over time (the
    /// wire-level seam the FWD-15/FWD-14 tests need: a stream that stays
    /// open past a deadline, or dribbles bytes toward a cap).
    async fn spawn_sse_responder_with_writer<F, Fut>(head: &str, writer: F) -> String
    where
        F: FnOnce(tokio::net::tcp::OwnedWriteHalf) -> Fut + Send + 'static,
        Fut: std::future::Future<Output = ()> + Send,
    {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("local addr");
        let head = head.to_string();
        tokio::spawn(async move {
            use tokio::io::{AsyncReadExt, AsyncWriteExt};
            let Ok((mut sock, _)) = listener.accept().await else {
                return;
            };
            let mut buf = vec![0u8; 8192];
            loop {
                let read = sock.read(&mut buf).await.unwrap_or(0);
                if read == 0 || String::from_utf8_lossy(&buf).contains("\r\n\r\n") {
                    break;
                }
            }
            let _ = sock.write_all(head.as_bytes()).await;
            let _ = sock.flush().await;
            let (_, write_half) = sock.into_split();
            writer(write_half).await;
        });
        format!("http://{addr}")
    }

    fn streaming_client(total_byte_cap: u64) -> TestArc<SharedHttpClient> {
        TestArc::new(
            SharedHttpClient::new(crate::client::HttpClientConfig {
                stream_total_byte_cap: total_byte_cap,
                ..crate::client::HttpClientConfig::default()
            })
            .expect("client builds"),
        )
    }

    /// Builds a client whose configured request/read timeout is the
    /// (test-scaled) old 30 s deadline: the FWD-15 wire test sends
    /// through it and asserts the stream outlives that deadline.
    fn client_with_timeout(timeout: Duration) -> TestArc<SharedHttpClient> {
        TestArc::new(
            SharedHttpClient::new(crate::client::HttpClientConfig {
                request_timeout: Some(timeout),
                connect_timeout: Some(Duration::from_secs(5)),
                read_timeout: Some(timeout),
                ..crate::client::HttpClientConfig::default()
            })
            .expect("client builds"),
        )
    }

    async fn call_forward(base_url: &str, ctx: OperationContext) -> ResponseEnvelope {
        call_forward_authed(base_url, ctx, &None).await
    }

    async fn call_forward_authed(
        base_url: &str,
        ctx: OperationContext,
        auth_scheme: &Option<HttpAuthScheme>,
    ) -> ResponseEnvelope {
        forward(
            &minimal_client(),
            base_url,
            "/x",
            "GET",
            auth_scheme,
            &TestHashMap::new(),
            "svc",
            &serde_json::json!({"type": "object"}),
            &[],
            json!({}),
            ctx,
        )
        .await
    }

    #[tokio::test]
    async fn vendor_json_content_types_decode_as_json() {
        for content_type in [
            "application/vnd.api+json",
            "application/problem+json",
            "application/hal+json; charset=utf-8",
            "application/json",
            "APPLICATION/JSON",
        ] {
            let base = spawn_responder(TestArc::new(move |_parts| {
                http_response(200, content_type, br#"{"ok":true}"#.to_vec())
            }))
            .await;
            let envelope = call_forward(&base, noop_context()).await;
            match envelope.result {
                Ok(Value::Object(map)) => assert_eq!(map["ok"], json!(true), "{content_type}"),
                other => panic!("{content_type}: expected JSON object, got {other:?}"),
            }
        }
    }

    #[tokio::test]
    async fn oversized_json_body_trips_the_response_cap() {
        let response = http_response(200, "application/json", vec![b'['; RESPONSE_BODY_CAP + 1]);
        let base = spawn_responder(TestArc::new(move |_| response.clone())).await;
        let envelope = call_forward(&base, noop_context()).await;
        match envelope.result {
            Err(err) => {
                assert_eq!(err.code, "HTTP_413");
                assert!(err.message.contains("response cap"));
            }
            other => panic!("expected cap error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn oversized_text_body_trips_the_response_cap() {
        let response = http_response(200, "text/plain", vec![b'a'; RESPONSE_BODY_CAP + 1]);
        let base = spawn_responder(TestArc::new(move |_| response.clone())).await;
        let envelope = call_forward(&base, noop_context()).await;
        match envelope.result {
            Err(err) => assert_eq!(err.code, "HTTP_413"),
            other => panic!("expected cap error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn oversized_binary_body_trips_the_response_cap() {
        let response = http_response(
            200,
            "application/octet-stream",
            vec![0u8; RESPONSE_BODY_CAP + 1],
        );
        let base = spawn_responder(TestArc::new(move |_| response.clone())).await;
        let envelope = call_forward(&base, noop_context()).await;
        match envelope.result {
            Err(err) => assert_eq!(err.code, "HTTP_413"),
            other => panic!("expected cap error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn non_sse_content_type_on_a_sub_stream_errors_loudly() {
        let base = spawn_responder(TestArc::new(|_parts| {
            http_response(200, "text/html", b"<html>hello</html>".to_vec())
        }))
        .await;
        let stream = forward_stream(
            &minimal_client(),
            &base,
            "/x",
            "GET",
            &None,
            &TestHashMap::new(),
            "svc",
            &serde_json::json!({"type": "object"}),
            &[],
            json!({}),
            noop_context(),
        );
        let envelopes = collect_stream(stream).await;
        assert_eq!(
            envelopes.len(),
            1,
            "exactly one loud error, not an empty stream"
        );
        match &envelopes[0].result {
            Err(err) => {
                assert_eq!(err.code, "INVALID_RESPONSE_TYPE");
                assert!(err.message.contains("text/html"));
                assert!(err.message.contains("text/event-stream"));
            }
            other => panic!("expected content-type error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn sse_content_type_still_streams() {
        let base = spawn_responder(TestArc::new(|_parts| {
            http_response(
                200,
                "text/event-stream; charset=utf-8",
                b"data: {\"n\":1}\n\ndata: done\n\n".to_vec(),
            )
        }))
        .await;
        let stream = forward_stream(
            &minimal_client(),
            &base,
            "/x",
            "GET",
            &None,
            &TestHashMap::new(),
            "svc",
            &serde_json::json!({"type": "object"}),
            &[],
            json!({}),
            noop_context(),
        );
        let envelopes = collect_stream(stream).await;
        assert_eq!(envelopes.len(), 2);
        assert!(envelopes[0].result.is_ok());
        assert!(envelopes[1].result.is_ok());
    }

    /// FWD-17: a valid-JSON payload surfaces as the decoded value
    /// itself; a non-JSON payload surfaces as the
    /// `{"data", "event"}` wrapper. Numbers-vs-quoted-strings are both
    /// valid JSON, so both decode (the raw-text fallback would have
    /// erased that distinction).
    #[tokio::test]
    async fn sse_payload_contract_decodes_json_and_wraps_non_json() {
        let base = spawn_responder(TestArc::new(|_parts| {
            http_response(
                200,
                "text/event-stream",
                b"data: 123\n\ndata: \"123\"\n\ndata: plain text\n\ndata: {\"n\":1}\n\n".to_vec(),
            )
        }))
        .await;
        let stream = forward_stream(
            &minimal_client(),
            &base,
            "/x",
            "GET",
            &None,
            &TestHashMap::new(),
            "svc",
            &serde_json::json!({"type": "object"}),
            &[],
            json!({}),
            noop_context(),
        );
        let envelopes = collect_stream(stream).await;
        assert_eq!(envelopes.len(), 4);
        assert_eq!(envelopes[0].result.clone().unwrap(), json!(123));
        assert_eq!(envelopes[1].result.clone().unwrap(), json!("123"));
        assert_eq!(
            envelopes[2].result.clone().unwrap(),
            json!({"data": "plain text", "event": null})
        );
        assert_eq!(envelopes[3].result.clone().unwrap(), json!({"n": 1}));
    }

    /// FWD-17: an upstream `event:` field on a non-JSON payload is
    /// carried in the wrapper; JSON payloads surface as themselves even
    /// under a named event, and the name is reset after dispatch (the
    /// second frame must not inherit the first frame's name).
    #[tokio::test]
    async fn sse_event_field_carrys_on_the_non_json_wrapper_and_resets() {
        let base = spawn_responder(TestArc::new(|_parts| {
            http_response(
                200,
                "text/event-stream",
                b"event: error\ndata: upstream exploded\n\nevent: custom\ndata: {\"n\":1}\n\ndata: after\n\n"
                    .to_vec(),
            )
        }))
        .await;
        let stream = forward_stream(
            &minimal_client(),
            &base,
            "/x",
            "GET",
            &None,
            &TestHashMap::new(),
            "svc",
            &serde_json::json!({"type": "object"}),
            &[],
            json!({}),
            noop_context(),
        );
        let envelopes = collect_stream(stream).await;
        assert_eq!(envelopes.len(), 3);
        assert_eq!(
            envelopes[0].result.clone().unwrap(),
            json!({"data": "upstream exploded", "event": "error"})
        );
        assert_eq!(envelopes[1].result.clone().unwrap(), json!({"n": 1}));
        assert_eq!(
            envelopes[2].result.clone().unwrap(),
            json!({"data": "after", "event": null}),
            "the name must not leak across frames"
        );
    }

    /// FWD-17 parser pin: WHATWG last-wins for repeated `event:` lines
    /// in one frame; an `event:`-only frame (no data) dispatches
    /// nothing but must not leak its name into a later frame.
    #[test]
    fn sse_parser_last_event_wins_and_name_does_not_leak_across_frames() {
        let mut parser = SseParser::new();
        let events = parser
            .feed(b"event: a\nevent: b\ndata: x\n\n", false)
            .expect("ascii only");
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].event.as_deref(), Some("b"));
        assert_eq!(events[0].data, "x");

        let events = parser
            .feed(b"event: orphan\n\n", false)
            .expect("ascii only");
        assert!(events.is_empty(), "no data field, nothing dispatched");

        let events = parser.feed(b"data: next\n\n", false).expect("ascii only");
        assert_eq!(events.len(), 1);
        assert_eq!(
            events[0].event.as_deref(),
            None,
            "an event:-only frame must not leak its name"
        );
    }

    #[tokio::test]
    async fn bearer_credential_with_control_character_fails_loudly() {
        let base = spawn_responder(TestArc::new(|_parts| {
            http_response(200, "application/json", b"{}".to_vec())
        }))
        .await;
        let ctx = ctx_with_capability("svc", "tok\u{0007}en-secret-marker".to_string());
        let envelope = call_forward_authed(&base, ctx, &Some(HttpAuthScheme::Bearer)).await;
        match envelope.result {
            Err(err) => {
                assert!(err
                    .message
                    .contains("refusing to send the request unauthenticated"));
                assert!(
                    !err.message.contains("secret-marker"),
                    "error must not echo credential material"
                );
                assert!(
                    !err.message.contains("tok\u{0007}en"),
                    "error must not echo credential material"
                );
            }
            other => panic!("expected loud credential error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn api_key_with_invalid_header_name_fails_loudly() {
        let base = "https://api.example.com".to_string();
        let ctx = ctx_with_capability("svc", "key-value".to_string());
        let mut ctx = ctx;
        ctx.capabilities = Capabilities::new().with_http_token("svc", "key-value".to_string());
        let result = build_request(
            &base,
            "/x",
            "GET",
            &Some(HttpAuthScheme::ApiKey {
                header_name: "bad header name".to_string(),
            }),
            &TestHashMap::new(),
            "svc",
            &serde_json::json!({"type": "object"}),
            &json!({}),
            &ctx,
        );
        match result {
            Ok(_) => panic!("invalid API-key header name must fail loudly"),
            Err(err) => {
                assert!(err.message.contains("invalid header name"));
                assert!(!err.message.contains("key-value"));
            }
        }
    }

    #[tokio::test]
    async fn default_header_with_invalid_value_fails_loudly() {
        let mut defaults = TestHashMap::new();
        defaults.insert("X-Trace".to_string(), "bad\u{0000}value".to_string());
        let err = build_request(
            "https://api.example.com",
            "/x",
            "GET",
            &None,
            &defaults,
            "svc",
            &serde_json::json!({"type": "object"}),
            &json!({}),
            &noop_context(),
        )
        .expect_err("invalid default-header value must fail loudly");
        assert!(err.message.contains("X-Trace"));
        assert!(err.message.contains("invalid value"));
    }

    #[tokio::test]
    async fn default_header_with_invalid_name_fails_loudly() {
        let mut defaults = TestHashMap::new();
        defaults.insert("bad header".to_string(), "v".to_string());
        let err = build_request(
            "https://api.example.com",
            "/x",
            "GET",
            &None,
            &defaults,
            "svc",
            &serde_json::json!({"type": "object"}),
            &json!({}),
            &noop_context(),
        )
        .expect_err("invalid default-header name must fail loudly");
        assert!(err.message.contains("bad header"));
    }

    #[tokio::test]
    async fn error_body_is_echoed_bounded_in_the_error_envelope() {
        let body = vec![b'x'; ERROR_BODY_ECHO_CAP * 3];
        let base = spawn_responder(TestArc::new(move |_| {
            http_response(429, "text/plain", body.clone())
        }))
        .await;
        let envelope = call_forward(&base, noop_context()).await;
        match envelope.result {
            Err(err) => {
                assert_eq!(err.code, "HTTP_429", "message was: {}", err.message);
                assert!(
                    err.message.contains("HTTP 429: Too Many Requests"),
                    "message was: {}",
                    err.message
                );
                assert!(
                    err.message.contains("[truncated]"),
                    "message was: {}",
                    err.message
                );
                assert!(
                    err.message.len() < ERROR_BODY_ECHO_CAP * 2,
                    "echo must stay bounded, was {}",
                    err.message.len()
                );
            }
            other => panic!("expected HTTP_429 error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn small_error_body_is_echoed_in_full() {
        let base = spawn_responder(TestArc::new(|_| {
            http_response(404, "text/plain", b"no such widget: id=42".to_vec())
        }))
        .await;
        let envelope = call_forward(&base, noop_context()).await;
        match envelope.result {
            Err(err) => {
                assert_eq!(err.code, "HTTP_404");
                assert!(err.message.contains("no such widget: id=42"));
            }
            other => panic!("expected HTTP_404, got {other:?}"),
        }
    }

    #[test]
    fn json_detection_covers_vendor_suffix_and_rejects_lookalikes() {
        for positive in [
            "application/json",
            "application/json; charset=utf-8",
            "application/vnd.api+json",
            "application/problem+json",
            "Application/Vnd.Api+JSON",
        ] {
            assert!(is_json_content_type(positive), "{positive} must be JSON");
        }
        for negative in [
            "text/json",
            "application/jsonx",
            "xapplication/json",
            "text/html",
        ] {
            assert!(
                !is_json_content_type(negative),
                "{negative} must not be JSON"
            );
        }
    }

    #[test]
    fn sse_detection_requires_exact_essence() {
        assert!(is_sse_content_type("text/event-stream"));
        assert!(is_sse_content_type("text/event-stream; charset=utf-8"));
        assert!(!is_sse_content_type("text/html"));
        assert!(!is_sse_content_type("text/event-streamx"));
    }

    #[tokio::test]
    async fn oversized_non2xx_error_body_is_not_echoed_unbounded() {
        let body = vec![b'e'; STATUS_BODY_DRAIN + 1];
        let base = spawn_responder(TestArc::new(move |_| {
            http_response(500, "text/plain", body.clone())
        }))
        .await;
        let envelope = call_forward(&base, noop_context()).await;
        match envelope.result {
            Err(err) => {
                assert_eq!(err.code, "HTTP_500");
                assert!(
                    err.message.len() < STATUS_BODY_DRAIN,
                    "echo must stay far below the drain budget"
                );
                assert!(err.message.contains("too large to echo"));
            }
            other => panic!("expected HTTP_500, got {other:?}"),
        }
    }

    /// FWD-15 wire test (deadline-scaled): the responder trickles
    /// `: keepalive` comments and one `data:` event past the configured
    /// total request timeout (a scaled stand-in for the 30 s default
    /// the streaming path used to inherit); the subscription must still
    /// be delivering events after that deadline has passed. With the
    /// fix, `forward_stream` sends through the derived no-total-timeout
    /// client, so the stream survives; without it, reqwest 0.13's total
    /// timeout rides into the body stream and kills the subscription at
    /// the deadline.
    #[tokio::test]
    async fn stream_survives_past_the_total_request_timeout() {
        let timeout = Duration::from_millis(500);
        let keepalive = Duration::from_millis(200);
        let total_keepalives = 6u32;
        let start = std::time::Instant::now();
        let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n";
        let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
            use tokio::io::AsyncWriteExt;
            for _ in 0..total_keepalives {
                tokio::time::sleep(keepalive).await;
                let _ = sock.write_all(b": keepalive\n\n").await;
                let _ = sock.flush().await;
            }
            let _ = sock.write_all(b"data: {\"late\":true}\n\n").await;
            let _ = sock.flush().await;
            let _ = sock.shutdown().await;
        })
        .await;
        let client = client_with_timeout(timeout);
        let stream = forward_stream(
            &client,
            &base,
            "/x",
            "GET",
            &None,
            &TestHashMap::new(),
            "svc",
            &serde_json::json!({"type": "object"}),
            &[],
            json!({}),
            noop_context(),
        );
        tokio::pin!(stream);
        let crossed = tokio::time::timeout(Duration::from_secs(5), stream.next())
            .await
            .expect("the stream must deliver within the test budget")
            .expect("stream must not end without the data event");
        assert!(
            start.elapsed() > timeout,
            "the event must arrive after the old total-request deadline (elapsed {:?}, timeout {:?})",
            start.elapsed(),
            timeout
        );
        match crossed.result {
            Ok(value) => assert_eq!(value, json!({"late": true})),
            other => panic!("expected the post-deadline data event, got {other:?}"),
        }
        assert!(
            tokio::time::timeout(Duration::from_millis(500), stream.next())
                .await
                .unwrap_or(None)
                .is_none(),
            "responder closed after the data event; stream must end"
        );
    }

    /// FWD-14 wire test: a stream whose total bytes exceed the
    /// configured per-subscription cap terminates with exactly one
    /// terminal error envelope (the stream-ends semantics: error frame,
    /// then end).
    #[tokio::test]
    async fn stream_exceeding_total_byte_cap_terminates_with_one_error() {
        let cap = 4096u64;
        let chunk = vec![b'a'; 1024];
        let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n";
        let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
            use tokio::io::AsyncWriteExt;
            for _ in 0..64 {
                let _ = sock.write_all(b"data: ").await;
                let _ = sock.write_all(&chunk).await;
                let _ = sock.write_all(b"\n\n").await;
                let _ = sock.flush().await;
            }
            let _ = sock.shutdown().await;
        })
        .await;
        let client = streaming_client(cap);
        let stream = forward_stream(
            &client,
            &base,
            "/x",
            "GET",
            &None,
            &TestHashMap::new(),
            "svc",
            &serde_json::json!({"type": "object"}),
            &[],
            json!({}),
            noop_context(),
        );
        let envelopes = collect_stream(stream).await;
        assert!(!envelopes.is_empty(), "events before the cap still flow");
        assert!(
            envelopes[..envelopes.len() - 1]
                .iter()
                .all(|e| e.result.is_ok()),
            "every envelope before the terminal one is an event"
        );
        let terminal = envelopes.last().expect("terminal envelope present");
        match &terminal.result {
            Err(err) => {
                assert_eq!(err.code, "HTTP_413");
                assert!(err.message.contains("total streamed-bytes cap"));
            }
            other => panic!("expected the terminal cap error, got {other:?}"),
        }
        let ok_count = envelopes[..envelopes.len() - 1].len();
        assert_eq!(
            envelopes.len(),
            ok_count + 1,
            "exactly one terminal error envelope after the last event"
        );
    }

    /// FWD-14 parser-boundary test: for a never-newline feed the
    /// cap check fires *before* the buffer takes the chunk's bytes, so
    /// the buffered bytes stay at or below the cap — the pre-fix shape
    /// (extend first, check after) buffered past the cap before erroring.
    /// A full-cap buffer remains legal (the cap is inclusive) as long as
    /// the arriving chunk completes a line.
    #[test]
    fn line_cap_trips_before_the_buffer_takes_the_overshooting_chunk() {
        let mut parser = SseParser::new();
        let seed = vec![b'x'; SSE_EVENT_BUFFER_CAP + 1];
        let oversized = parser.feed(&seed, false);
        assert!(
            matches!(oversized, Err(SseParseError::BufferOverflow)),
            "a single over-cap line trips at the pre-extend check"
        );
        let mut parser = SseParser::new();
        let half = vec![b'x'; SSE_EVENT_BUFFER_CAP / 2];
        let first = parser.feed(&half, false);
        assert!(first.is_ok(), "partial line under the cap buffers fine");
        let second = parser.feed(&seed, false);
        assert!(
            matches!(second, Err(SseParseError::BufferOverflow)),
            "the chunk that would push past the cap is rejected before extend"
        );
        let mut parser = SseParser::new();
        let at_cap = vec![b'x'; SSE_EVENT_BUFFER_CAP - 2];
        let ok = parser.feed(&at_cap, false);
        assert!(ok.is_ok(), "a partial line under the cap is legal");
        let framing = parser.feed(b"\n\n", false);
        assert!(
            framing.is_ok(),
            "the newline pair completes the event without tripping the cap"
        );
        let next = parser.feed(&vec![b'y'; SSE_EVENT_BUFFER_CAP], false);
        assert!(
            next.is_ok(),
            "the dispatched event drained the buffer; a fresh full-cap line is legal again"
        );
        let over = parser.feed(b"z", false);
        assert!(
            matches!(over, Err(SseParseError::BufferOverflow)),
            "one byte past a full buffer still trips before extend"
        );
    }
}