alkhttp 0.5.0

//! `from_openapi` adapter: parse an OpenAPI 3.x document into
//! [`HandlerRegistration`] bundles with reqwest-backed forwarding handlers
//! (ADR-051 for the input format).
//!
//! The forwarding handler is the no-env-vars credential injection point
//! (ADR-014): it reads `OperationContext.capabilities`, never
//! `std::env::var`. Provenance is `FromOpenAPI` (leaf,
//! `composition_authority: None`, `scoped_env: None`, `Internal` by
//! default — ADR-015/022). Imported error codes are prefixed `HTTP_<status>`
//! to avoid collision with the protocol-level codes (ADR-023).
//!
//! [ADR-051]: https://docs.rs/alkhttp (docs/architecture/decisions)

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

use alkcall::client::{AdapterError, OperationAdapter};
use alkcall::core::types::Capabilities;
use alkcall::registry::context::OperationContext;
use alkcall::registry::registration::{
    make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
};
use alkcall::registry::spec::{
    AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility,
};
use async_trait::async_trait;
use serde_json::Value;

use super::forward::{
    forward, forward_stream, validate_path_template, HttpAuthScheme, HttpServiceConfig,
    GATEWAY_BODY_KEY, HEADER_PARAM_IN_MARKER, HEADER_PARAM_MARKER_VALUE,
};
use super::openapi_spec::{
    collect_ignored_schema_keys, OpenAPISpec, Operation, Parameter, SUCCESS_RESPONSE_KEYS,
};
use crate::adapters::input_validation::bounded_join;
use crate::adapters::input_validation::CompiledInputSchema;
use crate::client::SharedHttpClient;

fn unbound_placeholders(path_template: &str, input_schema: &Value) -> Vec<String> {
    let mut unbound = Vec::new();
    let mut rest = path_template;
    while let Some(start) = rest.find('{') {
        let Some(end_rel) = rest[start..].find('}') else {
            break;
        };
        let name = &rest[start + 1..start + end_rel];
        if !name.is_empty() {
            let properties = input_schema
                .get("properties")
                .and_then(|p| p.as_object())
                .map(|p| p.keys().cloned().collect::<Vec<_>>())
                .unwrap_or_default();
            if !properties.iter().any(|k| k == name) {
                unbound.push(name.to_string());
            }
        }
        rest = &rest[start + end_rel + 1..];
    }
    unbound
}

fn oai14_ignored_keys(schemas: &[&Value]) -> Vec<String> {
    let mut found = Vec::new();
    for schema in schemas {
        collect_ignored_schema_keys(schema, &mut found);
    }
    found.sort();
    found.dedup();
    found
}

fn collision_message(batch: &str, kind: &str, first_path: &str) -> AdapterError {
    AdapterError::SchemaParse {
        message: format!(
            "duplicate {kind} in import batch: `{batch}` (first registered from \
             {first_path}; the registry would silently replace the earlier \
             registration — disambiguate the operationIds or paths)"
        ),
    }
}

fn reject_collisions(
    op_ids: Vec<String>,
    paths: Vec<String>,
    routes: Vec<(String, String)>,
) -> Result<(), AdapterError> {
    let mut seen_names: HashMap<&str, &str> = HashMap::new();
    let mut seen_routes: HashMap<&(String, String), &str> = HashMap::new();
    for ((op_id, path), route) in op_ids.iter().zip(paths.iter()).zip(routes.iter()) {
        if let Some(first_path) = seen_names.get(op_id.as_str()) {
            return Err(collision_message(op_id, "operationId", first_path));
        }
        if let Some(first_path) = seen_routes.get(route) {
            return Err(collision_message(op_id, "path+method", first_path));
        }
        seen_names.insert(op_id, path);
        seen_routes.insert(route, path);
    }
    Ok(())
}

/// The `from_openapi` adapter (ADR-051 input format): turns a parsed
/// OpenAPI document into forwarding-handler registrations under one
/// namespace — the credential injection point per ADR-014.
pub struct FromOpenAPI {
    spec: OpenAPISpec,
    config: HttpServiceConfig,
    http_client: Arc<SharedHttpClient>,
}

impl FromOpenAPI {
    /// Assemble the adapter over a parsed spec, service config, and the
    /// shared outbound client.
    pub fn new(
        spec: OpenAPISpec,
        config: HttpServiceConfig,
        http_client: Arc<SharedHttpClient>,
    ) -> Self {
        Self {
            spec,
            config,
            http_client,
        }
    }

    fn normalize_operation_id(op: &Operation, method: &str, path: &str) -> String {
        if let Some(id) = &op.operation_id {
            return id.clone();
        }
        let parts: Vec<&str> = path
            .split('/')
            .filter(|p| !p.is_empty() && !p.starts_with('{'))
            .collect();
        let base = if parts.is_empty() {
            "root".to_string()
        } else {
            parts.join("_")
        };
        format!("{method}_{base}")
    }

    fn detect_op_type(method: &str, op: &Operation) -> OperationType {
        // OAI-06: the success envelope may be declared under any 2XX key
        // (204, 206, 226...), under the class wildcard `2XX` (review 002
        // OAI-13), or under `default` — a stream declared there must
        // still classify as Sub, not fall through to a giant single-
        // string text body. [`SUCCESS_RESPONSE_KEYS`] is in precedence
        // order: concrete statuses outrank the wildcard, which outranks
        // `default`.
        let success = SUCCESS_RESPONSE_KEYS
            .iter()
            .find_map(|k| op.responses.get(*k));
        if let Some(resp) = success {
            if resp.content.contains_key("text/event-stream") {
                return OperationType::Sub;
            }
        }
        if method.eq_ignore_ascii_case("get") {
            OperationType::Query
        } else {
            OperationType::Mutation
        }
    }

    fn build_input_schema(
        &self,
        op: &Operation,
        path_parameters: &[Parameter],
    ) -> Result<Value, AdapterError> {
        let mut properties = serde_json::Map::new();
        let mut required = Vec::new();
        let locator = format!(
            "{} {}",
            op.operation_id.as_deref().unwrap_or("?"),
            self.config.namespace
        );

        // OAI-13: path-item-level parameters merge into every operation
        // under the path. The operation-level entries come second, so a
        // same-`name`-same-`in` override wins by last-insert-wins in the
        // loop below (OpenAPI override semantics), and the overridden
        // path-item entry's `required` cannot leak: the operation entry
        // is the authoritative declaration pushed last.
        let merged: Vec<&Parameter> = path_parameters.iter().chain(op.parameters.iter()).collect();
        for param in merged {
            let schema = match &param.schema {
                Some(s) => self.spec.resolve_refs_recursive(s)?,
                None => serde_json::json!({"type": "string"}),
            };
            if param.name == GATEWAY_BODY_KEY {
                return Err(AdapterError::SchemaParse {
                    message: format!(
                        "parameter named `{GATEWAY_BODY_KEY}` collides with the gateway's \
                         requestBody placeholder key; the declared parameter (path-item \
                         level or operation-level) would be diverted into the request \
                         body at call time — rename the parameter (review 001 OAI-07)"
                    ),
                });
            }
            match param.in_.as_str() {
                "header" => {
                    self.check_header_param_collisions(&param.name, &locator)?;
                    properties.insert(
                        param.name.clone(),
                        serde_json::json!({
                            HEADER_PARAM_IN_MARKER: HEADER_PARAM_MARKER_VALUE,
                            "schema": schema,
                        }),
                    );
                }
                "cookie" => {
                    return Err(AdapterError::SchemaParse {
                        message: format!(
                            "parameter `{}` on {} {} uses `in: cookie`, which the HTTP \
                             adapter does not support; cookies cannot be declared through \
                             the gateway contract — define a header or query parameter \
                             instead (review 001 OAI-03)",
                            param.name,
                            self.config.namespace,
                            op.operation_id.as_deref().unwrap_or("?")
                        ),
                    });
                }
                _ => {}
            }
            if param.in_ != "header" {
                properties.insert(param.name.clone(), schema);
            }
            if param.required {
                required.push(param.name.clone());
            }
        }

        if let Some(body) = &op.request_body {
            if let Some(json_schema) = body.content.get("application/json") {
                let resolved = self.spec.resolve_refs_recursive(json_schema)?;
                properties.insert(GATEWAY_BODY_KEY.to_string(), resolved);
                required.push(GATEWAY_BODY_KEY.to_string());
            }
        }

        if properties.is_empty() {
            return Ok(serde_json::json!({
                "type": "object",
                "additionalProperties": false,
            }));
        }

        // OAI-18: the generated input schema advertises closed-by-default
        // object shape explicitly (`additionalProperties: false`), so the
        // /schema advert states the rule call time enforces and a
        // schema-driven external validator reaches the same verdicts.
        Ok(serde_json::json!({
            "type": "object",
            "properties": properties,
            "required": required,
            "additionalProperties": false,
        }))
    }

    /// Header-precedence gate (review 002 OAI-19): `build_request` inserts
    /// header params first, then `default_headers`, then credential
    /// headers — later `insert` calls replace earlier ones, so a declared
    /// `in: header` parameter whose name matches a default/credential
    /// header would silently never deliver the peer's value upstream.
    /// `Authorization` on an authed namespace is rejected outright (the
    /// credential header always wins by construction); a collision with a
    /// configured `default_headers` key fails import with both names —
    /// the assembly's config is visible at import time, so silence about
    /// it would be a lie about the wire.
    fn check_header_param_collisions(&self, name: &str, locator: &str) -> Result<(), AdapterError> {
        if self.config.auth.is_some() && name.eq_ignore_ascii_case("authorization") {
            return Err(AdapterError::SchemaParse {
                message: format!(
                    "header parameter `{name}` on {locator} collides with the \
                     Authorization credential header the adapter injects from \
                     Capabilities; the peer-supplied value would be silently replaced \
                     by the outbound credential on every call — rename the parameter or \
                     remove the auth scheme from the service config (review 002 OAI-19)"
                ),
            });
        }
        let lower = name.to_ascii_lowercase();
        if self
            .config
            .default_headers
            .keys()
            .any(|k| k.to_ascii_lowercase() == lower)
        {
            return Err(AdapterError::SchemaParse {
                message: format!(
                    "header parameter `{name}` on {locator} collides with a configured \
                     default_headers entry of the same name; the default value would \
                     silently replace the peer-supplied header value at call time — \
                     rename the parameter or drop the default_headers entry \
                     (review 002 OAI-19)"
                ),
            });
        }
        if let Some(HttpAuthScheme::ApiKey { header_name }) = &self.config.auth {
            if header_name.eq_ignore_ascii_case(name) || header_name.to_ascii_lowercase() == lower {
                return Err(AdapterError::SchemaParse {
                    message: format!(
                        "header parameter `{name}` on {locator} collides with the API-key \
                         credential header `{header_name}`; the credential would silently \
                         replace the peer-supplied value at call time — rename the \
                         parameter (review 002 OAI-19)"
                    ),
                });
            }
        }
        Ok(())
    }

