//! The 6 fixed gateway endpoints (`/search`, `/schema`, `/call`,
//! `/batch`, `/subscribe`, `/publish`) — the sole HTTP invoke path
//! (ADR-042, ADR-047; `/publish` per ADR-068).
//!
//! Each endpoint delegates to `GatewayDispatch` (the shared dispatch
//! spine); auth is the shared `bearer_auth_middleware`; error mapping is
//! `gateway::error`. There is no per-operation `POST /{service}/{op}`
//! direct-call surface (ADR-047). `/publish` lives in this module too
//! (the review-001 "separate module" note is stale).
//!
//! The whole router carries an explicit request-body limit (2 MiB plus
//! 64 KiB framing headroom — GW-15): a raw-`Body` handler such as
//! `/publish`'s never consults axum's `DefaultBodyLimit` (that limit is
//! an extension extractors read), so the gateway installs its own cap
//! instead.
//!
//! Module status mapping note (GW-16 tracks the drift): the
//! hand-rolled pre-dispatch rejections below use 400/`INVALID_INPUT`
//! while mid-stream chunk errors map 422 through `gateway::error`;
//! normalizing them is GW-16, not GW-15.

use std::collections::VecDeque;
use std::convert::Infallible;
use std::sync::Arc;
use std::time::Duration;

use alkcall::core::auth::{Identity, IdentityProvider};
use alkcall::protocol::wire::{CallError, ResponseEnvelope};
use alkcall::registry::registration::OperationRegistry;
use alkcall::registry::spec::{AccessResult, Visibility};
use axum::body::Bytes;
use axum::extract::{FromRef, Query, State};
use axum::http::header::{CACHE_CONTROL, VARY};
use axum::http::{HeaderValue, StatusCode};
use axum::response::sse::{Event, KeepAlive};
use axum::response::{IntoResponse, Json, Response, Sse};
use axum::routing::{get, post};
use axum::Router;
use futures::stream::{self, BoxStream};
use futures::StreamExt;
use jsonschema::Validator;
use serde::Deserialize;
use serde_json::{json, Value};

use super::dispatch::GatewayDispatch;
use super::error::call_error_to_http_response_with_identity;
use super::schema_cache::{CompileFailed, PublishSchemaCache};
use super::MAX_BATCH_OPERATIONS;
use crate::server::auth::ResolvedIdentity;
use crate::server::state::RouterState;

const SERVICES_LIST: &str = "services/list";
const SERVICES_SCHEMA: &str = "services/schema";
const MAX_PUBLISH_LINE_BYTES: usize = 2 * 1024 * 1024;

/// The explicit request-body limit for the whole gateway router
/// (GW-15): the 2 MiB convention plus a 64 KiB framing headroom, so a
/// request can carry an at-cap line (2 MiB) plus its NDJSON first-line
/// header framing without the layer pre-empting the route's own
/// per-line cap error — the semantic 400/`INVALID_INPUT` the cap
/// exists to emit. axum's `DefaultBodyLimit` is a request extension
/// consulted by `FromRequest` extractors; the raw-`Body` hand-rolled
/// `/publish` path never consults it, so without this layer that route
/// has no whole-body cap at all. The layer pre-rejects an oversized
/// declared `Content-Length` and wraps the body in a counting stream
/// that errors when chunked (undeclared-length) uploads exceed the
/// limit; the middleware answers `413 Payload Too Large` in both
/// cases. This is a backstop behind the per-line cap, not a second
/// semantic: the per-line cap fires first on any single over-cap
/// line.
const GATEWAY_BODY_LIMIT: usize = MAX_PUBLISH_LINE_BYTES + 64 * 1024;

const GATEWAY_BODY_LIMIT_EXCEEDED: &str = "gateway request body limit exceeded";

/// SSE keep-alive interval on `/subscribe` (GW-13). Shared comment
/// frames (axum's KeepAlive::default) plus a `retry:` field on the
/// stream's first event reconnect the client on drops; 15 s sits under
/// the common LB/proxy idle timeouts (30-60 s).
const SSE_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(15);

type ByteStream = futures::stream::BoxStream<'static, Result<Bytes, axum::Error>>;

#[derive(Clone)]
pub(crate) struct GatewayState {
    registry: Arc<OperationRegistry>,
    identity_provider: Arc<dyn IdentityProvider>,
    publish_schemas: crate::gateway::schema_cache::PublishSchemaCache,
}

impl GatewayState {
    pub(crate) fn new(
        registry: Arc<OperationRegistry>,
        identity_provider: Arc<dyn IdentityProvider>,
        publish_schemas: crate::gateway::schema_cache::PublishSchemaCache,
    ) -> Self {
        Self {
            registry,
            identity_provider,
            publish_schemas,
        }
    }

    fn dispatch(&self) -> GatewayDispatch {
        GatewayDispatch::new(
            Arc::clone(&self.registry),
            Arc::clone(&self.identity_provider),
        )
    }
}

impl FromRef<RouterState> for GatewayState {
    fn from_ref(state: &RouterState) -> Self {
        GatewayState::new(
            Arc::clone(&state.registry),
            Arc::clone(&state.identity_provider),
            state.publish_schemas.clone(),
        )
    }
}

pub(crate) fn gateway_router() -> Router<RouterState> {
    Router::new()
        .route("/search", get(search_handler))
        .route("/schema", get(schema_handler))
        .route("/call", post(call_handler))
        .route("/batch", post(batch_handler))
        .route("/subscribe", post(subscribe_handler))
        .route("/publish", post(publish_handler))
        .layer(axum::middleware::from_fn(gateway_body_limit))
}

/// The `/call` and `/subscribe` request body: the target operation
/// (with or without a leading `/`) and its input object.
#[derive(Debug, Deserialize)]
pub struct CallRequest {
    /// The operation name, qualified (`namespace/op`); a leading slash
    /// is tolerated.
    pub operation: String,
    /// The operation input; defaults to null when the line omits it.
    #[serde(default = "Value::default")]
    pub input: Value,
}

/// The `GET /schema?name=<operation>` query parameters.
#[derive(Debug, Deserialize)]
pub struct SchemaQuery {
    /// The operation name to project, qualified (`namespace/op`).
    pub name: String,
}

pub(crate) async fn call_handler(
    State(state): State<GatewayState>,
    ResolvedIdentity(identity): ResolvedIdentity,
    Json(request): Json<CallRequest>,
) -> Response {
    if is_internal_op(&state.registry, &request.operation) {
        return not_found_response(&request.operation);
    }
    let dispatch = state.dispatch();
    let envelope = dispatch
        .invoke(identity.clone(), &request.operation, request.input)
        .await;
    envelope_to_response(envelope, identity.as_ref())
}

pub(crate) async fn search_handler(
    State(state): State<GatewayState>,
    ResolvedIdentity(identity): ResolvedIdentity,
) -> Response {
    let dispatch = state.dispatch();
    let envelope = dispatch
        .invoke(identity.clone(), SERVICES_LIST, json!({}))
        .await;
    discovery_get_response(envelope, identity.as_ref())
}

pub(crate) async fn schema_handler(
    State(state): State<GatewayState>,
    ResolvedIdentity(identity): ResolvedIdentity,
    Query(query): Query<SchemaQuery>,
) -> Response {
    if is_internal_op(&state.registry, &query.name) {
        return with_no_cache_headers(not_found_response(&query.name));
    }
    if let Some(forbidden) = access_check_for_op(&state.registry, &query.name, identity.as_ref()) {
        return with_no_cache_headers(forbidden_response(forbidden, identity.as_ref()));
    }
    let dispatch = state.dispatch();
    let envelope = dispatch
        .invoke(
            identity.clone(),
            SERVICES_SCHEMA,
            json!({ "name": query.name }),
        )
        .await;
    discovery_get_response(envelope, identity.as_ref())
}

pub(crate) async fn batch_handler(
    State(state): State<GatewayState>,
    ResolvedIdentity(identity): ResolvedIdentity,
    Json(requests): Json<Vec<CallRequest>>,
) -> Response {
    if requests.len() > MAX_BATCH_OPERATIONS {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({
                "code": "INVALID_INPUT",
                "message": format!(
                    "batch exceeds the maximum of {MAX_BATCH_OPERATIONS} operations"
                ),
            })),
        )
            .into_response();
    }
    let dispatch = state.dispatch();
    let mut results: Vec<Value> = Vec::with_capacity(requests.len());
    for request in requests {
        if is_internal_op(&state.registry, &request.operation) {
            results.push(not_found_envelope_json(&request.operation));
            continue;
        }
        let envelope = dispatch
            .invoke(identity.clone(), &request.operation, request.input)
            .await;
        results.push(envelope_to_json(envelope));
    }
    Json(json!({ "results": results })).into_response()
}

pub(crate) async fn subscribe_handler(
    State(state): State<GatewayState>,
    ResolvedIdentity(identity): ResolvedIdentity,
    Json(request): Json<CallRequest>,
) -> Response {
    let stream = if is_internal_op(&state.registry, &request.operation) {
        subscribe_stream_internal_error(request.operation)
    } else {
        let dispatch = state.dispatch();
        let envelope_stream =
            dispatch.invoke_streaming(identity, &request.operation, request.input);
        subscribe_stream_from_envelope_stream(envelope_stream)
    };
    Sse::new(stream)
        .keep_alive(
            KeepAlive::new()
                .interval(SSE_KEEP_ALIVE_INTERVAL)
                .event(keep_alive_event()),
        )
        .into_response()
}

/// The SSE projection stream of `POST /subscribe`: each item is one
/// wire-ready frame (a `data:` event or a keep-alive comment); the
/// stream is infallible — errors arrive as in-band `event: error`
/// frames (GW-04).
pub type SubscribeStream = BoxStream<'static, Result<Event, Infallible>>;

