reddb-io-server 1.24.12

use super::*;
use std::sync::{Mutex, OnceLock};

/// Issue #767 / S8 — context handed to streaming handlers so they can
/// emit `stream.opened` / `stream.closed` audit events without
/// re-deriving the principal or peeking back into the request headers.
///
/// `principal` is the stable label produced by [`principal_for`] (the
/// same value used by the QPS quota gate and the capacity registry).
/// `token` is the raw bearer string, retained so handlers can decide
/// whether the bearer credential's expiry is shorter than the lease's
/// snapshot-TTL deadline.
pub(crate) struct StreamAuditCtx<'a> {
    pub principal: &'a str,
    pub token: Option<&'a str>,
}

fn extract_bearer(headers: &BTreeMap<String, String>) -> Option<&str> {
    headers
        .get("authorization")
        .and_then(|v| v.strip_prefix("Bearer "))
        .map(|t| t.trim())
        .filter(|t| !t.is_empty())
}

const CATALOG_DEPRECATION_DATE: &str = "2026-08-08";
const CATALOG_DEPRECATION_LOG_INTERVAL: Duration = Duration::from_secs(60);

pub(crate) fn deprecated_catalog_response(
    endpoint: &'static str,
    mut response: HttpResponse,
) -> HttpResponse {
    warn_deprecated_catalog_endpoint(endpoint);

    response.extra_headers.push((
        "Deprecation",
        crate::server::header_escape_guard::HeaderEscapeGuard::header_value(
            CATALOG_DEPRECATION_DATE,
        )
        .expect("catalog deprecation date is a valid HTTP header value"),
    ));
    response.extra_headers.push((
        "Sunset",
        crate::server::header_escape_guard::HeaderEscapeGuard::header_value(
            CATALOG_DEPRECATION_DATE,
        )
        .expect("catalog sunset date is a valid HTTP header value"),
    ));
    response
}

fn warn_deprecated_catalog_endpoint(endpoint: &'static str) {
    static LAST_WARNED: OnceLock<Mutex<BTreeMap<&'static str, SystemTime>>> = OnceLock::new();

    let now = SystemTime::now();
    let mut last_warned = match LAST_WARNED
        .get_or_init(|| Mutex::new(BTreeMap::new()))
        .lock()
    {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    };
    let should_warn = last_warned
        .get(endpoint)
        .and_then(|last| now.duration_since(*last).ok())
        .map(|elapsed| elapsed >= CATALOG_DEPRECATION_LOG_INTERVAL)
        .unwrap_or(true);

    if should_warn {
        last_warned.insert(endpoint, now);
        tracing::warn!(
            target: "reddb::http",
            endpoint,
            deprecation = CATALOG_DEPRECATION_DATE,
            sunset = CATALOG_DEPRECATION_DATE,
            "deprecated catalog endpoint used"
        );
    }
}

impl RedDBServer {
    pub(crate) fn try_route_streaming<W: std::io::Write>(
        &self,
        request: &HttpRequest,
        writer: &mut W,
    ) -> io::Result<bool> {
        let request_path = Self::route_alias_target(&request.method, &request.path)
            .unwrap_or_else(|| request.path.clone());
        // Two streaming dispatch paths share this entry:
        //   * SSE for `ASK ... STREAM` queries (pre-existing).
        //   * NDJSON for clients that opt in with
        //     `Accept: application/x-ndjson` (issue #760).
        // Both ride the same auth + quota gate below so the streaming
        // surface inherits the non-streaming /query authorisation
        // model unchanged.
        let is_query_post = matches!(
            (request.method.as_str(), request_path.as_str()),
            ("POST", "/query")
        );
        let is_input_stream_post = matches!(
            (request.method.as_str(), request_path.as_str()),
            ("POST", "/streams/input")
        ) && content_type_is_ndjson(&request.headers);
        // Issue #805 / #750 — the dedicated read-only SELECT streaming
        // route. Unlike `/query` (which only streams when the client
        // opts in via an NDJSON Accept header or an ASK STREAM body),
        // `/query/stream` is inherently a streaming endpoint: any POST
        // to it streams NDJSON, so no content negotiation is needed.
        let is_select_stream_post = matches!(
            (request.method.as_str(), request_path.as_str()),
            ("POST", "/query/stream")
        );
        if !is_query_post && !is_input_stream_post && !is_select_stream_post {
            return Ok(false);
        }
        let is_sse = is_query_post && is_stream_ask_query_body(&request.body);
        let is_ndjson = is_query_post && wants_ndjson_response(&request.headers);
        if !is_sse && !is_ndjson && !is_input_stream_post && !is_select_stream_post {
            return Ok(false);
        }

        if !self.surface_allows(&request.method, &request_path) {
            writer.write_all(&json_error(404, "not found on this listener").to_http_bytes())?;
            writer.flush()?;
            return Ok(true);
        }

        if !self.is_authorized(&request.method, &request_path, &request.headers) {
            writer.write_all(&json_error(401, "unauthorized").to_http_bytes())?;
            writer.flush()?;
            return Ok(true);
        }

        // See `dispatch_route`: streaming statements need the caller's
        // identity installed for the privilege / RLS gates too.
        let _identity = crate::server::request_identity::RequestIdentityGuard::install(
            crate::server::handlers_auth::resolve_auth_caller(self, &request.headers)
                .map(|caller| (caller.id, caller.role)),
        );

        let principal = principal_for(&request.headers);
        match self.runtime.quota_bucket().consume(&principal) {
            crate::runtime::quota_bucket::QuotaOutcome::Throttled => {
                let retry = self.runtime.quota_bucket().retry_after_secs();
                let response = HttpResponse {
                    status: 429,
                    content_type: "application/json",
                    body: format!("{{\"error\":\"rate limited\",\"retry_after_secs\":{retry}}}")
                        .into_bytes(),
                    extra_headers: Vec::new(),
                };
                writer.write_all(&response.to_http_bytes())?;
                writer.flush()?;
                Ok(true)
            }
            crate::runtime::quota_bucket::QuotaOutcome::Granted
            | crate::runtime::quota_bucket::QuotaOutcome::NotConfigured => {
                // SSE wins precedence when both an ASK STREAM body and
                // an NDJSON Accept header are present — ASK STREAM is
                // the long-standing surface; NDJSON is opt-in for
                // ordinary SELECTs.
                let bearer = extract_bearer(&request.headers);
                let audit_ctx = StreamAuditCtx {
                    principal: &principal,
                    token: bearer,
                };
                if is_select_stream_post {
                    // Issue #805 / #750 — read-only SELECT streaming.
                    // Rides the same auth + quota gate above; the
                    // handler owns the read-only gate, descriptor-first
                    // framing, and chunked NDJSON body. Lease/audit/
                    // capacity plumbing is deferred to #750 siblings, so
                    // this branch does not acquire a capacity guard.
                    //
                    // Issue #807 / 750c — the cursor registry scopes each
                    // entry to the requesting `(tenant, principal)`, so the
                    // handler needs both here. `principal` is the same label
                    // the quota gate used above; `tenant` is resolved from
                    // the request headers.
                    let tenant = self.stream_tenant_for(&request.headers);
                    self.handle_query_select_stream(
                        request.body.clone(),
                        &principal,
                        &tenant,
                        writer,
                    )?;
                } else if is_sse {
                    self.handle_query_sse_stream(request.body.clone(), writer)?;
                } else if is_input_stream_post {
                    let cfg = crate::server::output_stream::StreamConfig::load(&self.runtime);
                    match self.stream_capacity.try_acquire(
                        &principal,
                        cfg.max_global_streams,
                        cfg.max_per_principal_streams,
                    ) {
                        Ok(_capacity_guard) => {
                            self.handle_query_ndjson_input_stream(
                                request.body.clone(),
                                &audit_ctx,
                                writer,
                            )?;
                        }
                        Err(err) => {
                            emit_capacity_refused_audit(&self.runtime, &principal, &err);
                            let response = stream_capacity_refusal_response(&err);
                            writer.write_all(&response.to_http_bytes())?;
                            writer.flush()?;
                        }
                    }
                } else {
                    // Issue #761 / S2 — capacity guard. Caps are read
                    // from `red.config` at OpenStream so subsequent
                    // KV mutations apply to future acquisitions only.
                    // The RAII guard is held for the duration of the
                    // handler call so success / mid-stream error /
                    // snapshot expiry / panic unwind all release the
                    // slot through Drop.
                    let cfg = crate::server::output_stream::StreamConfig::load(&self.runtime);
                    match self.stream_capacity.try_acquire(
                        &principal,
                        cfg.max_global_streams,
                        cfg.max_per_principal_streams,
                    ) {
                        Ok(_capacity_guard) => {
                            self.handle_query_ndjson_stream(
                                request.body.clone(),
                                &audit_ctx,
                                writer,
                            )?;
                        }
                        Err(err) => {
                            emit_capacity_refused_audit(&self.runtime, &principal, &err);
                            let response = stream_capacity_refusal_response(&err);
                            writer.write_all(&response.to_http_bytes())?;
                            writer.flush()?;
                        }
                    }
                }
                Ok(true)
            }
        }
    }

    /// Whether a parsed request would be handled by the streaming
    /// dispatch path ([`Self::try_route_streaming`]). The async axum edge
    /// (issue #931) calls this *before* dispatch to decide whether to
    /// bridge the response as a chunked/SSE stream or as a single
    /// buffered body — the detection here must stay in lock-step with the
    /// guards at the top of `try_route_streaming`.
    pub(crate) fn is_streaming_request(&self, request: &HttpRequest) -> bool {
        let request_path = Self::route_alias_target(&request.method, &request.path)
            .unwrap_or_else(|| request.path.clone());
        let is_query_post = matches!(
            (request.method.as_str(), request_path.as_str()),
            ("POST", "/query")
        );
        let is_input_stream_post = matches!(
            (request.method.as_str(), request_path.as_str()),
            ("POST", "/streams/input")
        ) && content_type_is_ndjson(&request.headers);
        let is_select_stream_post = matches!(
            (request.method.as_str(), request_path.as_str()),
            ("POST", "/query/stream")
        );
        if !is_query_post && !is_input_stream_post && !is_select_stream_post {
            return false;
        }
        let is_sse = is_query_post && is_stream_ask_query_body(&request.body);
        let is_ndjson = is_query_post && wants_ndjson_response(&request.headers);
        is_sse || is_ndjson || is_input_stream_post || is_select_stream_post
    }

    /// Issue #807 / 750c — resolve the tenant a `/query/stream` request runs
    /// under so a minted cursor can be scoped to it. Precedence mirrors the
    /// metrics surface: explicit `x-reddb-tenant` header, then the tenant
    /// carried by the bearer credential (validated JWT or auth-store token),
    /// then the ambient connection tenant, falling back to `"default"`.
    pub(crate) fn stream_tenant_for(&self, headers: &BTreeMap<String, String>) -> String {
        headers
            .get("x-reddb-tenant")
            .filter(|value| !value.trim().is_empty())
            .map(|value| value.trim().to_string())
            .or_else(|| self.stream_tenant_from_bearer(headers))
            .or_else(crate::runtime::impl_core::current_tenant)
            .unwrap_or_else(|| "default".to_string())
    }

    fn stream_tenant_from_bearer(&self, headers: &BTreeMap<String, String>) -> Option<String> {
        let token = headers
            .get("authorization")
            .and_then(|value| value.strip_prefix("Bearer "))?;
        if looks_like_jwt(token) {
            if let Some(validator) = self.runtime.oauth_validator() {
                if let Ok((tenant, _username, _role)) =
                    crate::wire::redwire::auth::validate_oauth_jwt_full(&validator, token)
                {
                    return tenant;
                }
            }
        }
        let auth_store = self.auth_store.as_ref()?;
        auth_store
            .validate_token_full(token)
            .and_then(|(id, _role)| id.tenant)
    }

    /// Dispatch one HTTP request and record its method/route/status into
    /// the operational telemetry counter (`reddb_http_requests_total`,
    /// issue #1239). The label normalization folds the raw path to its
    /// matched route *template* (or `__unmatched__`) so high-cardinality
    /// paths/ids/query strings never become labels; the status is folded to
    /// its class. Recording wraps the full dispatch so gate rejections
    /// (401/404/405/429) are counted too — every handled request increments.
    pub(crate) fn route(&self, request: HttpRequest) -> HttpResponse {
        let method_label = http_request_metrics::method_label(&request.method);
        let route_label = self.http_metric_route_label(&request.method, &request.path);
        let response = self.dispatch_route(request);
        self.http_request_metrics()
            .record(method_label, route_label, response.status);
        response
    }