    fn build_output_schema(&self, op: &Operation) -> Result<Value, AdapterError> {
        // Mirrors `detect_op_type`'s success-key sweep (OAI-06, OAI-13):
        // a stream declared under a non-200/201 2XX key, the `2XX`
        // wildcard, or `default` still governs the output schema shape.
        let success = SUCCESS_RESPONSE_KEYS
            .iter()
            .find_map(|k| op.responses.get(*k));
        let Some(resp) = success else {
            return Ok(serde_json::json!({}));
        };
        if let Some(json_schema) = resp.content.get("application/json") {
            return self.spec.resolve_refs_recursive(json_schema);
        }
        if let Some(sse_schema) = resp.content.get("text/event-stream") {
            return self.spec.resolve_refs_recursive(sse_schema);
        }
        Ok(serde_json::json!({}))
    }

    fn build_error_schemas(&self, op: &Operation) -> Result<Vec<ErrorDefinition>, AdapterError> {
        let mut out = Vec::new();
        for (code, resp) in &op.responses {
            let status: Option<u16> = code.parse::<u16>().ok();
            let is_2xx = matches!(status, Some(s) if (200..300).contains(&s));
            if is_2xx {
                continue;
            }
            // OAI-13: class wildcards map to the first legal concrete
            // status in their implied range — `4XX` → HTTP_400, `5XX` →
            // HTTP_500 (ADR-023 codes must be a concrete `HTTP_<status>`;
            // the runtime mapper synthesizes the actual status for any
            // unmapped one, so the projection is a faithful
            // representative, not a catch-all lie). `default` has no
            // implied range at all: an `HTTP_0` entry would be advertised
            // by `/search` yet never match a real status, so it is
            // dropped, loudly (review 001 OAI-06). A line recording this
            // mapping lives in ADR-066.
            let wildcard_status: Option<u16> = match code.as_str() {
                "4XX" => Some(400),
                "5XX" => Some(500),
                _ => None,
            };
            let (status_code, effective_status) = match (status, wildcard_status) {
                (Some(s), _) => (s, Some(s)),
                (None, Some(s)) => (s, Some(s)),
                (None, None) => {
                    tracing::warn!(
                        operation = %op.operation_id.as_deref().unwrap_or("?"),
                        namespace = %self.config.namespace,
                        response_key = %code,
                        "response key is not a concrete HTTP status; dropping it from the \
                         imported error schemas — unmapped upstream statuses surface as \
                         HTTP_<status> at call time (review 001 OAI-06)"
                    );
                    continue;
                }
            };
            let schema = if let Some(json_schema) = resp.content.get("application/json") {
                self.spec.resolve_refs_recursive(json_schema)?
            } else {
                serde_json::json!({})
            };
            out.push(ErrorDefinition {
                code: format!("HTTP_{status_code}"),
                description: format!("HTTP {status_code} response ({code} declared)"),
                schema,
                http_status: effective_status,
            });
        }
        Ok(out)
    }

    fn build_registration(
        &self,
        method: &str,
        path: &str,
        op: &Operation,
        path_parameters: &[Parameter],
    ) -> Result<HandlerRegistration, AdapterError> {
        validate_path_template(path)?;
        let name = Self::normalize_operation_id(op, method, path);
        let qualified_name = format!("{}/{name}", self.config.namespace);
        let op_type = Self::detect_op_type(method, op);
        let input_schema = self.build_input_schema(op, path_parameters)?;
        let output_schema = self.build_output_schema(op)?;
        let error_schemas = self.build_error_schemas(op)?;
        let ignored = oai14_ignored_keys(&[&input_schema, &output_schema]);
        if !ignored.is_empty() {
            tracing::warn!(
                namespace = %self.config.namespace,
                operation = %name,
                keys = %ignored.join(", "),
                "schema declares serialization keywords the HTTP adapter ignores: they \
                 change what a conforming OpenAPI client would send (polymorphic \
                 discriminator headers, XML wire annotations) but the adapter forwards \
                 JSON only — remove them or split the operation if the upstream \
                 requires that wire shape (review 002 OAI-14)"
            );
        }

        let spec = OperationSpec::new(
            qualified_name,
            op_type,
            Visibility::Internal,
            input_schema.clone(),
            output_schema,
            error_schemas,
            AccessControl::default(),
            None,
        );
        let input_validator = CompiledInputSchema::compile(&input_schema).map_err(|message| {
            AdapterError::SchemaParse {
                message: format!(
                    "operation `{}`: {message} (review 002 OAI-18: an un-validatable \
                         input contract must fail import, not register unenforced)",
                    spec.name
                ),
            }
        })?;
        let path_template = path.to_string();
        let method_upper = method.to_ascii_uppercase();
        let auth_scheme = self.config.auth.clone();
        let default_headers = self.config.default_headers.clone();
        let base_url = self.config.base_url.clone();
        let namespace = self.config.namespace.clone();
        let http_client = Arc::clone(&self.http_client);
        let error_status_codes: Vec<(u16, String)> = spec
            .error_schemas
            .iter()
            .map(|e| (e.http_status.unwrap_or(0), e.code.clone()))
            .collect();
        let unbound = unbound_placeholders(&path_template, &input_schema);
        if !unbound.is_empty() {
            // OAI-13 name-the-cause: after the path-item merge the only
            // remaining sources of an unbound placeholder are refs the
            // parser could not resolve or a parameter missing `name`/`in`
            // — both fail at parse time — so a live placeholder almost
            // always means the spec relies on a feature the adapter does
            // not model (e.g. path-item `parameters` under an older
            // build, or a `parameters` $ref the parse path skipped).
            // Name the placeholder and the parameter path so the
            // diagnosis is not a dead end.
            return Err(AdapterError::SchemaParse {
                message: format!(
                    "path {method} {} declares placeholder(s) {} with no \
                     matching parameter in the operation's resolved input schema. Every \
                     `parameters` source was merged (path-item level and operation level, \
                     review 002 OAI-13); a placeholder left unbound after the merge means \
                     a declared parameter was dropped — check that each entry under \
                     `paths` for this path's `parameters` (and the path-item's shared \
                     list) has `name` and `in`, and that its $ref, if any, resolves. \
                     The placeholder would otherwise render as a literal `{{}}` path segment",
                    bounded_join(&[path_template.to_string()]),
                    bounded_join(&unbound)
                ),
            });
        }

        let handler = if op_type == OperationType::Sub {
            let stream_handler =
                make_streaming_handler(move |input: Value, context: OperationContext| {
                    let path_template = path_template.clone();
                    let method_upper = method_upper.clone();
                    let auth_scheme = auth_scheme.clone();
                    let default_headers = default_headers.clone();
                    let base_url = base_url.clone();
                    let namespace = namespace.clone();
                    let http_client = Arc::clone(&http_client);
                    let error_status_codes = error_status_codes.clone();
                    let input_schema = input_schema.clone();
                    let input_validator = input_validator.clone();
                    forward_stream(
                        &http_client,
                        &base_url,
                        &path_template,
                        &method_upper,
                        &auth_scheme,
                        &default_headers,
                        &namespace,
                        &input_schema,
                        &error_status_codes,
                        Some(&input_validator),
                        input,
                        context,
                    )
                });
            HandlerKind::Stream(stream_handler)
        } else {
            let once_handler = make_handler(move |input: Value, context: OperationContext| {
                let path_template = path_template.clone();
                let method_upper = method_upper.clone();
                let auth_scheme = auth_scheme.clone();
                let default_headers = default_headers.clone();
                let base_url = base_url.clone();
                let namespace = namespace.clone();
                let http_client = Arc::clone(&http_client);
                let error_status_codes = error_status_codes.clone();
                let input_schema = input_schema.clone();
                let input_validator = input_validator.clone();
                async move {
                    forward(
                        &http_client,
                        &base_url,
                        &path_template,
                        &method_upper,
                        &auth_scheme,
                        &default_headers,
                        &namespace,
                        &input_schema,
                        &error_status_codes,
                        Some(&input_validator),
                        input,
                        context,
                    )
                    .await
                }
            });
            HandlerKind::Once(once_handler)
        };

        let capabilities = Capabilities::new();
        Ok(HandlerRegistration::new(
            spec,
            handler,
            OperationProvenance::FromOpenAPI,
            None,
            None,
            capabilities,
        ))
    }
}

#[async_trait]
impl OperationAdapter for FromOpenAPI {
    async fn import(&self) -> Result<Vec<HandlerRegistration>, AdapterError> {
        self.spec.validate_import_loud_features()?;
        let mut bundles = Vec::new();
        let mut op_ids = Vec::new();
        let mut paths = Vec::new();
        let mut routes = Vec::new();
        for (path, item) in &self.spec.paths {
            for (method, op) in &item.operations {
                let registration = self.build_registration(method, path, op, &item.parameters)?;
                let name = registration.spec.name.clone();
                let qualified_op_id = name
                    .rsplit_once('/')
                    .map(|(_, id)| id.to_string())
                    .unwrap_or_else(|| name.clone());
                op_ids.push(qualified_op_id);
                paths.push(path.clone());
                routes.push((path.clone(), method.to_ascii_uppercase()));
                bundles.push(registration);
            }
        }
        reject_collisions(op_ids, paths, routes)?;
        Ok(bundles)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::adapters::forward::{build_request, HttpAuthScheme};
    use crate::adapters::input_validation::CompiledInputSchema;
    use crate::client::HttpClientConfig;
    use alkcall::core::types::Capabilities;
    use alkcall::protocol::wire::ResponseEnvelope;
    use alkcall::registry::context::AbortPolicy;
    use alkcall::registry::env::OperationEnv;
    use futures::StreamExt;
    use reqwest::header::AUTHORIZATION;
    use reqwest::Method;
    use serde_json::json;
    use std::collections::HashMap;
    use std::time::Duration;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::TcpListener;

    fn noop_context(request_id: &str, capabilities: Capabilities) -> OperationContext {
        struct NoopEnv;
        #[async_trait]
        impl 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: request_id.to_string(),
            parent_request_id: None,
            identity: None,
            handler_identity: None,
            forwarded_for: None,
            capabilities,
            metadata: HashMap::new(),
            scoped_env: alkcall::registry::context::ScopedPeerEnv::empty(),
            env: Arc::new(NoopEnv),
            abort_policy: AbortPolicy::default(),
            deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
            internal: true,
            ownership: None,
        }
    }

