fix(gateway): fail-closed publish_schema compile + compile-once cache (GW-01 follow-up)

- new gateway::schema_cache — PublishSchemaCache: compile the op's
  publish_schema once per registration (value-keyed invalidation for
  hot reload), cache compile failures (logged once at error level,
  never retried per request)
- /publish compile failure is now fail-closed: the chunk stream
  terminates with INTERNAL (500), the error text stays in the log
  (no schema internals on the wire) — the per-request warn-and-skip
  unvalidated ingest path is removed
- schema resolution is lazy (first chunk poll, after invoke_sink's
  404/403/422 pre-checks — GW-11 order preserved) and keyed by schema
  value, so re-registration/hot reload is picked up (test)
- NdjsonChunkStream: first Err item is terminal (done + stream end),
  mirroring the wire pump's send(Err) + break — Ok chunks can never
  follow an error on the HTTP path either (found by spy-handler test)

Verified: cargo test (308), cargo test --all-features, clippy
--all-targets -D warnings (default + all-features), fmt --check.

Tasks: review-001-publish-schema-validation-robust

src/gateway/mod.rs

@@ -1,6 +1,7 @@
 pub mod dispatch;
 pub mod error;
 pub mod routes;
+pub(crate) mod schema_cache;
 
 pub use dispatch::GatewayDispatch;
 pub use routes::CallRequest;

src/gateway/routes.rs

@@ -33,6 +33,7 @@ use serde_json::{json, Value};
 
 use super::dispatch::GatewayDispatch;
 use super::error::call_error_to_http_response_with_identity;
+use super::schema_cache::{CompileFailed, PublishSchemaCache};
 use crate::server::auth::ResolvedIdentity;
 use crate::server::state::RouterState;
 
@@ -53,16 +54,19 @@ type ByteStream = futures::stream::BoxStream<'static, Result<Bytes, axum::Error>
 pub(crate) struct GatewayState {
     registry: Arc<OperationRegistry>,
     identity_provider: Arc<dyn IdentityProvider>,
+    publish_schemas: crate::gateway::schema_cache::PublishSchemaCache,
 }
 
 impl GatewayState {
     pub(crate) fn new(
         registry: Arc<OperationRegistry>,
         identity_provider: Arc<dyn IdentityProvider>,
+        publish_schemas: crate::gateway::schema_cache::PublishSchemaCache,
     ) -> Self {
         Self {
             registry,
             identity_provider,
+            publish_schemas,
         }
     }
 
@@ -79,6 +83,7 @@ impl FromRef<RouterState> for GatewayState {
         GatewayState::new(
             Arc::clone(&state.registry),
             Arc::clone(&state.identity_provider),
+            state.publish_schemas.clone(),
         )
     }
 }
@@ -219,13 +224,22 @@ pub type SubscribeStream = BoxStream<'static, Result<Event, Infallible>>;
 /// a client disconnect drops the body stream and the sink (and the
 /// handler's `PublishStream`) sees EOF, matching call-protocol
 /// write-half close semantics.