    /// Normalize `(method, path)` to a bounded static route label: the
    /// matched catalog route *template* (e.g. `/catalog/collections/:name`),
    /// or the reserved `__unmatched__` bucket. Alias paths are resolved to
    /// their canonical template first so an alias and its canonical share
    /// one label. This is the only place raw paths are inspected, and the
    /// output is always one of the static route patterns — never the path.
    fn http_metric_route_label(&self, method: &str, path: &str) -> &'static str {
        let Some(method) = route_catalog::RouteMethod::from_http_method(method) else {
            return http_request_metrics::UNMATCHED_ROUTE;
        };
        let catalog = routes::discovered_route_catalog();
        let canonical = catalog.resolve_alias(method, path);
        let lookup = canonical.as_deref().unwrap_or(path);
        match catalog.find(method, lookup) {
            Some(matched) => matched.spec.pattern,
            None => http_request_metrics::UNMATCHED_ROUTE,
        }
    }

    fn dispatch_route(&self, request: HttpRequest) -> HttpResponse {
        let request = Self::rewrite_route_alias(request);
        let HttpRequest {
            method,
            path,
            query,
            headers,
            body,
        } = request;

        // CORS preflight. Browsers send `OPTIONS` with no credentials
        // before a cross-origin request; it must be answered *before*
        // auth/surface checks (a 401/404 on preflight would block the
        // real request). 204 + the permissive `Access-Control-*` headers
        // that ride on every response (see `HttpResponse::to_http_bytes`).
        if method == "OPTIONS" {
            return HttpResponse {
                status: 204,
                content_type: "application/json",
                body: Vec::new(),
                extra_headers: Vec::new(),
            };
        }

        // PLAN.md Phase 6.2 — endpoint segregation. Listeners bound
        // to the dedicated admin / metrics ports refuse paths
        // outside their surface so accidentally exposing them does
        // not leak the rest of the API.
        if !self.surface_allows(&method, &path) {
            return json_error(404, "not found on this listener");
        }

        if !self.is_authorized(&method, &path, &headers) {
            return json_error(401, "unauthorized");
        }

        // Scope the runtime's thread-local identity (principal, role,
        // tenant) to this request so the privilege / column-policy / RLS
        // gates see who is calling. Without this every statement over
        // HTTP ran as an "embedded" caller with enforcement off.
        let _identity = crate::server::request_identity::RequestIdentityGuard::install(
            crate::server::handlers_auth::resolve_auth_caller(self, &headers)
                .map(|caller| (caller.id, caller.role)),
        );

        // PLAN.md Phase 4.4 — per-caller QPS quota. Health probes
        // skip the gate so probes never trip 429 on a hot instance.
        if !self.discovered_route_uses_middleware(
            &method,
            &path,
            route_catalog::RouteMiddleware::QuotaBypass,
        ) {
            let principal = principal_for(&headers);
            match self.runtime.quota_bucket().consume(&principal) {
                crate::runtime::quota_bucket::QuotaOutcome::Throttled => {
                    let retry = self.runtime.quota_bucket().retry_after_secs();
                    return HttpResponse {
                        status: 429,
                        content_type: "application/json",
                        body: format!(
                            "{{\"error\":\"rate limited\",\"retry_after_secs\":{retry}}}"
                        )
                        .into_bytes(),
                        extra_headers: Vec::new(),
                    };
                }
                crate::runtime::quota_bucket::QuotaOutcome::Granted
                | crate::runtime::quota_bucket::QuotaOutcome::NotConfigured => {}
            }
        }

        if let Some(deny) = self.discovered_route_ops_policy(&method, &path, &headers) {
            return deny;
        }

        // One catalog lookup serves both the handler dispatch and the
        // canonical-404 decision below.
        let matched = Self::discovered_route(&method, &path);
        if let Some(matched) = &matched {
            let request = route_catalog::RouteRequest {
                matched,
                method: &method,
                path: &path,
                query: &query,
                headers: &headers,
                body: &body,
            };
            if let Some(response) = self.route_discovered_buffered(&request) {
                return response;
            }
        }
        if Self::discovered_route_method_not_allowed(&method, &path) {
            return json_error(405, "method not allowed");
        }
        // A cataloged path whose handler declined (returned `None`) is a
        // canonical 404 — it must not fall through to the UI bundle surface.
        if matched.is_some() {
            return json_error(404, "not found");
        }

        // `red server --ui` (#1047, ADR 0051): no catalog route matched —
        // fall back to the static bundle surface. This runs last so a bundle
        // asset can never shadow a real endpoint; a missing asset returns
        // `None` and we drop to the canonical 404 below.
        if method == "GET" {
            if let Some(ui_dir) = self.ui_dir() {
                if let Some(response) = crate::server::ui_static::serve_bundle_asset(ui_dir, &path)
                {
                    return response;
                }
            }
        }
        json_error(404, format!("route not found: {} {}", method, path))
    }

    pub(crate) fn resolve_projection_payload(
        &self,
        payload: &JsonValue,
    ) -> Result<Option<RuntimeGraphProjection>, HttpResponse> {
        self.graph_use_cases()
            .resolve_projection(
                json_string_field(payload, "projection_name").as_deref(),
                crate::application::graph_payload::parse_inline_projection(payload),
            )
            .map_err(|err| json_error(400, err.to_string()))
    }

    /// Buffered dispatch: invoke the handler the matched command declared.
    /// The catalog is the only routing table — there is no id-keyed match
    /// here, so a route can only be reached through the handler its own
    /// declaration binds.
    fn route_discovered_buffered(
        &self,
        request: &route_catalog::RouteRequest<'_>,
    ) -> Option<HttpResponse> {
        (request.matched.spec.handler?)(self, request)
    }

    /// Routing 3/3 (#1643) — `/config/{key.path}` GET/PUT/DELETE. Re-runs the
    /// legacy `strip_prefix("/config/")` parse so multi-segment keys and the
    /// bare-`/config` fall-through (→ canonical 404) are preserved.
    pub(crate) fn handle_config_key_route(
        &self,
        method: &str,
        path: &str,
        body: &[u8],
    ) -> Option<HttpResponse> {
        let config_key = path.strip_prefix("/config/")?.trim_matches('/');
        if config_key.is_empty() {
            return None;
        }
        Some(match method {
            "GET" => self.handle_config_get_key(config_key),
            "PUT" => self.handle_config_set_key(config_key, body.to_vec()),
            "DELETE" => self.handle_config_delete_key(config_key),
            _ => json_error(405, "method not allowed"),
        })
    }

    /// Routing 3/3 (#1643) — the KV collection surface. Re-runs the legacy
    /// helper parsing (percent-decode, multi-segment keys, `/kv/` vs `/kvs/`
    /// watch alias) in the same order the legacy matcher used, returning
    /// `None` when none of the shapes match so the caller falls through to
    /// the canonical 404.
    pub(crate) fn handle_collection_kv_route(
        &self,
        method: &str,
        path: &str,
        query: &BTreeMap<String, String>,
        headers: &BTreeMap<String, String>,
        body: &[u8],
    ) -> Option<HttpResponse> {
        if let Some(collection) = collection_kv_invalidate_tags_path(path) {
            return Some(match method {
                "POST" => self.handle_invalidate_kv_tags(collection, body.to_vec()),
                _ => json_error(405, "method not allowed for KV tag invalidation endpoint"),
            });
        }
        if let Some((collection, key)) = collection_kv_watch_path(path) {
            return Some(match method {
                "GET" => self.handle_watch_kv(&collection, &key, query),
                _ => json_error(405, "method not allowed for KV watch endpoint"),
            });
        }
        if let Some((collection, key)) = collection_kv_path(path) {
            let policy_action = match method {
                "GET" => "select",
                "PUT" => "insert",
                "PATCH" => "update",
                "DELETE" => "delete",
                _ => return Some(json_error(405, "method not allowed for KV endpoint")),
            };
            if let Some(deny) =
                self.check_collection_http_policy(headers, policy_action, &collection)
            {
                return Some(deny);
            }
            return Some(match method {
                "GET" => self.handle_get_kv(&collection, &key),
                "PUT" => self.handle_put_kv(&collection, &key, body.to_vec()),
                "PATCH" => self.handle_patch_kv(&collection, &key, body.to_vec()),
                "DELETE" => self.handle_delete_kv(&collection, &key),
                _ => json_error(405, "method not allowed for KV endpoint"),
            });
        }
        None
    }

    /// Routing 3/3 (#1643) — index lifecycle actions
    /// (`POST /indexes/:name/:action`). Non-POST methods and unknown actions
    /// return `None` so the caller falls through to the canonical 404, exactly
    /// as the legacy `if method == "POST"` guard did.
    pub(crate) fn handle_index_action_route(
        &self,
        method: &str,
        name: &str,
        action: &str,
    ) -> Option<HttpResponse> {
        if method != "POST" {
            return None;
        }
        let result = match action {
            "enable" => self.admin_use_cases().set_index_enabled(name, true),
            "disable" => self.admin_use_cases().set_index_enabled(name, false),
            "warmup" => self.admin_use_cases().warmup_index(name),
            "building" => self.admin_use_cases().mark_index_building(name),
            "fail" => self.admin_use_cases().fail_index(name),
            "stale" => self.admin_use_cases().mark_index_stale(name),
            "ready" => self.admin_use_cases().mark_index_ready(name),
            _ => return None,
        };
        Some(match result {
            Ok(index) => json_response(200, crate::presentation::admin_json::index_json(&index)),
            Err(err) => json_error(400, err.to_string()),
        })
    }

    fn rewrite_route_alias(mut request: HttpRequest) -> HttpRequest {
        if let Some(target) = Self::route_alias_target(&request.method, &request.path) {
            request.path = target;
        }
        request
    }

    fn route_alias_target(method: &str, path: &str) -> Option<String> {
        let method = route_catalog::RouteMethod::from_http_method(method)?;
        routes::discovered_route_catalog().resolve_alias(method, path)
    }

    fn discovered_route(method: &str, path: &str) -> Option<route_catalog::RouteMatch<'static>> {
        let method = route_catalog::RouteMethod::from_http_method(method)?;
        routes::discovered_route_catalog().find(method, path)
    }

    fn discovered_route_method_not_allowed(method: &str, path: &str) -> bool {
        let Some(method) = route_catalog::RouteMethod::from_http_method(method) else {
            return false;
        };
        let catalog = routes::discovered_route_catalog();
        catalog.find(method, path).is_none() && catalog.path_exists(path)
    }

    fn discovered_route_uses_middleware(
        &self,
        method: &str,
        path: &str,
        middleware: route_catalog::RouteMiddleware,
    ) -> bool {
        Self::discovered_route(method, path)
            .map(|matched| matched.spec.middlewares.contains(&middleware))
            .unwrap_or(false)
    }

    fn discovered_route_ops_policy(
        &self,
        method: &str,
        path: &str,
        headers: &BTreeMap<String, String>,
    ) -> Option<HttpResponse> {
        let matched = Self::discovered_route(method, path)?;
        for middleware in matched.spec.middlewares {
            if let route_catalog::RouteMiddleware::OpsPolicy(action) = *middleware {
                return self.check_ops_http_policy(
                    headers,
                    action,
                    Self::ops_policy_resource_name(matched.spec.id),
                );
            }
        }
        None
    }

    fn ops_policy_resource_name(route_id: &'static str) -> &'static str {
        match route_id {
            "admin.audit" => "audit",
            "admin.blob_cache.stats" => "blob-cache-stats",
            "admin.status" => "admin-status",
            "metrics.scrape" => "metrics",
            "ops.backup.status" => "backup-status",
            "ops.cluster.status" => "cluster-status",
            "ops.ec.status" => "eventual-consistency",
            "ops.replication.status" => "replication-status",
            "ops.topology.graph" => "topology-graph",
            _ => route_id,
        }
    }

    fn listener_surface(&self) -> route_catalog::ListenerSurface {
        match self.options.surface {
            crate::server::ServerSurface::Public => route_catalog::ListenerSurface::Public,
            crate::server::ServerSurface::AdminOnly => route_catalog::ListenerSurface::Admin,
            crate::server::ServerSurface::MetricsOnly => route_catalog::ListenerSurface::Metrics,
        }
    }

    /// PLAN.md Phase 6.2 — return whether `path` is reachable on
    /// this listener given its `ServerSurface` (Public / AdminOnly /
    /// MetricsOnly). Cataloged routes use their declared listener
    /// surfaces; legacy routes keep the old path-prefix fallback until
    /// their families migrate.
    fn surface_allows(&self, method: &str, path: &str) -> bool {
        if let Some(matched) = Self::discovered_route(method, path) {
            return matched.spec.surfaces.contains(&self.listener_surface());
        }

        match self.options.surface {
            crate::server::ServerSurface::Public => true,
            crate::server::ServerSurface::AdminOnly => {
                path.starts_with("/admin/")
                    || path == "/metrics"
                    || path == "/admin"
                    || path.starts_with("/health/")
                    || path == "/health"
            }
            crate::server::ServerSurface::MetricsOnly => {
                path == "/metrics"
                    || path == "/api/v1/query"
                    || path == "/api/v1/query_range"
                    || path == "/api/v1/write"
                    || path.starts_with("/health/")
                    || path == "/health"
            }
        }
    }

    /// Pre-IAM HTTP authorization: the catalog supplies the route's auth
    /// class and the decision is made from the presented HTTP credentials
    /// alone. `CommandAuthorizer` / `IamCommandPolicyEngine` are deliberately
    /// *not* engaged here — routing HTTP through the policy engine is a
    /// behavior change owned by a later slice of Spec #2109, and wiring it in
    /// nominally would only cost a per-request principal hash and context
    /// build whose result is discarded.
    fn is_authorized(&self, method: &str, path: &str, headers: &BTreeMap<String, String>) -> bool {
        // Set when the matched route is an operator (`AdminToken`) route and
        // no `RED_ADMIN_TOKEN` is configured: instead of staying open, the
        // route falls through to the user-auth chain below and additionally
        // demands an admin-role bearer.
        let mut admin_role_required = false;
        let mut bearer_required = false;
        if let Some(matched) = Self::discovered_route(method, path) {
            if matches!(
                matched.spec.auth,
                route_catalog::RouteAuth::Public | route_catalog::RouteAuth::OptionalUser
            ) {
                return true;
            }
            // Operator read surfaces (`ops:*` capabilities: replication and
            // cluster status, topology) never answer an anonymous caller,
            // even when the store runs with `require_auth = false` — that
            // flag opens the data plane, not the operator plane. The
            // capability itself is enforced by the IAM policy middleware
            // when policies are installed.
            if matches!(
                matched.spec.auth,
                route_catalog::RouteAuth::OpsCapability(_)
            ) {
                bearer_required = true;
            }
            if matches!(matched.spec.auth, route_catalog::RouteAuth::AdminToken) {
                if let Some(expected) = read_admin_token() {
                    let presented = headers
                        .get("authorization")
                        .and_then(|v| v.strip_prefix("Bearer "))
                        .unwrap_or("");
                    return crate::crypto::constant_time_eq(
                        presented.as_bytes(),
                        expected.as_bytes(),
                    );
                }
                if path.starts_with("/v1/_admin/") {
                    return false;
                }
                admin_role_required = true;
            }
        }

        // `red server --ui` (#1047, ADR 0051) — the served bundle is inert
        // static assets with no embedded credential, so they are public by
        // design even on an authed database: the browser loads them and
        // then authenticates against the RedWire data endpoint. Only a
        // `GET` that resolves to a *real file* in the bundle bypasses auth,
        // so an authenticated API route is never accidentally opened (it
        // has no matching file). The 401-vs-asset decision is made here,
        // before dispatch, to keep the gate in lock-step with `route`.
        // Operator surfaces (`/admin/*`, `/v1/_admin/*`, `/metrics`) are
        // excluded so a bundle file can never relax the admin-token gate.
        if method == "GET"
            && !path.starts_with("/admin/")
            && !path.starts_with("/v1/_admin/")
            && path != "/metrics"
        {
            if let Some(ui_dir) = self.ui_dir() {
                if crate::server::ui_static::bundle_asset_exists(ui_dir, path) {
                    return true;
                }
            }
        }

        // PLAN.md Phase 6.1 — operator endpoints (/admin/*, /metrics)
        // are gated by a separate `RED_ADMIN_TOKEN` independent of
        // the application auth store. When the env var (or its
        // `_FILE` companion) is unset, the endpoints stay open so
        // dev installs and dashboards keep working; once set, every
        // hit needs `Authorization: Bearer <token>` with a
        // constant-time match.
        let requires_admin_token = path.starts_with("/v1/_admin/");
        let is_admin_surface =
            path.starts_with("/admin/") || requires_admin_token || path == "/metrics";
        if is_admin_surface {
            if let Some(expected) = read_admin_token() {
                let presented = headers
                    .get("authorization")
                    .and_then(|v| v.strip_prefix("Bearer "))
                    .unwrap_or("");
                if !crate::crypto::constant_time_eq(presented.as_bytes(), expected.as_bytes()) {
                    return false;
                }
                // Admin token matches — bypass the rest of the auth
                // chain. Operators want /admin/* to keep working when
                // the user-auth backend is down for maintenance.
                return true;
            }
            if requires_admin_token {
                return false;
            }
            // No operator token configured: `/admin/*` and `/metrics` are
            // only reachable by an admin-role bearer (or when the user auth
            // store itself is absent / disabled, i.e. an explicitly
            // unauthenticated dev install).
            admin_role_required = true;
        }

        // Public endpoints that never require authentication.
        if matches!(
            (method, path),
            ("GET", "/health")
                | ("GET", "/ready")
                | ("GET", "/ready/query")
                | ("GET", "/ready/write")
                | ("GET", "/ready/repair")
                | ("GET", "/ready/serverless")
                | ("GET", "/ready/serverless/query")
                | ("GET", "/ready/serverless/write")
                | ("GET", "/ready/serverless/repair")
                | ("POST", "/auth/login")
                | ("POST", "/auth/bootstrap")
                // Browser credential layer (issue #936): login takes a
                // password, refresh/logout authenticate via the HttpOnly
                // refresh cookie — none of them carry a bearer token, so
                // they must bypass the bearer gate (they enforce their
                // own credentials inside the handler).
                | ("POST", "/auth/browser/login")
                | ("POST", "/auth/browser/refresh")
                | ("POST", "/auth/browser/logout")
                // #752 — feature/capability discovery. Both levels are
                // unauthenticated: the UI must be able to discover what
                // the server offers (and what an *anonymous* caller may
                // do) before holding a token. `/auth/capabilities`
                // upgrades its answer when a valid bearer is presented.
                | ("GET", "/capabilities")
                | ("GET", "/auth/capabilities")
        ) {
            return true;
        }

        // If no auth store is configured, all requests are permitted.
        let auth_store = match &self.auth_store {
            Some(store) => store,
            None => return true,
        };

        // If auth is disabled in the config, allow everything.
        if !auth_store.is_enabled() {
            return true;
        }

        // Extract bearer token from Authorization header. JWT-shaped
        // tokens are routed through the configured OAuthValidator
        // first; fallback to AuthStore lookup for opaque API keys /
        // session tokens.
        let token = headers
            .get("authorization")
            .and_then(|v| v.strip_prefix("Bearer "));

        match token {
            Some(tok) => {
                let role = match resolve_bearer_role(tok, &self.runtime, auth_store.as_ref()) {
                    BearerOutcome::Valid(role) => role,
                    BearerOutcome::Invalid => return false,
                    BearerOutcome::NoMatch => return false,
                };
                // POST /auth/can is a self-introspection probe — any
                // authenticated principal (including read-only) can ask
                // what they themselves are permitted to do.
                let is_self_probe = matches!((method, path), ("POST", "/auth/can"));
                let is_write = !matches!(method, "GET" | "HEAD" | "OPTIONS");
                if admin_role_required {
                    role.can_admin()
                } else if is_self_probe {
                    role.can_read()
                } else if is_write {
                    role.can_write()
                } else {
                    role.can_read()
                }
            }
            None => {
                // No token: allow only if require_auth is false, and never
                // for operator routes (those need an admin bearer when no
                // `RED_ADMIN_TOKEN` is configured).
                !admin_role_required && !bearer_required && !auth_store.config().require_auth
            }
        }
    }
}

/// Outcome of resolving a bearer token. Distinguishes a hard reject
/// (signature/issuer/expiry failure) from "no match in any store"
/// because the audit log treats them differently.
pub(crate) enum BearerOutcome {
    Valid(crate::auth::Role),
    /// JWT-shaped + OAuth validator rejected (issuer mismatch, expired,
    /// bad signature, audience mismatch). Keep distinct so callers can
    /// surface `WWW-Authenticate: Bearer error="invalid_token"`.
    Invalid,
    /// Neither the OAuth validator nor the AuthStore recognized the
    /// token. 401 with a generic error.
    NoMatch,
}

/// Returns true when `token` is shaped like a compact JWT
/// (`header.payload.signature` with three base64url segments).
/// Cheap structural check — does NOT validate the signature.
pub(crate) fn looks_like_jwt(token: &str) -> bool {
    let mut parts = token.split('.');
    let (Some(h), Some(p), Some(s)) = (parts.next(), parts.next(), parts.next()) else {
        return false;
    };
    if parts.next().is_some() {
        return false;
    }
    if h.is_empty() || p.is_empty() || s.is_empty() {
        return false;
    }
    let valid_b64url = |chunk: &str| {
        chunk
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
    };
    valid_b64url(h) && valid_b64url(p) && valid_b64url(s)
}

/// Bearer-token resolution shared by `is_authorized` and the
/// `/auth/whoami` handler. Tries OAuth JWT validation first when the
/// token has JWT shape AND the runtime has an `OAuthValidator`
/// configured; falls back to `AuthStore::validate_token` for opaque
/// tokens (API keys, session tokens) and JWT-shaped tokens when no
/// validator is wired.
pub(crate) fn resolve_bearer_role(
    token: &str,
    runtime: &crate::runtime::RedDBRuntime,
    auth_store: &crate::auth::store::AuthStore,
) -> BearerOutcome {
    let token_fp = token_fingerprint(token);
    if looks_like_jwt(token) {
        if let Some(validator) = runtime.oauth_validator() {
            match crate::wire::redwire::auth::validate_oauth_jwt(&validator, token) {
                Ok((username, role)) => {
                    // F-04: `username` is a JWT-claim string in the
                    // federated case, attacker-influenceable. Strip
                    // CR/LF/control bytes through the LogField escaper.
                    tracing::info!(
                        target: "reddb::http_auth",
                        method = "oauth_jwt",
                        user = %reddb_wire::audit_safe_log_field(&username),
                        token_sha256 = %token_fp,
                        "JWT accepted"
                    );
                    return BearerOutcome::Valid(role);
                }
                Err(reason) => {
                    // F-04: `reason` quotes parts of the token; sanitize.
                    tracing::warn!(
                        target: "reddb::http_auth",
                        method = "oauth_jwt",
                        token_sha256 = %token_fp,
                        reason = %reddb_wire::audit_safe_log_field(&reason),
                        "JWT rejected"
                    );
                    return BearerOutcome::Invalid;
                }
            }
        }
        // JWT-shaped but no validator: fall through to AuthStore. This
        // matches the behaviour where an operator hands out long-lived
        // base64-segmented API keys that happen to look like JWTs.
    }
    match auth_store.validate_token(token) {
        Some((_username, role)) => BearerOutcome::Valid(role),
        None => {
            tracing::debug!(
                target: "reddb::http_auth",
                method = "bearer",
                token_sha256 = %token_fp,
                "bearer token unknown"
            );
            BearerOutcome::NoMatch
        }
    }
}