    fn config(namespace: &str, base_url: &str, auth: Option<HttpAuthScheme>) -> HttpServiceConfig {
        HttpServiceConfig {
            namespace: namespace.to_string(),
            base_url: base_url.to_string(),
            auth,
            default_headers: HashMap::new(),
        }
    }

    fn test_http_client() -> Arc<SharedHttpClient> {
        Arc::new(SharedHttpClient::new(HttpClientConfig::default()).unwrap())
    }

    fn adapter(spec: OpenAPISpec, config: HttpServiceConfig) -> FromOpenAPI {
        FromOpenAPI::new(spec, config, test_http_client())
    }

    fn minimal_spec_json() -> &'static str {
        r#"{
            "openapi": "3.0.0",
            "info": { "title": "Test", "version": "1.0.0" },
            "paths": {
                "/widgets": {
                    "get": {
                        "operationId": "listWidgets",
                        "responses": {
                            "200": {
                                "content": {
                                    "application/json": {
                                        "schema": { "type": "array", "items": { "type": "string" } }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }"#
    }

    /// OAI-18 advertised-vs-enforced round-trip (review 002 Test-gap 10):
    /// the chain-test that lets advertise/enforce drift surface as a
    /// CI failure instead of surviving reviews. A doc is imported; the
    /// `/schema`-exported `input_schema` (the exact value
    /// `OperationSpec::input_schema` carries, projected verbatim by
    /// `spec_to_json`) is compiled with the same runtime validator and
    /// driven over a matrix of violating inputs; every input the
    /// exported schema accepts must be accepted by `build_request`, and
    /// every input it rejects must be rejected as `INVALID_INPUT` —
    /// accept-sets exactly equal, both directions.
    #[tokio::test]
    async fn advertised_input_schema_accepts_exactly_what_build_request_accepts() {
        let doc = r#"{
            "openapi": "3.0.0",
            "info": { "title": "T", "version": "1" },
            "paths": {
                "/items": {
                    "get": {
                        "operationId": "listItems",
                        "parameters": [
                            {"name": "q", "in": "query", "schema": {"type": "string"}},
                            {"name": "n", "in": "query", "schema": {"type": "integer", "minimum": 1},
                             "required": true},
                            {"name": "level", "in": "query",
                             "schema": {"type": "string", "enum": ["low", "high"]}}
                        ],
                        "responses": {
                            "200": {"content": {"application/json": {"schema": {}}}}
                        }
                    }
                }
            }
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let imported = adapter(spec, config("items", "https://api.example.com", None));
        let bundles = imported.import().await.unwrap();
        assert_eq!(bundles.len(), 1);
        let advertised_schema = bundles[0].spec.input_schema.clone();

        for (input, expect_accepted) in [
            (json!({"n": 1}), true),
            (json!({"n": 1, "q": "rust"}), true),
            (json!({"n": 1, "level": "high"}), true),
            (json!({}), false),
            (json!({"q": "missing-n"}), false),
            (json!({"n": 0}), false),
            (json!({"n": 1, "level": "medium"}), false),
            (json!({"n": 1, "debug": true}), false),
            (json!({"n": 1, "q": {"object": 1}}), false),
            (json!({"n": 1, "n": 2}), true),
        ] {
            let advertised_verdict = jsonschema::options()
                .build(&advertised_schema)
                .unwrap()
                .is_valid(&input);
            assert_eq!(
                advertised_verdict, expect_accepted,
                "advertised schema verdict mismatch for {input}"
            );

            let result = build_request(
                "https://api.example.com",
                "/items",
                "GET",
                &None,
                &HashMap::new(),
                "items",
                &advertised_schema,
                Some(&CompiledInputSchema::compile(&advertised_schema).unwrap()),
                &input,
                &noop_context("rt-1", Capabilities::new()),
            );
            match (result, expect_accepted) {
                (Ok(_), true) => {}
                (Err(err), false) => {
                    assert_eq!(err.code, "INVALID_INPUT", "input was: {input}");
                }
                (Ok(_), false) => {
                    panic!("enforcement accepted what the advertised schema rejects: {input}")
                }
                (Err(err), true) => panic!(
                    "enforcement rejected what the advertised schema accepts: {input} — {}",
                    err.message
                ),
            }
        }
    }

    #[tokio::test]
    async fn import_minimal_doc_yields_one_registration() {
        let spec = OpenAPISpec::from_json(minimal_spec_json()).unwrap();
        let adapter = adapter(spec, config("widgets", "https://api.example.com", None));
        let bundles = adapter.import().await.unwrap();
        assert_eq!(bundles.len(), 1);
        assert_eq!(bundles[0].spec.name, "widgets/listWidgets");
        assert_eq!(bundles[0].spec.namespace, "widgets");
        assert_eq!(bundles[0].spec.op_type, OperationType::Query);
        assert_eq!(bundles[0].spec.visibility, Visibility::Internal);
        assert_eq!(bundles[0].provenance, OperationProvenance::FromOpenAPI);
        assert!(bundles[0].composition_authority.is_none());
        assert!(bundles[0].scoped_env.is_none());
    }

    #[tokio::test]
    async fn parse_failure_returns_schema_parse() {
        let result = OpenAPISpec::from_json("not json");
        assert!(matches!(result, Err(AdapterError::SchemaParse { .. })));
    }

    #[tokio::test]
    async fn missing_paths_returns_schema_parse() {
        let result = OpenAPISpec::from_json(r#"{"info":{"title":"x","version":"1"}}"#);
        match result {
            Err(AdapterError::SchemaParse { message }) => assert!(message.contains("paths")),
            other => panic!("expected SchemaParse, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn generated_operation_id_when_absent() {
        let doc = r#"{
            "openapi": "3.0.0",
            "info": { "title": "T", "version": "1" },
            "paths": { "/users/{id}/posts": { "get": {
                "parameters": [{"name": "id", "in": "path", "required": true, "schema": {"type": "string"}}],
                "responses": { "200": { "content": { "application/json": { "schema": {} } } } }
            } } }
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let adapter = adapter(spec, config("svc", "https://api.example.com", None));
        let bundles = adapter.import().await.unwrap();
        assert_eq!(bundles[0].spec.name, "svc/get_users_posts");
    }

    #[tokio::test]
    async fn duplicate_operation_ids_rejected_at_import() {
        let doc = r#"{
            "openapi": "3.0.0",
            "info": {"title": "T", "version": "1"},
            "paths": {
                "/a/x": {"get": {"operationId": "dup", "responses": {"200": {"content": {"application/json": {"schema": {}}}}}}},
                "/b/y": {"get": {"operationId": "dup", "responses": {"200": {"content": {"application/json": {"schema": {}}}}}}}
            }
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let result = adapter(spec, config("svc", "https://x", None))
            .import()
            .await;
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(
                    message.contains("duplicate operationId"),
                    "message was: {message}"
                );
                assert!(message.contains("dup"), "message was: {message}");
            }
            Ok(bundles) => panic!(
                "expected duplicate-operationId rejection, got {} bundles",
                bundles.len()
            ),
            Err(e) => panic!("expected SchemaParse, got {e}"),
        }
    }

    #[tokio::test]
    async fn generated_id_collision_between_template_and_literal_path_rejected() {
        let doc = r#"{
            "openapi": "3.0.0",
            "info": {"title": "T", "version": "1"},
            "paths": {
                "/x/{id}/y": {"get": {"parameters": [{"name": "id", "in": "path", "required": true, "schema": {"type": "string"}}], "responses": {"200": {"content": {"application/json": {"schema": {}}}}}}},
                "/x/y": {"get": {"responses": {"200": {"content": {"application/json": {"schema": {}}}}}}}
            }
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let result = adapter(spec, config("svc", "https://x", None))
            .import()
            .await;
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(
                    message.contains("duplicate operationId"),
                    "message was: {message}"
                );
                assert!(message.contains("get_x_y"), "message was: {message}");
            }
            Ok(bundles) => panic!(
                "expected generated-id collision rejection, got {} bundles",
                bundles.len()
            ),
            Err(e) => panic!("expected SchemaParse, got {e}"),
        }
    }

    #[test]
    fn route_collision_check_rejects_duplicate_path_method() {
        let result = reject_collisions(
            vec!["a".to_string(), "b".to_string()],
            vec!["/shared".to_string(), "/other".to_string()],
            vec![
                ("/shared".to_string(), "GET".to_string()),
                ("/shared".to_string(), "GET".to_string()),
            ],
        );
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(
                    message.contains("duplicate path+method"),
                    "message was: {message}"
                );
            }
            Ok(()) => panic!("expected path+method collision rejection"),
            other => panic!("expected SchemaParse, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn unresolved_path_placeholder_fails_import_loudly() {
        let doc = r#"{
            "openapi": "3.0.0",
            "info": { "title": "T", "version": "1" },
            "paths": { "/users/{id}/posts": { "get": { "responses": { "200": { "content": { "application/json": { "schema": {} } } } } } } }
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let adapter = adapter(spec, config("svc", "https://api.example.com", None));
        match adapter.import().await {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(message.contains("placeholder"), "message was: {message}");
                assert!(message.contains("id"), "message was: {message}");
            }
            Ok(bundles) => panic!(
                "expected unresolved-placeholder import error, got {} bundles",
                bundles.len()
            ),
            Err(e) => panic!("expected SchemaParse, got {e}"),
        }
    }

    #[tokio::test]
    async fn refd_parameters_and_request_body_resolve_into_op_schema() {
        let doc = r##"{
            "openapi": "3.0.0",
            "info": {"title": "T", "version": "1"},
            "components": {
                "schemas": {
                    "Widget": {"type": "object", "properties": {"name": {"type": "string"}}}
                },
                "parameters": {
                    "Id": {
                        "name": "id",
                        "in": "path",
                        "required": true,
                        "schema": {"type": "string"}
                    }
                },
                "requestBodies": {
                    "WidgetInput": {
                        "content": {
                            "application/json": {
                                "schema": {"$ref": "#/components/schemas/Widget"}
                            }
                        }
                    }
                }
            },
            "paths": {
                "/widgets/{id}": {"put": {
                    "operationId": "replaceWidget",
                    "parameters": [{"$ref": "#/components/parameters/Id"}],
                    "requestBody": {"$ref": "#/components/requestBodies/WidgetInput"},
                    "responses": {"200": {"content": {"application/json": {"schema": {}}}}}
                }}
            }
        }"##;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("svc", "https://x", None))
            .import()
            .await
            .unwrap();
        assert_eq!(bundles.len(), 1);
        let props = bundles[0]
            .spec
            .input_schema
            .get("properties")
            .unwrap()
            .as_object()
            .unwrap();
        assert!(props.contains_key("id"), "ref'd path param present");
        let body = props.get("body").unwrap();
        assert_eq!(body["type"], "object");
        assert!(
            body.get("properties")
                .and_then(|p| p.as_object())
                .unwrap()
                .contains_key("name"),
            "recursive $ref (requestBody -> schemas/Widget) fully expanded, not empty"
        );
    }

    #[tokio::test]
    async fn self_referential_schema_behind_refd_parameter_fails_cleanly_not_abort() {
        let doc = r##"{
            "openapi": "3.0.0",
            "info": {"title": "T", "version": "1"},
            "components": {
                "schemas": {
                    "Node": {"type": "object", "properties": {
                        "next": {"$ref": "#/components/schemas/Node"}
                    }}
                },
                "requestBodies": {
                    "NodeInput": {
                        "content": {
                            "application/json": {"schema": {"$ref": "#/components/requestBodies/NodeInput"}}
                        }
                    }
                }
            },
            "paths": {
                "/nodes": {"post": {
                    "operationId": "createNode",
                    "requestBody": {"$ref": "#/components/requestBodies/NodeInput"},
                    "responses": {"201": {"content": {"application/json": {"schema": {}}}}}
                }}
            }
        }"##;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let result = adapter(spec, config("svc", "https://x", None))
            .import()
            .await;
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(
                    message.contains("circular $ref"),
                    "clean resolver error, message was: {message}"
                );
            }
            Ok(bundles) => panic!(
                "expected clean circular-$ref error, got {} bundles",
                bundles.len()
            ),
            Err(e) => panic!("expected SchemaParse, got {e}"),
        }
    }

    #[tokio::test]
    async fn unterminated_path_template_fails_import_not_first_call() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/x{open":{"get":{"operationId":"x","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let result = adapter(spec, config("ns", "https://x", None))
            .import()
            .await;
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(
                    message.contains("unterminated placeholder"),
                    "message was: {message}"
                );
                assert!(message.contains("/x{open"), "message was: {message}");
            }
            Ok(bundles) => panic!(
                "unterminated path template must fail at import (JS-02), got {} bundles",
                bundles.len()
            ),
            Err(e) => panic!("expected SchemaParse, got {e}"),
        }
    }

    #[tokio::test]
    async fn header_param_colliding_with_authorization_on_authed_namespace_rejected() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/me":{"get":{
                "operationId":"me",
                "parameters":[{"name":"Authorization","in":"header","schema":{"type":"string"}}],
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}}"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let result = adapter(
            spec,
            config("svc", "https://x", Some(HttpAuthScheme::Bearer)),
        )
        .import()
        .await;
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(
                    message.contains("Authorization credential header"),
                    "message was: {message}"
                );
                assert!(message.contains("OAI-19"), "message was: {message}");
            }
            Ok(bundles) => panic!(
                "Authorization header param on authed namespace must be rejected, got {} bundles",
                bundles.len()
            ),
            Err(e) => panic!("expected SchemaParse, got {e}"),
        }
    }

    #[tokio::test]
    async fn header_param_colliding_with_default_headers_rejected() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/t":{"get":{
                "operationId":"t",
                "parameters":[{"name":"X-Tenant","in":"header","schema":{"type":"string"}}],
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}}"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let mut cfg = config("svc", "https://x", None);
        cfg.default_headers
            .insert("x-tenant".to_string(), "fixed".to_string());
        let result = adapter(spec, cfg).import().await;
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(
                    message.contains("default_headers"),
                    "message was: {message}"
                );
                assert!(message.contains("X-Tenant"), "message was: {message}");
            }
            Ok(bundles) => panic!(
                "header/default_headers collision must be rejected, got {} bundles",
                bundles.len()
            ),
            Err(e) => panic!("expected SchemaParse, got {e}"),
        }
    }

    #[tokio::test]
    async fn header_param_colliding_with_api_key_header_rejected() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/t":{"get":{
                "operationId":"t",
                "parameters":[{"name":"x-api-key","in":"header","schema":{"type":"string"}}],
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}}"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let auth = Some(HttpAuthScheme::ApiKey {
            header_name: "X-API-Key".to_string(),
        });
        let result = adapter(spec, config("svc", "https://x", auth))
            .import()
            .await;
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(
                    message.contains("API-key credential header"),
                    "message was: {message}"
                );
            }
            Ok(bundles) => panic!(
                "API-key header collision must be rejected, got {} bundles",
                bundles.len()
            ),
            Err(e) => panic!("expected SchemaParse, got {e}"),
        }
    }

    #[tokio::test]
    async fn header_param_without_config_collision_imports_cleanly() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/t":{"get":{
                "operationId":"t",
                "parameters":[{"name":"X-Trace-Id","in":"header","schema":{"type":"string"}}],
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}}"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("svc", "https://x", None))
            .import()
            .await
            .unwrap();
        assert_eq!(bundles.len(), 1);
    }

    #[tokio::test]
    async fn op_type_detection() {
        let get_doc = r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},"paths":{"/g":{"get":{"operationId":"g","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}}"#;
        let post_doc = r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},"paths":{"/p":{"post":{"operationId":"p","responses":{"201":{"content":{"application/json":{"schema":{}}}}}}}}}"#;
        let sse_doc = r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},"paths":{"/s":{"post":{"operationId":"s","responses":{"200":{"content":{"text/event-stream":{"schema":{}}}}}}}}}"##;

        let spec = OpenAPISpec::from_json(get_doc).unwrap();
        assert_eq!(
            adapter(spec, config("ns", "https://x", None))
                .import()
                .await
                .unwrap()[0]
                .spec
                .op_type,
            OperationType::Query
        );
        let spec = OpenAPISpec::from_json(post_doc).unwrap();
        assert_eq!(
            adapter(spec, config("ns", "https://x", None))
                .import()
                .await
                .unwrap()[0]
                .spec
                .op_type,
            OperationType::Mutation
        );
        let spec = OpenAPISpec::from_json(sse_doc).unwrap();
        assert_eq!(
            adapter(spec, config("ns", "https://x", None))
                .import()
                .await
                .unwrap()[0]
                .spec
                .op_type,
            OperationType::Sub
        );
    }

    #[tokio::test]
    async fn error_response_becomes_http_status_definition() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/x":{"get":{"operationId":"x","responses":{
                "200":{"content":{"application/json":{"schema":{}}}},
                "404":{"content":{"application/json":{"schema":{"type":"object","properties":{"msg":{"type":"string"}}}}}}
            }}}}
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("ns", "https://x", None))
            .import()
            .await
            .unwrap();
        let errors = &bundles[0].spec.error_schemas;
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0].code, "HTTP_404");
        assert_eq!(errors[0].http_status, Some(404));
    }

    #[tokio::test]
    async fn wildcard_error_keys_project_to_class_representative_status() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/x":{"get":{"operationId":"x","responses":{
                "200":{"content":{"application/json":{"schema":{}}}},
                "404":{"content":{"application/json":{"schema":{}}}},
                "4XX":{"content":{"application/json":{"schema":{"type":"object","properties":{"err":{"type":"string"}}}}}},
                "5XX":{"content":{"application/json":{"schema":{"type":"object"}}}},
                "default":{"content":{"application/json":{"schema":{}}}}
            }}}}
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("ns", "https://x", None))
            .import()
            .await
            .unwrap();
        let errors = &bundles[0].spec.error_schemas;
        let codes: Vec<&str> = errors.iter().map(|e| e.code.as_str()).collect();
        assert!(
            codes.contains(&"HTTP_400"),
            "`4XX` projects to the class representative HTTP_400 (OAI-13): {codes:?}"
        );
        assert!(
            codes.contains(&"HTTP_500"),
            "`5XX` projects to the class representative HTTP_500 (OAI-13): {codes:?}"
        );
        assert!(
            codes.contains(&"HTTP_404"),
            "the concrete 404 is retained alongside the wildcard: {codes:?}"
        );
        assert!(
            errors.iter().all(|e| e.code != "HTTP_0"),
            "/search must never advertise an HTTP_0 code that cannot match (OAI-06): {errors:?}"
        );
        assert!(errors.iter().all(|e| e.http_status.is_some()));
        let wildcard_4xx = errors
            .iter()
            .find(|e| e.code == "HTTP_400")
            .expect("4XX representative present");
        assert_eq!(wildcard_4xx.http_status, Some(400));
        assert!(
            wildcard_4xx
                .schema
                .get("properties")
                .and_then(|p| p.as_object())
                .is_some_and(|p| p.contains_key("err")),
            "the wildcard's declared payload schema is carried by the representative"
        );
    }

    #[tokio::test]
    async fn default_response_key_is_dropped_not_advertised_as_http_0() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/x":{"get":{"operationId":"x","responses":{
                "200":{"content":{"application/json":{"schema":{}}}},
                "404":{"content":{"application/json":{"schema":{}}}},
                "default":{"content":{"application/json":{"schema":{}}}}
            }}}}
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("ns", "https://x", None))
            .import()
            .await
            .unwrap();
        let errors = &bundles[0].spec.error_schemas;
        assert_eq!(
            errors.len(),
            1,
            "only the concrete 404 may be advertised; `default` has no implied status range"
        );
        assert_eq!(errors[0].code, "HTTP_404");
        assert!(
            errors.iter().all(|e| e.code != "HTTP_0"),
            "/search must never advertise an HTTP_0 code that cannot match (OAI-06): {errors:?}"
        );
        assert!(errors.iter().all(|e| e.http_status.is_some()));
    }

    #[tokio::test]
    async fn default_declared_sse_stream_classifies_as_subscription() {
        let doc = r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},"paths":{"/s":{"post":{"operationId":"s","responses":{"default":{"content":{"text/event-stream":{"schema":{}}}}}}}}}"##;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        assert_eq!(
            adapter(spec, config("ns", "https://x", None))
                .import()
                .await
                .unwrap()[0]
                .spec
                .op_type,
            OperationType::Sub,
            "a `default`-declared text/event-stream must classify as Sub, not fall \
             through to a giant single text body (OAI-06)"
        );
    }

    #[tokio::test]
    async fn non_200_2xx_sse_stream_classifies_as_subscription() {
        let doc = r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},"paths":{"/s":{"post":{"operationId":"s","responses":{"206":{"content":{"text/event-stream":{"schema":{}}}}}}}}}"##;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        assert_eq!(
            adapter(spec, config("ns", "https://x", None))
                .import()
                .await
                .unwrap()[0]
                .spec
                .op_type,
            OperationType::Sub
        );
    }

    #[tokio::test]
    async fn wildcard_2xx_sse_stream_classifies_as_subscription() {
        let doc = r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},"paths":{"/s":{"post":{"operationId":"s","responses":{"2XX":{"content":{"text/event-stream":{"schema":{}}}}}}}}}"##;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("ns", "https://x", None))
            .import()
            .await
            .unwrap();
        assert_eq!(
            bundles[0].spec.op_type,
            OperationType::Sub,
            "a `2XX`-declared text/event-stream must classify as Sub, not fall through \
             to a giant single text body (OAI-13)"
        );
        assert!(
            matches!(bundles[0].handler, HandlerKind::Stream(_)),
            "the Sub classification registers a streaming handler"
        );
    }

    #[tokio::test]
    async fn concrete_success_key_outranks_2xx_wildcard() {
        let doc = r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},"paths":{"/s":{"post":{"operationId":"s","responses":{
            "2XX":{"content":{"text/event-stream":{"schema":{}}}},
            "200":{"content":{"application/json":{"schema":{}}}}
        }}}}}"##;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("ns", "https://x", None))
            .import()
            .await
            .unwrap();
        assert_eq!(
            bundles[0].spec.op_type,
            OperationType::Mutation,
            "the concrete 200 JSON response governs; the 2XX SSE entry is not consulted"
        );
        let output = &bundles[0].spec.output_schema;
        assert_ne!(
            output.get("type").and_then(|t| t.as_str()),
            Some("string"),
            "output schema follows the concrete 200 JSON response, not the SSE wildcard"
        );
    }

    #[tokio::test]
    async fn wildcard_2xx_json_output_schema_resolves() {
        let doc = r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},"paths":{"/j":{"get":{"operationId":"j","responses":{
            "2XX":{"content":{"application/json":{"schema":{"type":"object","properties":{"v":{"type":"integer"}}}}}}
        }}}}}"##;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("ns", "https://x", None))
            .import()
            .await
            .unwrap();
        let output = &bundles[0].spec.output_schema;
        assert_eq!(
            output["properties"]["v"]["type"],
            "string".replace("string", "integer"),
            "the 2XX-declared JSON schema governs the output schema (OAI-13)"
        );
    }

    #[tokio::test]
    async fn shared_path_item_parameters_import_and_operation_overrides_win() {
        let doc = r##"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{
                "/users/{id}/posts": {
                    "parameters":[
                        {"name":"id","in":"path","required":true,"schema":{"type":"string","pattern":"^u-"}},
                        {"name":"verbose","in":"query","schema":{"type":"boolean"}}
                    ],
                    "get":{
                        "operationId":"listPosts",
                        "parameters":[
                            {"name":"verbose","in":"query","required":true,"schema":{"type":"string","maxLength":2}}
                        ],
                        "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
                    },
                    "delete":{
                        "operationId":"deletePost",
                        "responses":{"204":{"content":{"application/json":{"schema":{}}}}}
                    }
                }
            }
        }"##;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("svc", "https://x", None))
            .import()
            .await
            .unwrap();
        assert_eq!(
            bundles.len(),
            2,
            "both operations import, not a placeholder failure"
        );

        let get = bundles
            .iter()
            .find(|b| b.spec.name == "svc/listPosts")
            .expect("get op registered");
        let props = get
            .spec
            .input_schema
            .get("properties")
            .and_then(|p| p.as_object())
            .expect("input schema has properties");
        assert!(
            props.contains_key("id"),
            "the shared id placeholder is bound by the path-item parameter"
        );
        let verbose = props
            .get("verbose")
            .expect("verbose declared at both levels");
        assert_eq!(
            verbose["maxLength"], 2,
            "the operation-level entry override wins on shared name+in"
        );
        let get_required = get
            .spec
            .input_schema
            .get("required")
            .and_then(|r| r.as_array())
            .expect("required list present");
        assert!(
            get_required.iter().any(|v| v == "id"),
            "path-item required id flows into the merged requirements"
        );
        assert!(
            get_required.iter().any(|v| v == "verbose"),
            "the override's own required:true is honored"
        );

        let delete = bundles
            .iter()
            .find(|b| b.spec.name == "svc/deletePost")
            .expect("delete op registered");
        let delete_props = delete
            .spec
            .input_schema
            .get("properties")
            .and_then(|p| p.as_object())
            .expect("input schema has properties");
        assert!(delete_props.contains_key("id"));
        let verbose_schema = delete_props
            .get("verbose")
            .expect("inherited from path item");
        assert_eq!(
            verbose_schema["type"], "boolean",
            "no operation-level entry means the path-item declaration applies verbatim"
        );
    }

    #[tokio::test]
    async fn shared_path_item_parameters_defeat_the_misleading_placeholder_failure() {
        // The OAI-13 headline case: a Petstore-with-shared-params shape
        // previously failed the whole import with "declares placeholder(s)
        // id with no matching parameter", pointing nowhere near the cause.
        let doc = r##"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "components":{"parameters":{
                "Id":{"name":"id","in":"path","required":true,"schema":{"type":"integer"}}
            }},
            "paths":{
                "/pets/{id}": {
                    "parameters":[{"$ref":"#/components/parameters/Id"}],
                    "get":{
                        "operationId":"showPetById",
                        "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
                    }
                }
            }
        }"##;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("svc", "https://x", None))
            .import()
            .await
            .unwrap();
        assert_eq!(bundles.len(), 1);
        let props = bundles[0]
            .spec
            .input_schema
            .get("properties")
            .and_then(|p| p.as_object())
            .expect("input schema has properties");
        assert!(props.contains_key("id"));
        assert_eq!(props["id"]["type"], "integer");
    }

    #[tokio::test]
    async fn input_schema_from_params_and_body() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/u/{id}":{"get":{
                "operationId":"u",
                "parameters":[{"name":"id","in":"path","required":true,"schema":{"type":"string"}},{"name":"q","in":"query","schema":{"type":"string"}}],
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("ns", "https://x", None))
            .import()
            .await
            .unwrap();
        let schema = &bundles[0].spec.input_schema;
        let props = schema.get("properties").unwrap().as_object().unwrap();
        assert!(props.contains_key("id"));
        assert!(props.contains_key("q"));
        let required = schema.get("required").unwrap().as_array().unwrap();
        assert!(required.iter().any(|v| v == "id"));
    }

    #[tokio::test]
    async fn header_parameters_are_marked_and_cookies_rejected() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/track/{id}":{"get":{
                "operationId":"track",
                "parameters":[
                    {"name":"id","in":"path","required":true,"schema":{"type":"string"}},
                    {"name":"X-Trace-Id","in":"header","schema":{"type":"string"}},
                    {"name":"session","in":"cookie","schema":{"type":"string"}}
                ],
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let result = adapter(spec, config("ns", "https://x", None))
            .import()
            .await;
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(message.contains("cookie"), "message was: {message}");
                assert!(message.contains("session"), "message was: {message}");
            }
            Ok(bundles) => panic!(
                "expected cookie-parameter rejection, got {} bundles",
                bundles.len()
            ),
            Err(e) => panic!("expected SchemaParse, got {e}"),
        }

        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/track/{id}":{"get":{
                "operationId":"track",
                "parameters":[
                    {"name":"id","in":"path","required":true,"schema":{"type":"string"}},
                    {"name":"X-Trace-Id","in":"header","schema":{"type":"string"}}
                ],
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("ns", "https://x", None))
            .import()
            .await
            .unwrap();
        let props = bundles[0]
            .spec
            .input_schema
            .get("properties")
            .unwrap()
            .as_object()
            .unwrap();
        let header_prop = props.get("X-Trace-Id").expect("header param declared");
        assert_eq!(header_prop["wire"], "header");
        assert!(props.get("id").is_some());
        let q_prop = props
            .get("q")
            .map(|v| v.get("wire").is_none())
            .unwrap_or(true);
        assert!(q_prop, "query params carry no wire marker");
    }

    #[tokio::test]
    async fn header_parameter_flows_as_upstream_request_header_not_query() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/track":{"get":{
                "operationId":"track",
                "parameters":[{"name":"X-Trace-Id","in":"header","schema":{"type":"string"}}],
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}}"#;
        let (base, rx) = spawn_capturing_server().await;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("svc", &base, None))
            .import()
            .await
            .unwrap();
        let ctx = noop_context("req-hdr", Capabilities::new());
        let response = match &bundles[0].handler {
            HandlerKind::Once(h) => h(serde_json::json!({"X-Trace-Id": "t-9"}), ctx).await,
            _ => panic!("expected Once handler"),
        };
        assert!(response.result.is_ok(), "{:?}", response.result);
        let captured = rx.await.unwrap();
        assert!(
            !captured.query.contains("X-Trace-Id"),
            "header param must not land in the query string: {}",
            captured.query
        );
        assert_eq!(
            captured.headers.get("x-trace-id").map(String::as_str),
            Some("t-9")
        );
    }

    #[tokio::test]
    async fn parameter_named_body_is_rejected_at_import_even_without_request_body() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/x":{"post":{
                "operationId":"x",
                "parameters":[{"name":"body","in":"query","schema":{"type":"string"}}],
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}}"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let result = adapter(spec, config("ns", "https://x", None))
            .import()
            .await;
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(
                    message.contains("collides with the gateway"),
                    "message was: {message}"
                );
                assert!(message.contains("OAI-07"), "message was: {message}");
            }
            Ok(bundles) => panic!(
                "expected body-key collision rejection, got {} bundles",
                bundles.len()
            ),
            Err(e) => panic!("expected SchemaParse, got {e}"),
        }
    }

    #[tokio::test]
    async fn ref_resolution_in_input_schema() {
        let doc = r##"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "components":{"schemas":{"Widget":{"type":"object","properties":{"name":{"type":"string"}}}}},
            "paths":{"/w":{"post":{
                "operationId":"w",
                "requestBody":{"content":{"application/json":{"schema":{"$ref":"#/components/schemas/Widget"}}}},
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}
        }"##;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("ns", "https://x", None))
            .import()
            .await
            .unwrap();
        let props = bundles[0]
            .spec
            .input_schema
            .get("properties")
            .unwrap()
            .as_object()
            .unwrap();
        let body = props.get("body").unwrap();
        assert_eq!(body.get("type").unwrap(), "object");
        assert!(body
            .get("properties")
            .unwrap()
            .as_object()
            .unwrap()
            .contains_key("name"));
    }

    #[tokio::test]
    async fn build_request_injects_bearer_from_capabilities() {
        let caps = Capabilities::new().with_http_token("github", "tok-123".to_string());
        let ctx = noop_context("req-1", caps);
        let (method, url, body, headers) = build_request(
            "https://api.github.com",
            "/repos/{owner}/{repo}/issues",
            "GET",
            &Some(HttpAuthScheme::Bearer),
            &HashMap::new(),
            "github",
            &serde_json::json!({"type": "object", "additionalProperties": true}),
            None,
            &serde_json::json!({"owner":"a","repo":"b"}),
            &ctx,
        )
        .unwrap();
        assert_eq!(method, Method::GET);
        assert_eq!(url.path(), "/repos/a/b/issues");
        assert_eq!(url.host_str(), Some("api.github.com"));
        assert!(headers.get(AUTHORIZATION).is_none() || body.is_none());
        let auth = headers.get(AUTHORIZATION).unwrap();
        assert_eq!(auth.to_str().unwrap(), "Bearer tok-123");
    }

    #[tokio::test]
    async fn build_request_api_key_header_from_capabilities() {
        let caps = Capabilities::new().with_api_key("vastai", "key-xyz".to_string());
        let ctx = noop_context("req-2", caps);
        let (_, _, _, headers) = build_request(
            "https://api.vast.ai",
            "/machines",
            "GET",
            &Some(HttpAuthScheme::ApiKey {
                header_name: "X-API-Key".to_string(),
            }),
            &HashMap::new(),
            "vastai",
            &serde_json::json!({"type": "object", "additionalProperties": true}),
            None,
            &serde_json::json!({}),
            &ctx,
        )
        .unwrap();
        assert_eq!(
            headers.get("X-API-Key").unwrap().to_str().unwrap(),
            "key-xyz"
        );
    }

    #[tokio::test]
    async fn build_request_path_and_query_split() {
        let ctx = noop_context("req-3", Capabilities::new());
        let (_, url, _, _) = build_request(
            "https://api.example.com",
            "/widgets/{id}",
            "GET",
            &None,
            &HashMap::new(),
            "svc",
            &serde_json::json!({"type": "object", "additionalProperties": true}),
            None,
            &serde_json::json!({"id":42,"filter":"active"}),
            &ctx,
        )
        .unwrap();
        assert_eq!(url.path(), "/widgets/42");
        assert_eq!(url.query().unwrap(), "filter=active");
    }

    async fn spawn_echo_server(
        status: u16,
        body: &'static str,
        content_type: &'static str,
    ) -> String {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            loop {
                let (mut sock, _) = match listener.accept().await {
                    Ok(pair) => pair,
                    Err(_) => break,
                };
                let status_line = match status {
                    200 => "200 OK",
                    201 => "201 Created",
                    404 => "404 Not Found",
                    500 => "500 Internal Server Error",
                    _ => "200 OK",
                };
                let body_bytes = body.as_bytes();
                let response = format!(
                    "HTTP/1.1 {status_line}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                    body_bytes.len(),
                    body
                );
                let mut buf = [0u8; 4096];
                let _ = sock.read(&mut buf).await;
                sock.write_all(response.as_bytes()).await.unwrap();
                sock.flush().await.unwrap();
            }
        });
        format!("http://{addr}")
    }

    #[tokio::test]
    async fn integration_forwarding_handler_calls_external_endpoint() {
        let base = spawn_echo_server(200, r#"{"ok":true}"#, "application/json").await;
        let spec = OpenAPISpec::from_json(
            r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/data":{"get":{"operationId":"data","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}}"#,
        )
        .unwrap();
        let bundles = adapter(spec, config("svc", &base, None))
            .import()
            .await
            .unwrap();
        let registration = &bundles[0];
        let ctx = noop_context("req-10", Capabilities::new());
        let response = match &registration.handler {
            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
            _ => panic!("expected Once handler"),
        };
        assert_eq!(response.request_id, "req-10");
        match response.result {
            Ok(v) => assert_eq!(v, serde_json::json!({"ok":true})),
            Err(e) => panic!("expected Ok, got {e:?}"),
        }
    }

    #[tokio::test]
    async fn integration_non_2xx_returns_declared_error() {
        let base = spawn_echo_server(404, r#"{"error":"missing"}"#, "application/json").await;
        let spec = OpenAPISpec::from_json(
            r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/missing":{"get":{"operationId":"missing","responses":{
                "200":{"content":{"application/json":{"schema":{}}}},
                "404":{"content":{"application/json":{"schema":{"type":"object"}}}}
            }}}}}"#,
        )
        .unwrap();
        let bundles = adapter(spec, config("svc", &base, None))
            .import()
            .await
            .unwrap();
        let registration = &bundles[0];
        let ctx = noop_context("req-11", Capabilities::new());
        let response = match &registration.handler {
            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
            _ => panic!("expected Once handler"),
        };
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "HTTP_404");
                assert!(!e.retryable);
            }
            other => panic!("expected HTTP_404 error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn subscription_op_registration_is_handler_kind_stream() {
        let spec = OpenAPISpec::from_json(
            r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/stream":{"post":{"operationId":"stream","responses":{"200":{"content":{"text/event-stream":{"schema":{}}}}}}}}}"##,
        )
        .unwrap();
        let bundles = adapter(spec, config("svc", "https://x", None))
            .import()
            .await
            .unwrap();
        assert!(matches!(bundles[0].handler, HandlerKind::Stream(_)));
    }

    #[tokio::test]
    async fn query_op_registration_is_handler_kind_once() {
        let spec = OpenAPISpec::from_json(
            r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/data":{"get":{"operationId":"data","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}}"#,
        )
        .unwrap();
        let bundles = adapter(spec, config("svc", "https://x", None))
            .import()
            .await
            .unwrap();
        assert!(matches!(bundles[0].handler, HandlerKind::Once(_)));
    }

    #[tokio::test]
    async fn integration_sse_subscription_streams_responded_events() {
        let sse_body = "data: {\"n\":1}\n\ndata: {\"n\":2}\n\n";
        let base = spawn_echo_server(200, sse_body, "text/event-stream").await;
        let spec = OpenAPISpec::from_json(
            r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/stream":{"post":{"operationId":"stream","responses":{"200":{"content":{"text/event-stream":{"schema":{}}}}}}}}}"##,
        )
        .unwrap();
        let bundles = adapter(spec, config("svc", &base, None))
            .import()
            .await
            .unwrap();
        let registration = &bundles[0];
        let ctx = noop_context("req-12", Capabilities::new());
        let stream = match &registration.handler {
            HandlerKind::Stream(h) => h(serde_json::json!({}), ctx),
            _ => panic!("expected Stream handler"),
        };
        let collected: Vec<ResponseEnvelope> = stream.collect().await;
        assert_eq!(collected.len(), 2);
        assert_eq!(collected[0].result, Ok(serde_json::json!({"n":1})));
        assert_eq!(collected[1].result, Ok(serde_json::json!({"n":2})));
        assert_eq!(collected[0].request_id, "req-12");
        assert_eq!(collected[1].request_id, "req-12");
    }

    #[tokio::test]
    async fn integration_sse_subscription_http_error_returns_single_error_envelope() {
        let base = spawn_echo_server(404, r#"{"error":"missing"}"#, "application/json").await;
        let spec = OpenAPISpec::from_json(
            r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/stream":{"post":{"operationId":"stream","responses":{
                "200":{"content":{"text/event-stream":{"schema":{}}}},
                "404":{"content":{"application/json":{"schema":{"type":"object"}}}}
            }}}}}"##,
        )
        .unwrap();
        let bundles = adapter(spec, config("svc", &base, None))
            .import()
            .await
            .unwrap();
        let registration = &bundles[0];
        let ctx = noop_context("req-err", Capabilities::new());
        let stream = match &registration.handler {
            HandlerKind::Stream(h) => h(serde_json::json!({}), ctx),
            _ => panic!("expected Stream handler"),
        };
        let collected: Vec<ResponseEnvelope> = stream.collect().await;
        assert_eq!(collected.len(), 1);
        match &collected[0].result {
            Err(e) => assert_eq!(e.code, "HTTP_404"),
            other => panic!("expected HTTP_404 error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn integration_query_forwarding_unchanged_single_response() {
        let base = spawn_echo_server(200, r#"{"ok":true}"#, "application/json").await;
        let spec = OpenAPISpec::from_json(
            r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/data":{"get":{"operationId":"data","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}}"#,
        )
        .unwrap();
        let bundles = adapter(spec, config("svc", &base, None))
            .import()
            .await
            .unwrap();
        let registration = &bundles[0];
        let ctx = noop_context("req-q", Capabilities::new());
        let response = match &registration.handler {
            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
            _ => panic!("expected Once handler"),
        };
        assert_eq!(response.request_id, "req-q");
        assert_eq!(response.result, Ok(serde_json::json!({"ok":true})));
    }

    #[test]
    fn no_env_vars_read_in_build_request() {
        std::env::set_var("OPENAI_API_KEY", "should-not-be-used");
        let ctx = noop_context("req-13", Capabilities::new());
        let err = build_request(
            "https://api.openai.com",
            "/v1/chat",
            "POST",
            &Some(HttpAuthScheme::Bearer),
            &HashMap::new(),
            "openai",
            &serde_json::json!({"type": "object", "additionalProperties": true}),
            None,
            &serde_json::json!({"body":{"prompt":"hi"}}),
            &ctx,
        )
        .expect_err("absent capability is a loud error (FWD-16), never an unauthenticated send");
        assert_eq!(err.code, "INTERNAL");
        assert!(
            err.message
                .contains("capability for namespace `openai` is absent"),
            "message was: {}",
            err.message
        );
        assert!(
            !err.message.contains("should-not-be-used"),
            "error must not echo env material"
        );
        std::env::remove_var("OPENAI_API_KEY");
    }

    #[test]
    fn sse_frames_parse_multi_event_buffer() {
        let mut parser = crate::adapters::forward::SseParser::new();
        let events = parser
            .feed(b"data: a\n\ndata: b\n\n", false)
            .expect("parse errors impossible on ascii");
        assert_eq!(events.len(), 2);
        assert_eq!(events[0].data, "a");
        assert_eq!(events[1].data, "b");
        let tail = parser.feed(b"", false).expect("tail");
        assert!(tail.is_empty());
    }

    #[test]
    fn sse_frames_handle_partial_trailing_line() {
        let mut parser = crate::adapters::forward::SseParser::new();
        let events = parser
            .feed(b"data: a\n\ndata: par", false)
            .expect("parse errors impossible on ascii");
        assert_eq!(events.len(), 1);
        let rest = parser.feed(b"tial\n\n", false).expect("rest");
        assert_eq!(rest.len(), 1);
        assert_eq!(rest[0].data, "partial");
    }

    #[test]
    fn sse_frames_skip_comment_lines() {
        let mut parser = crate::adapters::forward::SseParser::new();
        let events = parser
            .feed(b": comment\ndata: x\n\n", false)
            .expect("parse errors impossible on ascii");
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].data, "x");
    }

    #[test]
    fn sse_frames_join_multi_line_data() {
        let mut parser = crate::adapters::forward::SseParser::new();
        let events = parser
            .feed(b"data: line1\ndata: line2\n\n", false)
            .expect("parse errors impossible on ascii");
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].data, "line1\nline2");
    }

    #[test]
    fn parse_sse_frames_strips_bom() {
        let mut parser = crate::adapters::forward::SseParser::new();
        let events = parser
            .feed("\u{feff}data: a\n\n".as_bytes(), false)
            .expect("parse errors impossible on bom+ascii");
        assert_eq!(events.len(), 1);
    }

    #[test]
    fn sse_multichunk_events_reassembled() {
        let mut parser = crate::adapters::forward::SseParser::new();
        let first = parser
            .feed(b"data: {\"n\":1}\n", false)
            .expect("parse errors impossible on ascii");
        assert!(first.is_empty(), "no blank line yet, event pending");
        let second = parser
            .feed(b"\ndata: {\"n\":2}\n\n", false)
            .expect("parse errors impossible on ascii");
        assert_eq!(
            second.len(),
            2,
            "the review's empirically-verified loss case"
        );
        assert_eq!(second[0].data, "{\"n\":1}");
        assert_eq!(second[1].data, "{\"n\":2}");
        let eof = parser.feed(b"", true).expect("eof");
        assert!(eof.is_empty(), "no event left pending");
    }

    #[test]
    fn sse_multichunk_split_utf8_char() {
        let payload = "{\"s\":\"héllo\"}";
        let bytes = format!("data: {payload}\n\n").into_bytes();
        let split = bytes.len() - payload.len() + 3;
        assert!(
            payload.as_bytes()[split - 8..].contains(&0xc3),
            "split inside multi-byte char"
        );
        let (head, tail) = bytes.split_at(split);
        let head = head.to_vec();
        let tail = tail.to_vec();
        let mut parser = crate::adapters::forward::SseParser::new();
        let first = parser.feed(&head, false).expect("first chunk");
        assert!(
            first.is_empty(),
            "frame incomplete until blank line arrives"
        );
        let second = parser.feed(&tail, false).expect("second chunk");
        assert_eq!(second.len(), 1);
        assert_eq!(second[0].data, payload);
    }

    #[test]
    fn sse_pending_event_dispatched_at_eof() {
        let mut parser = crate::adapters::forward::SseParser::new();
        let pending = parser.feed(b"data: tail-event\n", true).expect("eof feed");
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].data, "tail-event");
    }

    #[test]
    fn sse_oversized_partial_line_no_dispatch() {
        let big_line = "x".repeat(16 * 1024 * 1024);
        let body = format!("data: {big_line}");
        let mut parser = crate::adapters::forward::SseParser::new();
        let result = parser.feed(body.as_bytes(), false);
        assert!(
            matches!(
                result,
                Err(crate::adapters::forward::SseParseError::BufferOverflow)
            ),
            "unterminated oversized line must trip the cap"
        );
    }

    #[test]
    fn http_service_config_struct_fields() {
        let cfg = config(
            "ns",
            "https://api.example.com",
            Some(HttpAuthScheme::Bearer),
        );
        assert_eq!(cfg.namespace, "ns");
        assert_eq!(cfg.base_url, "https://api.example.com");
        assert!(matches!(cfg.auth, Some(HttpAuthScheme::Bearer)));
    }

    #[test]
    fn openapi_info_parsed_from_doc() {
        let spec = OpenAPISpec::from_json(minimal_spec_json()).unwrap();
        assert_eq!(spec.info.title, "Test");
        assert_eq!(spec.info.version, "1.0.0");
    }

    #[test]
    fn openapi_components_parsed_when_present() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "components":{"schemas":{"Foo":{"type":"object"}}},
            "paths":{"/x":{"get":{"operationId":"x","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        assert!(spec.components.is_some());
        assert!(spec
            .components
            .as_ref()
            .unwrap()
            .schemas
            .contains_key("Foo"));
    }

    #[tokio::test]
    async fn import_multiple_operations() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{
                "/a":{"get":{"operationId":"getA","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}},
                "/b":{"post":{"operationId":"postB","responses":{"201":{"content":{"application/json":{"schema":{}}}}}}}
            }
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("svc", "https://x", None))
            .import()
            .await
            .unwrap();
        assert_eq!(bundles.len(), 2);
        assert_eq!(bundles[0].spec.name, "svc/getA");
        assert_eq!(bundles[1].spec.name, "svc/postB");
    }

    #[tokio::test]
    async fn basic_auth_injection_from_capabilities() {
        let caps = Capabilities::new().with_http_token("svc", "dXNlcjpwYXNz".to_string());
        let ctx = noop_context("req-14", caps);
        let (_, _, _, headers) = build_request(
            "https://api.example.com",
            "/x",
            "GET",
            &Some(HttpAuthScheme::Basic),
            &HashMap::new(),
            "svc",
            &serde_json::json!({"type": "object", "additionalProperties": true}),
            None,
            &serde_json::json!({}),
            &ctx,
        )
        .unwrap();
        assert_eq!(
            headers.get(AUTHORIZATION).unwrap().to_str().unwrap(),
            "Basic dXNlcjpwYXNz"
        );
    }

    #[test]
    fn resolve_ref_rejects_external_refs() {
        let spec = OpenAPISpec::from_json(minimal_spec_json()).unwrap();
        let err = spec.resolve_ref("https://other/file.json").unwrap_err();
        assert!(matches!(err, AdapterError::SchemaParse { .. }));
    }

    #[tokio::test]
    async fn resolve_ref_missing_target_returns_schema_parse() {
        let spec = OpenAPISpec::from_json(minimal_spec_json()).unwrap();
        let err = spec
            .resolve_ref("#/components/schemas/Missing")
            .unwrap_err();
        assert!(matches!(err, AdapterError::SchemaParse { .. }));
    }

    #[test]
    fn default_headers_applied_to_request() {
        let ctx = noop_context("req-15", Capabilities::new());
        let mut defaults = HashMap::new();
        defaults.insert("X-Trace".to_string(), "abc".to_string());
        let (_, _, _, headers) = build_request(
            "https://api.example.com",
            "/x",
            "GET",
            &None,
            &defaults,
            "svc",
            &serde_json::json!({"type": "object", "additionalProperties": true}),
            None,
            &serde_json::json!({}),
            &ctx,
        )
        .unwrap();
        assert_eq!(headers.get("X-Trace").unwrap().to_str().unwrap(), "abc");
    }

    #[derive(serde::Deserialize)]
    struct CapturedRequest {
        method: String,
        path: String,
        query: String,
        headers: HashMap<String, String>,
        body: String,
    }

    async fn spawn_capturing_server() -> (String, tokio::sync::oneshot::Receiver<CapturedRequest>) {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let (tx, rx) = tokio::sync::oneshot::channel();
        tokio::spawn(async move {
            let (mut sock, _) = listener.accept().await.unwrap();
            let mut buf = vec![0u8; 8192];
            let n = sock.read(&mut buf).await.unwrap();
            let raw = String::from_utf8_lossy(&buf[..n]).to_string();
            let mut lines = raw.split("\r\n");
            let request_line = lines.next().unwrap_or("");
            let mut parts = request_line.split_whitespace();
            let method = parts.next().unwrap_or("").to_string();
            let raw_path = parts.next().unwrap_or("");
            let (path, query) = match raw_path.split_once('?') {
                Some((p, q)) => (p.to_string(), q.to_string()),
                None => (raw_path.to_string(), String::new()),
            };
            let mut headers = HashMap::new();
            for line in lines.by_ref() {
                if line.is_empty() {
                    break;
                }
                if let Some((k, v)) = line.split_once(':') {
                    headers.insert(k.to_lowercase(), v.trim().to_string());
                }
            }
            let body = lines.collect::<Vec<_>>().join("\r\n");
            let _ = tx.send(CapturedRequest {
                method,
                path,
                query,
                headers,
                body,
            });
            let response =
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 2\r\n\r\n{}";
            sock.write_all(response.as_bytes()).await.unwrap();
            sock.flush().await.unwrap();
        });
        (format!("http://{addr}"), rx)
    }

    #[tokio::test]
    async fn integration_forwarding_handler_sends_body_and_query() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/items/{id}":{"post":{
                "operationId":"updateItem",
                "parameters":[{"name":"id","in":"path","required":true,"schema":{"type":"string"}}],
                "requestBody":{"content":{"application/json":{"schema":{"type":"object"}}}},
                "responses":{"200":{"content":{"application/json":{"schema":{}}}}}
            }}}
        }"#;
        let (base, rx) = spawn_capturing_server().await;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("svc", &base, None))
            .import()
            .await
            .unwrap();
        let registration = &bundles[0];
        let ctx = noop_context("req-16", Capabilities::new());
        let response = match &registration.handler {
            HandlerKind::Once(h) => {
                h(serde_json::json!({"id":"42","body":{"name":"widget"}}), ctx).await
            }
            _ => panic!("expected Once handler"),
        };
        assert!(
            response.result.is_ok(),
            "expected Ok, got {:?}",
            response.result
        );
        let captured = rx.await.unwrap();
        assert_eq!(captured.method, "POST");
        assert_eq!(captured.path, "/items/42");
        assert_eq!(captured.query, "");
        assert_eq!(
            captured.headers.get("content-type").unwrap(),
            "application/json"
        );
        assert!(captured.body.contains("\"name\":\"widget\""));
    }

    #[tokio::test]
    async fn integration_bearer_token_injected_on_outbound_request() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/me":{"get":{"operationId":"me","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}
        }"#;
        let (base, rx) = spawn_capturing_server().await;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("openai", &base, Some(HttpAuthScheme::Bearer)))
            .import()
            .await
            .unwrap();
        let registration = &bundles[0];
        let caps = Capabilities::new().with_http_token("openai", "sk-test-token".to_string());
        let ctx = noop_context("req-17", caps);
        let _ = match &registration.handler {
            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
            _ => panic!("expected Once handler"),
        };
        let captured = rx.await.unwrap();
        assert_eq!(
            captured.headers.get("authorization").unwrap(),
            "Bearer sk-test-token"
        );
    }

    #[tokio::test]
    async fn import_returns_empty_vec_for_paths_with_no_http_methods() {
        let doc = r#"{
            "openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/x":{"summary":"no methods here"}}
        }"#;
        let spec = OpenAPISpec::from_json(doc).unwrap();
        let bundles = adapter(spec, config("svc", "https://x", None))
            .import()
            .await
            .unwrap();
        assert!(bundles.is_empty());
    }

    #[tokio::test]
    async fn text_response_returned_as_string() {
        let base = spawn_echo_server(200, "hello world", "text/plain").await;
        let spec = OpenAPISpec::from_json(
            r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/t":{"get":{"operationId":"t","responses":{"200":{"content":{"text/plain":{"schema":{}}}}}}}}}"#,
        )
        .unwrap();
        let bundles = adapter(spec, config("svc", &base, None))
            .import()
            .await
            .unwrap();
        let registration = &bundles[0];
        let ctx = noop_context("req-18", Capabilities::new());
        let response = match &registration.handler {
            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
            _ => panic!("expected Once handler"),
        };
        match response.result {
            Ok(Value::String(s)) => assert_eq!(s, "hello world"),
            other => panic!("expected String, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn undeclared_error_status_returns_http_status_code() {
        let base = spawn_echo_server(500, "boom", "text/plain").await;
        let spec = OpenAPISpec::from_json(
            r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
            "paths":{"/x":{"get":{"operationId":"x","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}}"#,
        )
        .unwrap();
        let bundles = adapter(spec, config("svc", &base, None))
            .import()
            .await
            .unwrap();
        let registration = &bundles[0];
        let ctx = noop_context("req-19", Capabilities::new());
        let response = match &registration.handler {
            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
            _ => panic!("expected Once handler"),
        };
        match response.result {
            Err(e) => assert_eq!(e.code, "HTTP_500"),
            other => panic!("expected HTTP_500, got {other:?}"),
        }
    }

    fn minimal_spec_yaml() -> &'static str {
        r#"
openapi: 3.0.0
info:
  title: Test
  version: 1.0.0
paths:
  /widgets:
    get:
      operationId: listWidgets
      responses:
        "200":
          content:
            application/json:
              schema:
                type: array
                items:
                  type: string
"#
    }

    #[tokio::test]
    async fn from_yaml_parses_minimal_doc_yields_one_registration() {
        let spec = OpenAPISpec::from_yaml(minimal_spec_yaml()).unwrap();
        let adapter = adapter(spec, config("widgets", "https://api.example.com", None));
        let bundles = adapter.import().await.unwrap();
        assert_eq!(bundles.len(), 1);
        assert_eq!(bundles[0].spec.name, "widgets/listWidgets");
        assert_eq!(bundles[0].spec.namespace, "widgets");
        assert_eq!(bundles[0].spec.op_type, OperationType::Query);
        assert_eq!(bundles[0].spec.visibility, Visibility::Internal);
        assert_eq!(bundles[0].provenance, OperationProvenance::FromOpenAPI);
        assert!(bundles[0].composition_authority.is_none());
        assert!(bundles[0].scoped_env.is_none());
    }

    #[tokio::test]
    async fn from_yaml_resolves_refs_in_input_schema() {
        let doc = r##"
openapi: 3.0.0
info:
  title: T
  version: "1"
components:
  schemas:
    Widget:
      type: object
      properties:
        name:
          type: string
paths:
  /w:
    post:
      operationId: w
      requestBody:
        content:
          application/json:
            schema:
              $ref: "#/components/schemas/Widget"
      responses:
        "200":
          content:
            application/json:
              schema: {}
"##;
        let spec = OpenAPISpec::from_yaml(doc).unwrap();
        let bundles = adapter(spec, config("ns", "https://x", None))
            .import()
            .await
            .unwrap();
        let props = bundles[0]
            .spec
            .input_schema
            .get("properties")
            .unwrap()
            .as_object()
            .unwrap();
        let body = props.get("body").unwrap();
        assert_eq!(body.get("type").unwrap(), "object");
        assert!(body
            .get("properties")
            .unwrap()
            .as_object()
            .unwrap()
            .contains_key("name"));
    }

    #[tokio::test]
    async fn from_str_json_doc_succeeds_via_json_path() {
        let spec = OpenAPISpec::from_str(minimal_spec_json()).unwrap();
        assert_eq!(spec.info.title, "Test");
        assert_eq!(spec.info.version, "1.0.0");
        let adapter = adapter(spec, config("widgets", "https://api.example.com", None));
        let bundles = adapter.import().await.unwrap();
        assert_eq!(bundles.len(), 1);
        assert_eq!(bundles[0].spec.name, "widgets/listWidgets");
    }

    #[tokio::test]
    async fn from_str_yaml_doc_succeeds_via_yaml_fallback() {
        let spec = OpenAPISpec::from_str(minimal_spec_yaml()).unwrap();
        assert_eq!(spec.info.title, "Test");
        assert_eq!(spec.info.version, "1.0.0");
        let adapter = adapter(spec, config("widgets", "https://api.example.com", None));
        let bundles = adapter.import().await.unwrap();
        assert_eq!(bundles.len(), 1);
        assert_eq!(bundles[0].spec.name, "widgets/listWidgets");
    }

    #[tokio::test]
    async fn from_str_json_doc_preserves_yes_string_against_yaml_coercion() {
        // The JSON-first correctness guard from ADR-051 §2: a JSON doc with a
        // string field whose value is "yes" must survive `from_str` with the
        // string intact. JSON parses first; the YAML path never runs on a
        // valid JSON doc.
        let doc_with_default = r#"{
            "openapi": "3.0.0",
            "info": { "title": "T", "version": "1" },
            "paths": {
                "/x": {
                    "get": {
                        "operationId": "x",
                        "parameters": [
                            {"name": "active", "in": "query", "schema": {"type": "string", "default": "yes"}}
                        ],
                        "responses": {
                            "200": {"content": {"application/json": {"schema": {}}}}
                        }
                    }
                }
            }
        }"#;
        let spec = OpenAPISpec::from_str(doc_with_default).unwrap();
        let adapter = adapter(spec, config("ns", "https://x", None));
        let bundles = adapter.import().await.unwrap();
        let active = &bundles[0].spec.input_schema["properties"]["active"];
        assert_eq!(active["type"], "string");
        assert_eq!(active["default"], Value::String("yes".to_string()));
        assert_ne!(active["default"], Value::Bool(true));
    }

    #[tokio::test]
    async fn from_yaml_preserves_bare_yes_as_string_yaml_1_2_behavior() {
        // Documents the behavior of `yaml_serde` 0.10.x (ADR-051 §3): it
        // implements the YAML 1.2 core schema, NOT YAML 1.1. Under YAML 1.2,
        // only `true`/`false` (and case variants) are booleans — bare
        // `yes`/`no`/`on`/`off`/`y`/`n` are plain strings.
        let doc = r#"
openapi: 3.0.0
info:
  title: T
  version: "1"
paths:
  /x:
    get:
      operationId: x
      parameters:
        - name: active
          in: query
          schema:
            type: string
            default: yes
      responses:
        "200":
          content:
            application/json:
              schema: {}
"#;
        let spec = OpenAPISpec::from_yaml(doc).unwrap();
        let adapter = adapter(spec, config("ns", "https://x", None));
        let bundles = adapter.import().await.unwrap();
        let active = &bundles[0].spec.input_schema["properties"]["active"];
        assert_eq!(active["default"], Value::String("yes".to_string()));
        assert_ne!(active["default"], Value::Bool(true));
    }

    #[tokio::test]
    async fn from_yaml_malformed_returns_schema_parse() {
        let result = OpenAPISpec::from_yaml("openapi: 3.0.0\ninfo: [unclosed");
        match result {
            Err(AdapterError::SchemaParse { message }) => {
                assert!(message.contains("YAML"), "message was: {message}");
            }
            other => panic!("expected SchemaParse, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn from_str_malformed_both_returns_schema_parse() {
        // Neither valid JSON (unterminated string) nor valid YAML (an
        // unterminated flow mapping that both parsers reject).
        let malformed = "openapi: 3.0.0\ninfo: {title: \"T\npaths: [";
        let result = OpenAPISpec::from_str(malformed);
        assert!(matches!(result, Err(AdapterError::SchemaParse { .. })));
    }

    #[tokio::test]
    async fn from_yaml_missing_paths_returns_schema_parse() {
        let doc = r#"
openapi: 3.0.0
info:
  title: x
  version: "1"
"#;
        let result = OpenAPISpec::from_yaml(doc);
        match result {
            Err(AdapterError::SchemaParse { message }) => assert!(message.contains("paths")),
            other => panic!("expected SchemaParse, got {other:?}"),
        }
    }
}