alkhttp/src/gateway/routes.rs

//! 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).

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 crate::server::auth::ResolvedIdentity;
use crate::server::state::RouterState;

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

/// 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>,
}

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

    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),
        )
    }
}

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))
}

#[derive(Debug, Deserialize)]
pub struct CallRequest {
    pub operation: String,
    #[serde(default = "Value::default")]
    pub input: Value,
}

#[derive(Debug, Deserialize)]
pub struct SchemaQuery {
    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()
}

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.
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, validator): (String, Value, Option<jsonschema::Validator>) = {
        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();
        };
        let validator = state
            .registry
            .registration(op.strip_prefix('/').unwrap_or(&op))
            .and_then(|reg| reg.spec.publish_schema.clone())
            .and_then(|schema| match jsonschema::options().build(&schema) {
                Ok(v) => Some(v),
                Err(e) => {
                    tracing::warn!(
                        operation = %op,
                        error = %e,
                        "publish_schema failed to compile; chunks will not be validated"
                    );
                    None
                }
            });
        (op, chunk.clone(), validator)
    };

    let chunks = NdjsonChunkStream::new(header_lines, validator, 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())
}

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()
}

/// 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. Lines are capped at `MAX_PUBLISH_LINE_BYTES` (in
/// addition to axum's own 2 MiB default body limit); a byte-level read
/// failure surfaces as a terminal error.
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,
        }
    }

    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);
                }
                return Ok(Some(std::mem::take(&mut self.buffer)));
            }
            match self.bytes.next().await {
                Some(Ok(bytes)) => self.buffer.extend_from_slice(&bytes),
                Some(Err(_)) => {
                    self.done = true;
                    return Err(LineError::Read);
                }
                None => {
                    self.done = true;
                    continue;
                }
            }
            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 {
                    self.done = true;
                    self.buffer.clear();
                    self.pending.clear();
                    return Err(LineError::LineCap);
                }
                self.pending.push_back(line.to_vec());
            }
        }
    }
}

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(INVALID_INPUT)` item — the same shape an initiator-side
/// `call.error` produces on the wire.
struct NdjsonChunkStream {
    lines: std::pin::Pin<Box<BufferedLines>>,
    validator: Option<jsonschema::Validator>,
    pending_first: Option<Value>,
}

impl NdjsonChunkStream {
    fn new(
        lines: BufferedLines,
        validator: Option<jsonschema::Validator>,
        pending_first: Option<Value>,
    ) -> Self {
        Self {
            lines: Box::pin(lines),
            validator,
            pending_first,
        }
    }

    fn validate_chunk(
        &mut self,
        value: Value,
    ) -> std::task::Poll<Option<Result<Value, CallError>>> {
        if let Some(validator) = &self.validator {
            if !Validator::is_valid(validator, &value) {
                return std::task::Poll::Ready(Some(Err(CallError::invalid_input(
                    "published chunk failed publish_schema validation",
                )
                .with_details(json!({ "chunk": value })))));
            }
        }
        std::task::Poll::Ready(Some(Ok(value)))
    }
}

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.pending_first.is_some() {
            let first = self.pending_first.take().unwrap_or(Value::Null);
            return self.validate_chunk(first);
        }
        let line = {
            let mut next = Box::pin(self.lines.next_line());
            match std::future::Future::poll(next.as_mut(), cx) {
                std::task::Poll::Ready(Ok(Some(line))) => line,
                std::task::Poll::Ready(Ok(None)) => return std::task::Poll::Ready(None),
                std::task::Poll::Ready(Err(e)) => {
                    return std::task::Poll::Ready(Some(Err(CallError::invalid_input(e.message()))))
                }
                std::task::Poll::Pending => return std::task::Poll::Pending,
            }
        };
        let value = match serde_json::from_slice::<Value>(&line) {
            Ok(v) => v,
            Err(e) => {
                return std::task::Poll::Ready(Some(Err(CallError::invalid_input(format!(
                    "publish line is not valid JSON: {e}"
                )))))
            }
        };
        self.validate_chunk(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),
        };
        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}"
        );
    }

    #[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}"
        );
    }
}