fn token_fingerprint(token: &str) -> String {
    let digest = crate::crypto::sha256(token.as_bytes());
    crate::utils::to_hex_prefix(&digest, 8)
}

// Chain-integrity + capacity-audit handlers (relocated in routing 3/3,
// #1643) so they sit with their handler siblings above the test module
// rather than below it.
/// Issue #524 — `GET /collections/:name/chain-tip`. Returns the cached chain
/// tip JSON. 404 when the collection is not a `KIND blockchain` or has no
/// rows yet (the engine guarantees a genesis row on creation, so the 404
/// branch effectively means "wrong kind / collection absent").
pub(crate) fn handle_chain_tip(
    runtime: &crate::runtime::RedDBRuntime,
    collection: &str,
) -> HttpResponse {
    let Some(tip) = runtime.chain_tip_for_collection(collection) else {
        return json_error(
            404,
            format!("chain-tip: collection '{collection}' is not a blockchain or has no rows"),
        );
    };
    let server_time = crate::runtime::blockchain_kind::now_ms();
    let mut hex = String::with_capacity(64);
    for b in tip.hash.iter() {
        hex.push_str(&format!("{b:02x}"));
    }
    let mut obj = crate::json::Map::new();
    obj.insert(
        "block_height".to_string(),
        crate::json::Value::Number(tip.height as f64),
    );
    obj.insert("hash".to_string(), crate::json::Value::String(hex));
    obj.insert(
        "timestamp".to_string(),
        crate::json::Value::Number(tip.timestamp_ms as f64),
    );
    obj.insert(
        "server_time".to_string(),
        crate::json::Value::Number(server_time as f64),
    );
    json_response(200, crate::json::Value::Object(obj))
}

/// Issue #525 — admin-token gate for verify-chain + clear-integrity endpoints.
/// When `RED_ADMIN_TOKEN` is unset the endpoints stay open (dev installs).
/// When set, callers must present a matching `Authorization: Bearer <token>`.
pub(crate) fn admin_token_ok(headers: &BTreeMap<String, String>) -> bool {
    let Some(expected) = read_admin_token() else {
        return true;
    };
    let presented = headers
        .get("authorization")
        .and_then(|v| v.strip_prefix("Bearer "))
        .unwrap_or("");
    crate::crypto::constant_time_eq(presented.as_bytes(), expected.as_bytes())
}

/// Issue #525 — `POST /collections/:name/verify-chain`.
pub(crate) fn handle_verify_chain(
    runtime: &crate::runtime::RedDBRuntime,
    collection: &str,
) -> HttpResponse {
    let Some(outcome) = runtime.verify_chain_for_collection(collection) else {
        return json_error(
            404,
            format!(
                "verify-chain: collection '{collection}' is not a blockchain or does not exist"
            ),
        );
    };
    let mut obj = crate::json::Map::new();
    obj.insert(
        "checked".to_string(),
        crate::json::Value::Number(outcome.checked as f64),
    );
    obj.insert("ok".to_string(), crate::json::Value::Bool(outcome.ok));
    obj.insert(
        "first_bad_height".to_string(),
        match outcome.first_bad_height {
            Some(h) => crate::json::Value::Number(h as f64),
            None => crate::json::Value::Null,
        },
    );
    json_response(200, crate::json::Value::Object(obj))
}

/// Issue #525 — `POST /collections/:name/clear-integrity-flag`.
pub(crate) fn handle_clear_integrity_flag(
    runtime: &crate::runtime::RedDBRuntime,
    collection: &str,
) -> HttpResponse {
    if !runtime.clear_chain_integrity_flag(collection) {
        return json_error(
            404,
            format!(
                "clear-integrity-flag: collection '{collection}' is not a blockchain or does not exist"
            ),
        );
    }
    let mut obj = crate::json::Map::new();
    obj.insert("ok".to_string(), crate::json::Value::Bool(true));
    obj.insert(
        "collection".to_string(),
        crate::json::Value::String(collection.to_string()),
    );
    json_response(200, crate::json::Value::Object(obj))
}

/// Issue #767 / S8 — record a `stream.closed { reason: capacity_refused }`
/// audit event whenever the capacity guard refuses an open. The brief
/// requires audit coverage of every state transition; capacity_refused
/// is the only one that never produces a lease handle.
fn emit_capacity_refused_audit(
    runtime: &crate::runtime::RedDBRuntime,
    principal: &str,
    err: &crate::server::output_stream::AcquireError,
) {
    use crate::server::output_stream::AcquireError;
    let (limit, current) = match err {
        AcquireError::GlobalExhausted { limit, current } => (*limit, *current),
        AcquireError::PrincipalExhausted { limit, current, .. } => (*limit, *current),
    };
    crate::server::output_stream::audit_stream_capacity_refused(
        runtime,
        principal,
        err.code(),
        limit,
        current,
    );
}

#[cfg(test)]
mod tests {
    use super::*;

    fn request(path: &str) -> HttpRequest {
        request_with("GET", path, Vec::new())
    }

    fn request_with(method: &str, path: &str, body: Vec<u8>) -> HttpRequest {
        HttpRequest {
            method: method.to_string(),
            path: path.to_string(),
            query: BTreeMap::new(),
            headers: BTreeMap::new(),
            body,
        }
    }

    fn request_with_bearer(method: &str, path: &str, body: Vec<u8>, token: &str) -> HttpRequest {
        let mut request = request_with(method, path, body);
        request
            .headers
            .insert("authorization".to_string(), format!("Bearer {token}"));
        request
    }

    fn server_with_open_store() -> (RedDBServer, std::sync::Arc<crate::auth::store::AuthStore>) {
        let store = std::sync::Arc::new(crate::auth::store::AuthStore::new(
            crate::auth::AuthConfig {
                enabled: true,
                session_ttl_secs: 60,
                require_auth: false,
                auto_encrypt_storage: false,
                vault_enabled: false,
                cert: Default::default(),
                oauth: Default::default(),
            },
        ));
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        runtime.set_auth_store(std::sync::Arc::clone(&store));
        let server = RedDBServer::new(runtime).with_auth(std::sync::Arc::clone(&store));
        (server, store)
    }

    #[test]
    fn operator_read_surfaces_never_answer_anonymous_callers() {
        let (server, store) = server_with_open_store();
        // `require_auth = false` opens the data plane; the operator plane
        // (cluster status, replication status, topology) still needs a
        // credential.
        for path in [
            "/cluster/status",
            "/replication/status",
            "/v1/topology/graph",
        ] {
            let response = server.route(request(path));
            assert_eq!(response.status, 401, "{path} answered an anonymous caller");
        }
        store
            .create_user("viewer", "pw", crate::auth::Role::Read)
            .expect("user");
        let session = store.authenticate("viewer", "pw").expect("login");
        let response = server.route(request_with_bearer(
            "GET",
            "/cluster/status",
            Vec::new(),
            &session.token,
        ));
        assert_ne!(
            response.status, 401,
            "a read bearer is enough for a read surface"
        );
        assert_ne!(response.status, 403);
    }

    #[test]
    fn credential_posts_require_a_json_content_type() {
        let (server, _store) = server_with_open_store();
        let body = br#"{"username":"nobody","password":"x"}"#.to_vec();
        let response = server.route(request_with("POST", "/auth/login", body.clone()));
        assert_eq!(
            response.status, 415,
            "a bodied POST without a JSON content type"
        );
        let mut request = request_with("POST", "/auth/login", body);
        request.headers.insert(
            "content-type".to_string(),
            "application/json; charset=utf-8".to_string(),
        );
        let response = server.route(request);
        assert_ne!(response.status, 415);
    }

    #[test]
    fn http_request_counter_collapses_raw_paths_to_route_template() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        // Two distinct high-cardinality paths under the same dynamic route
        // must both fold to the single `/catalog/collections/:name` label —
        // the raw collection names never appear as labels. Status class is
        // left unasserted (the collections may not exist); the point is the
        // route dimension collapses regardless of outcome.
        let _ = server.route(request("/catalog/collections/users"));
        let _ = server.route(request("/catalog/collections/orders"));

        let snapshot = server.http_request_metrics().snapshot();
        let collapsed: u64 = snapshot
            .iter()
            .filter(|(labels, _)| {
                labels.method == "GET" && labels.route == "/catalog/collections/:name"
            })
            .map(|(_, count)| *count)
            .sum();
        assert_eq!(
            collapsed, 2,
            "both raw collection paths must collapse to one route-template series: {snapshot:?}"
        );