/// `POST /publish` (ADR-068): the body is NDJSON — one published chunk
/// per line. OQ-02 resolution: the first line carries
/// `{ "operation": "/{service}/{op}", "chunk": {...} }` (subsequent
/// lines are chunk values only); a terminal error is a plain HTTP
/// status + JSON body (not an NDJSON line). The body is streamed (never
/// fully buffered): the first non-blank line names the operation and
/// carries the first chunk, then the remainder streams into the sink —
/// a client disconnect drops the body stream and the sink (and the
/// handler's `PublishStream`) sees EOF, matching call-protocol
/// write-half close semantics.
///
/// Chunk validation (GW-01, ADR-046 §4): each chunk is validated
/// against the op's `publish_schema` via the compile-once
/// [`PublishSchemaCache`]. A schema that **fails to compile** is
/// fail-closed, not fail-open (review-001 post-remediation follow-up):
/// the chunk stream terminates with `INTERNAL` before any chunk reaches
/// the handler — the unvalidated-ingest hole the per-request
/// `warn`-and-skip opened is shut. The compile error itself is logged
/// at error level and never echoed on the wire.
pub(crate) async fn publish_handler(
    State(state): State<GatewayState>,
    ResolvedIdentity(identity): ResolvedIdentity,
    body: axum::body::Body,
) -> Response {
    let mut header_lines = BufferedLines::new(Box::pin(body.into_data_stream()));
    let (operation, first_chunk) = {
        let line = match header_lines.next_line().await {
            Ok(Some(line)) => line,
            Ok(None) => {
                return invalid_input_response("empty publish body: expected NDJSON chunks")
            }
            Err(e) => return invalid_input_response(e.message()),
        };
        let value = match serde_json::from_slice::<Value>(&line) {
            Ok(v) => v,
            Err(e) => {
                return invalid_input_response(&format!(
                    "first publish line is not valid JSON: {e}"
                ))
            }
        };
        let op = value
            .get("operation")
            .and_then(|o| o.as_str())
            .map(str::to_string);
        let Some(op) = op else {
            return missing_header_field_response();
        };
        let Some(chunk) = value.get("chunk") else {
            return missing_header_field_response();
        };
        (op, chunk.clone())
    };

    let chunks = NdjsonChunkStream::new(
        header_lines,
        Arc::clone(&state.registry),
        state.publish_schemas.clone(),
        operation.clone(),
        Some(first_chunk),
    );
    let dispatch = state.dispatch();
    let envelope = dispatch
        .invoke_sink(identity.clone(), &operation, Value::Null, Box::pin(chunks))
        .await;
    envelope_to_response(envelope, identity.as_ref())
}

/// The per-request `publish_schema` situation for a `/publish` call
/// (ADR-046 §4). Resolution is lazy — the chunk stream resolves against
/// the [`PublishSchemaCache`] on its first poll, which runs only after
/// `invoke_sink`'s pre-checks have admitted the operation (an unknown,
/// internal, or wrong-type op is rejected before any schema resolution
/// or compile happens, preserving the GW-11 dispatch-owned pre-check
/// order and keeping 404/403/422 intact).
enum PublishSchemaState {
    /// Not yet resolved: the stream holds the registry + cache until
    /// the first chunk is polled.
    Unresolved {
        registry: Arc<OperationRegistry>,
        cache: PublishSchemaCache,
        operation: String,
    },
    /// The op registers no `publish_schema`; chunks pass as-is (the
    /// wire path's behavior for schema-less Pub ops).
    Unvalidated,
    /// Chunks are validated against this compiled validator.
    Validated(std::sync::Arc<jsonschema::Validator>),
    /// The op's `publish_schema` failed to compile (GW-01 follow-up):
    /// fail-closed — the stream yields one terminal `INTERNAL` error
    /// and ends, so no chunk ever reaches the handler unvalidated.
    Failed(CompileFailed),
}

impl PublishSchemaState {
    fn resolve(&mut self) {
        let Self::Unresolved {
            registry,
            cache,
            operation,
        } = self
        else {
            return;
        };
        let operation = std::mem::take(operation);
        *self = match cache.validator(registry, &operation) {
            Ok(None) => Self::Unvalidated,
            Ok(Some(validator)) => Self::Validated(validator),
            Err(compile_failed) => {
                tracing::debug!(
                    operation = %operation,
                    "rejecting /publish: the operation's publish_schema could not be compiled \
                     (fail-closed)"
                );
                Self::Failed(compile_failed)
            }
        };
    }

    /// Validate one chunk; also flips a terminal compile failure into
    /// stream-end after the first error item (an `Err` from this stream
    /// is terminal — the handler sees the error and the stream closes,
    /// mirroring the wire pump's `send(Err)` + `break`).
    fn validate_chunk(
        &mut self,
        value: Value,
    ) -> std::task::Poll<Option<Result<Value, CallError>>> {
        self.resolve();
        match self {
            Self::Unvalidated => std::task::Poll::Ready(Some(Ok(value))),
            Self::Validated(validator) => {
                if Validator::is_valid(validator, &value) {
                    std::task::Poll::Ready(Some(Ok(value)))
                } else {
                    std::task::Poll::Ready(Some(Err(CallError::invalid_input(
                        "published chunk failed publish_schema validation",
                    )
                    .with_details(json!({ "chunk": value })))))
                }
            }
            Self::Failed(failed) => {
                let error = failed.call_error();
                *self = Self::Unvalidated;
                std::task::Poll::Ready(Some(Err(error)))
            }
            Self::Unresolved { .. } => unreachable!("resolve() runs before validation"),
        }
    }
}

fn missing_header_field_response() -> Response {
    invalid_input_response("first publish line must carry {\"operation\": ..., \"chunk\": ...}")
}

fn invalid_input_response(message: &str) -> Response {
    (
        StatusCode::BAD_REQUEST,
        Json(json!({ "code": "INVALID_INPUT", "message": message })),
    )
        .into_response()
}

/// The explicit gateway request-body limit (GW-15). See
/// [`GATEWAY_BODY_LIMIT`] for why this is a layer and not axum's
/// `DefaultBodyLimit` extension.
async fn gateway_body_limit(req: axum::extract::Request, next: axum::middleware::Next) -> Response {
    let (parts, body) = req.into_parts();

    if let Some(len) = parts
        .headers
        .get(axum::http::header::CONTENT_LENGTH)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.parse::<usize>().ok())
    {
        if len > GATEWAY_BODY_LIMIT {
            return gateway_body_limit_exceeded();
        }
    }

    let exceeded = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let limited_req = axum::extract::Request::from_parts(
        parts,
        axum::body::Body::from_stream(LimitedBody {
            inner: body.into_data_stream(),
            remaining: GATEWAY_BODY_LIMIT,
            exceeded: Arc::clone(&exceeded),
        }),
    );

    let response = next.run(limited_req).await;

    if exceeded.load(std::sync::atomic::Ordering::Relaxed) {
        return gateway_body_limit_exceeded();
    }
    response
}

fn gateway_body_limit_exceeded() -> Response {
    (
        StatusCode::PAYLOAD_TOO_LARGE,
        "gateway request body limit exceeded",
    )
        .into_response()
}

struct LimitedBody {
    inner: axum::body::BodyDataStream,
    remaining: usize,
    exceeded: Arc<std::sync::atomic::AtomicBool>,
}

impl futures::Stream for LimitedBody {
    type Item = Result<Bytes, std::io::Error>;

    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        let this = &mut *self;
        match std::pin::Pin::new(&mut this.inner).poll_next(cx) {
            std::task::Poll::Ready(Some(Ok(data))) => {
                let len = data.len();
                if len > this.remaining {
                    this.remaining = 0;
                    this.exceeded
                        .store(true, std::sync::atomic::Ordering::Relaxed);
                    return std::task::Poll::Ready(Some(Err(std::io::Error::other(
                        GATEWAY_BODY_LIMIT_EXCEEDED,
                    ))));
                }
                this.remaining -= len;
                std::task::Poll::Ready(Some(Ok(data)))
            }
            std::task::Poll::Ready(Some(Err(_))) => std::task::Poll::Ready(Some(Err(
                std::io::Error::other(GATEWAY_BODY_LIMIT_EXCEEDED),
            ))),
            std::task::Poll::Pending => std::task::Poll::Pending,
            std::task::Poll::Ready(None) => std::task::Poll::Ready(None),
        }
    }
}

/// A newline-framed reader over the request body's byte stream. Blank
/// lines are skipped; a trailing unterminated line is yielded as a
/// final line. Each chunk read extends the buffer, drains every
/// complete `\n`-terminated line into the pending queue (capped at
/// `MAX_PUBLISH_LINE_BYTES`), then caps the unterminated tail **even
/// when no `\n` was seen** (GW-15): the tail can therefore never grow
/// past the cap across chunks, and the trailing-EOF `mem::take` path
/// re-checks the cap before yielding. A cap breach or a byte-level
/// read failure is terminal and aborts the whole reader (pending
/// lines included — a breach poisons the body framing).
struct BufferedLines {
    bytes: ByteStream,
    buffer: Vec<u8>,
    pending: VecDeque<Vec<u8>>,
    done: bool,
}

impl BufferedLines {
    fn new(bytes: ByteStream) -> Self {
        Self {
            bytes,
            buffer: Vec::new(),
            pending: VecDeque::new(),
            done: false,
        }
    }

    fn abort_on_cap(&mut self) -> LineError {
        self.done = true;
        self.buffer.clear();
        self.pending.clear();
        LineError::LineCap
    }

    async fn next_line(&mut self) -> Result<Option<Vec<u8>>, LineError> {
        loop {
            if let Some(line) = self.pending.pop_front() {
                if line.iter().all(|b| b.is_ascii_whitespace()) {
                    continue;
                }
                return Ok(Some(line));
            }
            if self.done {
                if self.buffer.iter().all(|b| b.is_ascii_whitespace()) {
                    self.buffer.clear();
                    return Ok(None);
                }
                if self.buffer.len() > MAX_PUBLISH_LINE_BYTES {
                    return Err(self.abort_on_cap());
                }
                return Ok(Some(std::mem::take(&mut self.buffer)));
            }
            match self.bytes.next().await {
                Some(Ok(bytes)) => {
                    self.buffer.extend_from_slice(&bytes);
                    while let Some(pos) = self.buffer.iter().position(|b| *b == b'\n') {
                        let line: Vec<u8> = self.buffer.drain(..=pos).collect();
                        let line = &line[..line.len() - 1];
                        if line.len() > MAX_PUBLISH_LINE_BYTES {
                            return Err(self.abort_on_cap());
                        }
                        self.pending.push_back(line.to_vec());
                    }
                    if self.buffer.len() > MAX_PUBLISH_LINE_BYTES {
                        return Err(self.abort_on_cap());
                    }
                }
                Some(Err(_)) => {
                    self.done = true;
                    return Err(LineError::Read);
                }
                None => {
                    self.done = true;
                    continue;
                }
            }
        }
    }
}

enum LineError {
    Read,
    LineCap,
}

impl LineError {
    fn message(&self) -> &'static str {
        match self {
            Self::Read => "publish body read failed",
            Self::LineCap => "publish line exceeds the per-line cap",
        }
    }
}