+///
+/// Chunk validation (GW-01, ADR-046 §4): each chunk is validated
+/// against the op's `publish_schema` via the compile-once
+/// [`PublishSchemaCache`]. A schema that **fails to compile** is
+/// fail-closed, not fail-open (review-001 post-remediation follow-up):
+/// the chunk stream terminates with `INTERNAL` before any chunk reaches
+/// the handler — the unvalidated-ingest hole the per-request
+/// `warn`-and-skip opened is shut. The compile error itself is logged
+/// at error level and never echoed on the wire.
 pub(crate) async fn publish_handler(
     State(state): State<GatewayState>,
     ResolvedIdentity(identity): ResolvedIdentity,
     body: axum::body::Body,
 ) -> Response {
     let mut header_lines = BufferedLines::new(Box::pin(body.into_data_stream()));
-    let (operation, first_chunk, validator): (String, Value, Option<jsonschema::Validator>) = {
+    let (operation, first_chunk) = {
         let line = match header_lines.next_line().await {
             Ok(Some(line)) => line,
             Ok(None) => {
@@ -251,25 +265,16 @@ pub(crate) async fn publish_handler(
         let Some(chunk) = value.get("chunk") else {
             return missing_header_field_response();
         };
-        let validator = state
+        (op, chunk.clone())
-            .registry
-            .registration(op.strip_prefix('/').unwrap_or(&op))
-            .and_then(|reg| reg.spec.publish_schema.clone())
-            .and_then(|schema| match jsonschema::options().build(&schema) {
-                Ok(v) => Some(v),
-                Err(e) => {
-                    tracing::warn!(
-                        operation = %op,
-                        error = %e,
-                        "publish_schema failed to compile; chunks will not be validated"
-                    );
-                    None
-                }
-            });
-        (op, chunk.clone(), validator)
     };
 
-    let chunks = NdjsonChunkStream::new(header_lines, validator, Some(first_chunk));
+    let chunks = NdjsonChunkStream::new(
+        header_lines,
+        Arc::clone(&state.registry),
+        state.publish_schemas.clone(),
+        operation.clone(),
+        Some(first_chunk),
+    );
     let dispatch = state.dispatch();
     let envelope = dispatch
         .invoke_sink(identity.clone(), &operation, Value::Null, Box::pin(chunks))
@@ -277,6 +282,88 @@ pub(crate) async fn publish_handler(
     envelope_to_response(envelope, identity.as_ref())
 }
 
+/// The per-request `publish_schema` situation for a `/publish` call
+/// (ADR-046 §4). Resolution is lazy — the chunk stream resolves against
+/// the [`PublishSchemaCache`] on its first poll, which runs only after
+/// `invoke_sink`'s pre-checks have admitted the operation (an unknown,
+/// internal, or wrong-type op is rejected before any schema resolution
+/// or compile happens, preserving the GW-11 dispatch-owned pre-check
+/// order and keeping 404/403/422 intact).
+enum PublishSchemaState {
+    /// Not yet resolved: the stream holds the registry + cache until
+    /// the first chunk is polled.
+    Unresolved {
+        registry: Arc<OperationRegistry>,
+        cache: PublishSchemaCache,
+        operation: String,
+    },
+    /// The op registers no `publish_schema`; chunks pass as-is (the
+    /// wire path's behavior for schema-less Pub ops).
+    Unvalidated,
+    /// Chunks are validated against this compiled validator.
+    Validated(std::sync::Arc<jsonschema::Validator>),
+    /// The op's `publish_schema` failed to compile (GW-01 follow-up):
+    /// fail-closed — the stream yields one terminal `INTERNAL` error
+    /// and ends, so no chunk ever reaches the handler unvalidated.
+    Failed(CompileFailed),
+}
+
+impl PublishSchemaState {
+    fn resolve(&mut self) {
+        let Self::Unresolved {
+            registry,
+            cache,
+            operation,
+        } = self
+        else {
+            return;
+        };
+        let operation = std::mem::take(operation);
+        *self = match cache.validator(registry, &operation) {
+            Ok(None) => Self::Unvalidated,
+            Ok(Some(validator)) => Self::Validated(validator),
+            Err(compile_failed) => {
+                tracing::debug!(
+                    operation = %operation,
+                    "rejecting /publish: the operation's publish_schema could not be compiled \
+                     (fail-closed)"
+                );
+                Self::Failed(compile_failed)
+            }
+        };
+    }
+
+    /// Validate one chunk; also flips a terminal compile failure into
+    /// stream-end after the first error item (an `Err` from this stream
+    /// is terminal — the handler sees the error and the stream closes,
+    /// mirroring the wire pump's `send(Err)` + `break`).
+    fn validate_chunk(
+        &mut self,
+        value: Value,
+    ) -> std::task::Poll<Option<Result<Value, CallError>>> {
+        self.resolve();
+        match self {
+            Self::Unvalidated => std::task::Poll::Ready(Some(Ok(value))),
+            Self::Validated(validator) => {
+                if Validator::is_valid(validator, &value) {
+                    std::task::Poll::Ready(Some(Ok(value)))
+                } else {
+                    std::task::Poll::Ready(Some(Err(CallError::invalid_input(
+                        "published chunk failed publish_schema validation",
+                    )
+                    .with_details(json!({ "chunk": value })))))
+                }
+            }
+            Self::Failed(failed) => {
+                let error = failed.call_error();
+                *self = Self::Unvalidated;
+                std::task::Poll::Ready(Some(Err(error)))
+            }
+            Self::Unresolved { .. } => unreachable!("resolve() runs before validation"),
+        }
+    }
+}
+
 fn missing_header_field_response() -> Response {
     invalid_input_response("first publish line must carry {\"operation\": ..., \"chunk\": ...}")
 }
@@ -371,40 +458,53 @@ impl LineError {
 /// validated against the op's compiled `publish_schema` — matching the
 /// wire dispatcher's per-chunk validation (alkcall ADR-046). Malformed
 /// JSON, a schema violation, or a body-read failure yields a terminal
-/// `Err(INVALID_INPUT)` item — the same shape an initiator-side
+/// `Err` item — the same shape an initiator-side `call.error` produces
-/// `call.error` produces on the wire.
+/// on the wire, and equally terminal: like the wire pump's
+/// `send(Err)` + `break`, the first `Err` ends the stream. A schema
+/// that failed to compile yields that terminal `Err(INTERNAL)` on the
+/// first chunk — fail-closed (review-001 post-remediation follow-up).
 struct NdjsonChunkStream {
     lines: std::pin::Pin<Box<BufferedLines>>,
-    validator: Option<jsonschema::Validator>,
+    schema_state: PublishSchemaState,
     pending_first: Option<Value>,
+    done: bool,
 }
 
 impl NdjsonChunkStream {
     fn new(
         lines: BufferedLines,
-        validator: Option<jsonschema::Validator>,
+        registry: Arc<OperationRegistry>,
+        cache: PublishSchemaCache,
+        operation: String,
         pending_first: Option<Value>,
     ) -> Self {
         Self {
             lines: Box::pin(lines),
-            validator,
+            schema_state: PublishSchemaState::Unresolved {
+                registry,
+                cache,
+                operation,
+            },
             pending_first,
+            done: false,
         }
     }
 
-    fn validate_chunk(
+    /// `Some(item)` from `validate_chunk` where an `Err` item sets
+    /// `done` — the stream does not yield after an error (the wire
+    /// pump's semantics; the dropping mpsc sender below is the same
+    /// contract).
+    fn poll_validated(
         &mut self,
         value: Value,
     ) -> std::task::Poll<Option<Result<Value, CallError>>> {
-        if let Some(validator) = &self.validator {
+        match self.schema_state.validate_chunk(value) {
-            if !Validator::is_valid(validator, &value) {
+            std::task::Poll::Ready(Some(Err(e))) => {
-                return std::task::Poll::Ready(Some(Err(CallError::invalid_input(
+                self.done = true;
-                    "published chunk failed publish_schema validation",
+                std::task::Poll::Ready(Some(Err(e)))
-                )
-                .with_details(json!({ "chunk": value })))));
             }
+            other => other,
         }
-        std::task::Poll::Ready(Some(Ok(value)))
     }
 }
 
@@ -415,30 +515,44 @@ impl futures::Stream for NdjsonChunkStream {
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Option<Self::Item>> {
+        if self.done {
+            return std::task::Poll::Ready(None);
+        }
         if self.pending_first.is_some() {
             let first = self.pending_first.take().unwrap_or(Value::Null);
-            return self.validate_chunk(first);
+            return self.poll_validated(first);
         }
         let line = {
             let mut next = Box::pin(self.lines.next_line());
-            match std::future::Future::poll(next.as_mut(), cx) {
+            let polled = match std::future::Future::poll(next.as_mut(), cx) {
-                std::task::Poll::Ready(Ok(Some(line))) => line,
+                std::task::Poll::Ready(Ok(Some(line))) => Ok(Some(line)),
-                std::task::Poll::Ready(Ok(None)) => return std::task::Poll::Ready(None),
+                std::task::Poll::Ready(Ok(None)) => Ok(None),
-                std::task::Poll::Ready(Err(e)) => {
+                std::task::Poll::Ready(Err(e)) => Err(e.message()),
-                    return std::task::Poll::Ready(Some(Err(CallError::invalid_input(e.message()))))
+                std::task::Poll::Pending => {
+                    drop(next);
+                    return std::task::Poll::Pending;
+                }
+            };
+            drop(next);
+            match polled {
+                Ok(Some(line)) => line,
+                Ok(None) => return std::task::Poll::Ready(None),
+                Err(message) => {
+                    self.done = true;
+                    return std::task::Poll::Ready(Some(Err(CallError::invalid_input(message))));
                 }
-                std::task::Poll::Pending => return std::task::Poll::Pending,
             }
         };
         let value = match serde_json::from_slice::<Value>(&line) {
             Ok(v) => v,
             Err(e) => {
+                self.done = true;
                 return std::task::Poll::Ready(Some(Err(CallError::invalid_input(format!(
                     "publish line is not valid JSON: {e}"
-                )))))
+                )))));
             }
         };
-        self.validate_chunk(value)
+        self.poll_validated(value)
     }
 }
 
@@ -862,6 +976,7 @@ mod tests {
                 crate::websocket::DEFAULT_WS_MAX_SESSIONS,
             )),
             ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
+            publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(),
         };
         let auth_state = Arc::clone(&provider);
         gateway_router()
@@ -2243,4 +2358,217 @@ mod tests {
             "the prematurely terminated body reaches the handler as an INVALID_INPUT Err item: {resp}"
         );
     }
+
+    // --- publish_schema fail-closed (GW-01 follow-up) ----------------------
+
+    fn registry_with_broken_publish_schema() -> Arc<OperationRegistry> {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                OperationSpec::new(
+                    "ingest/broken",
+                    OperationType::Pub,
+                    Visibility::External,
+                    json!({}),
+                    json!({}),
+                    vec![],
+                    AccessControl::default(),
+                    None,
+                )
+                .with_publish_schema(json!({ "required": "n" })), // `required` must be an array
+                HandlerKind::Sink(make_sink_handler(
+                    |_unused_input, ctx, mut chunks| async move {
+                        use futures::StreamExt;
+                        let mut collected: Vec<Value> = Vec::new();
+                        while let Some(chunk) = chunks.next().await {
+                            match chunk {
+                                Ok(v) => collected.push(v),
+                                Err(e) => return ResponseEnvelope::error(ctx.request_id, e),
+                            }
+                        }
+                        ResponseEnvelope::ok(
+                            ctx.request_id,
+                            json!({ "count": collected.len(), "chunks": collected }),
+                        )
+                    },
+                )),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        Arc::new(registry)
+    }
+
+    #[tokio::test]
+    async fn publish_with_uncompilable_schema_fails_closed_with_internal() {
+        let router = build_router(registry_with_broken_publish_schema(), unused_provider());
+        let body = ndjson(&[
+            json!({ "operation": "ingest/broken", "chunk": { "n": 1 } }),
+            json!({ "n": 2 }),
+        ]);
+        let req = raw_request("POST", "/publish", body);
+        let (status, resp) = send(router, req).await;
+        assert_eq!(
+            status,
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "an un-compilable publish_schema must fail the request loudly (500), not skip validation: {resp}"
+        );
+        assert_eq!(
+            resp.get("code"),
+            Some(&json!("INTERNAL")),
+            "fail-closed maps to INTERNAL, not INVALID_INPUT (this is a server-side defect): {resp}"
+        );
+        assert!(
+            resp.get("message")
+                .and_then(|m| m.as_str())
+                .map(|m| m.contains("failed to compile"))
+                .unwrap_or(false),
+            "the wire message names the compile failure without echoing schema internals: {resp}"
+        );
+    }
+
+    #[tokio::test]
+    async fn publish_with_uncompilable_schema_never_delivers_chunks_to_the_handler() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                OperationSpec::new(
+                    "ingest/spy",
+                    OperationType::Pub,
+                    Visibility::External,
+                    json!({}),
+                    json!({}),
+                    vec![],
+                    AccessControl::default(),
+                    None,
+                )
+                .with_publish_schema(json!({ "type": 123 })), // non-string `type` never compiles
+                HandlerKind::Sink(make_sink_handler(|_input, ctx, mut chunks| async move {
+                    use futures::StreamExt;
+                    let received: Vec<Result<Value, CallError>> = chunks.by_ref().collect().await;
+                    ResponseEnvelope::ok(
+                        ctx.request_id,
+                        json!({
+                            "chunk_count": received.len(),
+                            "ok_chunks": received.iter().filter(|r| r.is_ok()).count(),
+                            "error_codes": received
+                                .iter()
+                                .filter_map(|r| r.as_ref().err().map(|e| e.code.clone()))
+                                .collect::<Vec<String>>(),
+                        }),
+                    )
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let router = build_router(Arc::new(registry), unused_provider());
+        let body = ndjson(&[
+            json!({ "operation": "ingest/spy", "chunk": { "n": 1 } }),
+            json!({ "n": 2 }),
+        ]);
+        let (status, resp) = send(router, raw_request("POST", "/publish", body)).await;
+        let output = resp.get("output").expect("spy handler output");
+        assert_eq!(
+            output.get("ok_chunks"),
+            Some(&json!(0)),
+            "no chunk may reach the handler when the publish_schema failed to compile: {resp}"
+        );
+        assert_eq!(
+            output.get("chunk_count"),
+            Some(&json!(1)),
+            "exactly one item flows: the terminal INTERNAL error (stream ends after it): {resp}"
+        );
+        assert_eq!(
+            output.get("error_codes"),
+            Some(&json!(["INTERNAL"])),
+            "the terminal item is the fail-closed INTERNAL compile error: {resp}"
+        );
+        let _ = status;
+    }
+
+    #[tokio::test]
+    async fn publish_hot_reload_replacement_schema_is_picked_up() {
+        let cache = PublishSchemaCache::new();
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                OperationSpec::new(
+                    "ingest/typed",
+                    OperationType::Pub,
+                    Visibility::External,
+                    json!({}),
+                    json!({}),
+                    vec![],
+                    AccessControl::default(),
+                    None,
+                )
+                .with_publish_schema(json!({
+                    "type": "object",
+                    "properties": { "n": { "type": "integer" } },
+                    "required": ["n"],
+                    "additionalProperties": false
+                })),
+                HandlerKind::Sink(make_sink_handler(|input, ctx, _chunks| async move {
+                    ResponseEnvelope::ok(ctx.request_id, input)
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+
+        let v1 = cache
+            .validator(&registry, "ingest/typed")
+            .unwrap()
+            .expect("compiled validator");
+        assert!(!v1.is_valid(&json!({ "n": 1, "extra": true })));
+
+        // Hot reload: re-register the same op with a schema that drops
+        // `additionalProperties`; the same cache entry must recompile.
+        registry
+            .register(HandlerRegistration::new(
+                OperationSpec::new(
+                    "ingest/typed",
+                    OperationType::Pub,
+                    Visibility::External,
+                    json!({}),
+                    json!({}),
+                    vec![],
+                    AccessControl::default(),
+                    None,
+                )
+                .with_publish_schema(json!({
+                    "type": "object",
+                    "properties": { "n": { "type": "integer" } },
+                    "required": ["n"]
+                })),
+                HandlerKind::Sink(make_sink_handler(|input, ctx, _chunks| async move {
+                    ResponseEnvelope::ok(ctx.request_id, input)
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+
+        let v2 = cache
+            .validator(&registry, "ingest/typed")
+            .unwrap()
+            .expect("recompiled validator");
+        assert!(
+            !Arc::ptr_eq(&v1, &v2),
+            "a changed schema value must invalidate the cached validator"
+        );
+        assert!(
+            v2.is_valid(&json!({ "n": 1, "extra": true })),
+            "the recompiled schema must reflect the new registration"
+        );
+    }
 }

src/gateway/schema_cache.rs

@@ -0,0 +1,258 @@
+//! Compile-once cache for operation `publish_schema` validators
+//! (ADR-046 §4, ADR-068; review-001 post-remediation follow-up).
+//!
+//! The `/publish` route validates each chunk against the op's
+//! `publish_schema`. Compiling the schema per request is both a
+//! per-request CPU/allocation cost on the hot path and — worse — the
+//! GW-01 follow-up fail-open: when compilation fails, chunks flowed to
+//! the handler **unvalidated** behind only a `tracing::warn`, quietly
+//! reopening the transport-dependent validation gap GW-01 closed.
+//!
+//! This cache compiles each op's schema **once** and remembers the
+//! outcome — [`PublishSchemaCache::validator`] returns either a shared
+//! compiled validator or a cached compile failure. The failure is not
+//! retried until the registration changes: an un-compilable schema is a
+//! deployment defect (registry assembly), not a transient condition.
+//!
+//! Invalidation is by schema **value**: the raw `serde_json::Value` is
+//! stored alongside the compiled validator, and a request whose op's
+//! current `publish_schema` differs from the cached one recompiles. This
+//! makes re-registration (hot reload) correct without lifetime coupling
+//! to registry internals: a new schema value is picked up on the next
+//! `/publish` request to that op.
+//!
+//! The registry is defined not to mutate after assembly
+//! (`CachedOpenAPIDoc` relies on the same invariant), so in practice
+//! each schema compiles once per process — the value comparison is
+//! defense in depth for re-assembly paths.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use alkcall::protocol::wire::CallError;
+use alkcall::registry::registration::OperationRegistry;
+use jsonschema::Validator;
+use parking_lot::RwLock;
+use serde_json::Value;
+
+/// Compiled-validator (or recorded compile failure) per operation name.
+#[derive(Clone, Default)]
+pub(crate) struct PublishSchemaCache {
+    entries: Arc<RwLock<HashMap<String, CacheEntry>>>,
+}
+
+#[derive(Clone)]
+enum CacheEntry {
+    Compiled {
+        schema: Value,
+        validator: Arc<Validator>,
+    },
+    /// Compile failed; the error was already logged at compile-attempt
+    /// time and every subsequent request fails closed. Held only so a
+    /// schema whose value still equals the failing one is not
+    /// recompiled (and re-logged) on every request.
+    Failed { schema: Value },
+}
+
+impl PublishSchemaCache {
+    pub(crate) fn new() -> Self {
+        Self::default()
+    }
+
+    /// The compiled validator for `op`'s current `publish_schema`, from
+    /// the cache when the schema value is unchanged, compiled otherwise.
+    ///
+    /// `Ok(None)` means the op has no `publish_schema` (chunks flow
+    /// unvalidated, matching the wire path). `Err` is a **failed**
+    /// compile — cached, loud, never a validator: the caller must fail
+    /// the publish rather than skip validation.
+    pub(crate) fn validator(
+        &self,
+        registry: &OperationRegistry,
+        op: &str,
+    ) -> Result<Option<Arc<Validator>>, CompileFailed> {
+        let Some(schema) = registry
+            .registration(op)
+            .and_then(|reg| reg.spec.publish_schema.clone())
+        else {
+            return Ok(None);
+        };
+        if let Some(cached) = Self::lookup(&self.entries, op, &schema) {
+            return cached;
+        }
+        Self::compile(&self.entries, op, schema)
+    }
+
+    fn lookup(
+        entries: &RwLock<HashMap<String, CacheEntry>>,
+        op: &str,
+        schema: &Value,
+    ) -> Option<Result<Option<Arc<Validator>>, CompileFailed>> {
+        let guard = entries.read();
+        match guard.get(op) {
+            Some(CacheEntry::Compiled {
+                schema: cached_schema,
+                validator,
+            }) if cached_schema == schema => Some(Ok(Some(Arc::clone(validator)))),
+            Some(CacheEntry::Failed {
+                schema: cached_schema,
+            }) if cached_schema == schema => Some(Err(CompileFailed {
+                operation: op.to_string(),
+            })),
+            _ => None,
+        }
+    }
+
+    fn compile(
+        entries: &RwLock<HashMap<String, CacheEntry>>,
+        op: &str,
+        schema: Value,
+    ) -> Result<Option<Arc<Validator>>, CompileFailed> {
+        match jsonschema::options().build(&schema) {
+            Ok(validator) => {
+                let validator = Arc::new(validator);
+                entries.write().insert(
+                    op.to_string(),
+                    CacheEntry::Compiled {
+                        schema,
+                        validator: Arc::clone(&validator),
+                    },
+                );
+                Ok(Some(validator))
+            }
+            Err(error) => {
+                tracing::error!(
+                    operation = %op,
+                    error = %error,
+                    "publish_schema failed to compile; every /publish request to this operation \
+                     fails (fail-closed) until the schema is corrected and re-registered"
+                );
+                entries
+                    .write()
+                    .insert(op.to_string(), CacheEntry::Failed { schema });
+                Err(CompileFailed {
+                    operation: op.to_string(),
+                })
+            }
+        }
+    }
+}
+
+/// A `publish_schema` that could not be compiled (review-001
+/// post-remediation, GW-01 follow-up): the publish request fails closed
+/// with `INTERNAL`; the compile error itself stays in the server log —
+/// an untrusted schema's error text must not be echoed on the wire
+/// (same discipline as the `/openapi.json` cache-miss 500).
+#[derive(Debug)]
+pub(crate) struct CompileFailed {
+    operation: String,
+}
+
+impl CompileFailed {
+    pub(crate) fn call_error(&self) -> CallError {
+        CallError::internal(format!(
+            "publish_schema for operation '{}' failed to compile; publishing is disabled for \
+             this operation",
+            self.operation
+        ))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alkcall::core::types::Capabilities;
+    use alkcall::registry::registration::{
+        make_sink_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+    };
+    use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
+    use serde_json::json;
+
+    fn registry_with_pub_op(name: &str, schema: Option<Value>) -> OperationRegistry {
+        let mut registry = OperationRegistry::new();
+        let mut spec = OperationSpec::new(
+            name,
+            OperationType::Pub,
+            Visibility::External,
+            json!({}),
+            json!({}),
+            vec![],
+            AccessControl::default(),
+            None,
+        );
+        if let Some(schema) = schema {
+            spec = spec.with_publish_schema(schema);
+        }
+        registry
+            .register(HandlerRegistration::new(
+                spec,
+                HandlerKind::Sink(make_sink_handler(|_input, ctx, _chunks| async move {
+                    alkcall::protocol::wire::ResponseEnvelope::ok(ctx.request_id, Value::Null)
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        registry
+    }
+
+    #[test]
+    fn no_publish_schema_yields_ok_none() {
+        let cache = PublishSchemaCache::new();
+        let registry = registry_with_pub_op("ingest/push", None);
+        assert!(matches!(
+            cache.validator(&registry, "ingest/push"),
+            Ok(None)
+        ));
+    }
+
+    #[test]
+    fn unknown_op_yields_ok_none() {
+        let cache = PublishSchemaCache::new();
+        let registry = OperationRegistry::new();
+        assert!(matches!(cache.validator(&registry, "no/such"), Ok(None)));
+    }
+
+    #[test]
+    fn valid_schema_compiles_and_validates() {
+        let cache = PublishSchemaCache::new();
+        let registry = registry_with_pub_op(
+            "ingest/typed",
+            Some(json!({ "type": "object", "required": ["n"] })),
+        );
+        let validator = cache.validator(&registry, "ingest/typed").unwrap().unwrap();
+        assert!(validator.is_valid(&json!({ "n": 1 })));
+        assert!(!validator.is_valid(&json!({})));
+    }
+
+    #[test]
+    fn repeated_lookups_return_the_same_cached_validator() {
+        let cache = PublishSchemaCache::new();
+        let registry = registry_with_pub_op(
+            "ingest/typed",
+            Some(json!({ "type": "object", "required": ["n"] })),
+        );
+        let first = cache.validator(&registry, "ingest/typed").unwrap().unwrap();
+        let second = cache.validator(&registry, "ingest/typed").unwrap().unwrap();
+        assert!(
+            Arc::ptr_eq(&first, &second),
+            "cache must serve the same compiled validator"
+        );
+    }
+
+    #[test]
+    fn uncompilable_schema_fails_closed_and_stays_failed() {
+        let cache = PublishSchemaCache::new();
+        let registry = registry_with_pub_op(
+            "ingest/broken",
+            Some(json!({ "required": "n" })), // `required` must be an array
+        );
+        assert!(cache.validator(&registry, "ingest/broken").is_err());
+        assert!(
+            cache.validator(&registry, "ingest/broken").is_err(),
+            "the cached failure must persist — no retry, no validator"
+        );
+    }
+}

src/server/adapter.rs

@@ -127,6 +127,7 @@ impl HttpAdapter {
             ws_sessions: Arc::clone(&ws_sessions),
             ws_session_slots: Arc::clone(&ws_session_slots),
             ws_idle_timeout,
+            publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(),
         };
         let router = build_router(state, None);
         Self {
@@ -154,6 +155,7 @@ impl HttpAdapter {
             ws_sessions: Arc::clone(&self.ws_sessions),
             ws_session_slots: Arc::clone(&self.ws_session_slots),
             ws_idle_timeout: self.ws_idle_timeout,
+            publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(),
         };
         // `extra_routes` is borrowed, not consumed (SRV-05): a builder
         // call after `with_extra_routes` must keep the custom routes in
@@ -172,6 +174,7 @@ impl HttpAdapter {
             ws_sessions: Arc::clone(&self.ws_sessions),
             ws_session_slots: Arc::clone(&self.ws_session_slots),
             ws_idle_timeout: self.ws_idle_timeout,
+            publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(),
         };
         self.router = build_router(state, Some(routes.clone()));
         self.extra_routes = Some(routes);
@@ -195,6 +198,7 @@ impl HttpAdapter {
             ws_sessions: Arc::clone(&self.ws_sessions),
             ws_session_slots: Self::rebuild_session_slots(max_sessions),
             ws_idle_timeout: self.ws_idle_timeout,
+            publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(),
         };
         self.router = build_router(state, self.extra_routes.clone());
         self
@@ -217,6 +221,7 @@ impl HttpAdapter {
             ws_sessions: Arc::clone(&self.ws_sessions),
             ws_session_slots: Arc::clone(&self.ws_session_slots),
             ws_idle_timeout: self.ws_idle_timeout,
+            publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(),
         };
         self.router = build_router(state, self.extra_routes.clone());
         self
@@ -1025,6 +1030,7 @@ mod tests {
                 crate::websocket::DEFAULT_WS_MAX_SESSIONS,
             )),
             ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
+            publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(),
         }
     }
 

src/server/state.rs

@@ -42,6 +42,10 @@ pub(crate) struct RouterState {
     pub(crate) ws_session_slots: Arc<tokio::sync::Semaphore>,
     /// Idle-read timeout for the WS pumps (WS-01); `None` disables.
     pub(crate) ws_idle_timeout: Option<std::time::Duration>,
+    /// Compile-once `publish_schema` validator cache for the `/publish`
+    /// route (GW-01 follow-up); built empty at adapter construction and
+    /// populated lazily per operation.
+    pub(crate) publish_schemas: crate::gateway::schema_cache::PublishSchemaCache,
 }
 
 impl axum::extract::FromRef<RouterState> for crate::websocket::SessionState {
@@ -102,6 +106,7 @@ mod tests {
                 crate::websocket::DEFAULT_WS_MAX_SESSIONS,
             )),
             ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
+            publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(),
         };
         let extracted: DecoyConfig = axum::extract::FromRef::from_ref(&state);
         assert!(matches!(extracted, DecoyConfig::Redirect { .. }));

tasks/gateway/review-001-publish-schema-validation-robust.md

@@ -1,7 +1,7 @@
 ---
 id: review-001-publish-schema-validation-robust
 name: Fix /publish schema validation fail-open + per-request recompilation (post-remediation)
-status: pending
+status: completed
 depends_on: []
 scope: narrow
 risk: high
@@ -41,11 +41,11 @@ schema originates), with the gateway cache as defense in depth.
 
 ## Acceptance Criteria
 
-- [ ] Compiled validator (or error) cached per registration; no per-request recompile (test: compile count / perf shape not asserted, but code path is registration-keyed)
+- [x] Compiled validator (or error) cached per registration; no per-request recompile (test: compile count / perf shape not asserted, but code path is registration-keyed)
-- [ ] Un-compilable `publish_schema` → request fails loudly (500/INTERNAL, error logged), chunks never flow unvalidated (test)
+- [x] Un-compilable `publish_schema` → request fails loudly (500/INTERNAL, error logged), chunks never flow unvalidated (test)
-- [ ] Schema re-registration (hot reload) picks up the new schema (test)
+- [x] Schema re-registration (hot reload) picks up the new schema (test)
-- [ ] A `publish_schema`-registered Pub op still rejects an invalid chunk (existing GW-01 gate stays green)
+- [x] A `publish_schema`-registered Pub op still rejects an invalid chunk (existing GW-01 gate stays green)
-- [ ] `cargo test` and `cargo clippy --all-targets -- -D warnings` pass
+- [x] `cargo test` and `cargo clippy --all-targets -- -D warnings` pass
 
 ## References
 
@@ -55,13 +55,78 @@ schema originates), with the gateway cache as defense in depth.
 
 ## Notes
 
-> Agent fills during implementation. Found in the post-remediation
+Mechanism: new `src/gateway/schema_cache.rs` —
-> sweep — not a Review 001 finding itself. Fail-open is the priority
+`PublishSchemaCache`, a cloneable `Arc<RwLock<HashMap<String,
-> half; caching is the efficiency half. Registration-side (adapter)
+CacheEntry>>>` held on `RouterState` (built empty per
-> rejection at import is the preferred end state if alkcall's
+`HttpAdapter`/router construction; `GatewayState` extracts it via
-> registration surface allows observing `publish_schema` at
+`FromRef`). Each entry stores the compiled `Arc<jsonschema::Validator>`
-> HandlerRegistration time.
+(or a `Failed` marker) **plus the raw schema `Value`** that produced
+it. Lookup compares the registry's current `publish_schema` value
+against the cached one: equal → serve the cached validator/failure;
+different → recompile. This gives hot-reload correctness by value
+equality without lifetime-coupling to registry internals; under the
+documented assemble-then-serve invariant (the same one
+`CachedOpenAPIDoc` relies on) each schema compiles exactly once per
+process, and the value check is defense in depth for re-assembly
+paths.
+
+Fail-closed semantics: a failed compile is **cached as a failure** —
+the error is logged once at `error` level (never retried or re-logged
+per request), and the wire message is a generic INTERNAL naming the
+compile failure without echoing schema internals (an untrusted
+schema's error text must not reach the wire; same discipline as the
+`/openapi.json` cache-miss 500).
+
+Resolution is **lazy**, resolved against the cache inside the chunk
+stream's first poll, not in the handler prologue: `invoke_sink` owns
+the 404/403/422 pre-checks (GW-11), so an unknown/internal/wrong-type
+op is rejected before any schema lookup or compile happens — no cache
+population or error-log spam from ops a caller cannot reach.
+
+Stream terminality fix (found by the spy-handler test): the old
+`NdjsonChunkStream` yielded `Err` items but kept streaming — a handler
+that drained the stream (instead of aborting on first `Err`) would
+have kept receiving **unvalidated** chunks after a violation or a
+compile failure. The wire pump (`alkcall dispatch.rs:599-600`) does
+`send(Err)` + `break`, i.e. the error item is terminal and the channel
+closes. `NdjsonChunkStream` now mirrors that exactly: the first `Err`
+item (schema violation, bad JSON, read failure, or compile failure)
+sets `done` and the stream ends — `Ok` chunks can never flow after an
+error on this transport either.
+
+Adapter-side registration-time rejection was considered and **not**
+pursued here: alkcall's `OperationRegistry::register` does not observe
+`publish_schema` compilability, and adding that would be an alkcall
+surface change (this crate consumes `alkcall = "0.1.1"` from crates-
+io) — the gateway cache is the defense in depth the task prescribed.
+Worth noting for alkcall: the identical compile-fail-open pattern
+exists on the wire path (`alkcall/src/protocol/dispatch.rs:356-366`,
+warn-and-skip) — upstream fix candidate, out of scope for this crate.
 
 ## Summary
 
-> Filled on completion.
+Implemented in three parts:
+
+- **`src/gateway/schema_cache.rs`**: `PublishSchemaCache` — compile-
+  once, value-keyed invalidation, cached failures, `error!`-level compile
+  logging, generic non-leaking wire error (`CompileFailed::call_error`
+  → `CallError::internal`). 5 unit tests (no-schema/unknown-op →
+  `Ok(None)`; compile+validate; `Arc::ptr_eq` cache identity; failure
+  stays failed).
+- **`src/gateway/routes.rs`**: `publish_handler` uses the cache;
+  `PublishSchemaState` (Unresolved → Unvalidated/Validated/Failed)
+  resolves lazily on first chunk poll, after `invoke_sink`'s
+  pre-checks; `NdjsonChunkStream` is terminal-on-first-`Err` (wire
+  parity). Route tests: uncompilable schema → 500 `INTERNAL` with
+  non-echoing message; spy handler receives 0 `Ok` chunks and exactly
+  one terminal `INTERNAL` error; hot-reload re-registration picks up
+  the replacement schema (validator identity + semantics asserted);
+  all pre-existing GW-01 gates (reject-invalid, accept-valid,
+  first-chunk validation, no-schema passthrough) stay green.
+- **`src/server/{state,adapter}.rs`** + `gateway/mod.rs`: cache plumbed
+  through `RouterState`/`GatewayState` at all construction sites.
+
+Verification: `cargo test` 308 passed (default), `--all-features` all
+suites green (379+), `cargo clippy --all-targets -- -D warnings`
+clean (default + all-features), `cargo fmt --check` clean.
+`--no-default-features` warnings are pre-existing on the base commit.