        // The raw paths must not have leaked into the snapshot as labels.
        assert!(
            snapshot
                .iter()
                .all(|(labels, _)| !labels.route.contains("users")
                    && !labels.route.contains("orders")),
            "raw collection names must never become route labels: {snapshot:?}"
        );
    }

    #[test]
    fn http_request_counter_records_unmatched_paths_in_reserved_bucket() {
        use crate::server::http_request_metrics::{HttpRequestLabels, UNMATCHED_ROUTE};

        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        // Unknown high-cardinality paths must fold to the reserved bucket so
        // a 404 scan cannot blow up the `route` dimension.
        let _ = server.route(request("/no/such/path/aaaa"));
        let _ = server.route(request("/no/such/path/bbbb"));

        let unmatched = server.http_request_metrics().count(HttpRequestLabels {
            method: "GET",
            route: UNMATCHED_ROUTE,
            status: "4xx",
        });
        assert_eq!(
            unmatched, 2,
            "unmatched paths collapse to one bounded bucket"
        );
    }

    #[test]
    fn principal_inflight_refusal_has_structured_backoff_shape() {
        use crate::server::http_principal_limiter::PrincipalCapExceeded;

        let err = PrincipalCapExceeded {
            // Embed a quote-bearing principal so the test also proves the
            // tainted-field path escapes hostile header-derived labels.
            principal: "replica:ev\"il".to_string(),
            limit: 64,
            current: 64,
        };
        let response = principal_inflight_refusal_response(&err, 5);
        assert_eq!(response.status, 429);
        assert_eq!(response.content_type, "application/json");

        // `Retry-After: 5` rides as a guard-validated header so clients
        // with only header-level backoff still get a hint.
        assert!(
            response
                .extra_headers
                .iter()
                .any(|(name, value)| name.eq_ignore_ascii_case("Retry-After")
                    && value.to_str().ok() == Some("5")),
            "expected Retry-After: 5, got {:?}",
            response.extra_headers
        );

        let body: crate::json::Value =
            crate::json::from_slice(&response.body).expect("refusal body is valid JSON");
        assert_eq!(body["ok"], crate::json::Value::Bool(false));
        assert_eq!(body["retry_after_secs"].as_i64(), Some(5));
        let error = &body["error"];
        assert_eq!(
            error["code"],
            crate::json::Value::String(
                crate::server::http_principal_limiter::PRINCIPAL_INFLIGHT_CODE.to_string()
            )
        );
        assert_eq!(error["limit"].as_i64(), Some(64));
        assert_eq!(error["current"].as_i64(), Some(64));
        // The principal round-trips intact through the tainted-field escape.
        assert_eq!(
            error["principal"],
            crate::json::Value::String("replica:ev\"il".to_string())
        );
    }

    #[test]
    fn fresh_server_health_is_ready_when_query_endpoint_works() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let query = server.route(request_with(
            "POST",
            "/query",
            br#"{"query":"SELECT 1"}"#.to_vec(),
        ));
        assert_eq!(
            query.status,
            200,
            "fresh server should accept queries: {}",
            String::from_utf8_lossy(&query.body)
        );

        let health = server.route(request("/health"));
        let health_body = String::from_utf8_lossy(&health.body);
        assert_eq!(
            health.status, 200,
            "fresh server should not report degraded health after queries work: {}",
            health_body
        );
        assert!(
            health_body.contains("\"state\":\"healthy\""),
            "fresh server should report healthy state after queries work: {health_body}"
        );
        assert!(
            health_body.contains("\"issues\":[]"),
            "fresh server should not report startup drift issues after queries work: {health_body}"
        );

        let ready = server.route(request("/ready"));
        assert_eq!(
            ready.status,
            200,
            "fresh server should be ready after queries work: {}",
            String::from_utf8_lossy(&ready.body)
        );

        let query_ready = server.route(request("/ready/query"));
        assert_eq!(
            query_ready.status,
            200,
            "fresh server should report query readiness after /query works: {}",
            String::from_utf8_lossy(&query_ready.body)
        );
    }

    #[test]
    fn discovered_lifecycle_health_routes_dispatch() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        for path in ["/health/live", "/health/ready", "/health/startup"] {
            let response = server.route(request(path));
            assert_eq!(
                response.status,
                200,
                "expected discovered health route {path} to dispatch: {}",
                String::from_utf8_lossy(&response.body)
            );
        }
    }

    #[test]
    fn canonical_v1_alias_rewrites_to_legacy_dispatch_path() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let response = server.route(request("/v1/capabilities"));

        assert_eq!(
            response.status,
            200,
            "expected /v1/capabilities alias to dispatch to /capabilities: {}",
            String::from_utf8_lossy(&response.body)
        );
    }

    #[test]
    fn dynamic_alias_rewrites_with_captured_params() {
        assert_eq!(
            RedDBServer::route_alias_target("GET", "/v1/ai/models/embedding-small"),
            Some("/ai/models/embedding-small".to_string())
        );
    }

    #[test]
    fn auth_routes_dispatch_from_catalog() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let matched = RedDBServer::discovered_route("POST", "/auth/login")
            .expect("auth login route should be cataloged");
        assert_eq!(matched.spec.id, "auth.login");

        let mut login = request_with(
            "POST",
            "/v1/auth/login",
            br#"{"username":"admin","password":"pw"}"#.to_vec(),
        );
        login
            .headers
            .insert("content-type".to_string(), "application/json".to_string());
        let response = server.route(login);
        assert_eq!(
            response.status,
            501,
            "canonical auth alias should dispatch to auth handler: {}",
            String::from_utf8_lossy(&response.body)
        );
    }

    #[test]
    fn dynamic_auth_alias_rewrites_with_captured_params() {
        assert_eq!(
            RedDBServer::route_alias_target("DELETE", "/v1/auth/users/cloud-admin"),
            Some("/auth/users/cloud-admin".to_string())
        );
        assert_eq!(
            RedDBServer::route_alias_target("DELETE", "/v1/auth/tenants/acme/users/alice"),
            Some("/auth/tenants/acme/users/alice".to_string())
        );
    }

    #[test]
    fn query_routes_dispatch_from_catalog_aliases() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let response = server.route(request("/v1/query"));

        assert_eq!(
            response.status,
            405,
            "canonical query discovery alias should dispatch through the catalog: {}",
            String::from_utf8_lossy(&response.body)
        );
        assert!(
            response
                .extra_headers
                .iter()
                .any(|(name, value)| *name == "Allow" && value == "POST"),
            "GET /v1/query should keep the query discovery Allow header"
        );
    }

    /// The two commands the catalog declares without a buffered handler
    /// (`streams.input`, `streams.query.output`) are served here instead —
    /// `try_route_streaming` answers them before buffered dispatch runs.
    #[test]
    fn streaming_served_commands_are_recognized_by_the_streaming_entry() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let mut input = request_with("POST", "/streams/input", b"{}\n".to_vec());
        input.headers.insert(
            "content-type".to_string(),
            "application/x-ndjson".to_string(),
        );
        assert!(server.is_streaming_request(&input));

        let output = request_with(
            "POST",
            "/query/stream",
            br#"{"query":"SELECT 1 as n"}"#.to_vec(),
        );
        assert!(server.is_streaming_request(&output));
    }

    #[test]
    fn streaming_detection_recognizes_canonical_query_alias() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);
        let mut request = request_with(
            "POST",
            "/v1/query",
            br#"{"query":"SELECT 1 as n"}"#.to_vec(),
        );
        request
            .headers
            .insert("accept".to_string(), "application/x-ndjson".to_string());

        assert!(server.is_streaming_request(&request));
    }

    #[test]
    fn metrics_route_dispatches_from_catalog() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let response = server.route(request("/metrics"));

        assert_eq!(
            response.status,
            200,
            "expected catalog-dispatched /metrics response: {}",
            String::from_utf8_lossy(&response.body)
        );
    }

    #[test]
    fn prometheus_namespace_alias_rewrites_to_grafana_compatible_path() {
        assert_eq!(
            RedDBServer::route_alias_target("GET", "/prometheus/api/v1/query"),
            Some("/api/v1/query".to_string())
        );
    }

    #[test]
    fn get_grpc_explains_grpc_contract() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let response = server.route(request("/grpc"));

        assert_eq!(
            response.status,
            200,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        let json: crate::json::Value =
            crate::json::from_slice(&response.body).expect("grpc discovery json");
        assert_eq!(json.get("service"), Some(&crate::json!("reddb.v1.RedDB")));
        assert!(
            json.get("examples")
                .and_then(crate::json::Value::as_object)
                .and_then(|examples| examples.get("query"))
                .and_then(crate::json::Value::as_str)
                .unwrap_or_default()
                .contains("grpcurl"),
            "GET /grpc should advertise a grpcurl query example: {json:?}"
        );
    }

    #[test]
    fn get_query_explains_post_contract() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let response = server.route(request("/query"));

        assert_eq!(
            response.status,
            405,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        assert!(
            response
                .extra_headers
                .iter()
                .any(|(name, value)| *name == "Allow" && value == "POST"),
            "GET /query should advertise Allow: POST"
        );
        let json: crate::json::Value =
            crate::json::from_slice(&response.body).expect("query contract json");
        assert_eq!(json.get("ok"), Some(&crate::json!(false)));
        assert_eq!(json.get("code"), Some(&crate::json!("method_not_allowed")));
        assert!(
            json.get("examples")
                .and_then(crate::json::Value::as_object)
                .and_then(|examples| examples.get("json_query"))
                .and_then(crate::json::Value::as_str)
                .unwrap_or_default()
                .contains("SELECT 1"),
            "GET /query should teach the minimal POST /query shape: {json:?}"
        );
    }

    #[test]
    fn root_endpoint_is_self_describing_for_first_run() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let response = server.route(request("/"));

        assert_eq!(
            response.status,
            200,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        let json: crate::json::Value =
            crate::json::from_slice(&response.body).expect("root discovery json");
        assert_eq!(json.get("service"), Some(&crate::json!("reddb")));
        assert_eq!(json.get("ok"), Some(&crate::json!(true)));
        assert!(
            json.get("endpoints")
                .and_then(crate::json::Value::as_object)
                .and_then(|endpoints| endpoints.get("query"))
                .and_then(crate::json::Value::as_str)
                == Some("POST /query"),
            "root discovery should advertise /query: {json:?}"
        );
        assert!(
            json.get("examples")
                .and_then(crate::json::Value::as_object)
                .and_then(|examples| examples.get("http_json_query"))
                .and_then(crate::json::Value::as_str)
                .unwrap_or_default()
                .contains("SELECT 1"),
            "root discovery should include a minimal query example: {json:?}"
        );
    }

    #[test]
    fn deprecated_catalog_endpoint_returns_deprecation_headers() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let response = server.route(request("/catalog/indexes/status"));

        assert_eq!(response.status, 200);
        assert!(response.extra_headers.iter().any(|(name, value)| {
            *name == "Deprecation" && value.to_str().ok() == Some(CATALOG_DEPRECATION_DATE)
        }));
        assert!(response.extra_headers.iter().any(|(name, value)| {
            *name == "Sunset" && value.to_str().ok() == Some(CATALOG_DEPRECATION_DATE)
        }));
    }

    #[test]
    fn canonical_catalog_endpoint_is_not_deprecated() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let response = server.route(request("/catalog"));

        assert_eq!(response.status, 200);
        assert!(response
            .extra_headers
            .iter()
            .all(|(name, _)| *name != "Deprecation" && *name != "Sunset"));
    }

    #[test]
    fn admin_users_requires_shared_secret_and_creates_regular_user() {
        let _guard = crate::server::RED_ADMIN_TOKEN_ENV_LOCK
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let previous = std::env::var_os("RED_ADMIN_TOKEN");
        std::env::set_var("RED_ADMIN_TOKEN", "admin-secret");

        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let auth = std::sync::Arc::new(crate::auth::AuthStore::new(crate::auth::AuthConfig {
            enabled: true,
            require_auth: true,
            ..crate::auth::AuthConfig::default()
        }));
        let server = RedDBServer::new(runtime).with_auth(std::sync::Arc::clone(&auth));
        let body =
            br#"{"username":"system","password":"pw","role":"admin","tenant_id":"acme"}"#.to_vec();

        let unauthorized = server.route(request_with("POST", "/v1/_admin/users", body.clone()));
        assert_eq!(unauthorized.status, 401);

        let created = server.route(request_with_bearer(
            "POST",
            "/v1/_admin/users",
            body,
            "admin-secret",
        ));
        assert_eq!(
            created.status,
            201,
            "{}",
            String::from_utf8_lossy(&created.body)
        );

        let user = auth.get_user(Some("acme"), "system").unwrap();
        assert_eq!(user.role, crate::auth::Role::Admin);

        let alias_body =
            br#"{"username":"alias-system","password":"pw","role":"admin","tenant_id":"acme"}"#
                .to_vec();
        let alias_created = server.route(request_with_bearer(
            "POST",
            "/v1/admin/system-users",
            alias_body,
            "admin-secret",
        ));
        assert_eq!(
            alias_created.status,
            201,
            "{}",
            String::from_utf8_lossy(&alias_created.body)
        );
        let alias_user = auth.get_user(Some("acme"), "alias-system").unwrap();
        assert_eq!(alias_user.role, crate::auth::Role::Admin);

        match previous {
            Some(value) => std::env::set_var("RED_ADMIN_TOKEN", value),
            None => std::env::remove_var("RED_ADMIN_TOKEN"),
        }
    }

    #[test]
    fn v1_keyed_routes_split_kv_config_and_vault_domains() {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime);

        let kv_put = server.route(request_with(
            "PUT",
            "/v1/kv/sessions/token",
            br#"{"value":"abc","ttl_ms":1000}"#.to_vec(),
        ));
        assert_eq!(
            kv_put.status,
            200,
            "{}",
            String::from_utf8_lossy(&kv_put.body)
        );

        let kv_get = server.route(request("/v1/kv/sessions/token"));
        assert_eq!(kv_get.status, 200);
        let body = String::from_utf8_lossy(&kv_get.body);
        assert!(body.contains("\"collection\":\"sessions\""), "{body}");
        assert!(body.contains("\"key\":\"token\""), "{body}");

        let config_put = server.route(request_with(
            "PUT",
            "/v1/config/app/feature",
            br#"{"value":"on"}"#.to_vec(),
        ));
        assert_eq!(
            config_put.status,
            200,
            "{}",
            String::from_utf8_lossy(&config_put.body)
        );

        let config_ttl = server.route(request_with(
            "PUT",
            "/v1/config/app/temporary",
            br#"{"value":"on","ttl_ms":1000}"#.to_vec(),
        ));
        assert_eq!(config_ttl.status, 400);
        assert!(
            String::from_utf8_lossy(&config_ttl.body).contains("CONFIG does not support TTL"),
            "{}",
            String::from_utf8_lossy(&config_ttl.body)
        );

        let vault_counter = server.route(request_with(
            "POST",
            "/v1/vault/secrets/api_key/incr",
            br#"{"by":1}"#.to_vec(),
        ));
        assert_eq!(vault_counter.status, 405);
    }

    // ---------- Issue #736 — Red UI collection metadata contract ----------
    //
    // Contract tests for `GET /catalog/collections/{name}`. Each test
    // creates a collection of a given model kind, fetches the
    // metadata, and pins the shape the Red UI relies on:
    //   - `model` + `primary_capability` mirror the model kind
    //   - `capabilities[]` lists the action surfaces for that model
    //   - `schema`, `indexes`, `retention`, `tenant_scope`,
    //     `entity_count`, `model_specific`, `actions` all present
    //   - empty collections still classify by model without probing
    //   - not-found returns 404 with the authorization-safe body
    //
    // The runtime is in-memory and has no auth store wired, so the
    // server's principal resolver returns `(None, false)` and the
    // coarse action gates default to "allowed". That is the right
    // contract for "auth not configured" mode.

    fn ui_metadata(server: &RedDBServer, name: &str) -> crate::json::Value {
        let response = server.route(request(&format!("/catalog/collections/{name}")));
        assert_eq!(
            response.status,
            200,
            "ui_metadata expected 200 for {name}: {}",
            String::from_utf8_lossy(&response.body)
        );
        crate::json::from_slice(&response.body).expect("ui metadata JSON")
    }

    fn ddl(server: &RedDBServer, sql: &str) {
        let response = server.route(request_with(
            "POST",
            "/query",
            format!("{{\"query\":{}}}", crate::json!(sql)).into_bytes(),
        ));
        assert!(
            (200..300).contains(&response.status),
            "DDL `{sql}` failed: {}",
            String::from_utf8_lossy(&response.body)
        );
    }

    fn assert_metadata_envelope(payload: &crate::json::Value, name: &str, model: &str) {
        assert_eq!(payload.get("ok"), Some(&crate::json!(true)));
        let collection = payload.get("collection").expect("collection object");
        assert_eq!(collection.get("name"), Some(&crate::json!(name)));
        assert_eq!(collection.get("model"), Some(&crate::json!(model)));
        assert_eq!(
            collection.get("primary_capability"),
            Some(&crate::json!(model))
        );
        let caps = collection
            .get("capabilities")
            .and_then(crate::json::Value::as_array)
            .expect("capabilities array");
        assert!(
            !caps.is_empty(),
            "capabilities must be non-empty for {model}"
        );
        assert!(caps.iter().any(|c| c.as_str() == Some("describe")));
        assert!(collection.get("schema").is_some(), "schema object missing");
        assert!(
            collection.get("indexes").is_some(),
            "indexes object missing"
        );
        assert!(collection.get("retention").is_some(), "retention missing");
        assert!(
            collection.get("tenant_scope").is_some(),
            "tenant_scope missing"
        );
        assert!(
            collection.get("entity_count").is_some(),
            "entity_count missing"
        );
        let actions = collection
            .get("actions")
            .and_then(crate::json::Value::as_object)
            .expect("actions object");
        for required in [
            "read",
            "write",
            "delete",
            "drop_collection",
            "alter_collection",
        ] {
            assert!(
                actions.contains_key(required),
                "actions.{required} missing for {model}"
            );
        }
        let alter = actions.get("alter_collection").unwrap();
        assert_eq!(
            alter.get("allowed"),
            Some(&crate::json!("unknown")),
            "alter_collection must be unknown until #740/#741: {alter}"
        );
        let tenant = collection.get("tenant_scope").unwrap();
        assert_eq!(tenant.get("kind"), Some(&crate::json!("unknown")));
    }

    fn fresh_server() -> RedDBServer {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        RedDBServer::new(runtime)
    }

    #[test]
    fn ui_metadata_table_contract() {
        let server = fresh_server();
        ddl(
            &server,
            "CREATE TABLE accounts (id INTEGER PRIMARY KEY, name TEXT)",
        );
        let payload = ui_metadata(&server, "accounts");
        assert_metadata_envelope(&payload, "accounts", "table");

        let collection = payload.get("collection").unwrap();
        // Empty table classifies as `model=table` without probing rows.
        assert_eq!(
            collection
                .get("entity_count")
                .and_then(crate::json::Value::as_i64),
            Some(0)
        );
        // Strict schema produces a typed column list from the contract.
        let schema = collection.get("schema").unwrap();
        let mode = schema
            .get("mode")
            .and_then(crate::json::Value::as_str)
            .expect("schema.mode");
        assert!(
            matches!(mode, "strict" | "semi_structured"),
            "table schema.mode unexpected: {mode}"
        );
        let columns = schema
            .get("columns")
            .and_then(crate::json::Value::as_array)
            .expect("schema.columns must be populated for tables");
        assert!(!columns.is_empty(), "table columns must be exposed");
        // Capability tags include table-specific actions.
        let caps: Vec<&str> = collection
            .get("capabilities")
            .and_then(crate::json::Value::as_array)
            .unwrap()
            .iter()
            .filter_map(|c| c.as_str())
            .collect();
        for required in ["select", "insert", "update", "delete", "create_index"] {
            assert!(
                caps.contains(&required),
                "table capabilities missing {required}: {caps:?}"
            );
        }
    }

    #[test]
    fn ui_metadata_document_contract() {
        let server = fresh_server();
        ddl(&server, "CREATE DOCUMENT notes");
        let payload = ui_metadata(&server, "notes");
        assert_metadata_envelope(&payload, "notes", "document");
        let caps: Vec<&str> = payload
            .get("collection")
            .unwrap()
            .get("capabilities")
            .and_then(crate::json::Value::as_array)
            .unwrap()
            .iter()
            .filter_map(|c| c.as_str())
            .collect();
        assert!(caps.contains(&"json_path"));
    }

    #[test]
    fn ui_metadata_queue_contract() {
        let server = fresh_server();
        ddl(&server, "CREATE QUEUE jobs");
        let payload = ui_metadata(&server, "jobs");
        assert_metadata_envelope(&payload, "jobs", "queue");

        let collection = payload.get("collection").unwrap();
        let actions = collection
            .get("actions")
            .and_then(crate::json::Value::as_object)
            .unwrap();
        for queue_action in ["push", "peek", "ack", "nack", "purge", "dlq_move"] {
            assert!(
                actions.contains_key(queue_action),
                "queue action {queue_action} missing"
            );
        }
        let model_specific = collection.get("model_specific").unwrap();
        assert!(
            model_specific.get("queue_mode").is_some(),
            "queue_mode must be exposed"
        );
    }

    #[test]
    fn ui_metadata_vector_contract() {
        let server = fresh_server();
        ddl(
            &server,
            "CREATE COLLECTION embeddings KIND vector.turbo DIM 8 METRIC cosine",
        );
        let payload = ui_metadata(&server, "embeddings");
        assert_metadata_envelope(&payload, "embeddings", "vector");
        let model_specific = payload
            .get("collection")
            .unwrap()
            .get("model_specific")
            .unwrap();
        assert_eq!(
            model_specific
                .get("dimension")
                .and_then(crate::json::Value::as_i64),
            Some(8)
        );
        assert_eq!(model_specific.get("metric"), Some(&crate::json!("cosine")));
        let actions = payload
            .get("collection")
            .unwrap()
            .get("actions")
            .and_then(crate::json::Value::as_object)
            .unwrap();
        assert!(actions.contains_key("search"));
        assert!(actions.contains_key("upsert"));
    }

    #[test]
    fn ui_metadata_graph_contract() {
        let server = fresh_server();
        ddl(&server, "CREATE GRAPH social");
        let payload = ui_metadata(&server, "social");
        assert_metadata_envelope(&payload, "social", "graph");
        let actions = payload
            .get("collection")
            .unwrap()
            .get("actions")
            .and_then(crate::json::Value::as_object)
            .unwrap();
        assert!(actions.contains_key("traverse"));
        assert!(actions.contains_key("subgraph"));
    }

    #[test]
    fn ui_metadata_kv_contract() {
        let server = fresh_server();
        ddl(&server, "CREATE KV sessions");
        let payload = ui_metadata(&server, "sessions");
        assert_metadata_envelope(&payload, "sessions", "kv");
        let actions = payload
            .get("collection")
            .unwrap()
            .get("actions")
            .and_then(crate::json::Value::as_object)
            .unwrap();
        for kv_action in ["increment", "compare_and_set", "list_by_prefix"] {
            assert!(
                actions.contains_key(kv_action),
                "kv action {kv_action} missing"
            );
        }
    }

    #[test]
    fn ui_metadata_timeseries_contract() {
        let server = fresh_server();
        ddl(&server, "CREATE TIMESERIES events");
        let payload = ui_metadata(&server, "events");
        assert_metadata_envelope(&payload, "events", "timeseries");
        let model_specific = payload
            .get("collection")
            .unwrap()
            .get("model_specific")
            .unwrap();
        assert!(model_specific.get("session_key").is_some());
        assert!(model_specific.get("session_gap_ms").is_some());
    }

    #[test]
    fn ui_metadata_empty_collection_classifies_without_rows() {
        // Acceptance criterion: empty collections classify by model
        // without probing data rows. We never insert any rows.
        let server = fresh_server();
        ddl(&server, "CREATE TABLE empty_t (id INTEGER PRIMARY KEY)");
        ddl(&server, "CREATE QUEUE empty_q");
        ddl(&server, "CREATE KV empty_kv");

        for (name, model) in [
            ("empty_t", "table"),
            ("empty_q", "queue"),
            ("empty_kv", "kv"),
        ] {
            let payload = ui_metadata(&server, name);
            let collection = payload.get("collection").unwrap();
            assert_eq!(collection.get("model"), Some(&crate::json!(model)));
            assert_eq!(
                collection
                    .get("entity_count")
                    .and_then(crate::json::Value::as_i64),
                Some(0),
                "empty collection {name} should report 0 entities without probing"
            );
        }
    }

    #[test]
    fn ui_metadata_not_found_is_authorization_safe() {
        let server = fresh_server();
        let response = server.route(request("/catalog/collections/does_not_exist"));
        assert_eq!(response.status, 404);
        let body: crate::json::Value =
            crate::json::from_slice(&response.body).expect("not-found JSON");
        // Authorization-safe envelope: identical for "absent" and
        // "not-visible-to-this-principal" so the endpoint cannot be
        // used to enumerate collections by probing names.
        assert_eq!(body.get("ok"), Some(&crate::json!(false)));
        assert_eq!(
            body.get("error"),
            Some(&crate::json!("not_found_or_not_visible"))
        );
    }

    // -----------------------------------------------------------------
    // #740 — Red UI security read contracts: /auth/tenants,
    // /auth/policies, /auth/can.
    // -----------------------------------------------------------------

    fn auth_server(require_auth: bool) -> (RedDBServer, std::sync::Arc<crate::auth::AuthStore>) {
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let auth = std::sync::Arc::new(crate::auth::AuthStore::new(crate::auth::AuthConfig {
            enabled: true,
            require_auth,
            ..crate::auth::AuthConfig::default()
        }));
        let server = RedDBServer::new(runtime).with_auth(std::sync::Arc::clone(&auth));
        (server, auth)
    }

    fn issue_token(
        auth: &crate::auth::AuthStore,
        tenant: Option<&str>,
        username: &str,
        role: crate::auth::Role,
    ) -> String {
        auth.create_user_in_tenant(tenant, username, "pw", role)
            .expect("create user");
        auth.authenticate_in_tenant(tenant, username, "pw")
            .expect("authenticate")
            .token
    }

    fn parse_json(body: &[u8]) -> crate::json::Value {
        crate::json::from_slice(body).expect("json parse")
    }

    fn attach_policy_json(auth: &crate::auth::AuthStore, user: &str, policy_json: &str) {
        let policy = crate::auth::policies::Policy::from_json_str(policy_json).unwrap();
        let policy_id = policy.id.clone();
        auth.put_policy(policy).unwrap();
        auth.attach_policy(
            crate::auth::store::PrincipalRef::User(crate::auth::UserId::platform(user)),
            &policy_id,
        )
        .unwrap();
    }

    #[test]
    fn auth_user_delete_honors_user_lifecycle_deny_policy() {
        let (server, auth) = auth_server(true);
        auth.create_user("cloud-admin", "pw", crate::auth::Role::Admin)
            .unwrap();
        let customer_token = issue_token(&auth, None, "customer-admin", crate::auth::Role::Admin);
        attach_policy_json(
            &auth,
            "customer-admin",
            r#"{"id":"customer-allow-all","version":1,"statements":[{"effect":"allow","actions":["*"],"resources":["*"]}]}"#,
        );
        attach_policy_json(
            &auth,
            "customer-admin",
            r#"{"id":"protect-cloud-admin","version":1,"statements":[{"effect":"deny","actions":["user:delete"],"resources":["user:cloud-admin"]}]}"#,
        );

        let response = server.route(request_with_bearer(
            "DELETE",
            "/auth/users/cloud-admin",
            Vec::new(),
            &customer_token,
        ));

        assert_eq!(
            response.status,
            403,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        assert!(auth.get_user(None, "cloud-admin").is_some());
    }

    #[test]
    fn iam_policy_delete_honors_policy_deny() {
        let (server, auth) = auth_server(true);
        let customer_token = issue_token(&auth, None, "customer-admin", crate::auth::Role::Admin);
        attach_policy_json(
            &auth,
            "customer-admin",
            r#"{"id":"customer-allow-all","version":1,"statements":[{"effect":"allow","actions":["*"],"resources":["*"]}]}"#,
        );
        attach_policy_json(
            &auth,
            "customer-admin",
            r#"{"id":"protect-cloud-admin","version":1,"statements":[{"effect":"deny","actions":["policy:drop"],"resources":["policy:protect-cloud-admin"]}]}"#,
        );

        let response = server.route(request_with_bearer(
            "DELETE",
            "/admin/policies/protect-cloud-admin",
            Vec::new(),
            &customer_token,
        ));

        assert_eq!(
            response.status,
            403,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        assert!(auth.get_policy("protect-cloud-admin").is_some());
    }

    #[test]
    fn auth_tenants_platform_admin_sees_every_tenant() {
        let (server, auth) = auth_server(true);
        let admin_token = issue_token(&auth, None, "admin", crate::auth::Role::Admin);
        // Seed users in two tenants + one platform user.
        auth.create_user_in_tenant(Some("acme"), "alice", "pw", crate::auth::Role::Read)
            .unwrap();
        auth.create_user_in_tenant(Some("widgets"), "bob", "pw", crate::auth::Role::Read)
            .unwrap();

        let response = server.route(request_with_bearer(
            "GET",
            "/auth/tenants",
            Vec::new(),
            &admin_token,
        ));
        assert_eq!(
            response.status,
            200,
            "{}",
            String::from_utf8_lossy(&response.body)
        );

        let body = parse_json(&response.body);
        let tenants = body
            .get("tenants")
            .and_then(crate::json::Value::as_array)
            .expect("tenants array");
        let ids: Vec<Option<&str>> = tenants
            .iter()
            .map(|t| t.get("id").and_then(crate::json::Value::as_str))
            .collect();
        // Platform admin sees the implicit platform entry + every tenant.
        assert!(ids.contains(&None), "platform entry missing: {ids:?}");
        assert!(ids.contains(&Some("acme")), "acme missing: {ids:?}");
        assert!(ids.contains(&Some("widgets")), "widgets missing: {ids:?}");
    }

    #[test]
    fn auth_tenants_tenant_admin_sees_only_own_tenant() {
        let (server, auth) = auth_server(true);
        // Platform tenant has 'platform_user'; acme has 'alice_admin' as admin.
        auth.create_user_in_tenant(None, "platform_user", "pw", crate::auth::Role::Read)
            .unwrap();
        auth.create_user_in_tenant(Some("widgets"), "bob", "pw", crate::auth::Role::Read)
            .unwrap();
        let token = issue_token(&auth, Some("acme"), "alice_admin", crate::auth::Role::Admin);

        let response = server.route(request_with_bearer(
            "GET",
            "/auth/tenants",
            Vec::new(),
            &token,
        ));
        assert_eq!(
            response.status,
            200,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        let body = parse_json(&response.body);
        let tenants = body
            .get("tenants")
            .and_then(crate::json::Value::as_array)
            .unwrap();
        let ids: Vec<Option<&str>> = tenants
            .iter()
            .map(|t| t.get("id").and_then(crate::json::Value::as_str))
            .collect();
        assert_eq!(
            ids,
            vec![Some("acme")],
            "tenant admin must see only own tenant: {ids:?}"
        );
    }

    #[test]
    fn auth_tenants_non_admin_sees_only_self_scope() {
        let (server, auth) = auth_server(true);
        auth.create_user_in_tenant(Some("widgets"), "bob", "pw", crate::auth::Role::Read)
            .unwrap();
        let token = issue_token(&auth, Some("acme"), "carol", crate::auth::Role::Read);
        let response = server.route(request_with_bearer(
            "GET",
            "/auth/tenants",
            Vec::new(),
            &token,
        ));
        assert_eq!(response.status, 200);
        let body = parse_json(&response.body);
        let tenants = body
            .get("tenants")
            .and_then(crate::json::Value::as_array)
            .unwrap();
        let ids: Vec<Option<&str>> = tenants
            .iter()
            .map(|t| t.get("id").and_then(crate::json::Value::as_str))
            .collect();
        assert_eq!(ids, vec![Some("acme")]);
    }

    #[test]
    fn auth_tenants_unauthenticated_requires_auth() {
        let (server, _auth) = auth_server(true);
        let response = server.route(request("/auth/tenants"));
        // Routing middleware gates this — no bearer with require_auth on.
        assert_eq!(response.status, 401);
    }

    #[test]
    fn auth_tenants_auth_disabled_returns_full_set() {
        // require_auth = false, no bearer → handler treats it as
        // platform-admin visibility.
        let (server, auth) = auth_server(false);
        auth.create_user_in_tenant(Some("acme"), "alice", "pw", crate::auth::Role::Read)
            .unwrap();
        let response = server.route(request("/auth/tenants"));
        assert_eq!(
            response.status,
            200,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        let body = parse_json(&response.body);
        let tenants = body
            .get("tenants")
            .and_then(crate::json::Value::as_array)
            .unwrap();
        let ids: Vec<Option<&str>> = tenants
            .iter()
            .map(|t| t.get("id").and_then(crate::json::Value::as_str))
            .collect();
        assert!(ids.contains(&Some("acme")));
    }

    #[test]
    fn auth_policies_platform_admin_sees_every_policy() {
        let (server, auth) = auth_server(true);
        let admin_token = issue_token(&auth, None, "admin", crate::auth::Role::Admin);
        // Two policies in different tenants.
        let policy_json = |id: &str, tenant: Option<&str>| {
            let tenant_field = match tenant {
                Some(t) => format!(",\"tenant\":\"{t}\""),
                None => String::new(),
            };
            format!(
                r#"{{"id":"{id}","version":1,"statements":[{{"effect":"Allow","actions":["*"],"resources":["*"]}}]{tenant_field}}}"#
            )
        };
        auth.put_policy(
            crate::auth::policies::Policy::from_json_str(&policy_json("p_platform", None)).unwrap(),
        )
        .unwrap();
        auth.put_policy(
            crate::auth::policies::Policy::from_json_str(&policy_json("p_acme", Some("acme")))
                .unwrap(),
        )
        .unwrap();
        auth.put_policy(
            crate::auth::policies::Policy::from_json_str(&policy_json(
                "p_widgets",
                Some("widgets"),
            ))
            .unwrap(),
        )
        .unwrap();

        let response = server.route(request_with_bearer(
            "GET",
            "/auth/policies",
            Vec::new(),
            &admin_token,
        ));
        assert_eq!(
            response.status,
            200,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        let body = parse_json(&response.body);
        let policies = body
            .get("policies")
            .and_then(crate::json::Value::as_array)
            .unwrap();
        let ids: Vec<&str> = policies
            .iter()
            .filter_map(|p| p.get("id").and_then(crate::json::Value::as_str))
            .collect();
        for required in ["p_platform", "p_acme", "p_widgets"] {
            assert!(
                ids.contains(&required),
                "platform admin missing {required}: {ids:?}"
            );
        }
    }

    #[test]
    fn auth_policies_tenant_admin_sees_own_plus_platform() {
        let (server, auth) = auth_server(true);
        let policy_json = |id: &str, tenant: Option<&str>| {
            let tenant_field = match tenant {
                Some(t) => format!(",\"tenant\":\"{t}\""),
                None => String::new(),
            };
            format!(
                r#"{{"id":"{id}","version":1,"statements":[{{"effect":"Allow","actions":["*"],"resources":["*"]}}]{tenant_field}}}"#
            )
        };
        auth.put_policy(
            crate::auth::policies::Policy::from_json_str(&policy_json("p_platform", None)).unwrap(),
        )
        .unwrap();
        auth.put_policy(
            crate::auth::policies::Policy::from_json_str(&policy_json("p_acme", Some("acme")))
                .unwrap(),
        )
        .unwrap();
        auth.put_policy(
            crate::auth::policies::Policy::from_json_str(&policy_json(
                "p_widgets",
                Some("widgets"),
            ))
            .unwrap(),
        )
        .unwrap();
        let token = issue_token(&auth, Some("acme"), "alice_admin", crate::auth::Role::Admin);

        let response = server.route(request_with_bearer(
            "GET",
            "/auth/policies",
            Vec::new(),
            &token,
        ));
        assert_eq!(response.status, 200);
        let body = parse_json(&response.body);
        let policies = body
            .get("policies")
            .and_then(crate::json::Value::as_array)
            .unwrap();
        let ids: Vec<&str> = policies
            .iter()
            .filter_map(|p| p.get("id").and_then(crate::json::Value::as_str))
            .collect();
        assert!(
            ids.contains(&"p_platform"),
            "platform policy hidden: {ids:?}"
        );
        assert!(ids.contains(&"p_acme"), "own tenant policy hidden: {ids:?}");
        assert!(
            !ids.contains(&"p_widgets"),
            "leaked other-tenant policy: {ids:?}"
        );
    }

    #[test]
    fn auth_can_batch_returns_results_per_check() {
        let (server, auth) = auth_server(true);
        let admin_token = issue_token(&auth, None, "admin", crate::auth::Role::Admin);
        // Attach allow-all so admin user has a matching Allow.
        auth.put_policy(
            crate::auth::policies::Policy::from_json_str(
                r#"{"id":"allow_all","version":1,"statements":[{"effect":"Allow","actions":["*"],"resources":["*"]}]}"#,
            )
            .unwrap(),
        )
        .unwrap();
        auth.attach_policy(
            crate::auth::store::PrincipalRef::User(crate::auth::UserId::from_parts(None, "admin")),
            "allow_all",
        )
        .unwrap();

        let body = br#"{"checks":[
            {"action":"read","resource":{"kind":"collection","name":"accounts"}},
            {"action":"delete","resource":{"kind":"collection","name":"audit"}}
        ]}"#
        .to_vec();
        let response = server.route(request_with_bearer("POST", "/auth/can", body, &admin_token));
        assert_eq!(
            response.status,
            200,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        let body = parse_json(&response.body);
        let results = body
            .get("results")
            .and_then(crate::json::Value::as_array)
            .expect("results array");
        assert_eq!(results.len(), 2);
        for r in results {
            assert_eq!(r.get("allowed"), Some(&crate::json!(true)));
            assert!(
                r.get("reason")
                    .and_then(crate::json::Value::as_str)
                    .is_some(),
                "reason must be present and a string: {r:?}"
            );
        }
    }

    #[test]
    fn auth_can_denies_when_no_matching_policy_in_policy_only_mode() {
        let (server, auth) = auth_server(true);
        auth.set_enforcement_mode(crate::auth::enforcement_mode::PolicyEnforcementMode::PolicyOnly);
        let token = issue_token(&auth, Some("acme"), "alice", crate::auth::Role::Read);

        let body = br#"{"checks":[
            {"action":"write","resource":{"kind":"collection","name":"secret"}}
        ]}"#
        .to_vec();
        let response = server.route(request_with_bearer("POST", "/auth/can", body, &token));
        assert_eq!(response.status, 200);
        let body = parse_json(&response.body);
        let result = &body
            .get("results")
            .and_then(crate::json::Value::as_array)
            .unwrap()[0];
        assert_eq!(result.get("allowed"), Some(&crate::json!(false)));
        let reason = result
            .get("reason")
            .and_then(crate::json::Value::as_str)
            .unwrap();
        assert!(
            reason.contains("default deny") || reason.starts_with("deny"),
            "reason should explain the deny: {reason}"
        );
    }

    #[test]
    fn auth_can_accepts_single_check_form() {
        let (server, auth) = auth_server(true);
        let token = issue_token(&auth, None, "admin", crate::auth::Role::Admin);
        let body =
            br#"{"action":"read","resource":{"kind":"collection","name":"accounts"}}"#.to_vec();
        let response = server.route(request_with_bearer("POST", "/auth/can", body, &token));
        assert_eq!(
            response.status,
            200,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        let body = parse_json(&response.body);
        let results = body
            .get("results")
            .and_then(crate::json::Value::as_array)
            .expect("results array");
        assert_eq!(results.len(), 1, "single-check form must yield one result");
    }

    #[test]
    fn auth_can_unauthenticated_requires_auth() {
        let (server, _auth) = auth_server(true);
        let body = br#"{"checks":[{"action":"read","resource":{"kind":"collection","name":"x"}}]}"#
            .to_vec();
        let response = server.route(request_with("POST", "/auth/can", body));
        // Gated by the routing middleware before reaching the handler.
        assert_eq!(response.status, 401);
    }

    #[test]
    fn auth_can_auth_disabled_returns_results() {
        let (server, _auth) = auth_server(false);
        let body = br#"{"checks":[{"action":"read","resource":{"kind":"collection","name":"x"}}]}"#
            .to_vec();
        let response = server.route(request_with("POST", "/auth/can", body));
        assert_eq!(
            response.status,
            200,
            "{}",
            String::from_utf8_lossy(&response.body)
        );
        let body = parse_json(&response.body);
        let results = body
            .get("results")
            .and_then(crate::json::Value::as_array)
            .unwrap();
        assert_eq!(results.len(), 1);
    }

    #[test]
    fn ui_metadata_readiness_exact_arm_still_wins() {
        // The exact-match arm for `/catalog/collections/readiness`
        // must not be shadowed by the new dynamic strip_prefix
        // handler (#736). Hitting it must keep returning the
        // readiness payload, not a 404 from the metadata handler.
        let server = fresh_server();
        let response = server.route(request("/catalog/collections/readiness"));
        assert_eq!(
            response.status,
            200,
            "readiness arm must still match: {}",
            String::from_utf8_lossy(&response.body)
        );
    }

    // ───────────────── Issue #760 — output streaming (NDJSON) ─────────────────

    fn ndjson_query(server: &RedDBServer, query: &str) -> Vec<u8> {
        let mut request = request_with(
            "POST",
            "/query",
            format!(r#"{{"query":"{query}"}}"#).into_bytes(),
        );
        request
            .headers
            .insert("accept".to_string(), "application/x-ndjson".to_string());
        let mut buf: Vec<u8> = Vec::new();
        let handled = server
            .try_route_streaming(&request, &mut buf)
            .expect("streaming dispatch");
        assert!(handled, "Accept: application/x-ndjson should be streamed");
        buf
    }

    fn split_response(raw: &[u8]) -> (String, Vec<u8>) {
        let pos = raw
            .windows(4)
            .position(|w| w == b"\r\n\r\n")
            .expect("header terminator");
        (
            String::from_utf8_lossy(&raw[..pos]).to_string(),
            raw[pos + 4..].to_vec(),
        )
    }

    fn decode_chunked_body(body: &[u8]) -> Vec<u8> {
        // Tiny ad-hoc chunked decoder for the HTTP/1.1 framing we
        // emit in `output_stream::write_chunk`. Sufficient for tests
        // — does not handle trailers or chunk extensions.
        let mut out = Vec::new();
        let mut i = 0;
        while i < body.len() {
            let crlf = body[i..]
                .windows(2)
                .position(|w| w == b"\r\n")
                .expect("chunk size line terminator");
            let size_hex = std::str::from_utf8(&body[i..i + crlf]).expect("utf8 hex");
            let size = usize::from_str_radix(size_hex.trim(), 16).expect("hex chunk size");
            i += crlf + 2;
            if size == 0 {
                break;
            }
            out.extend_from_slice(&body[i..i + size]);
            i += size + 2; // skip trailing \r\n
        }
        out
    }

    #[test]
    fn wants_ndjson_response_matches_canonical_and_alternates() {
        let mut headers = BTreeMap::new();
        assert!(!wants_ndjson_response(&headers));

        headers.insert("accept".to_string(), "application/json".to_string());
        assert!(!wants_ndjson_response(&headers));

        for accept in &[
            "application/x-ndjson",
            "application/ndjson",
            "text/ndjson",
            "Application/X-NDJSON",
            "application/x-ndjson; q=1.0",
            "text/html, application/x-ndjson;q=0.9",
        ] {
            headers.insert("accept".to_string(), (*accept).to_string());
            assert!(wants_ndjson_response(&headers), "Accept={accept}");
        }
    }

    #[test]
    fn ndjson_round_trip_emits_row_lines_and_end_envelope() {
        // End-to-end acceptance criterion #1 — NDJSON wire shape.
        let server = fresh_server();
        let raw = ndjson_query(&server, "SELECT 1 as n");
        let (head, body) = split_response(&raw);
        assert!(
            head.contains("Content-Type: application/x-ndjson"),
            "headers should advertise NDJSON: {head}"
        );
        assert!(
            head.contains("Transfer-Encoding: chunked"),
            "headers should advertise chunked transfer: {head}"
        );
        let decoded = decode_chunked_body(&body);
        let text = String::from_utf8(decoded).expect("ndjson body is utf8");
        let mut lines: Vec<&str> = text.lines().collect();
        let end_line = lines.pop().expect("at least one line");
        assert!(
            end_line.starts_with("{\"end\":"),
            "last line must be end envelope: {end_line}"
        );
        // Issue #766 — first line is the open_ack envelope.
        let open_ack = lines.remove(0);
        assert!(
            open_ack.starts_with("{\"open_ack\":"),
            "first line must be open_ack: {open_ack}"
        );
        assert!(
            open_ack.contains("\"resumable\":"),
            "open_ack must carry resumable flag: {open_ack}"
        );
        assert!(
            !lines.is_empty(),
            "expected at least one row line before end: {text}"
        );
        for row_line in &lines {
            assert!(
                row_line.starts_with("{\"row\":"),
                "non-terminal line must be a row envelope: {row_line}"
            );
        }
    }

    #[test]
    fn ndjson_path_does_not_regress_non_streaming_query() {
        // Acceptance criterion #3 — without the streaming Accept the
        // legacy materialising path keeps the prior wire shape.
        let server = fresh_server();
        let request = request_with("POST", "/query", br#"{"query":"SELECT 1 as n"}"#.to_vec());
        let mut buf: Vec<u8> = Vec::new();
        let handled = server
            .try_route_streaming(&request, &mut buf)
            .expect("dispatch");
        assert!(!handled, "no Accept header → no streaming dispatch");

        let response = server.route(request);
        assert_eq!(response.status, 200);
        let body = String::from_utf8_lossy(&response.body);
        assert!(
            body.contains("\"ok\":true"),
            "legacy shape preserved: {body}"
        );
        assert!(
            body.contains("\"descriptor\":"),
            "descriptor metadata still present: {body}"
        );
    }

    #[test]
    fn ndjson_refuses_when_session_has_active_begin() {
        // Acceptance criterion #4 — OpenStream against a session
        // with an active `BEGIN` is rejected with
        // `stream_in_transaction_unsupported`. We set a non-zero
        // connection id on this thread, execute `BEGIN` to populate
        // runtime's Transaction State registry, then invoke the
        // streaming dispatch and expect a structured 409 refusal.
        use crate::runtime::impl_core::{clear_current_connection_id, set_current_connection_id};

        let server = fresh_server();
        set_current_connection_id(760_001);
        // Open a transaction on this synthetic connection. The
        // runtime stamps a `TxnContext` keyed by the current id.
        let _ = server
            .runtime
            .execute_query("BEGIN")
            .expect("BEGIN should succeed on a fresh runtime");
        // Sanity check — the accessor sees the open transaction.
        assert!(server.runtime.connection_in_transaction(760_001));

        let mut request = request_with("POST", "/query", br#"{"query":"SELECT 1"}"#.to_vec());
        request
            .headers
            .insert("accept".to_string(), "application/x-ndjson".to_string());
        let mut buf: Vec<u8> = Vec::new();
        let handled = server
            .try_route_streaming(&request, &mut buf)
            .expect("dispatch");
        assert!(handled, "should still be streaming-dispatched");

        // The refusal is a regular HTTP response, not an NDJSON body
        // — headers have not been sent yet at this gate.
        let text = String::from_utf8_lossy(&buf);
        assert!(
            text.starts_with("HTTP/1.1 409"),
            "expected 409 refusal, got: {text}"
        );
        assert!(
            text.contains("stream_in_transaction_unsupported"),
            "structured code missing: {text}"
        );

        // Clean up — rollback so the test does not leak transaction
        // state into other tests run on this thread.
        let _ = server.runtime.execute_query("ROLLBACK");
        clear_current_connection_id();
    }

    // ───────────────── Issue #761 / S2 — capacity guard ─────────────────

    fn ndjson_request_with_bearer(server_query: &str, token: Option<&str>) -> HttpRequest {
        let mut request = request_with(
            "POST",
            "/query",
            format!(r#"{{"query":"{server_query}"}}"#).into_bytes(),
        );
        request
            .headers
            .insert("accept".to_string(), "application/x-ndjson".to_string());
        if let Some(token) = token {
            request
                .headers
                .insert("authorization".to_string(), format!("Bearer {token}"));
        }
        request
    }

    fn dispatch_streaming(server: &RedDBServer, request: &HttpRequest) -> Vec<u8> {
        let mut buf: Vec<u8> = Vec::new();
        let handled = server
            .try_route_streaming(request, &mut buf)
            .expect("streaming dispatch");
        assert!(handled);
        buf
    }

    #[test]
    fn ndjson_global_capacity_exhausted_returns_429_with_structured_body() {
        // Acceptance criterion #1 + #6 — global cap fires with
        // `server_stream_capacity_exhausted`, HTTP 429, Retry-After,
        // and a structured body carrying `limit` and `current`.
        let server = fresh_server();
        server
            .runtime
            .execute_query("SET CONFIG stream.max_global = 1")
            .expect("set global cap");
        server
            .runtime
            .execute_query("SET CONFIG stream.max_per_principal = 1")
            .expect("set per-principal cap");

        // Manually hold one slot to simulate an in-flight stream.
        let cap_registry = Arc::clone(server.stream_capacity());
        let _hold = cap_registry.try_acquire("anon", 1, 1).expect("first slot");

        let request = ndjson_request_with_bearer("SELECT 1", None);
        let raw = dispatch_streaming(&server, &request);
        let text = String::from_utf8_lossy(&raw);
        assert!(
            text.starts_with("HTTP/1.1 429"),
            "expected 429 refusal, got: {text}"
        );
        assert!(
            text.contains("Retry-After: 1"),
            "Retry-After header missing: {text}"
        );
        assert!(
            text.contains("server_stream_capacity_exhausted"),
            "structured code missing: {text}"
        );
        assert!(text.contains("\"limit\":1"), "limit missing: {text}");
        assert!(text.contains("\"current\":1"), "current missing: {text}");
    }

    #[test]
    fn ndjson_per_principal_capacity_isolated_across_principals() {
        // Acceptance criteria #2 + #3 — Alice's quota does not affect
        // Bob, and Alice's overflow is refused with
        // `principal_stream_quota_exhausted` even though global has
        // room.
        let server = fresh_server();
        server
            .runtime
            .execute_query("SET CONFIG stream.max_global = 100")
            .expect("set global cap");
        server
            .runtime
            .execute_query("SET CONFIG stream.max_per_principal = 1")
            .expect("set per-principal cap");

        let cap_registry = Arc::clone(server.stream_capacity());
        // Pre-load: Alice's single slot already held.
        let alice_principal = super::super::routing::principal_for(&{
            let mut h = BTreeMap::new();
            h.insert(
                "authorization".to_string(),
                "Bearer alice-token".to_string(),
            );
            h
        });
        let _hold = cap_registry
            .try_acquire(&alice_principal, 100, 1)
            .expect("Alice's first slot");

        // Alice's second open is refused with the principal-scoped code.
        let alice_request = ndjson_request_with_bearer("SELECT 1", Some("alice-token"));
        let raw = dispatch_streaming(&server, &alice_request);
        let text = String::from_utf8_lossy(&raw);
        assert!(
            text.starts_with("HTTP/1.1 429"),
            "expected 429, got: {text}"
        );
        assert!(
            text.contains("principal_stream_quota_exhausted"),
            "expected per-principal code: {text}"
        );
        assert!(text.contains("\"principal\":"), "principal missing: {text}");

        // Bob is unaffected — his stream completes with 200.
        let bob_request = ndjson_request_with_bearer("SELECT 1 as n", Some("bob-token"));
        let raw = dispatch_streaming(&server, &bob_request);
        let text = String::from_utf8_lossy(&raw);
        assert!(
            text.starts_with("HTTP/1.1 200"),
            "Bob should not be affected by Alice's quota: {text}"
        );
    }

    #[test]
    fn ndjson_capacity_slot_released_on_stream_end() {
        // Acceptance criterion #4 — releasing a slot on stream end
        // (normal completion) frees it for both counters. After a
        // successful round-trip the same principal can open a new
        // stream against a cap-of-1 configuration.
        let server = fresh_server();
        server
            .runtime
            .execute_query("SET CONFIG stream.max_global = 1")
            .expect("set global cap");
        server
            .runtime
            .execute_query("SET CONFIG stream.max_per_principal = 1")
            .expect("set per-principal cap");

        for _ in 0..3 {
            let request = ndjson_request_with_bearer("SELECT 1 as n", Some("solo-token"));
            let raw = dispatch_streaming(&server, &request);
            let text = String::from_utf8_lossy(&raw);
            assert!(
                text.starts_with("HTTP/1.1 200"),
                "slot was not released between streams: {text}"
            );
        }
        let (global, per_principal) = server.stream_capacity().snapshot();
        assert_eq!(global, 0, "global counter leaked across streams");
        assert!(
            per_principal.is_empty(),
            "per-principal map leaked: {per_principal:?}"
        );
    }

    // ───────────────── Issue #763 / S4 — input streaming (NDJSON) ─────────────────

    fn input_stream_request(body: &str) -> HttpRequest {
        let mut request = request_with("POST", "/streams/input", body.as_bytes().to_vec());
        request.headers.insert(
            "content-type".to_string(),
            "application/x-ndjson".to_string(),
        );
        request
    }

    fn dispatch_input_stream(server: &RedDBServer, body: &str) -> Vec<u8> {
        let request = input_stream_request(body);
        let mut buf: Vec<u8> = Vec::new();
        let handled = server
            .try_route_streaming(&request, &mut buf)
            .expect("streaming dispatch");
        assert!(
            handled,
            "POST /streams/input with x-ndjson should be streamed"
        );
        buf
    }

    #[test]
    fn ndjson_input_round_trip_commits_rows_and_returns_end_envelope() {
        // Acceptance criterion #1 — round-trip NDJSON insert. After the
        // stream ends every row is visible to non-streaming readers
        // (criterion #2 — chunked auto-commit is durable).
        let server = fresh_server();
        ddl(
            &server,
            "CREATE TABLE rows763 (id INTEGER PRIMARY KEY, name TEXT)",
        );

        let body = concat!(
            "{\"open\":{\"target\":\"rows763\",\"columns\":[\"id\",\"name\"]}}\n",
            "{\"row\":{\"id\":1,\"name\":\"alice\"}}\n",
            "{\"row\":{\"id\":2,\"name\":\"bob\"}}\n",
            "{\"row\":{\"id\":3,\"name\":\"carol\"}}\n",
        );
        let raw = dispatch_input_stream(&server, body);
        let (head, body_bytes) = split_response(&raw);
        assert!(
            head.contains("Content-Type: application/x-ndjson"),
            "headers should advertise NDJSON: {head}"
        );
        let decoded = decode_chunked_body(&body_bytes);
        let text = String::from_utf8(decoded).expect("ndjson body is utf8");
        // Acceptance criterion #4 — server is silent on success: the
        // only frame in the body is the terminal `{"end": …}` line.
        let lines: Vec<&str> = text.lines().collect();
        assert_eq!(
            lines.len(),
            1,
            "server should emit exactly one frame on success: {text}"
        );
        let end_line = lines[0];
        assert!(
            end_line.starts_with("{\"end\":"),
            "single line must be end envelope: {end_line}"
        );
        assert!(end_line.contains("\"row_count\":3"));

        // Acceptance criterion #1 + #2 — rows are durable / visible.
        let response = server.route(request_with(
            "POST",
            "/query",
            br#"{"query":"SELECT id, name FROM rows763"}"#.to_vec(),
        ));
        assert_eq!(response.status, 200);
        let body_text = String::from_utf8_lossy(&response.body);
        assert!(body_text.contains("alice"), "alice missing: {body_text}");
        assert!(body_text.contains("bob"), "bob missing: {body_text}");
        assert!(body_text.contains("carol"), "carol missing: {body_text}");
    }

    #[test]
    fn ndjson_input_malformed_row_terminates_with_error_envelope() {
        // Acceptance criterion #3 — malformed row terminates the stream
        // with a structured error envelope carrying chunk seq + reason
        // + recoverable_rid. Earlier rows up to the failure remain
        // visible.
        let server = fresh_server();
        ddl(
            &server,
            "CREATE TABLE rows763_err (id INTEGER PRIMARY KEY, name TEXT)",
        );
        // Force per-chunk-of-1 commits so the row before the malformed
        // one is durably committed before the failure point.
        server
            .runtime
            .execute_query("SET CONFIG stream.chunk.max_rows = 1")
            .expect("set chunk size");

        let body = concat!(
            "{\"open\":{\"target\":\"rows763_err\",\"columns\":[\"id\",\"name\"]}}\n",
            "{\"row\":{\"id\":1,\"name\":\"alice\"}}\n",
            "not valid json at all\n",
            "{\"row\":{\"id\":3,\"name\":\"carol\"}}\n",
        );
        let raw = dispatch_input_stream(&server, body);
        let (_head, body_bytes) = split_response(&raw);
        let decoded = decode_chunked_body(&body_bytes);
        let text = String::from_utf8(decoded).expect("utf8");
        let last = text.lines().last().unwrap_or("");
        assert!(
            last.starts_with("{\"error\":"),
            "stream must end with error envelope: {text}"
        );
        assert!(last.contains("\"code\":\"invalid_row\""));
        assert!(last.contains("\"chunk_seq\":"));
        assert!(last.contains("\"recoverable_rid\":"));

        // Acceptance criterion #3 — alice survives the mid-stream failure.
        let response = server.route(request_with(
            "POST",
            "/query",
            br#"{"query":"SELECT id, name FROM rows763_err"}"#.to_vec(),
        ));
        let body_text = String::from_utf8_lossy(&response.body);
        assert!(
            body_text.contains("alice"),
            "alice should survive the failure: {body_text}"
        );
        assert!(
            !body_text.contains("carol"),
            "carol after the failure must not be committed: {body_text}"
        );
    }

    #[test]
    fn ndjson_input_refuses_when_session_has_active_begin() {
        // Acceptance criterion #5 — opening an input stream inside
        // `BEGIN` refuses with `stream_in_transaction_unsupported`.
        use crate::runtime::impl_core::{clear_current_connection_id, set_current_connection_id};

        let server = fresh_server();
        set_current_connection_id(763_001);
        let _ = server
            .runtime
            .execute_query("BEGIN")
            .expect("BEGIN should succeed");
        assert!(server.runtime.connection_in_transaction(763_001));

        let body = "{\"open\":{\"target\":\"any\",\"columns\":[\"id\"]}}\n";
        let request = input_stream_request(body);
        let mut buf: Vec<u8> = Vec::new();
        let handled = server
            .try_route_streaming(&request, &mut buf)
            .expect("dispatch");
        assert!(handled);
        let text = String::from_utf8_lossy(&buf);
        assert!(
            text.starts_with("HTTP/1.1 409"),
            "expected 409 refusal, got: {text}"
        );
        assert!(
            text.contains("stream_in_transaction_unsupported"),
            "structured code missing: {text}"
        );

        let _ = server.runtime.execute_query("ROLLBACK");
        clear_current_connection_id();
    }

    #[test]
    fn ndjson_input_rejects_unsafe_identifier_before_streaming_headers() {
        // SQL-injection guard: identifiers must match the safe
        // character class. A crafted target/column name returns a
        // non-streaming 400 so the failure is unmistakable.
        let server = fresh_server();
        let body = "{\"open\":{\"target\":\"rows; DROP TABLE x\",\"columns\":[\"id\"]}}\n";
        let request = input_stream_request(body);
        let mut buf: Vec<u8> = Vec::new();
        let handled = server
            .try_route_streaming(&request, &mut buf)
            .expect("dispatch");
        assert!(handled);
        let text = String::from_utf8_lossy(&buf);
        assert!(
            text.starts_with("HTTP/1.1 400"),
            "expected 400 refusal, got: {text}"
        );
        assert!(text.contains("safe identifier"), "message: {text}");
    }

    #[test]
    fn ndjson_invalid_query_surfaces_structured_error_line() {
        // Acceptance criterion #2 — mid-stream failures surface as
        // structured NDJSON error envelopes, not silent disconnects.
        let server = fresh_server();
        let raw = ndjson_query(&server, "SELECT NOT_A_REAL_COLUMN FROM nowhere");
        let (_head, body) = split_response(&raw);
        let decoded = decode_chunked_body(&body);
        let text = String::from_utf8(decoded).expect("utf8");
        assert!(
            text.lines().any(|l| l.starts_with("{\"error\":")),
            "expected an error envelope in the stream: {text}"
        );
        assert!(
            text.lines().last().unwrap_or("").starts_with("{\"end\":"),
            "stream must still close with an end envelope: {text}"
        );
    }

    // ───────────── Issue #766 / S7 — resume coordinator ─────────────

    fn ndjson_query_body(server: &RedDBServer, body: Vec<u8>) -> Vec<u8> {
        let mut request = request_with("POST", "/query", body);
        request
            .headers
            .insert("accept".to_string(), "application/x-ndjson".to_string());
        let mut buf: Vec<u8> = Vec::new();
        let handled = server
            .try_route_streaming(&request, &mut buf)
            .expect("streaming dispatch");
        assert!(handled);
        buf
    }

    fn decode_ndjson(raw: &[u8]) -> Vec<String> {
        let (_head, body) = split_response(raw);
        let decoded = decode_chunked_body(&body);
        String::from_utf8(decoded)
            .expect("ndjson utf8")
            .lines()
            .map(str::to_string)
            .collect()
    }

    fn extract_json_field(line: &str, key: &str) -> Option<String> {
        // Crude extractor sufficient for tests; finds `"key":`<value> up
        // to the next `,` or `}` boundary at the same depth (no nesting).
        let needle = format!("\"{key}\":");
        let start = line.find(&needle)? + needle.len();
        let tail = &line[start..];
        let mut depth_obj = 0;
        let mut depth_arr = 0;
        let mut in_str = false;
        let mut escape = false;
        let mut end = tail.len();
        for (i, ch) in tail.char_indices() {
            if escape {
                escape = false;
                continue;
            }
            match ch {
                '\\' if in_str => escape = true,
                '"' => in_str = !in_str,
                '{' if !in_str => depth_obj += 1,
                '}' if !in_str => {
                    if depth_obj == 0 {
                        end = i;
                        break;
                    }
                    depth_obj -= 1;
                }
                '[' if !in_str => depth_arr += 1,
                ']' if !in_str => depth_arr -= 1,
                ',' if !in_str && depth_obj == 0 && depth_arr == 0 => {
                    end = i;
                    break;
                }
                _ => {}
            }
        }
        Some(tail[..end].trim().trim_matches('"').to_string())
    }

    #[test]
    fn ndjson_open_ack_marks_resumable_for_plain_select() {
        // Acceptance criterion #1 — OpenAck signals resumable: true
        // for an RID-ordered SELECT.
        let server = fresh_server();
        ddl(
            &server,
            "CREATE TABLE rows766 (id INTEGER PRIMARY KEY, name TEXT)",
        );
        ddl(
            &server,
            "INSERT INTO rows766 (id, name) VALUES (1, 'a'), (2, 'b')",
        );

        let raw = ndjson_query(&server, "SELECT id, name FROM rows766");
        let lines = decode_ndjson(&raw);
        let open_ack = &lines[0];
        assert!(
            open_ack.starts_with("{\"open_ack\":"),
            "first line must be open_ack: {open_ack}"
        );
        assert!(
            open_ack.contains("\"resumable\":true"),
            "plain SELECT should be resumable: {open_ack}"
        );
    }

    #[test]
    fn ndjson_open_ack_marks_non_resumable_for_aggregation() {
        // Acceptance criterion #1 — aggregations are not resumable.
        let server = fresh_server();
        ddl(
            &server,
            "CREATE TABLE rows766_agg (id INTEGER PRIMARY KEY, name TEXT)",
        );
        ddl(
            &server,
            "INSERT INTO rows766_agg (id, name) VALUES (1, 'a')",
        );

        let raw = ndjson_query(&server, "SELECT COUNT(*) FROM rows766_agg");
        let lines = decode_ndjson(&raw);
        let open_ack = &lines[0];
        assert!(
            open_ack.contains("\"resumable\":false"),
            "aggregation must not be resumable: {open_ack}"
        );
    }

    #[test]
    fn ndjson_resume_continues_from_resume_after_rid_when_hash_matches() {
        // Acceptance criterion #2 — happy-path resume. Open a stream,
        // record the row hash for the prefix, then re-open with
        // `resume_after_rid` + matching `prefix_hash` and observe only
        // the suffix rows being delivered.
        let server = fresh_server();
        ddl(
            &server,
            "CREATE TABLE rows766_res (id INTEGER PRIMARY KEY, name TEXT)",
        );
        ddl(
            &server,
            "INSERT INTO rows766_res (id, name) VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd')",
        );

        // Open #1 — full stream, capture snapshot_lsn and per-row rids.
        let raw = ndjson_query(&server, "SELECT id, name FROM rows766_res");
        let lines = decode_ndjson(&raw);
        let open_ack = lines.first().expect("open_ack");
        let snapshot_lsn: u64 = extract_json_field(open_ack, "snapshot_lsn")
            .expect("snapshot_lsn")
            .parse()
            .expect("snapshot_lsn integer");

        // Extract row lines (excluding open_ack and end).
        let row_lines: Vec<&str> = lines
            .iter()
            .filter(|l| l.starts_with("{\"row\":"))
            .map(String::as_str)
            .collect();
        assert!(row_lines.len() >= 2, "need ≥2 rows: {lines:?}");

        // Pick the rid of the first row as the resume boundary —
        // resume should deliver rows with rid > that value.
        let first_row = row_lines[0];
        let rid_str = extract_json_field(first_row, "rid").expect("rid present in row metadata");
        let resume_after_rid: u64 = rid_str.parse().expect("rid integer");

        // Compute prefix_hash by hashing the first row's exact bytes.
        use crate::server::output_stream::PrefixHasher;
        let mut hasher = PrefixHasher::new();
        hasher.update(first_row.as_bytes());
        let prefix_hash = hasher.finalize_hex();

        // Resume.
        let body = format!(
            "{{\"query\":\"SELECT id, name FROM rows766_res\",\"resume\":{{\"snapshot_lsn\":{snapshot_lsn},\"resume_after_rid\":{resume_after_rid},\"prefix_hash\":\"{prefix_hash}\"}}}}"
        );
        let raw = ndjson_query_body(&server, body.into_bytes());
        let lines = decode_ndjson(&raw);
        let row_count_in_resume = lines.iter().filter(|l| l.starts_with("{\"row\":")).count();
        let suffix_expected = row_lines.len() - 1;
        assert_eq!(
            row_count_in_resume, suffix_expected,
            "resume delivered wrong row count: lines={lines:?}"
        );
        let end_line = lines.last().expect("end envelope");
        assert!(
            end_line.contains(&format!("\"row_count\":{suffix_expected}")),
            "end row_count must match suffix length: {end_line}"
        );
        assert!(
            !lines.iter().any(|l| l.starts_with("{\"error\":")),
            "happy-path resume must not emit an error envelope: {lines:?}"
        );
    }

    // ───────────── Issue #767 / S8 — lease decoupling + audit ─────────────

    fn read_audit_log(server: &RedDBServer) -> String {
        let logger = server.runtime.audit_log();
        assert!(
            logger.wait_idle(std::time::Duration::from_secs(2)),
            "audit writer did not drain in time"
        );
        std::fs::read_to_string(logger.path()).unwrap_or_default()
    }

    #[test]
    fn ndjson_end_envelope_carries_opaque_lease_handle_not_internal_id() {
        // Acceptance criterion #3 — wire-visible handle is opaque,
        // hex-encoded, ≥128 bits. The deprecated `lease_id` field is
        // gone from the envelope so clients cannot regress to it.
        let server = fresh_server();
        let raw = ndjson_query(&server, "SELECT 1 as n");
        let (_head, body) = split_response(&raw);
        let decoded = decode_chunked_body(&body);
        let text = String::from_utf8(decoded).expect("utf8");
        let end_line = text.lines().last().expect("at least one line");
        assert!(end_line.starts_with("{\"end\":"));
        assert!(
            end_line.contains("\"lease_handle\":"),
            "end envelope must carry lease_handle: {end_line}"
        );
        assert!(
            !end_line.contains("\"lease_id\":"),
            "end envelope must not leak internal lease_id: {end_line}"
        );
        // Extract handle value and check shape.
        let needle = "\"lease_handle\":\"";
        let start = end_line.find(needle).expect("lease_handle field") + needle.len();
        let rest = &end_line[start..];
        let end = rest.find('"').expect("handle terminator");
        let handle = &rest[..end];
        assert_eq!(
            handle.len(),
            32,
            "lease_handle must be 32-char hex (128 bits): {handle}"
        );
        assert!(handle.chars().all(|c| c.is_ascii_hexdigit()));
    }

    /// Extract the `lease_handle` value (32-char hex) from an NDJSON
    /// `{"end": …}` envelope. Returns `None` when the envelope does not
    /// carry one (e.g. a non-streaming HTTP error response). Tests use
    /// the handle as a unique key when grepping the shared audit log —
    /// `RedDBOptions::in_memory()` writes `.audit.log` next to the
    /// (unique-per-test) data path's *parent* `/tmp`, so multiple
    /// tests running in the same process share the file.
    fn extract_lease_handle(raw: &[u8]) -> Option<String> {
        let (_head, body) = split_response(raw);
        let decoded = decode_chunked_body(&body);
        let text = String::from_utf8(decoded).ok()?;
        let needle = "\"lease_handle\":\"";
        let start = text.find(needle)? + needle.len();
        let rest = &text[start..];
        let end = rest.find('"')?;
        Some(rest[..end].to_string())
    }

    #[test]
    fn ndjson_stream_emits_open_and_close_audit_events() {
        // Acceptance criterion #5 — every state transition produces
        // an audit event. After a successful stream we expect both a
        // `stream.opened` and a `stream.closed` line carrying our
        // stream's unique lease_handle, with `reason=ok`. We filter
        // by handle (not by action count) because the audit file is
        // shared across in-process tests.
        let server = fresh_server();
        let raw = ndjson_query(&server, "SELECT 1 as n");
        let handle = extract_lease_handle(&raw).expect("end envelope carries handle");
        let body = read_audit_log(&server);
        let opened: Vec<&str> = body
            .lines()
            .filter(|l| {
                l.contains("\"action\":\"stream.opened\"")
                    && l.contains(&format!("\"lease_handle\":\"{handle}\""))
            })
            .collect();
        let closed: Vec<&str> = body
            .lines()
            .filter(|l| {
                l.contains("\"action\":\"stream.closed\"")
                    && l.contains(&format!("\"lease_handle\":\"{handle}\""))
            })
            .collect();
        assert_eq!(
            opened.len(),
            1,
            "expected exactly one stream.opened for handle {handle}"
        );
        assert_eq!(
            closed.len(),
            1,
            "expected exactly one stream.closed for handle {handle}"
        );
        assert!(
            opened[0].contains("\"snapshot_lsn\":") && opened[0].contains("\"query_hash\":"),
            "stream.opened missing required fields: {}",
            opened[0]
        );
        assert!(
            closed[0].contains("\"reason\":\"ok\""),
            "stream.closed must record reason=ok: {}",
            closed[0]
        );
    }

    #[test]
    fn ndjson_resume_refuses_on_prefix_hash_mismatch() {
        // Acceptance criterion #4 — wrong prefix_hash returns
        // `prefix_hash_mismatch` and delivers zero rows.
        let server = fresh_server();
        ddl(
            &server,
            "CREATE TABLE rows766_mis (id INTEGER PRIMARY KEY, name TEXT)",
        );
        ddl(
            &server,
            "INSERT INTO rows766_mis (id, name) VALUES (1, 'a'), (2, 'b'), (3, 'c')",
        );

        let raw = ndjson_query(&server, "SELECT id, name FROM rows766_mis");
        let lines = decode_ndjson(&raw);
        let open_ack = lines.first().expect("open_ack");
        let snapshot_lsn: u64 = extract_json_field(open_ack, "snapshot_lsn")
            .expect("snapshot_lsn")
            .parse()
            .expect("integer");
        let first_row = lines
            .iter()
            .find(|l| l.starts_with("{\"row\":"))
            .expect("first row");
        let resume_after_rid: u64 = extract_json_field(first_row, "rid")
            .expect("rid")
            .parse()
            .expect("u64");

        let body = format!(
            "{{\"query\":\"SELECT id, name FROM rows766_mis\",\"resume\":{{\"snapshot_lsn\":{snapshot_lsn},\"resume_after_rid\":{resume_after_rid},\"prefix_hash\":\"{}\"}}}}",
            "deadbeef".repeat(8)
        );
        let raw = ndjson_query_body(&server, body.into_bytes());
        let lines = decode_ndjson(&raw);
        assert!(
            lines
                .iter()
                .any(|l| l.starts_with("{\"error\":") && l.contains("prefix_hash_mismatch")),
            "expected prefix_hash_mismatch envelope: {lines:?}"
        );
        let rows_emitted = lines.iter().filter(|l| l.starts_with("{\"row\":")).count();
        assert_eq!(
            rows_emitted, 0,
            "no rows must be delivered on hash mismatch: {lines:?}"
        );
        let end_line = lines.last().expect("end");
        assert!(
            end_line.contains("\"row_count\":0"),
            "end must show 0 rows: {end_line}"
        );
    }

    #[test]
    fn ndjson_capacity_refusal_emits_capacity_refused_audit_event() {
        // Acceptance criterion #5 — capacity-refused is a state
        // transition; the audit log must carry a `stream.closed` row
        // with `reason: capacity_refused` even though no lease was
        // ever issued.
        let server = fresh_server();
        server
            .runtime
            .execute_query("SET CONFIG stream.max_global = 1")
            .expect("set cap");
        server
            .runtime
            .execute_query("SET CONFIG stream.max_per_principal = 1")
            .expect("set cap");
        // Hold the only slot so the next open is refused.
        let cap_registry = Arc::clone(server.stream_capacity());
        let _hold = cap_registry.try_acquire("anon", 1, 1).unwrap();

        let request = ndjson_request_with_bearer("SELECT 1", None);
        let raw = dispatch_streaming(&server, &request);
        assert!(String::from_utf8_lossy(&raw).starts_with("HTTP/1.1 429"));

        let body = read_audit_log(&server);
        assert!(
            body.lines().any(|l| {
                l.contains("\"action\":\"stream.closed\"")
                    && l.contains("\"reason\":\"capacity_refused\"")
            }),
            "audit log missing capacity_refused close event: {body}"
        );
    }

    #[test]
    fn ndjson_resume_refuses_on_expired_snapshot() {
        // Acceptance criterion #3 — an expired snapshot lease (no
        // entry / past TTL) refuses with `snapshot_expired`. We seed
        // the registry with a stale opened_at_ms so the TTL has
        // already elapsed.
        let server = fresh_server();
        ddl(
            &server,
            "CREATE TABLE rows766_exp (id INTEGER PRIMARY KEY, name TEXT)",
        );
        ddl(
            &server,
            "INSERT INTO rows766_exp (id, name) VALUES (1, 'a'), (2, 'b')",
        );

        // Seed a snapshot LSN we'll claim to resume from, with
        // opened_at_ms = 0 (way in the past) and ttl = 1ms so any
        // wall-clock check expires it.
        let bogus_snapshot_lsn: u64 = 1;
        server.lease_registry().record(bogus_snapshot_lsn, 0, 1);

        let body = format!(
            "{{\"query\":\"SELECT id, name FROM rows766_exp\",\"resume\":{{\"snapshot_lsn\":{bogus_snapshot_lsn},\"resume_after_rid\":0,\"prefix_hash\":\"{}\"}}}}",
            "0".repeat(64)
        );
        let raw = ndjson_query_body(&server, body.into_bytes());
        let lines = decode_ndjson(&raw);
        assert!(
            lines
                .iter()
                .any(|l| l.starts_with("{\"error\":") && l.contains("snapshot_expired")),
            "expected snapshot_expired envelope: {lines:?}"
        );
        assert_eq!(
            lines.iter().filter(|l| l.starts_with("{\"row\":")).count(),
            0,
            "no rows must be delivered on snapshot expiry: {lines:?}"
        );
    }

    #[test]
    fn ndjson_token_expired_during_lease_audit_emitted_when_jwt_exp_is_short() {
        // Acceptance criterion #4 — a JWT bearer whose `exp` lands
        // inside the lease's snapshot-TTL window triggers the dedicated
        // audit event exactly once per stream, with `lease_continued`
        // = true. The stream itself still completes successfully —
        // the lease, not the token, governs subsequent chunks (PRD
        // #759 lease-decoupling property).
        let server = fresh_server();
        // Synthetic JWT — header.payload.signature where payload
        // sets `exp: 1` (Unix epoch, well in the past). Auth is not
        // configured on this server, so the bearer is unverified;
        // but the audit emit consults the `exp` claim purely for
        // bookkeeping.
        let token = "eyJhbGciOiJIUzI1NiJ9.eyJleHAiOjF9.sig";
        let request = ndjson_request_with_bearer("SELECT 1 as n", Some(token));
        let raw = dispatch_streaming(&server, &request);
        assert!(
            String::from_utf8_lossy(&raw).starts_with("HTTP/1.1 200"),
            "stream must complete despite bearer expiry: {}",
            String::from_utf8_lossy(&raw)
        );
        let handle = extract_lease_handle(&raw).expect("end envelope carries handle");

        let body = read_audit_log(&server);
        let token_events: Vec<&str> = body
            .lines()
            .filter(|l| {
                l.contains("\"action\":\"stream.token_expired_during_lease\"")
                    && l.contains(&format!("\"lease_handle\":\"{handle}\""))
            })
            .collect();
        assert_eq!(
            token_events.len(),
            1,
            "expected exactly one token-expired event for handle {handle}"
        );
        assert!(
            token_events[0].contains("\"lease_continued\":true"),
            "token-expired event must mark lease_continued=true: {}",
            token_events[0]
        );
        // And we still get the open/close pair for this handle.
        assert!(body
            .lines()
            .any(|l| l.contains("\"action\":\"stream.opened\"")
                && l.contains(&format!("\"lease_handle\":\"{handle}\""))));
        assert!(body
            .lines()
            .any(|l| l.contains("\"action\":\"stream.closed\"")
                && l.contains(&format!("\"lease_handle\":\"{handle}\""))));
    }

    #[test]
    fn ndjson_opaque_token_does_not_emit_token_expired_event() {
        // Counterpoint — an opaque (non-JWT) bearer carries no `exp`
        // claim so the audit emitter has nothing to record. The
        // detector must not false-positive on api-key shapes.
        let server = fresh_server();
        let request = ndjson_request_with_bearer("SELECT 1 as n", Some("opaque-api-key"));
        let raw = dispatch_streaming(&server, &request);
        assert!(String::from_utf8_lossy(&raw).starts_with("HTTP/1.1 200"));
        let handle = extract_lease_handle(&raw).expect("end envelope carries handle");
        let body = read_audit_log(&server);
        // Filter strictly by our stream's handle — the shared audit
        // log may carry token-expired rows from other tests, but
        // none should match this lease handle.
        assert!(
            !body.lines().any(|l| {
                l.contains("\"action\":\"stream.token_expired_during_lease\"")
                    && l.contains(&format!("\"lease_handle\":\"{handle}\""))
            }),
            "opaque tokens must not trigger the token-expired audit for this handle: {body}"
        );
    }

    #[test]
    fn ndjson_resume_refuses_on_non_resumable_query() {
        // Acceptance criterion #5 — attempting resume on a
        // non-resumable plan returns `not_resumable`. The check fires
        // before the lease lookup, so a bogus snapshot_lsn does not
        // shadow the not_resumable code.
        let server = fresh_server();
        ddl(
            &server,
            "CREATE TABLE rows766_nores (id INTEGER PRIMARY KEY, name TEXT)",
        );
        ddl(
            &server,
            "INSERT INTO rows766_nores (id, name) VALUES (1, 'a')",
        );

        let body = format!(
            "{{\"query\":\"SELECT COUNT(*) FROM rows766_nores\",\"resume\":{{\"snapshot_lsn\":1,\"resume_after_rid\":0,\"prefix_hash\":\"{}\"}}}}",
            "0".repeat(64)
        );
        let raw = ndjson_query_body(&server, body.into_bytes());
        let lines = decode_ndjson(&raw);
        assert!(
            lines
                .iter()
                .any(|l| l.starts_with("{\"error\":") && l.contains("not_resumable")),
            "expected not_resumable envelope: {lines:?}"
        );
    }

    #[test]
    fn ndjson_lease_outlives_short_jwt_exp_acceptance_criterion_one() {
        // Acceptance criterion #1 — a token expiring mid-flight does
        // not terminate the stream. In the shim slice, materialisation
        // already de facto decouples the bearer from chunk delivery;
        // this test pins that property at the wire layer. The stream
        // must complete with `reason=ok` even though the JWT `exp` is
        // already in the past at the moment the open call returns.
        let server = fresh_server();
        let token = "eyJhbGciOiJIUzI1NiJ9.eyJleHAiOjF9.sig";
        let request = ndjson_request_with_bearer("SELECT 1 as n", Some(token));
        let raw = dispatch_streaming(&server, &request);
        let handle = extract_lease_handle(&raw).expect("end envelope carries handle");
        let (_head, http_body) = split_response(&raw);
        let decoded = decode_chunked_body(&http_body);
        let text = String::from_utf8(decoded).unwrap();
        let end_line = text.lines().last().unwrap_or("");
        assert!(
            end_line.starts_with("{\"end\":"),
            "stream must close with end envelope despite token expiry: {text}"
        );
        assert!(
            text.lines().any(|l| l.starts_with("{\"row\":")),
            "row data must still flow when bearer credential is expired: {text}"
        );
        // Close audit reason is `ok` — filter by handle so the shared
        // audit file's other rows don't break the assertion.
        let audit = read_audit_log(&server);
        let close = audit
            .lines()
            .find(|l| {
                l.contains("\"action\":\"stream.closed\"")
                    && l.contains(&format!("\"lease_handle\":\"{handle}\""))
            })
            .expect("close event recorded");
        assert!(
            close.contains("\"reason\":\"ok\""),
            "lease must close cleanly when token expired mid-flight: {close}"
        );
    }

    #[test]
    fn ndjson_input_stream_emits_open_and_close_audit_events() {
        // Symmetric coverage for the input-stream surface. A
        // successful insert path produces the same opened/closed
        // pair, distinguishable from output streams only by the
        // executed query (recorded via query_hash).
        let server = fresh_server();
        ddl(
            &server,
            "CREATE TABLE rows767 (id INTEGER PRIMARY KEY, name TEXT)",
        );
        let body = concat!(
            "{\"open\":{\"target\":\"rows767\",\"columns\":[\"id\",\"name\"]}}\n",
            "{\"row\":{\"id\":1,\"name\":\"alice\"}}\n",
        );
        let _ = dispatch_input_stream(&server, body);
        let log = read_audit_log(&server);
        assert!(
            log.lines()
                .any(|l| l.contains("\"action\":\"stream.opened\"")),
            "input stream missing stream.opened: {log}"
        );
        assert!(
            log.lines().any(|l| {
                l.contains("\"action\":\"stream.closed\"") && l.contains("\"reason\":\"ok\"")
            }),
            "input stream missing stream.closed ok: {log}"
        );
    }

    // ---------------------------------------------------------------
    // Security-review regressions (pre-launch hardening)
    // ---------------------------------------------------------------

    /// Restores an environment variable when dropped.
    struct SecurityEnvGuard {
        name: &'static str,
        previous: Option<String>,
    }

    impl SecurityEnvGuard {
        fn unset(name: &'static str) -> Self {
            let previous = std::env::var(name).ok();
            std::env::remove_var(name);
            Self { name, previous }
        }
    }

    impl Drop for SecurityEnvGuard {
        fn drop(&mut self) {
            match self.previous.take() {
                Some(value) => std::env::set_var(self.name, value),
                None => std::env::remove_var(self.name),
            }
        }
    }

    fn auth_enabled_config() -> crate::auth::AuthConfig {
        crate::auth::AuthConfig {
            enabled: true,
            session_ttl_secs: 60,
            require_auth: true,
            auto_encrypt_storage: false,
            vault_enabled: false,
            cert: Default::default(),
            oauth: Default::default(),
        }
    }

    /// Server with auth enabled plus an admin and a write-role user.
    /// Returns `(server, store, admin_key, writer_key)`.
    fn server_with_users() -> (
        RedDBServer,
        std::sync::Arc<crate::auth::store::AuthStore>,
        String,
        String,
    ) {
        use crate::auth::Role;
        let store = std::sync::Arc::new(crate::auth::store::AuthStore::new(auth_enabled_config()));
        store.create_user("root", "pw", Role::Admin).expect("admin");
        store.create_user("bob", "pw", Role::Write).expect("writer");
        let admin_key = store
            .create_api_key("root", "admin-key", Role::Admin)
            .expect("admin key")
            .key;
        let writer_key = store
            .create_api_key("bob", "writer-key", Role::Write)
            .expect("writer key")
            .key;
        let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        let server = RedDBServer::new(runtime).with_auth(std::sync::Arc::clone(&store));
        (server, store, admin_key, writer_key)
    }

    fn authed_request(method: &str, path: &str, body: &str, bearer: Option<&str>) -> HttpRequest {
        let mut request = request_with(method, path, body.as_bytes().to_vec());
        if let Some(token) = bearer {
            request
                .headers
                .insert("authorization".to_string(), format!("Bearer {token}"));
        }
        request
    }

    #[test]
    fn admin_routes_require_an_admin_bearer_when_no_admin_token_is_configured() {
        let _token = SecurityEnvGuard::unset("RED_ADMIN_TOKEN");
        let _token_file = SecurityEnvGuard::unset("RED_ADMIN_TOKEN_FILE");
        let (server, _store, admin_key, writer_key) = server_with_users();

        // `RouteAuth::AdminToken` used to return `true` outright when no
        // operator token was configured, so `/admin/*` (shutdown, restore,
        // failover, IAM policies) was reachable anonymously even under
        // `require_auth`. It must now fall through to the bearer chain.
        assert_eq!(
            server
                .route(authed_request("POST", "/admin/readonly", "{}", None))
                .status,
            401,
            "anonymous callers must not reach /admin/* without an operator token"
        );
        assert_eq!(
            server
                .route(authed_request(
                    "POST",
                    "/admin/readonly",
                    "{}",
                    Some(&writer_key)
                ))
                .status,
            401,
            "a write-role bearer must not reach /admin/*"
        );
        // Operators are not locked out: an admin bearer still passes.
        assert_ne!(
            server
                .route(authed_request(
                    "POST",
                    "/admin/readonly",
                    "{}",
                    Some(&admin_key)
                ))
                .status,
            401,
            "an admin bearer must still reach /admin/*"
        );
    }

    #[test]
    fn write_role_user_cannot_mint_an_api_key_for_another_principal() {
        let (server, store, _admin_key, writer_key) = server_with_users();

        // The handler took no headers at all: any write-role bearer could
        // name `root` in the body and receive an admin key.
        assert_eq!(
            server
                .route(authed_request(
                    "POST",
                    "/auth/api-keys",
                    r#"{"username":"root","name":"pivot","role":"admin"}"#,
                    Some(&writer_key),
                ))
                .status,
            403,
            "a write-role principal must not mint keys for another user"
        );
        assert_eq!(
            server
                .route(authed_request(
                    "POST",
                    "/auth/api-keys",
                    r#"{"username":"bob","name":"self","role":"admin"}"#,
                    Some(&writer_key),
                ))
                .status,
            403,
            "an API key must not exceed the caller's own role"
        );
        // Self-service at or below the caller's own role keeps working.
        assert_eq!(
            server
                .route(authed_request(
                    "POST",
                    "/auth/api-keys",
                    r#"{"username":"bob","name":"ci","role":"read"}"#,
                    Some(&writer_key),
                ))
                .status,
            201,
            "self-service key issuance must still work"
        );

        let root_keys = store
            .list_users()
            .iter()
            .find(|u| u.username == "root")
            .map(|u| u.api_keys.len())
            .unwrap_or_default();
        assert_eq!(root_keys, 1, "root must still own exactly its own key");
    }

    #[test]
    fn unauthenticated_callers_cannot_create_or_delete_principals() {
        let (server, store, _admin_key, _writer_key) = server_with_users();

        let created = server.route(authed_request(
            "POST",
            "/auth/users",
            r#"{"username":"evil","password":"pw","role":"admin"}"#,
            None,
        ));
        assert!(
            matches!(created.status, 401 | 403),
            "anonymous principal creation must be refused, got {}",
            created.status
        );
        assert!(
            !store.list_users().iter().any(|u| u.username == "evil"),
            "no principal may be created by an unauthenticated caller"
        );

        let deleted = server.route(authed_request("DELETE", "/auth/users/root", "", None));
        assert!(
            matches!(deleted.status, 401 | 403),
            "anonymous principal deletion must be refused, got {}",
            deleted.status
        );
        assert!(
            store.list_users().iter().any(|u| u.username == "root"),
            "the admin must survive an unauthenticated delete"
        );
    }
}

/// Match `/collections/:name/kvs/invalidate_tags`.
fn collection_kv_invalidate_tags_path(path: &str) -> Option<&str> {
    let prefix = "/collections/";
    let trimmed = path.strip_prefix(prefix)?;
    let collection = trimmed.strip_suffix("/kvs/invalidate_tags")?;
    let collection = collection.trim_matches('/');
    if collection.is_empty() {
        None
    } else {
        Some(collection)
    }
}

/// Match `/collections/:name/kvs/:key` to extract collection and key.
fn collection_kv_path(path: &str) -> Option<(String, String)> {
    let prefix = "/collections/";
    let trimmed = path.strip_prefix(prefix)?;
    let (collection, rest) = trimmed.split_once("/kvs/")?;
    let collection = collection.trim_matches('/');
    let key = rest.trim_matches('/');
    if collection.is_empty() || key.is_empty() {
        None
    } else {
        let collection = percent_decode_path_segment(collection).ok()?;
        let key = percent_decode_path_segment(key).ok()?;
        Some((collection, key))
    }
}

/// Match `/collections/:name/kv/:key/watch` or `/collections/:name/kvs/:key/watch`.
fn collection_kv_watch_path(path: &str) -> Option<(String, String)> {
    let prefix = "/collections/";
    let trimmed = path.strip_prefix(prefix)?;
    let (collection, rest) = trimmed
        .split_once("/kv/")
        .or_else(|| trimmed.split_once("/kvs/"))?;
    let key = rest.strip_suffix("/watch")?.trim_matches('/');
    let collection = collection.trim_matches('/');
    if collection.is_empty() || key.is_empty() {
        None
    } else {
        let collection = percent_decode_path_segment(collection).ok()?;
        let key = percent_decode_path_segment(key).ok()?;
        Some((collection, key))
    }
}

/// Issue #760 — does the request's `Accept` header opt in to NDJSON
/// output streaming? Matches the canonical `application/x-ndjson` and
/// the two common alternates (`application/ndjson`, `text/ndjson`),
/// case-insensitively, ignoring `q=` weight suffixes. A plain
/// `application/json` Accept is *not* a streaming opt-in — that
/// continues to ride the materialising one-shot path.
/// Issue #763 — does the request carry an NDJSON body? Used to gate
/// the input-stream entry point. Matches `application/x-ndjson` and
/// the same alternates accepted by [`wants_ndjson_response`].
pub(crate) fn content_type_is_ndjson(headers: &BTreeMap<String, String>) -> bool {
    let Some(raw) = headers.get("content-type") else {
        return false;
    };
    let token = raw
        .split(';')
        .next()
        .unwrap_or("")
        .trim()
        .to_ascii_lowercase();
    matches!(
        token.as_str(),
        "application/x-ndjson" | "application/ndjson" | "text/ndjson"
    )
}

pub(crate) fn wants_ndjson_response(headers: &BTreeMap<String, String>) -> bool {
    let Some(raw) = headers.get("accept") else {
        return false;
    };
    raw.split(',').any(|part| {
        let token = part
            .split(';')
            .next()
            .unwrap_or("")
            .trim()
            .to_ascii_lowercase();
        matches!(
            token.as_str(),
            "application/x-ndjson" | "application/ndjson" | "text/ndjson"
        )
    })
}

/// PLAN.md Phase 4.4 — derive a stable principal label from a
/// request's headers. Bearer tokens are hashed (sha256 prefix) so
/// the metrics label never leaks the raw token. Unauthenticated
/// requests share the `anon` bucket; replica RPCs that pass a
/// `x-reddb-replica-id` header get their own `replica:<id>` bucket.
pub(crate) fn principal_for(headers: &BTreeMap<String, String>) -> String {
    // `x-reddb-replica-id` is attacker-controlled and used to be consulted
    // first, so rotating it per request defeated both the QPS bucket and the
    // per-principal in-flight cap while growing the bucket map without
    // bound. It now only labels a bucket for a caller that already presented
    // a bearer, and the bearer still decides the identity — a replica gets
    // its own bucket, an anonymous client cannot mint one.
    let replica = headers
        .get("x-reddb-replica-id")
        .map(|value| value.trim())
        .filter(|value| !value.is_empty());
    if let Some(auth) = headers.get("authorization") {
        if let Some(token) = auth.strip_prefix("Bearer ") {
            let trimmed = token.trim();
            if !trimmed.is_empty() {
                let digest = crate::crypto::sha256(trimmed.as_bytes());
                let bearer = crate::utils::to_hex_prefix(&digest, 8);
                return match replica {
                    Some(replica) => format!("replica:{bearer}:{replica}"),
                    None => format!("bearer:{bearer}"),
                };
            }
        }
    }
    "anon".to_string()
}

/// Issue #761 / S2 — build the structured 429 refusal handed back when
/// `try_route_streaming` cannot acquire a stream slot. Both cap-fired
/// codes share the same response shape so clients can branch on the
/// `code` field without re-parsing.
///
/// Wire shape:
///   `Retry-After: <seconds>`  header
///   `{"ok": false, "error": {"code": <code>, "limit": N, "current": M,
///                            "principal": "..."?}}`
pub(crate) fn stream_capacity_refusal_response(
    err: &crate::server::output_stream::AcquireError,
) -> HttpResponse {
    use crate::json::{Map, Value as JsonValue};
    use crate::server::output_stream::AcquireError;

    let mut error_obj = Map::new();
    error_obj.insert(
        "code".to_string(),
        JsonValue::String(err.code().to_string()),
    );
    error_obj.insert(
        "message".to_string(),
        crate::json_field::SerializedJsonField::tainted(&err.message()),
    );
    match err {
        AcquireError::GlobalExhausted { limit, current } => {
            error_obj.insert("limit".to_string(), JsonValue::Number(*limit as f64));
            error_obj.insert("current".to_string(), JsonValue::Number(*current as f64));
        }
        AcquireError::PrincipalExhausted {
            principal,
            limit,
            current,
        } => {
            error_obj.insert(
                "principal".to_string(),
                crate::json_field::SerializedJsonField::tainted(principal),
            );
            error_obj.insert("limit".to_string(), JsonValue::Number(*limit as f64));
            error_obj.insert("current".to_string(), JsonValue::Number(*current as f64));
        }
    }
    let mut root = Map::new();
    root.insert("ok".to_string(), JsonValue::Bool(false));
    root.insert("error".to_string(), JsonValue::Object(error_obj));

    let mut response = json_response(429, JsonValue::Object(root));
    // `Retry-After: 1` (seconds) — a stream slot frees the moment any
    // other stream ends, so a one-second hint matches the actual
    // recovery cadence; the structured `current`/`limit` fields are
    // the load-bearing signal for clients with smarter backoff.
    if let Ok(retry_after) =
        crate::server::header_escape_guard::HeaderEscapeGuard::header_value("1")
    {
        response = response.with_header("Retry-After", retry_after);
    }
    response
}

/// Issue #934 / PRD #930 — structured refusal for a request rejected by the
/// per-principal concurrent in-flight cap at the async edge. Shares the
/// `{"ok":false,"error":{code,limit,current,principal}}` shape with the
/// stream-capacity refusal so a client can branch on `error.code` without
/// re-parsing prose, and carries a `Retry-After` header for clients with
/// only header-level backoff. The principal label is emitted through the
/// tainted-field path so a hostile `x-reddb-replica-id` header can't break
/// out of the JSON string.
pub(crate) fn principal_inflight_refusal_response(
    err: &crate::server::http_principal_limiter::PrincipalCapExceeded,
    retry_after_secs: u64,
) -> HttpResponse {
    use crate::json::{Map, Value as JsonValue};

    let mut error_obj = Map::new();
    error_obj.insert(
        "code".to_string(),
        JsonValue::String(
            crate::server::http_principal_limiter::PRINCIPAL_INFLIGHT_CODE.to_string(),
        ),
    );
    error_obj.insert(
        "message".to_string(),
        JsonValue::String("per-principal in-flight request cap exceeded".to_string()),
    );
    error_obj.insert(
        "principal".to_string(),
        crate::json_field::SerializedJsonField::tainted(&err.principal),
    );
    error_obj.insert("limit".to_string(), JsonValue::Number(err.limit as f64));
    error_obj.insert("current".to_string(), JsonValue::Number(err.current as f64));

    let mut root = Map::new();
    root.insert("ok".to_string(), JsonValue::Bool(false));
    root.insert("error".to_string(), JsonValue::Object(error_obj));
    root.insert(
        "retry_after_secs".to_string(),
        JsonValue::Number(retry_after_secs as f64),
    );

    let mut response = json_response(429, JsonValue::Object(root));
    if let Ok(retry_after) = crate::server::header_escape_guard::HeaderEscapeGuard::header_value(
        &retry_after_secs.to_string(),
    ) {
        response = response.with_header("Retry-After", retry_after);
    }
    response
}

/// PLAN.md Phase 6.1 — read the operator admin token. Honors
/// `RED_ADMIN_TOKEN` and the `RED_ADMIN_TOKEN_FILE` companion via
/// the shared `crate::utils::env_with_file_fallback` helper.
pub(crate) fn read_admin_token() -> Option<String> {
    crate::utils::env_with_file_fallback("RED_ADMIN_TOKEN")
}