/// The initiator-side chunk stream fed to `invoke_sink` (GW-01, GW-06):
/// NDJSON lines parsed lazily straight from the request body, each
/// validated against the op's compiled `publish_schema` — matching the
/// wire dispatcher's per-chunk validation (alkcall ADR-046). Malformed
/// JSON, a schema violation, or a body-read failure yields a terminal
/// `Err` item — the same shape an initiator-side `call.error` produces
/// on the wire, and equally terminal: like the wire pump's
/// `send(Err)` + `break`, the first `Err` ends the stream. A schema
/// that failed to compile yields that terminal `Err(INTERNAL)` on the
/// first chunk — fail-closed (review-001 post-remediation follow-up).
struct NdjsonChunkStream {
    lines: std::pin::Pin<Box<BufferedLines>>,
    schema_state: PublishSchemaState,
    pending_first: Option<Value>,
    done: bool,
}

impl NdjsonChunkStream {
    fn new(
        lines: BufferedLines,
        registry: Arc<OperationRegistry>,
        cache: PublishSchemaCache,
        operation: String,
        pending_first: Option<Value>,
    ) -> Self {
        Self {
            lines: Box::pin(lines),
            schema_state: PublishSchemaState::Unresolved {
                registry,
                cache,
                operation,
            },
            pending_first,
            done: false,
        }
    }

    /// `Some(item)` from `validate_chunk` where an `Err` item sets
    /// `done` — the stream does not yield after an error (the wire
    /// pump's semantics; the dropping mpsc sender below is the same
    /// contract).
    fn poll_validated(
        &mut self,
        value: Value,
    ) -> std::task::Poll<Option<Result<Value, CallError>>> {
        match self.schema_state.validate_chunk(value) {
            std::task::Poll::Ready(Some(Err(e))) => {
                self.done = true;
                std::task::Poll::Ready(Some(Err(e)))
            }
            other => other,
        }
    }
}

impl futures::Stream for NdjsonChunkStream {
    type Item = Result<Value, CallError>;

    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        if self.done {
            return std::task::Poll::Ready(None);
        }
        if self.pending_first.is_some() {
            let first = self.pending_first.take().unwrap_or(Value::Null);
            return self.poll_validated(first);
        }
        let line = {
            let mut next = Box::pin(self.lines.next_line());
            let polled = match std::future::Future::poll(next.as_mut(), cx) {
                std::task::Poll::Ready(Ok(Some(line))) => Ok(Some(line)),
                std::task::Poll::Ready(Ok(None)) => Ok(None),
                std::task::Poll::Ready(Err(e)) => Err(e.message()),
                std::task::Poll::Pending => {
                    drop(next);
                    return std::task::Poll::Pending;
                }
            };
            drop(next);
            match polled {
                Ok(Some(line)) => line,
                Ok(None) => return std::task::Poll::Ready(None),
                Err(message) => {
                    self.done = true;
                    return std::task::Poll::Ready(Some(Err(CallError::invalid_input(message))));
                }
            }
        };
        let value = match serde_json::from_slice::<Value>(&line) {
            Ok(v) => v,
            Err(e) => {
                self.done = true;
                return std::task::Poll::Ready(Some(Err(CallError::invalid_input(format!(
                    "publish line is not valid JSON: {e}"
                )))));
            }
        };
        self.poll_validated(value)
    }
}

/// The SSE projection of a streaming envelope (GW-04): each `Ok`
/// envelope becomes a `data:` frame carrying the output JSON plus the
/// `retry:` reconnect hint; the first `Err` envelope becomes an
/// `event:error` frame carrying the serialized `CallError` **and ends
/// the stream** — matching the wire dispatcher's `call.error`-is-
/// terminal semantics and http-server.md's documented contract
/// (an `Err` is terminal; the stream does not continue after it).
/// A quiet-but-alive stream is kept alive by axum comment frames at
/// `SSE_KEEP_ALIVE_INTERVAL` (GW-13).
///
/// The `scan` closure yields the error frame and flags the stream done
/// (subsequent polls return `None`), so the frame that ends the stream
/// is still written — the error event is emitted, not swallowed.
fn subscribe_stream_from_envelope_stream(
    stream: BoxStream<'static, ResponseEnvelope>,
) -> SubscribeStream {
    Box::pin(stream.scan(false, |done, envelope| {
        std::future::ready(if *done {
            None
        } else {
            let item = match envelope.result {
                Ok(output) => {
                    let data =
                        serde_json::to_string(&output).unwrap_or_else(|_| "null".to_string());
                    Event::default().retry(SSE_KEEP_ALIVE_INTERVAL).data(data)
                }
                Err(error) => {
                    *done = true;
                    let payload = serde_json::to_value(&error).unwrap_or(Value::Null);
                    let data =
                        serde_json::to_string(&payload).unwrap_or_else(|_| "null".to_string());
                    Event::default()
                        .event("error")
                        .retry(SSE_KEEP_ALIVE_INTERVAL)
                        .data(data)
                }
            };
            Some(Ok::<_, Infallible>(item))
        })
    }))
}

/// The keep-alive comment frame: an SSE comment (colon-prefixed, no
/// event/data) plus a `retry:` hint, emitted on quiet streams so
/// LB/proxy idle timeouts do not kill the connection.
fn keep_alive_event() -> Event {
    Event::default().retry(SSE_KEEP_ALIVE_INTERVAL)
}

pub(crate) fn subscribe_stream_internal_error(operation: String) -> SubscribeStream {
    Box::pin(stream::once(async move { error_event(&operation) }))
}

fn envelope_to_response(envelope: ResponseEnvelope, identity: Option<&Identity>) -> Response {
    match envelope.result {
        Ok(output) => {
            let body = envelope_to_ok_json(&envelope.request_id, &output);
            (StatusCode::OK, Json(body)).into_response()
        }
        Err(error) => call_error_to_http_response_with_identity(&error, identity),
    }
}

/// The per-identity GET endpoints (`/search`, `/schema`) are
/// AccessControl-filtered per caller and auth-dependent (200 vs 403/404
/// on the same name), so no shared cache may store or reuse the
/// response (GW-02). The header pair applies to every response these
/// routes emit, including the denial (error) paths.
fn discovery_get_response(envelope: ResponseEnvelope, identity: Option<&Identity>) -> Response {
    with_no_cache_headers(envelope_to_response(envelope, identity))
}

fn with_no_cache_headers(mut response: Response) -> Response {
    let headers = response.headers_mut();
    headers.insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
    headers.insert(VARY, HeaderValue::from_static("Authorization"));
    response
}

fn envelope_to_json(envelope: ResponseEnvelope) -> Value {
    match envelope.result {
        Ok(output) => envelope_to_ok_json(&envelope.request_id, &output),
        Err(error) => envelope_to_error_json(&envelope.request_id, &error),
    }
}

fn envelope_to_ok_json(request_id: &str, output: &Value) -> Value {
    json!({
        "request_id": request_id,
        "result": "ok",
        "output": output,
    })
}

fn envelope_to_error_json(request_id: &str, error: &CallError) -> Value {
    json!({
        "request_id": request_id,
        "result": "error",
        "error": serde_json::to_value(error).unwrap_or(Value::Null),
    })
}

fn not_found_envelope_json(operation: &str) -> Value {
    let error = CallError::not_found(operation);
    json!({
        "request_id": uuid::Uuid::new_v4().to_string(),
        "result": "error",
        "error": serde_json::to_value(&error).unwrap_or(Value::Null),
    })
}

fn not_found_response(operation: &str) -> Response {
    let error = CallError::not_found(operation);
    call_error_to_http_response_with_identity(&error, None)
}

fn forbidden_response(message: String, identity: Option<&Identity>) -> Response {
    let error = CallError::forbidden(message);
    call_error_to_http_response_with_identity(&error, identity)
}

fn access_check_for_op(
    registry: &OperationRegistry,
    operation: &str,
    identity: Option<&Identity>,
) -> Option<String> {
    let name = operation.strip_prefix('/').unwrap_or(operation);
    let reg = registry.registration(name)?;
    if let AccessResult::Forbidden(message) = reg.spec.access_control.check(identity, None, None) {
        return Some(message);
    }
    None
}

fn is_internal_op(registry: &OperationRegistry, operation: &str) -> bool {
    let name = operation.strip_prefix('/').unwrap_or(operation);
    match registry.registration(name) {
        Some(reg) => reg.spec.visibility == Visibility::Internal,
        None => false,
    }
}

fn error_event(operation: &str) -> Result<Event, Infallible> {
    let error = CallError::not_found(operation);
    let payload = serde_json::to_value(&error).unwrap_or(Value::Null);
    let data = serde_json::to_string(&payload).unwrap_or_else(|_| "null".to_string());
    Ok(Event::default().event("error").data(data))
}

#[cfg(test)]
mod tests {
    use super::*;
    use alkcall::core::auth::IdentityProvider;
    use alkcall::core::types::Capabilities;
    use alkcall::registry::discovery::{
        services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
    };
    use alkcall::registry::registration::{
        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
    };
    use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType};
    use axum::body::Body;
    use axum::http::Request;
    use axum::middleware::from_fn_with_state;
    use http_body_util::BodyExt;
    use std::collections::HashMap;
    use std::sync::Mutex as StdMutex;
    use tower::ServiceExt;

    struct StaticIdentityProvider {
        tokens: StdMutex<HashMap<String, Identity>>,
    }

    impl StaticIdentityProvider {
        fn new() -> Self {
            Self {
                tokens: StdMutex::new(HashMap::new()),
            }
        }

        fn with_token(self, token: &str, identity: Identity) -> Self {
            self.tokens
                .lock()
                .unwrap()
                .insert(token.to_string(), identity);
            self
        }
    }

    impl IdentityProvider for StaticIdentityProvider {
        fn resolve_from_fingerprint(&self, _fp: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
            let token_str = String::from_utf8_lossy(&token.raw);
            self.tokens.lock().unwrap().get(token_str.as_ref()).cloned()
        }
    }

    fn identity_with_scopes(id: &str, scopes: &[&str]) -> Identity {
        Identity {
            id: id.to_string(),
            scopes: scopes.iter().map(|s| s.to_string()).collect(),
            resources: HashMap::new(),
        }
    }

    fn external_spec(name: &str, acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Query,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            acl,
            None,
        )
    }

    fn internal_spec(name: &str) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Query,
            Visibility::Internal,
            json!({}),
            json!({}),
            vec![],
            AccessControl::default(),
            None,
        )
    }

    fn echo_handler() -> alkcall::registry::registration::Handler {
        make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) })
    }

    fn registry_with_echo() -> Arc<OperationRegistry> {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("echo/run", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    fn registry_with_restricted_op() -> Arc<OperationRegistry> {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec(
                    "admin/run",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    fn registry_with_internal_op() -> Arc<OperationRegistry> {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                internal_spec("secret/op"),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    fn subscription_spec(name: &str, visibility: Visibility, acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Sub,
            visibility,
            json!({}),
            json!({}),
            vec![],
            acl,
            None,
        )
    }

    fn multi_event_streaming_handler(
        outputs: Vec<Value>,
    ) -> alkcall::registry::registration::StreamingHandler {
        make_streaming_handler(move |_input, ctx| {
            let request_id = ctx.request_id.clone();
            let outputs = outputs.clone();
            futures::stream::iter(
                outputs
                    .into_iter()
                    .map(move |o| ResponseEnvelope::ok(request_id.clone(), o)),
            )
        })
    }

    fn error_streaming_handler(error: CallError) -> HandlerKind {
        HandlerKind::Stream(make_streaming_handler(move |_input, ctx| {
            let request_id = ctx.request_id.clone();
            let error = error.clone();
            futures::stream::iter(vec![ResponseEnvelope::error(request_id, error)])
        }))
    }

    fn registry_with_subscription_stream(
        name: &str,
        outputs: Vec<Value>,
    ) -> Arc<OperationRegistry> {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                subscription_spec(name, Visibility::External, AccessControl::default()),
                HandlerKind::Stream(multi_event_streaming_handler(outputs)),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    fn registry_with_subscription_error(name: &str, error: CallError) -> Arc<OperationRegistry> {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                subscription_spec(name, Visibility::External, AccessControl::default()),
                error_streaming_handler(error),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    fn registry_with_discovery_and_ops(
        inner_ops: Vec<HandlerRegistration>,
    ) -> Arc<OperationRegistry> {
        let mut inner = OperationRegistry::new();
        for op in inner_ops {
            inner.register(op).unwrap();
        }
        let inner = Arc::new(inner);
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                services_list_spec(),
                HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        registry
            .register(HandlerRegistration::new(
                services_schema_spec(),
                HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        for spec in inner.list_operations() {
            let name = spec.name.clone();
            let reg = inner.registration(&name).unwrap();
            registry
                .register(HandlerRegistration::new(
                    reg.spec.clone(),
                    reg.handler.clone(),
                    reg.provenance,
                    reg.composition_authority.clone(),
                    reg.scoped_env.clone(),
                    reg.capabilities.clone(),
                ))
                .unwrap();
        }
        Arc::new(registry)
    }

    fn unused_provider() -> Arc<dyn IdentityProvider> {
        Arc::new(StaticIdentityProvider::new())
    }

    fn build_router(
        registry: Arc<OperationRegistry>,
        provider: Arc<dyn IdentityProvider>,
    ) -> axum::Router {
        let state = RouterState {
            registry: Arc::clone(&registry),
            identity_provider: Arc::clone(&provider),
            decoy: crate::server::DecoyConfig::NotFound,
            openapi_doc: crate::server::adapter::CachedOpenAPIDoc::new(&registry),
            ws_sessions: Arc::new(crate::websocket::WsSessions::new()),
            ws_session_slots: Arc::new(tokio::sync::Semaphore::new(
                crate::websocket::DEFAULT_WS_MAX_SESSIONS,
            )),
            ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
            publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(),
        };
        let auth_state = Arc::clone(&provider);
        gateway_router()
            .route_layer(from_fn_with_state(
                auth_state,
                crate::server::auth::bearer_auth_middleware,
            ))
            .with_state(state)
    }

    // The gateway deadline is the module's real 30 s constant (GW-05
    // asserts enforcement in dispatch.rs); these SSE tests only touch
    // streaming ops, which are exempt from it.

    fn auth_header(token: &str) -> (&'static str, String) {
        ("authorization", format!("Bearer {token}"))
    }

    async fn send(router: axum::Router, req: Request<Body>) -> (StatusCode, Value) {
        let resp = router.oneshot(req).await.unwrap();
        let status = resp.status();
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let body: Value = if bytes.is_empty() {
            Value::Null
        } else {
            serde_json::from_slice(&bytes).unwrap_or(Value::Null)
        };
        (status, body)
    }

    fn json_request(method: &str, uri: &str, body: Value) -> Request<Body> {
        Request::builder()
            .method(method)
            .uri(uri)
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap()
    }

    #[tokio::test]
    async fn call_round_trip_external_op_returns_200_with_json_body() {
        let router = build_router(registry_with_echo(), unused_provider());
        let req = json_request(
            "POST",
            "/call",
            json!({ "operation": "echo/run", "input": { "msg": "hi" } }),
        );
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body.get("result"), Some(&json!("ok")));
        assert_eq!(body.get("output"), Some(&json!({ "msg": "hi" })));
    }

    #[tokio::test]
    async fn call_internal_op_returns_404() {
        let router = build_router(registry_with_internal_op(), unused_provider());
        let req = json_request(
            "POST",
            "/call",
            json!({ "operation": "secret/op", "input": {} }),
        );
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::NOT_FOUND);
        assert_eq!(body.get("code"), Some(&json!("NOT_FOUND")));
    }

    #[tokio::test]
    async fn call_unauthorized_restricted_op_returns_403() {
        let provider: Arc<dyn IdentityProvider> = Arc::new(
            StaticIdentityProvider::new()
                .with_token("user-tok", identity_with_scopes("user", &["user"])),
        );
        let router = build_router(registry_with_restricted_op(), provider);
        let (k, v) = auth_header("user-tok");
        let req = Request::builder()
            .method("POST")
            .uri("/call")
            .header("content-type", "application/json")
            .header(k, v)
            .body(Body::from(
                serde_json::to_vec(&json!({ "operation": "admin/run", "input": {} })).unwrap(),
            ))
            .unwrap();
        let (status, _body) = send(router, req).await;
        assert_eq!(status, StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn call_unauthenticated_restricted_op_returns_401() {
        let router = build_router(registry_with_restricted_op(), unused_provider());
        let req = json_request(
            "POST",
            "/call",
            json!({ "operation": "admin/run", "input": {} }),
        );
        let (status, _body) = send(router, req).await;
        assert_eq!(status, StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn search_returns_only_access_control_allowed_ops() {
        let ops = vec![
            HandlerRegistration::new(
                external_spec("public/echo", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ),
            HandlerRegistration::new(
                external_spec(
                    "admin/secret",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ),
        ];
        let discovery = registry_with_discovery_and_ops(ops);
        let provider: Arc<dyn IdentityProvider> = Arc::new(
            StaticIdentityProvider::new()
                .with_token("user-tok", identity_with_scopes("regular", &["user"])),
        );
        let router = build_router(discovery, provider);
        let (k, v) = auth_header("user-tok");
        let req = Request::builder()
            .method("GET")
            .uri("/search")
            .header(k, v)
            .body(Body::empty())
            .unwrap();
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        let ops = body
            .get("output")
            .and_then(|o| o.get("operations"))
            .and_then(|o| o.as_array())
            .expect("operations array");
        let names: Vec<&str> = ops
            .iter()
            .filter_map(|o| o.get("name").and_then(|n| n.as_str()))
            .collect();
        assert!(names.contains(&"public/echo"));
        assert!(!names.contains(&"admin/secret"));
    }

    #[tokio::test]
    async fn schema_returns_full_spec_for_authorized_op() {
        let ops = vec![HandlerRegistration::new(
            external_spec("echo/run", AccessControl::default()),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        )];
        let discovery = registry_with_discovery_and_ops(ops);
        let router = build_router(discovery, unused_provider());
        let req = Request::builder()
            .method("GET")
            .uri("/schema?name=echo%2Frun")
            .body(Body::empty())
            .unwrap();
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        let output = body.get("output").expect("output");
        assert_eq!(output.get("name"), Some(&json!("echo/run")));
        assert_eq!(output.get("namespace"), Some(&json!("echo")));
        assert!(output.get("input_schema").is_some());
        assert!(output.get("output_schema").is_some());
    }

    #[tokio::test]
    async fn schema_for_unauthorized_op_returns_403() {
        let ops = vec![HandlerRegistration::new(
            external_spec(
                "admin/secret",
                AccessControl {
                    required_scopes: vec!["admin".to_string()],
                    ..Default::default()
                },
            ),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        )];
        let discovery = registry_with_discovery_and_ops(ops);
        let provider: Arc<dyn IdentityProvider> = Arc::new(
            StaticIdentityProvider::new()
                .with_token("user-tok", identity_with_scopes("regular", &["user"])),
        );
        let router = build_router(discovery, provider);
        let (k, v) = auth_header("user-tok");
        let req = Request::builder()
            .method("GET")
            .uri("/schema?name=admin%2Fsecret")
            .header(k, v)
            .body(Body::empty())
            .unwrap();
        let (status, _body) = send(router, req).await;
        assert_eq!(status, StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn schema_unknown_op_returns_404() {
        let ops = vec![HandlerRegistration::new(
            external_spec("echo/run", AccessControl::default()),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        )];
        let discovery = registry_with_discovery_and_ops(ops);
        let router = build_router(discovery, unused_provider());
        let req = Request::builder()
            .method("GET")
            .uri("/schema?name=no%2Fsuch")
            .body(Body::empty())
            .unwrap();
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::NOT_FOUND);
        assert_eq!(body.get("code"), Some(&json!("NOT_FOUND")));
    }

    #[tokio::test]
    async fn schema_internal_op_returns_404_unauthenticated() {
        let router = build_router(registry_with_internal_op(), unused_provider());
        let req = Request::builder()
            .method("GET")
            .uri("/schema?name=secret%2Fop")
            .body(Body::empty())
            .unwrap();
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::NOT_FOUND);
        assert_eq!(body.get("code"), Some(&json!("NOT_FOUND")));
    }

    #[tokio::test]
    async fn schema_internal_op_returns_404_for_unauthorized_identity() {
        let provider: Arc<dyn IdentityProvider> = Arc::new(
            StaticIdentityProvider::new()
                .with_token("user-tok", identity_with_scopes("user", &["user"])),
        );
        let router = build_router(registry_with_internal_op(), provider);
        let (k, v) = auth_header("user-tok");
        let req = Request::builder()
            .method("GET")
            .uri("/schema?name=secret%2Fop")
            .header(k, v)
            .body(Body::empty())
            .unwrap();
        let (status, _body) = send(router, req).await;
        assert_eq!(status, StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn schema_internal_op_returns_404_for_anonymous_identity() {
        let router = build_router(registry_with_internal_op(), unused_provider());
        let (k, v) = auth_header("unknown-tok");
        let req = Request::builder()
            .method("GET")
            .uri("/schema?name=secret%2Fop")
            .header(k, v)
            .body(Body::empty())
            .unwrap();
        let (status, _body) = send(router, req).await;
        assert_eq!(status, StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn batch_returns_array_of_results_in_order() {
        let router = build_router(registry_with_echo(), unused_provider());
        let req = json_request(
            "POST",
            "/batch",
            json!([
                { "operation": "echo/run", "input": { "n": 1 } },
                { "operation": "echo/run", "input": { "n": 2 } },
            ]),
        );
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        let results = body
            .get("results")
            .and_then(|r| r.as_array())
            .expect("results array");
        assert_eq!(results.len(), 2);
        assert_eq!(results[0].get("output"), Some(&json!({ "n": 1 })));
        assert_eq!(results[1].get("output"), Some(&json!({ "n": 2 })));
    }

    #[tokio::test]
    async fn batch_internal_op_returns_not_found_in_array() {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                internal_spec("secret/op"),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        registry
            .register(HandlerRegistration::new(
                external_spec("echo/run", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let router = build_router(Arc::new(registry), unused_provider());
        let req = json_request(
            "POST",
            "/batch",
            json!([
                { "operation": "echo/run", "input": {} },
                { "operation": "secret/op", "input": {} },
            ]),
        );
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        let results = body
            .get("results")
            .and_then(|r| r.as_array())
            .expect("results array");
        assert_eq!(results.len(), 2);
        assert_eq!(results[0].get("result"), Some(&json!("ok")));
        assert_eq!(results[1].get("result"), Some(&json!("error")));
        assert_eq!(
            results[1].get("error").and_then(|e| e.get("code")),
            Some(&json!("NOT_FOUND"))
        );
        assert!(
            results[1]
                .get("request_id")
                .map(|id| !id.is_null())
                .unwrap_or(false),
            "internal-op entries must carry a generated request_id, not null"
        );
    }
    #[tokio::test]
    async fn batch_exceeding_operation_cap_returns_400_invalid_input() {
        let router = build_router(registry_with_echo(), unused_provider());
        let requests: Vec<Value> = (0..MAX_BATCH_OPERATIONS + 1)
            .map(|i| json!({ "operation": "echo/run", "input": { "n": i } }))
            .collect();
        let req = json_request("POST", "/batch", json!(requests));
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::BAD_REQUEST);
        assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
    }

    #[tokio::test]
    async fn batch_at_cap_dispatches_all_entries() {
        let router = build_router(registry_with_echo(), unused_provider());
        let requests: Vec<Value> = (0..MAX_BATCH_OPERATIONS)
            .map(|i| json!({ "operation": "echo/run", "input": { "n": i } }))
            .collect();
        let req = json_request("POST", "/batch", json!(requests));
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        let results = body
            .get("results")
            .and_then(|r| r.as_array())
            .expect("results array");
        assert_eq!(results.len(), MAX_BATCH_OPERATIONS);
    }

    #[tokio::test]
    async fn subscribe_on_subscription_streams_multiple_data_frames() {
        let router = build_router(
            registry_with_subscription_stream(
                "events/stream",
                vec![json!({ "n": 1 }), json!({ "n": 2 }), json!({ "n": 3 })],
            ),
            unused_provider(),
        );
        let req = json_request(
            "POST",
            "/subscribe",
            json!({ "operation": "events/stream", "input": {} }),
        );
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let ctype = resp
            .headers()
            .get(axum::http::header::CONTENT_TYPE)
            .map(|v| v.to_str().unwrap().to_string());
        assert!(
            ctype
                .as_deref()
                .unwrap_or("")
                .starts_with("text/event-stream"),
            "expected text/event-stream, got {ctype:?}"
        );
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let body = String::from_utf8_lossy(&bytes);
        let data_frames = body.matches("data:").count();
        assert_eq!(data_frames, 3, "expected 3 data frames, got: {body}");
        assert!(body.contains("\"n\":1"), "expected n=1, got: {body}");
        assert!(body.contains("\"n\":2"), "expected n=2, got: {body}");
        assert!(body.contains("\"n\":3"), "expected n=3, got: {body}");
    }

    #[tokio::test]
    async fn subscribe_on_subscription_that_yields_error_emits_error_event_then_closes() {
        let router = build_router(
            registry_with_subscription_error("events/fail", CallError::internal("handler blew up")),
            unused_provider(),
        );
        let req = json_request(
            "POST",
            "/subscribe",
            json!({ "operation": "events/fail", "input": {} }),
        );
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let body = String::from_utf8_lossy(&bytes);
        assert!(
            body.contains("event:error") || body.contains("event: error"),
            "expected error event, got: {body}"
        );
        assert!(
            body.contains("INTERNAL"),
            "expected INTERNAL code, got: {body}"
        );
        assert!(
            body.contains("handler blew up"),
            "expected error message, got: {body}"
        );
        let data_frames = body.matches("data:").count();
        assert_eq!(
            data_frames, 1,
            "expected exactly one data frame (the error payload), got: {body}"
        );
    }

    #[tokio::test]
    async fn subscribe_stream_is_terminal_after_an_error_event() {
        let router = build_router(
            registry_with_subscription_stream_continuing_after_error(
                "events/continue",
                CallError::internal("mid-stream failure"),
            ),
            unused_provider(),
        );
        let req = json_request(
            "POST",
            "/subscribe",
            json!({ "operation": "events/continue", "input": {} }),
        );
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let body = String::from_utf8_lossy(&bytes);
        let error_events = body.matches("event:").count();
        assert_eq!(error_events, 1, "exactly one error event, got: {body}");
        let data_frames = body.matches("data:").count();
        assert_eq!(
            data_frames, 1,
            "the error frame is the last event — no post-error data frames, got: {body}"
        );
        assert!(
            !body.contains("\"after\":true"),
            "the post-error envelope must not reach the wire, got: {body}"
        );
    }

    #[tokio::test]
    async fn subscribe_stream_carries_retry_field_and_keep_alive_comment() {
        let router = build_router(
            registry_with_subscription_stream("events/quiet", vec![json!({ "n": 1 })]),
            unused_provider(),
        );
        let req = json_request(
            "POST",
            "/subscribe",
            json!({ "operation": "events/quiet", "input": {} }),
        );
        let resp = router.oneshot(req).await.unwrap();
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let body = String::from_utf8_lossy(&bytes);
        assert!(
            body.contains("retry: 15000"),
            "expected a retry: hint on stream events, got: {body}"
        );
        assert!(
            body.contains(':'),
            "expected a keep-alive comment frame, got: {body}"
        );
    }

    fn registry_with_subscription_stream_continuing_after_error(
        name: &str,
        error: CallError,
    ) -> Arc<OperationRegistry> {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                subscription_spec(name, Visibility::External, AccessControl::default()),
                HandlerKind::Stream(make_streaming_handler(move |_input, ctx| {
                    let request_id = ctx.request_id.clone();
                    let error = error.clone();
                    futures::stream::iter(vec![
                        ResponseEnvelope::error(request_id.clone(), error),
                        ResponseEnvelope::ok(request_id, json!({ "after": true })),
                    ])
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    #[tokio::test]
    async fn subscribe_response_content_type_is_text_event_stream() {
        let router = build_router(
            registry_with_subscription_stream("events/stream", vec![json!({ "ok": true })]),
            unused_provider(),
        );
        let req = json_request(
            "POST",
            "/subscribe",
            json!({ "operation": "events/stream", "input": {} }),
        );
        let resp = router.oneshot(req).await.unwrap();
        let ctype = resp
            .headers()
            .get(axum::http::header::CONTENT_TYPE)
            .map(|v| v.to_str().unwrap().to_string());
        assert_eq!(
            ctype.as_deref(),
            Some("text/event-stream"),
            "expected text/event-stream, got {ctype:?}"
        );
    }

    #[tokio::test]
    async fn subscribe_internal_op_emits_error_event() {
        let router = build_router(registry_with_internal_op(), unused_provider());
        let req = json_request(
            "POST",
            "/subscribe",
            json!({ "operation": "secret/op", "input": {} }),
        );
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let body = String::from_utf8_lossy(&bytes);
        assert!(
            body.contains("event:error") || body.contains("event: error"),
            "expected error event, got: {body}"
        );
        assert!(
            body.contains("NOT_FOUND"),
            "expected NOT_FOUND, got: {body}"
        );
    }

    #[tokio::test]
    async fn subscribe_unknown_op_emits_not_found_error_event() {
        let router = build_router(
            registry_with_subscription_stream("events/stream", vec![json!({})]),
            unused_provider(),
        );
        let req = json_request(
            "POST",
            "/subscribe",
            json!({ "operation": "no/such", "input": {} }),
        );
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let body = String::from_utf8_lossy(&bytes);
        assert!(
            body.contains("event:error") || body.contains("event: error"),
            "expected error event, got: {body}"
        );
        assert!(
            body.contains("NOT_FOUND"),
            "expected NOT_FOUND, got: {body}"
        );
    }

    #[tokio::test]
    async fn subscribe_on_query_op_emits_invalid_operation_type_error_event() {
        let router = build_router(registry_with_echo(), unused_provider());
        let req = json_request(
            "POST",
            "/subscribe",
            json!({ "operation": "echo/run", "input": {} }),
        );
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let body = String::from_utf8_lossy(&bytes);
        assert!(
            body.contains("event:error") || body.contains("event: error"),
            "expected error event, got: {body}"
        );
        assert!(
            body.contains("INVALID_OPERATION_TYPE"),
            "expected INVALID_OPERATION_TYPE, got: {body}"
        );
    }

    #[test]
    fn is_internal_op_returns_false_for_unknown() {
        let registry = OperationRegistry::new();
        assert!(!is_internal_op(&registry, "no/such"));
        assert!(!is_internal_op(&registry, "/no/such"));
    }

    #[test]
    fn is_internal_op_detects_registered_internal_op() {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                internal_spec("secret/op"),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        assert!(is_internal_op(&registry, "secret/op"));
        assert!(is_internal_op(&registry, "/secret/op"));
    }

    #[test]
    fn is_internal_op_false_for_external_op() {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("echo/run", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        assert!(!is_internal_op(&registry, "echo/run"));
    }

    #[test]
    fn envelope_to_ok_json_shape() {
        let env = ResponseEnvelope::ok("req-1", json!({ "v": 1 }));
        let v = envelope_to_json(env);
        assert_eq!(v.get("request_id"), Some(&json!("req-1")));
        assert_eq!(v.get("result"), Some(&json!("ok")));
        assert_eq!(v.get("output"), Some(&json!({ "v": 1 })));
    }

    #[test]
    fn envelope_to_error_json_shape() {
        let env = ResponseEnvelope::not_found("req-2", "no/such");
        let v = envelope_to_json(env);
        assert_eq!(v.get("result"), Some(&json!("error")));
        assert_eq!(
            v.get("error").and_then(|e| e.get("code")),
            Some(&json!("NOT_FOUND"))
        );
    }

    #[tokio::test]
    async fn call_error_envelope_carries_retry_after_on_retryable_503() {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("flaky/op", AccessControl::default()),
                HandlerKind::Once(make_handler(|_input, ctx| async move {
                    ResponseEnvelope::error(
                        ctx.request_id,
                        CallError::new("HTTP_503", "overloaded", true)
                            .with_details(json!({ "retry_after": "30" })),
                    )
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let router = build_router(Arc::new(registry), unused_provider());
        let req = json_request(
            "POST",
            "/call",
            json!({ "operation": "flaky/op", "input": {} }),
        );
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
        let retry_after = resp
            .headers()
            .get(axum::http::header::RETRY_AFTER)
            .map(|v| v.to_str().unwrap().to_string());
        assert_eq!(
            retry_after.as_deref(),
            Some("30"),
            "a retryable HTTP_503 from a handler must carry Retry-After on the gateway error path"
        );
    }

    #[tokio::test]
    async fn call_with_leading_slash_in_operation_dispatches() {
        let router = build_router(registry_with_echo(), unused_provider());
        let req = json_request(
            "POST",
            "/call",
            json!({ "operation": "/echo/run", "input": {} }),
        );
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body.get("result"), Some(&json!("ok")));
    }

    #[tokio::test]
    async fn call_unknown_op_returns_404() {
        let router = build_router(registry_with_echo(), unused_provider());
        let req = json_request(
            "POST",
            "/call",
            json!({ "operation": "no/such", "input": {} }),
        );
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::NOT_FOUND);
        assert_eq!(body.get("code"), Some(&json!("NOT_FOUND")));
    }

    #[tokio::test]
    async fn search_unauthenticated_lists_default_acl_ops_only() {
        let ops = vec![
            HandlerRegistration::new(
                external_spec("public/echo", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ),
            HandlerRegistration::new(
                external_spec(
                    "admin/secret",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ),
        ];
        let discovery = registry_with_discovery_and_ops(ops);
        let router = build_router(discovery, unused_provider());
        let req = Request::builder()
            .method("GET")
            .uri("/search")
            .body(Body::empty())
            .unwrap();
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        let ops = body
            .get("output")
            .and_then(|o| o.get("operations"))
            .and_then(|o| o.as_array())
            .expect("operations array");
        let names: Vec<&str> = ops
            .iter()
            .filter_map(|o| o.get("name").and_then(|n| n.as_str()))
            .collect();
        assert!(names.contains(&"public/echo"));
        assert!(!names.contains(&"admin/secret"));
    }

    #[tokio::test]
    async fn gateway_router_mounts_at_expected_paths() {
        let ops = vec![HandlerRegistration::new(
            external_spec("echo/run", AccessControl::default()),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        )];
        let discovery = registry_with_discovery_and_ops(ops);
        let router = build_router(discovery, unused_provider());
        let req = json_request(
            "POST",
            "/call",
            json!({ "operation": "echo/run", "input": {} }),
        );
        let resp = router.clone().oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let req = Request::builder()
            .method("GET")
            .uri("/search")
            .body(Body::empty())
            .unwrap();
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn search_and_schema_carry_no_store_and_vary_authorization() {
        let discovery = registry_with_discovery_and_ops(vec![HandlerRegistration::new(
            external_spec("echo/run", AccessControl::default()),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        )]);
        let router = build_router(discovery, unused_provider());
        for uri in ["/search", "/schema?name=echo%2Frun"] {
            let req = Request::builder()
                .method("GET")
                .uri(uri)
                .body(Body::empty())
                .unwrap();
            let resp = router.clone().oneshot(req).await.unwrap();
            assert_eq!(resp.status(), StatusCode::OK);
            let cache_control = resp
                .headers()
                .get(axum::http::header::CACHE_CONTROL)
                .map(|v| v.to_str().unwrap().to_string());
            assert_eq!(
                cache_control.as_deref(),
                Some("no-store"),
                "GET {uri} must not be cacheable (GW-02)"
            );
            let vary = resp
                .headers()
                .get(axum::http::header::VARY)
                .map(|v| v.to_str().unwrap().to_string());
            assert_eq!(
                vary.as_deref(),
                Some("Authorization"),
                "GET {uri} is per-identity; it must Vary on Authorization (GW-02)"
            );
        }
    }

    #[tokio::test]
    async fn schema_denials_carry_no_store_and_vary_authorization() {
        let ops = vec![HandlerRegistration::new(
            external_spec(
                "admin/secret",
                AccessControl {
                    required_scopes: vec!["admin".to_string()],
                    ..Default::default()
                },
            ),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        )];
        let discovery = registry_with_discovery_and_ops(ops);
        let router = build_router(discovery, unused_provider());
        let req = Request::builder()
            .method("GET")
            .uri("/schema?name=admin%2Fsecret")
            .body(Body::empty())
            .unwrap();
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(
            resp.status(),
            StatusCode::UNAUTHORIZED,
            "FORBIDDEN with no identity maps to 401 (gateway error mapping)"
        );
        assert_eq!(
            resp.headers()
                .get(axum::http::header::CACHE_CONTROL)
                .map(|v| v.to_str().unwrap()),
            Some("no-store")
        );
        assert_eq!(
            resp.headers()
                .get(axum::http::header::VARY)
                .map(|v| v.to_str().unwrap()),
            Some("Authorization")
        );
    }
    // --- /publish (ADR-068) -------------------------------------------------

    use alkcall::registry::registration::make_sink_handler;

    fn publish_registry() -> Arc<OperationRegistry> {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "ingest/push",
                    OperationType::Pub,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Sink(make_sink_handler(|input, ctx, mut chunks| async move {
                    let mut collected: Vec<Value> = Vec::new();
                    use futures::StreamExt;
                    while let Some(chunk) = chunks.next().await {
                        match chunk {
                            Ok(v) => collected.push(v),
                            Err(e) => return ResponseEnvelope::error(ctx.request_id, e),
                        }
                    }
                    ResponseEnvelope::ok(
                        ctx.request_id,
                        json!({
                            "count": collected.len(),
                            "chunks": collected,
                            "seed": input,
                        }),
                    )
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        registry
            .register(HandlerRegistration::new(
                external_spec("echo/run", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "ingest/typed",
                    OperationType::Pub,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                )
                .with_publish_schema(json!({
                    "type": "object",
                    "properties": { "n": { "type": "integer" } },
                    "required": ["n"],
                    "additionalProperties": false
                })),
                HandlerKind::Sink(make_sink_handler(
                    |_unused_input, ctx, mut chunks| async move {
                        let mut collected: Vec<Value> = Vec::new();
                        use futures::StreamExt;
                        while let Some(chunk) = chunks.next().await {
                            match chunk {
                                Ok(v) => collected.push(v),
                                Err(e) => return ResponseEnvelope::error(ctx.request_id, e),
                            }
                        }
                        ResponseEnvelope::ok(
                            ctx.request_id,
                            json!({ "count": collected.len(), "chunks": collected }),
                        )
                    },
                )),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "secret/pub",
                    OperationType::Pub,
                    Visibility::Internal,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Sink(make_sink_handler(|input, ctx, _chunks| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    fn raw_request(method: &str, uri: &str, body: Vec<u8>) -> Request<Body> {
        Request::builder()
            .method(method)
            .uri(uri)
            .header("content-type", "application/x-ndjson")
            .body(Body::from(body))
            .unwrap()
    }

    fn ndjson(lines: &[Value]) -> Vec<u8> {
        let mut out = Vec::new();
        for l in lines {
            out.extend_from_slice(serde_json::to_string(l).unwrap().as_bytes());
            out.push(b'\n');
        }
        out
    }

    #[tokio::test]
    async fn publish_multi_chunk_sink_round_trip_returns_final_envelope() {
        let router = build_router(publish_registry(), unused_provider());
        let body = ndjson(&[
            json!({ "operation": "ingest/push", "chunk": { "n": 1 } }),
            json!({ "n": 2 }),
            json!({ "n": 3 }),
        ]);
        let req = raw_request("POST", "/publish", body);
        let (status, resp) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(resp.get("result"), Some(&json!("ok")));
        let output = resp.get("output").expect("output");
        assert_eq!(output["count"], 3);
        assert_eq!(output["chunks"][0], json!({ "n": 1 }));
        assert_eq!(output["chunks"][1], json!({ "n": 2 }));
        assert_eq!(output["chunks"][2], json!({ "n": 3 }));
    }

    #[tokio::test]
    async fn publish_internal_op_returns_404() {
        let router = build_router(publish_registry(), unused_provider());
        let body = ndjson(&[json!({ "operation": "secret/pub", "chunk": {} })]);
        let req = raw_request("POST", "/publish", body);
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::NOT_FOUND);
        assert_eq!(body.get("code"), Some(&json!("NOT_FOUND")));
    }

    fn pub_spec(name: &str, acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Pub,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            acl,
            None,
        )
    }

    #[tokio::test]
    async fn publish_unauthorized_restricted_op_returns_403() {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                pub_spec(
                    "ingest/push",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Sink(make_sink_handler(|input, ctx, _chunks| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let provider: Arc<dyn IdentityProvider> = Arc::new(
            StaticIdentityProvider::new()
                .with_token("user-tok", identity_with_scopes("user", &["user"])),
        );
        let router = build_router(Arc::new(registry), provider);
        let body = ndjson(&[json!({ "operation": "ingest/push", "chunk": {} })]);
        let (k, v) = auth_header("user-tok");
        let req = Request::builder()
            .method("POST")
            .uri("/publish")
            .header("content-type", "application/x-ndjson")
            .header(k, v)
            .body(Body::from(body))
            .unwrap();
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::FORBIDDEN);
        let _ = body;
    }

    #[tokio::test]
    async fn publish_non_pub_op_unauthenticated_maps_invalid_operation_type_to_401() {
        let router = build_router(publish_registry(), unused_provider());
        let body = ndjson(&[json!({ "operation": "echo/run", "chunk": {} })]);
        let req = raw_request("POST", "/publish", body);
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::UNAUTHORIZED);
        assert_eq!(body.get("code"), Some(&json!("INVALID_OPERATION_TYPE")));
    }

    #[tokio::test]
    async fn publish_non_pub_op_returns_422_invalid_operation_type() {
        let provider: Arc<dyn IdentityProvider> = Arc::new(
            StaticIdentityProvider::new()
                .with_token("user-tok", identity_with_scopes("user", &["user"])),
        );
        let router = build_router(publish_registry(), provider);
        let body = ndjson(&[json!({ "operation": "echo/run", "chunk": {} })]);
        let (k, v) = auth_header("user-tok");
        let req = Request::builder()
            .method("POST")
            .uri("/publish")
            .header("content-type", "application/x-ndjson")
            .header(k, v)
            .body(Body::from(body))
            .unwrap();
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
        assert_eq!(body.get("code"), Some(&json!("INVALID_OPERATION_TYPE")));
    }

    #[tokio::test]
    async fn publish_unknown_op_returns_404() {
        let router = build_router(publish_registry(), unused_provider());
        let body = ndjson(&[json!({ "operation": "no/such", "chunk": {} })]);
        let req = raw_request("POST", "/publish", body);
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::NOT_FOUND);
        assert_eq!(body.get("code"), Some(&json!("NOT_FOUND")));
    }

    #[tokio::test]
    async fn publish_missing_operation_in_first_line_returns_400() {
        let router = build_router(publish_registry(), unused_provider());
        let body = ndjson(&[json!({ "chunk": {} })]);
        let req = raw_request("POST", "/publish", body);
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::BAD_REQUEST);
        assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
    }

    #[tokio::test]
    async fn publish_empty_body_returns_400() {
        let router = build_router(publish_registry(), unused_provider());
        let req = raw_request("POST", "/publish", Vec::new());
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::BAD_REQUEST);
        assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
    }

    #[tokio::test]
    async fn publish_first_line_missing_chunk_returns_400_invalid_input() {
        let router = build_router(publish_registry(), unused_provider());
        let body = ndjson(&[json!({ "operation": "ingest/push" })]);
        let req = raw_request("POST", "/publish", body);
        let (status, body) = send(router, req).await;
        assert_eq!(status, StatusCode::BAD_REQUEST);
        assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
    }

    #[tokio::test]
    async fn publish_invalid_later_line_yields_handler_chunk_error() {
        let router = build_router(publish_registry(), unused_provider());
        let mut body = ndjson(&[json!({ "operation": "ingest/push", "chunk": { "n": 1 } })]);
        body.extend_from_slice(b"not-json\n");
        let req = raw_request("POST", "/publish", body);
        let (status, resp) = send(router, req).await;
        assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
        assert_eq!(
            resp.get("code"),
            Some(&json!("INVALID_INPUT")),
            "the malformed chunk line terminates the stream as INVALID_INPUT: {resp}"
        );
    }

    #[tokio::test]
    async fn publish_error_envelope_maps_to_http_status() {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "ingest/fail",
                    OperationType::Pub,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Sink(make_sink_handler(|_input, ctx, mut chunks| async move {
                    use futures::StreamExt;
                    while let Some(c) = chunks.next().await {
                        if c.is_err() {
                            break;
                        }
                    }
                    ResponseEnvelope::forbidden(ctx.request_id, "ingest denied")
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let router = build_router(Arc::new(registry), unused_provider());
        let body = ndjson(&[json!({ "operation": "ingest/fail", "chunk": {} })]);
        let req = raw_request("POST", "/publish", body);
        let (status, resp) = send(router, req).await;
        assert_eq!(
            status,
            StatusCode::UNAUTHORIZED,
            "FORBIDDEN with no identity maps to 401 (gateway error mapping)"
        );
        assert_eq!(resp.get("code"), Some(&json!("FORBIDDEN")));
    }

    #[tokio::test]
    async fn publish_schema_registered_op_rejects_invalid_chunk() {
        let router = build_router(publish_registry(), unused_provider());
        let body = ndjson(&[
            json!({ "operation": "ingest/typed", "chunk": { "n": 1 } }),
            json!({ "n": "not-an-integer" }),
            json!({ "n": 3 }),
        ]);
        let req = raw_request("POST", "/publish", body);
        let (status, resp) = send(router, req).await;
        assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
        assert_eq!(
            resp.get("code"),
            Some(&json!("INVALID_INPUT")),
            "the chunk stream terminates with the schema violation: {resp}"
        );
        assert_eq!(
            resp.get("message"),
            Some(&json!("published chunk failed publish_schema validation"))
        );
    }

    #[tokio::test]
    async fn publish_schema_registered_op_accepts_valid_chunks() {
        let router = build_router(publish_registry(), unused_provider());
        let body = ndjson(&[
            json!({ "operation": "ingest/typed", "chunk": { "n": 1 } }),
            json!({ "n": 2 }),
        ]);
        let req = raw_request("POST", "/publish", body);
        let (status, resp) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(resp.get("result"), Some(&json!("ok")));
        assert_eq!(resp["output"]["count"], 2);
    }

    #[tokio::test]
    async fn publish_op_without_schema_accepts_arbitrary_chunks() {
        let router = build_router(publish_registry(), unused_provider());
        let body = ndjson(&[
            json!({ "operation": "ingest/push", "chunk": { "anything": true } }),
            json!({ "n": null }),
        ]);
        let req = raw_request("POST", "/publish", body);
        let (status, resp) = send(router, req).await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(resp["output"]["count"], 2);
    }

    #[tokio::test]
    async fn publish_first_chunk_validated_against_publish_schema() {
        let router = build_router(publish_registry(), unused_provider());
        let body = ndjson(&[json!({ "operation": "ingest/typed", "chunk": { "wrong": 1 } })]);
        let req = raw_request("POST", "/publish", body);
        let (status, resp) = send(router, req).await;
        assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
        assert_eq!(resp.get("code"), Some(&json!("INVALID_INPUT")));
    }

    #[tokio::test]
    async fn publish_line_exceeding_cap_yields_invalid_input_chunk_error() {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "ingest/big",
                    OperationType::Pub,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Sink(make_sink_handler(|_input, ctx, mut chunks| async move {
                    use futures::StreamExt;
                    let mut last_error = None;
                    while let Some(chunk) = chunks.next().await {
                        if let Err(e) = chunk {
                            last_error = Some(e);
                            break;
                        }
                    }
                    let error = last_error.expect("an oversized line must produce an error item");
                    ResponseEnvelope::error(ctx.request_id, error)
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let router = build_router(Arc::new(registry), unused_provider());
        let mut body = ndjson(&[json!({ "operation": "ingest/big", "chunk": { "n": 1 } })]);
        let oversized = vec![b'a'; MAX_PUBLISH_LINE_BYTES + 1];
        body.extend_from_slice(&oversized);
        body.push(b'\n');
        let (status, resp) = send(router, raw_request("POST", "/publish", body)).await;
        assert_eq!(status, StatusCode::BAD_REQUEST);
        assert_eq!(resp.get("code"), Some(&json!("INVALID_INPUT")));
        assert!(
            resp.get("message")
                .and_then(|m| m.as_str())
                .map(|m| m.contains("publish line exceeds the per-line cap"))
                .unwrap_or(false),
            "expected the line-cap message, got: {resp}"
        );
    }

    fn registry_with_cap_witness_sink() -> Arc<OperationRegistry> {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "ingest/big",
                    OperationType::Pub,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Sink(make_sink_handler(|_input, ctx, mut chunks| async move {
                    use futures::StreamExt;
                    let mut last_error = None;
                    while let Some(chunk) = chunks.next().await {
                        if let Err(e) = chunk {
                            last_error = Some(e);
                            break;
                        }
                    }
                    let error = last_error.expect("a capped line must produce an error item");
                    ResponseEnvelope::error(ctx.request_id, error)
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    #[tokio::test]
    async fn publish_streamed_never_newline_body_over_cap_is_rejected_pre_extend() {
        let router = build_router(registry_with_cap_witness_sink(), unused_provider());
        let first_line = serde_json::to_vec(&json!({
            "operation": "ingest/big",
            "chunk": { "n": 1 }
        }))
        .unwrap();
        let mut body = first_line;
        body.push(b'\n');
        body.extend_from_slice(&vec![b'a'; MAX_PUBLISH_LINE_BYTES + 1]);
        let chunks: Vec<Bytes> = body.chunks(64 * 1024).map(Bytes::copy_from_slice).collect();
        let req = Request::builder()
            .method("POST")
            .uri("/publish")
            .header("content-type", "application/x-ndjson")
            .header("transfer-encoding", "chunked")
            .body(Body::from_stream(futures::stream::iter(
                chunks.into_iter().map(Ok::<_, std::convert::Infallible>),
            )))
            .unwrap();
        let (status, resp) = send(router, req).await;
        assert_eq!(
            status,
            StatusCode::UNPROCESSABLE_ENTITY,
            "a streamed never-newline body over the cap must surface the line cap before \
             any further chunk: {resp}"
        );
        assert_eq!(resp.get("code"), Some(&json!("INVALID_INPUT")));
        assert!(
            resp.get("message")
                .and_then(|m| m.as_str())
                .map(|m| m.contains("publish line exceeds the per-line cap"))
                .unwrap_or(false),
            "expected the pre-extend line-cap message, got: {resp}"
        );
        assert_eq!(resp.get("retryable"), Some(&json!(false)));
    }

    #[tokio::test]
    async fn publish_eof_without_newline_over_cap_is_rejected() {
        let router = build_router(registry_with_cap_witness_sink(), unused_provider());
        let first_line = serde_json::to_vec(&json!({
            "operation": "ingest/big",
            "chunk": { "n": 1 }
        }))
        .unwrap();
        let mut body = first_line;
        body.push(b'\n');
        body.extend_from_slice(&vec![b'a'; MAX_PUBLISH_LINE_BYTES + 1]);
        let (status, resp) = send(router, raw_request("POST", "/publish", body)).await;
        assert_eq!(
            status,
            StatusCode::BAD_REQUEST,
            "EOF with an over-cap unterminated tail must not yield the buffer: {resp}"
        );
        assert_eq!(resp.get("code"), Some(&json!("INVALID_INPUT")));
        assert!(
            resp.get("message")
                .and_then(|m| m.as_str())
                .map(|m| m.contains("publish line exceeds the per-line cap"))
                .unwrap_or(false),
            "expected the line-cap message, got: {resp}"
        );
    }

    #[tokio::test]
    async fn publish_body_over_gateway_limit_returns_413() {
        let router = build_router(publish_registry(), unused_provider());
        let mut body = serde_json::to_vec(&json!({
            "operation": "ingest/push",
            "chunk": { "n": 0 }
        }))
        .unwrap();
        body.push(b'\n');
        let filler = vec![b'a'; 4096 - 10];
        for _ in 0..600 {
            body.extend_from_slice(b"{\"n\":\"");
            body.extend_from_slice(&filler);
            body.extend_from_slice(b"\"}\n");
        }
        let chunks: Vec<Bytes> = body.chunks(64 * 1024).map(Bytes::copy_from_slice).collect();
        let req = Request::builder()
            .method("POST")
            .uri("/publish")
            .header("content-type", "application/x-ndjson")
            .body(Body::from_stream(futures::stream::iter(
                chunks.into_iter().map(Ok::<_, std::convert::Infallible>),
            )))
            .unwrap();
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(
            resp.status(),
            StatusCode::PAYLOAD_TOO_LARGE,
            "a chunked upload of many under-cap lines over the body limit must be cut off with 413"
        );
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        assert_eq!(
            &bytes[..],
            GATEWAY_BODY_LIMIT_EXCEEDED.as_bytes(),
            "the layer answers plain text 413, not an error envelope"
        );
    }

    #[tokio::test]
    async fn publish_declared_content_length_over_gateway_limit_returns_413() {
        let router = build_router(publish_registry(), unused_provider());
        let req = Request::builder()
            .method("POST")
            .uri("/publish")
            .header("content-type", "application/x-ndjson")
            .header("content-length", (GATEWAY_BODY_LIMIT + 1).to_string())
            .body(Body::from(vec![b'a'; 64 * 1024]))
            .unwrap();
        let (status, resp) = send(router, req).await;
        assert_eq!(
            status,
            StatusCode::PAYLOAD_TOO_LARGE,
            "a declared content-length over the limit must be pre-rejected with 413"
        );
        assert_eq!(
            resp,
            Value::Null,
            "the layer answers plain text 413, not an error envelope: {resp}"
        );
    }

    #[tokio::test]
    async fn publish_line_cap_breach_when_batched_with_complete_lines_is_still_rejected() {
        let router = build_router(registry_with_cap_witness_sink(), unused_provider());
        let mut body = ndjson(&[
            json!({ "operation": "ingest/big", "chunk": { "n": 1 } }),
            json!({ "n": 2 }),
        ]);
        body.extend_from_slice(&vec![b'a'; MAX_PUBLISH_LINE_BYTES + 1]);
        body.push(b'\n');
        let (status, resp) = send(router, raw_request("POST", "/publish", body)).await;
        assert_eq!(
            status,
            StatusCode::BAD_REQUEST,
            "an over-cap line batched with complete lines must still abort the request with the \
             cap error (today's status; GW-16 normalizes): {resp}"
        );
        assert_eq!(resp.get("code"), Some(&json!("INVALID_INPUT")));
    }

    #[tokio::test]
    async fn publish_body_at_line_cap_within_limit_still_round_trips() {
        let router = build_router(publish_registry(), unused_provider());
        let first_line = serde_json::to_vec(&json!({
            "operation": "ingest/push",
            "chunk": { "n": 1 }
        }))
        .unwrap();
        let mut body = first_line;
        body.push(b'\n');
        body.extend_from_slice(b"\"");
        body.extend_from_slice(&vec![b'a'; MAX_PUBLISH_LINE_BYTES - 2]);
        body.extend_from_slice(b"\"");
        body.push(b'\n');
        let (status, resp) = send(router, raw_request("POST", "/publish", body)).await;
        assert_eq!(
            status,
            StatusCode::OK,
            "a body under the limit with at-cap lines must pass the layer: {resp}"
        );
        assert_eq!(
            resp.get("result"),
            Some(&json!("ok")),
            "a second line exactly at the per-line cap must be accepted (the cap is >, not >=): {resp}"
        );
    }

    #[tokio::test]
    async fn publish_client_disconnect_before_dispatch_signals_error_item() {
        let router = build_router(publish_registry(), unused_provider());
        let first_line = serde_json::to_vec(&json!({
            "operation": "ingest/push",
            "chunk": { "n": 1 }
        }))
        .unwrap();
        let mut body = first_line.clone();
        body.push(b'\n');
        body.extend_from_slice(
            b"not-json
",
        );
        let req = raw_request("POST", "/publish", body);
        let (status, resp) = send(router, req).await;
        assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
        assert_eq!(
            resp.get("code"),
            Some(&json!("INVALID_INPUT")),
            "the prematurely terminated body reaches the handler as an INVALID_INPUT Err item: {resp}"
        );
    }

    // --- publish_schema fail-closed (GW-01 follow-up) ----------------------

    fn registry_with_broken_publish_schema() -> Arc<OperationRegistry> {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "ingest/broken",
                    OperationType::Pub,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                )
                .with_publish_schema(json!({ "required": "n" })), // `required` must be an array
                HandlerKind::Sink(make_sink_handler(
                    |_unused_input, ctx, mut chunks| async move {
                        use futures::StreamExt;
                        let mut collected: Vec<Value> = Vec::new();
                        while let Some(chunk) = chunks.next().await {
                            match chunk {
                                Ok(v) => collected.push(v),
                                Err(e) => return ResponseEnvelope::error(ctx.request_id, e),
                            }
                        }
                        ResponseEnvelope::ok(
                            ctx.request_id,
                            json!({ "count": collected.len(), "chunks": collected }),
                        )
                    },
                )),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    #[tokio::test]
    async fn publish_with_uncompilable_schema_fails_closed_with_internal() {
        let router = build_router(registry_with_broken_publish_schema(), unused_provider());
        let body = ndjson(&[
            json!({ "operation": "ingest/broken", "chunk": { "n": 1 } }),
            json!({ "n": 2 }),
        ]);
        let req = raw_request("POST", "/publish", body);
        let (status, resp) = send(router, req).await;
        assert_eq!(
            status,
            StatusCode::INTERNAL_SERVER_ERROR,
            "an un-compilable publish_schema must fail the request loudly (500), not skip validation: {resp}"
        );
        assert_eq!(
            resp.get("code"),
            Some(&json!("INTERNAL")),
            "fail-closed maps to INTERNAL, not INVALID_INPUT (this is a server-side defect): {resp}"
        );
        assert!(
            resp.get("message")
                .and_then(|m| m.as_str())
                .map(|m| m.contains("failed to compile"))
                .unwrap_or(false),
            "the wire message names the compile failure without echoing schema internals: {resp}"
        );
    }

    #[tokio::test]
    async fn publish_with_uncompilable_schema_never_delivers_chunks_to_the_handler() {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "ingest/spy",
                    OperationType::Pub,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                )
                .with_publish_schema(json!({ "type": 123 })), // non-string `type` never compiles
                HandlerKind::Sink(make_sink_handler(|_input, ctx, mut chunks| async move {
                    use futures::StreamExt;
                    let received: Vec<Result<Value, CallError>> = chunks.by_ref().collect().await;
                    ResponseEnvelope::ok(
                        ctx.request_id,
                        json!({
                            "chunk_count": received.len(),
                            "ok_chunks": received.iter().filter(|r| r.is_ok()).count(),
                            "error_codes": received
                                .iter()
                                .filter_map(|r| r.as_ref().err().map(|e| e.code.clone()))
                                .collect::<Vec<String>>(),
                        }),
                    )
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let router = build_router(Arc::new(registry), unused_provider());
        let body = ndjson(&[
            json!({ "operation": "ingest/spy", "chunk": { "n": 1 } }),
            json!({ "n": 2 }),
        ]);
        let (status, resp) = send(router, raw_request("POST", "/publish", body)).await;
        let output = resp.get("output").expect("spy handler output");
        assert_eq!(
            output.get("ok_chunks"),
            Some(&json!(0)),
            "no chunk may reach the handler when the publish_schema failed to compile: {resp}"
        );
        assert_eq!(
            output.get("chunk_count"),
            Some(&json!(1)),
            "exactly one item flows: the terminal INTERNAL error (stream ends after it): {resp}"
        );
        assert_eq!(
            output.get("error_codes"),
            Some(&json!(["INTERNAL"])),
            "the terminal item is the fail-closed INTERNAL compile error: {resp}"
        );
        let _ = status;
    }

    #[tokio::test]
    async fn publish_hot_reload_replacement_schema_is_picked_up() {
        let cache = PublishSchemaCache::new();
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "ingest/typed",
                    OperationType::Pub,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                )
                .with_publish_schema(json!({
                    "type": "object",
                    "properties": { "n": { "type": "integer" } },
                    "required": ["n"],
                    "additionalProperties": false
                })),
                HandlerKind::Sink(make_sink_handler(|input, ctx, _chunks| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();

        let v1 = cache
            .validator(&registry, "ingest/typed")
            .unwrap()
            .expect("compiled validator");
        assert!(!v1.is_valid(&json!({ "n": 1, "extra": true })));

        // Hot reload: re-register the same op with a schema that drops
        // `additionalProperties`; the same cache entry must recompile.
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "ingest/typed",
                    OperationType::Pub,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                )
                .with_publish_schema(json!({
                    "type": "object",
                    "properties": { "n": { "type": "integer" } },
                    "required": ["n"]
                })),
                HandlerKind::Sink(make_sink_handler(|input, ctx, _chunks| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();

        let v2 = cache
            .validator(&registry, "ingest/typed")
            .unwrap()
            .expect("recompiled validator");
        assert!(
            !Arc::ptr_eq(&v1, &v2),
            "a changed schema value must invalidate the cached validator"
        );
        assert!(
            v2.is_valid(&json!({ "n": 1, "extra": true })),
            "the recompiled schema must reflect the new registration"
        );
    }
}