refactor(gateway): migrate to alkcall 0.2 promoted gateway module

Bump the alkcall dependency to 0.2 (with the gateway feature) and
converge on the promoted shared pieces:

- The local dispatch spine (gateway/dispatch.rs, 721 lines) is deleted;
  GatewayDispatch, schema_disclosure_denial, and DEFAULT_DEADLINE are
  re-exported from alkcall::gateway (alkcall ADR-048). The 30 s default
  deadline preserves the previous behavior exactly.
- gateway/schema_cache.rs (PublishSchemaCache) is deleted: alkcall CF-003
  compiles publish_schema at registration time and exposes
  OperationRegistry::publish_validator; the /publish chunk stream
  resolves against it. Un-compilable schemas are now rejected at
  registration, so the two end-to-end fail-closed tests were reworked
  into a registration-rejection test (a stronger guarantee).
- schema_disclosure_denial consumers (to_mcp, routes) use alkcall's
  promoted implementation; the alkhttp-local copy is gone (ADR-071
  updated: the guard stays as defense-in-depth, the implementation no
  longer forks).
- CF-001: from_wss drop monitor and the WS overlay tests use
  CallError::connection_closed; the review-001-ws-eof-signal race tests
  now assert retryable CONNECTION_CLOSED on both resolution paths (the
  tolerated non-retryable INTERNAL write-failure outcome is gone).
- Added CHANGELOG.md (Keep a Changelog), Unreleased section records the
  bump and convergence.

Verification: cargo test default 453 ok, wss 470 ok, mcp 526 ok,
all-features 575 ok; clippy -D warnings clean (default + all-features,
all-targets); fmt clean; cargo doc warning-free.

Net: -1093 lines.

CHANGELOG.md

@@ -0,0 +1,46 @@
+# Changelog
+
+All notable changes to this crate are documented here. The format is
+based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and
+this crate adheres to [Semantic Versioning](https://semver.org/).
+
+## [Unreleased]
+
+### Changed
+
+- **`alkcall` dependency bumped to 0.2** with the `gateway` feature
+  enabled. The transport-neutral dispatch spine now comes from
+  `alkcall::gateway` (promoted from this crate per alkcall ADR-048):
+  `GatewayDispatch`, `schema_disclosure_denial`, and
+  `DEFAULT_DEADLINE` are re-exported from `alkhttp::gateway`; the
+  local copy (`gateway/dispatch.rs`) was deleted. The spine is used
+  with the 30 s default deadline (identical behavior to the local
+  version). alkcall 0.2 also carries the consumer-findings fixes
+  CF-001..004 documented in
+  `alkcall/docs/reviews/consumer-findings-ledger.md`.
+- **`/publish` chunk validation uses the registry-owned validator**
+  (CF-003). The local compile-once `PublishSchemaCache`
+  (`gateway/schema_cache.rs`) is removed — `publish_schema` is
+  compiled at **registration time** by alkcall and exposed as
+  `OperationRegistry::publish_validator`; an un-compilable schema is
+  now a registration error, so the fail-open window this cache closed
+  compiled at **registration time** by alkcall and exposed as
+  `OperationRegistry::publish_validator`; an un-compilable schema is
+  now a registration error, so the fail-open window this cache closed
+  GW-side cannot exist upstream either. The `/publish` chunk stream
+  resolves against the registry's cached validator.
+- The from_wss drop monitor and the WS overlay tests use
+  `CallError::connection_closed` (CF-001), and the
+  review-001-ws-eof-signal race tests now assert retryable
+  `CONNECTION_CLOSED` on **both** resolution paths (the drop monitor
+  and the undelivered-request write failure) — the previously
+  tolerated non-retryable `INTERNAL: failed to write request frame`
+  outcome on the race path is gone.
+
+### Notes
+
+- **CF-004 convergence:** the alkhttp-local `services/schema`
+  defense-in-depth guard remains, but its shared check
+  (`schema_disclosure_denial`) is now alkcall's promoted
+  implementation (ADR-048); the alkhttp-local copy is gone. See
+  ADR-071 for the updated disposition.

Cargo.lock

@@ -27,9 +27,9 @@ dependencies = [
 
 [[package]]
 name = "alkcall"
-version = "0.1.1"
+version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "49664a13ea571d7a6e59c9ed52d165dfdfe389c9a93bb301cc1fb86f29bae8bd"
+checksum = "83e91782140beee66656cda7c5bfb1248f76302225b51538d3523b51d178ac7d"
 dependencies = [
  "async-trait",
  "bytes",

Cargo.toml

@@ -22,12 +22,13 @@ h2 = ["dep:hyper", "hyper-util/http2", "hyper/http2"]
 http1 = ["dep:hyper", "hyper-util/http1", "hyper/http1"]
 
 [dependencies]
-alkcall = "0.1.1"
+alkcall = { version = "0.2", features = ["gateway"] }
 arc-swap = "1"
 axum = { version = "0.8", features = ["ws"] }
 hyper = { version = "1", optional = true, features = ["server"] }
 hyper-util = { version = "0.1", features = ["server", "service", "tokio"] }
 httpdate = "1"
+jsonschema = { version = "0.46", default-features = false }
 reqwest = { version = "0.13", default-features = false, features = ["json", "stream", "rustls"] }
 reqwest-middleware = "0.5"
 reqwest-retry = "0.9"
@@ -45,7 +46,6 @@ http = "1"
 http-body-util = "0.1"
 url = "2"
 percent-encoding = "2"
-jsonschema = { version = "0.46", default-features = false }
 parking_lot = "0.12"
 rmcp = { version = "1.8", optional = true, default-features = false, features = [
     "client",

docs/architecture/README.md

@@ -68,7 +68,7 @@ types — the former `alknet-core` and `alknet-call` merged).
 | [068](decisions/068-gateway-publish-endpoint.md) | Gateway `/publish` Endpoint | 6th gateway endpoint for `OperationType::Pub` (producer→consumer streaming); NDJSON request body → `call.published` chunks |
 | [069](decisions/069-webtransport-out-of-scope.md) | WebTransport Out of Scope | h3/WebTransport removed from alkhttp scope entirely (an alknet concern); supersedes the deferral framing of ADR-044 |
 | [070](decisions/070-from-wss-consumer-adapter.md) | `from_wss` Consumer Adapter | Import a remote node's operations over WSS — same-protocol importer, channels-over-WS as transport; `wss` feature gate |
-| [071](decisions/071-dispatch-schema-guard.md) | Dispatch-Spine `services/schema` Op-Path Guard | `GatewayDispatch` applies the GET `/schema` visibility+ACL checks to the meta-op's inner `name` (review-002 PRJ-16); alkcall CF-004 is the complete fix, this stays as defense-in-depth |
+| [071](decisions/071-dispatch-schema-guard.md) | Dispatch-Spine `services/schema` Op-Path Guard | `GatewayDispatch` applies the GET `/schema` visibility+ACL checks to the meta-op's inner `name` (review-002 PRJ-16); alkcall CF-004 is the complete fix, this stays as defense-in-depth (now via the promoted `alkcall::gateway` shared check — spine + guard themselves live in alkcall per ADR-048) |
 
 ## Relevant Open Questions
 

docs/architecture/decisions/071-dispatch-schema-guard.md

@@ -53,23 +53,53 @@ registry rejects non-`Pub` operations before the handler runs and the
 sink ignores the input, so nothing is projected there.
 
 The visibility + ACL check itself is one shared function
-(`gateway::dispatch::schema_disclosure_denial`) used by the HTTP GET
+(`schema_disclosure_denial`) used by the HTTP GET `/schema` route, the
-`/schema` route, the dispatch-spine guard, and the MCP `schema` tool,
+dispatch-spine guard, and the MCP `schema` tool, so the transports
-so the transports cannot drift.
+cannot drift.
 
 ## Consequences
 
 - **Transport-level invariant (PRJ-16's wording):** no alkhttp
   transport can fetch, through any path, a spec the GET `/schema`
   route would deny for the same identity.
-- **When CF-004 lands, the guard stays.** The alkcall-side handler
+- **When CF-004 landed, the guard stayed — and was promoted.** The
-  check is the complete fix for every transport (wire, overlays, peer
+  alkcall-side handler check is the complete fix for every transport
-  composition); this per-transport check remains as defense-in-depth.
+  (wire, overlays, peer composition; alkcall commit `8cb2a6e`). The
-  Do not remove it.
+  alkhttp spine, the GET `/schema` route, the MCP `schema` tool, and
-- Wire-observable behavior changes only for the previously-leaking
+  the spine's `services/schema` op-path guard all now converge on the
-  requests: they now get the same 404/403 the GET route returns.
+  **shared promoted implementation**: `alkcall::gateway::
-  Legitimate `services/schema` calls (allowed inner names) are
+  schema_disclosure_denial` (alkcall ADR-048). The local copy was
-  unaffected.
+  deleted on the alkcall-0.2 bump — one implementation, so the
+  transports still cannot drift, and no alkhttp-local fork remains.
+  Do not re-introduce a local copy.
+- **The dispatch spine itself is promoted too.** `GatewayDispatch`
+  (the `invoke`/`invoke_streaming`/`invoke_sink` spine with the
+  root-`OperationContext` invariants and the inner-`name` guard) is
+  `alkcall::gateway::GatewayDispatch` (feature `gateway`, enabled
+  unconditionally by the alkcall dependency — the gateway endpoints
+  are this crate's sole invoke path, ADR-047). The alkhttp-local copy
+  was deleted; only the HTTP-specific layers remain here (the
+  CallError → HTTP mapping in `gateway::error`, NDJSON/SSE framing,
+  body caps, batch envelopes, and this ADR's original
+  FORBIDDEN-vs-spec-404 split on the GET route).
+- **`/publish` chunk validation is registry-backed.** The local
+  compile-once `PublishSchemaCache` was deleted: alkcall CF-003
+  compiles `publish_schema` at registration time and exposes the
+  cached validator as `OperationRegistry::publish_validator`. The
+  `/publish` chunk stream resolves against that cache; the
+  fail-closed guarantee moved upstream with it (an un-compilable
+  schema is a registration error, so an unvalidated-ingest path
+  cannot be constructed).
+- **CF-001 resolution:** alkcall's `CallError::connection_closed`
+  (retryable `CONNECTION_CLOSED`) covers the write-failure race; the
+  from_wss drop-monitor race tests assert retryable-only (the
+  previously-tolerated non-retryable `INTERNAL: failed to write
+  request frame` outcome is gone from the wire vocabulary on the
+  undelivered-request path).
+- Wire-observable behavior is unchanged from the pre-promotion guard:
+  previously-leaking requests still get the same 404/403 the GET
+  route returns; legitimate `services/schema` calls (allowed inner
+  names) are unaffected.
 - The guard matches the outer registration's `spec.name` against the
   `services/schema` constant rather than the raw request string, so
   leading-slash variants (`/services/schema`) hit the same check.

src/adapters/from_wss.rs

@@ -33,7 +33,8 @@
 //! fails in-flight calls retryable: alkcall's client-side read pump only
 //! routes envelopes and does not observe EOF, so the adapter owns drop
 //! semantics — the [`WssSession`] monitor awaits the WS read pump's EOF
-//! signal and fails all pending calls with retryable `CONNECTION_CLOSED`.
+//! signal and fails all pending calls with retryable `CONNECTION_CLOSED`
+//! (alkcall's `CallError::connection_closed`, CF-001).
 //! The EOF signal is lossless (a retained watch value, WS-02): EOF is
 //! observed even if it fires before the monitor starts, or after the
 //! session was forgotten (fire-and-forget import). Once EOF has been
@@ -86,7 +87,7 @@ const PENDING_SWEEP_INTERVAL: std::time::Duration = std::time::Duration::from_mi
 const PENDING_SWEEP_MAX_POST_EOF: u32 = 8;
 
 fn connection_closed_error() -> CallError {
-    CallError::new("CONNECTION_CLOSED", "from_wss connection dropped", true)
+    CallError::connection_closed("from_wss connection dropped")
 }
 
 /// The `from_wss` consumer adapter (wss feature): dials a remote WS
@@ -978,15 +979,16 @@ mod tests {
             .expect("join");
         match response.result {
             Err(e) => {
-                if e.code == "INTERNAL" && e.message.contains("failed to write request frame") {
+                assert!(
-                    // The call registered but the mux was torn down
+                    e.retryable,
-                    // before the request frame was written — the
+                    "drop error must be retryable regardless of whether the call resolved via \
-                    // write-failure path resolved it (no hang; the
+                     the drop monitor (CONNECTION_CLOSED) or the CF-001 write-failure mapping \
-                    // retryable variant is asserted by the sweep test
+                     (also CONNECTION_CLOSED post-CF-001), got {e:?}"
-                    // and the pre-drop test).
+                );
-                } else {
+                assert_eq!(
-                    assert!(e.retryable, "drop error must be retryable, got {e:?}");
+                    e.code, "CONNECTION_CLOSED",
-                }
+                    "both resolution paths share the retryable wire code post-CF-001: {e:?}"
+                );
             }
             Ok(_) => panic!("expected Err after connection drop"),
         }

src/adapters/to_mcp.rs

@@ -292,15 +292,15 @@ fn json_type_name(value: &Value) -> &'static str {
 /// invisible (NOT_FOUND regardless of caller), ACL-forbidden ops
 /// return FORBIDDEN. The MCP `schema` tool must never return the full
 /// spec of an op the caller could not call. The check is the shared
-/// `schema_disclosure_denial` (review-002 PRJ-16) so the HTTP route,
+/// `schema_disclosure_denial` (review-002 PRJ-16, promoted to alkcall
-/// the dispatch-spine `services/schema` guard, and this tool cannot
+/// ADR-048) so the HTTP route, the dispatch-spine `services/schema`
-/// drift.
+/// guard, and this tool cannot drift.
 fn schema_visibility_and_access_denial(
     registry: &OperationRegistry,
     operation: &str,
     identity: Option<&Identity>,
 ) -> Option<CallError> {
-    crate::gateway::schema_disclosure_denial(registry, operation, identity)
+    alkcall::gateway::schema_disclosure_denial(registry, operation, identity)
 }
 
 fn map_search_response(response: ResponseEnvelope, query: Option<&str>) -> CallToolResult {

src/gateway/dispatch.rs

@@ -1,721 +0,0 @@
-//! Shared dispatch spine for the `to_openapi` / `to_mcp` gateway
-//! projections.
-//!
-//! Thin concrete struct (not a trait — the alknet research ruled out a
-//! trait with an associated output type). Holds `Arc<OperationRegistry>`
-//! and exposes an `invoke()` family returning the neutral
-//! `ResponseEnvelope` — identity is supplied per-call, resolved upstream
-//! in the auth middleware. Each
-//! gateway maps the envelope to its own wire shape (`to_openapi` → HTTP
-//! `Response`, `to_mcp` → `CallToolResult`).
-//!
-//! # Security invariants
-//!
-//! - `internal: false` — ACL runs against the caller's `identity`, not a
-//!   handler's composition authority (alkcall ADR-017).
-//! - `forwarded_for: None` — wire-ingress only.
-//!
-//! The root `OperationContext` is constructed identically for both
-//! gateways, making them provably identical on the security axis (auth,
-//! authority, ACL); they diverge only on wire-framing.
-//!
-//! # Deadline
-//!
-//! Once-op invokes ([`GatewayDispatch::invoke`]) and sink invokes
-//! ([`GatewayDispatch::invoke_sink`]; GW-17) are bounded by
-//! `DEFAULT_TIMEOUT` (30 s): the registry invoke is wrapped in
-//! `tokio::time::timeout` and a hung handler surfaces as a `TIMEOUT`
-//! error envelope (`504` under the gateway's error mapping), not an
-//! indefinitely-held HTTP request. The sink wrapper bounds the whole
-//! dispatch — chunk upload pacing and the handler's final-completion
-//! await alike — so the final envelope always arrives (or the deadline
-//! trips) within the window http-server.md documents. Streaming
-//! dispatch sets `deadline: None` (subscriptions are unbounded per
-//! alkcall ADR-021).
-//!
-//! # The `services/schema` op-path guard (review-002 PRJ-16)
-//!
-//! The registry-level pre-checks fire on the **outer** operation name
-//! only, but `services/schema` is an External op whose *input* names
-//! another operation — and the underlying handler does a bare
-//! `registry.registration(name)` → spec projection with no visibility
-//! or AccessControl check of its own (alkcall CF-004). Without a guard,
-//! `POST /call {"operation":"services/schema","input":{"name":X}}`
-//! returns the full spec of any Internal op, defeating the GET
-//! `/schema` fix (GW-02/SRV-02) through the op path.
-//! [`GatewayDispatch::invoke`] and [`GatewayDispatch::invoke_streaming`]
-//! therefore apply the same is-internal + access-control checks to the
-//! inner `name` that GET `/schema` applies
-//! (404 for Internal, FORBIDDEN for ACL denial) before dispatching.
-//! This is the alkhttp-local layer of the fix; the complete fix is the
-//! alkcall-side handler check (CF-004,
-//! `alkcall/docs/reviews/consumer-findings-ledger.md`). When CF-004
-//! lands, this guard remains as defense-in-depth: the per-transport
-//! check stays so no alkhttp transport can fetch a spec the GET
-//! `/schema` route would deny for the same identity.
-//!
-//! Sink dispatch (`invoke_sink`, the `/publish` path) cannot reach
-//! `services/schema`: the registry rejects non-`Pub` operations before
-//! the handler runs, and the sink handler never reads the input, so
-//! no spec is projected there.
-
-use std::collections::HashMap;
-use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::Arc;
-use std::time::{Duration, Instant};
-
-use alkcall::core::auth::Identity;
-use alkcall::core::types::Capabilities;
-use alkcall::protocol::wire::{CallError, ResponseEnvelope};
-use alkcall::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
-use alkcall::registry::env::LocalOperationEnv;
-use alkcall::registry::registration::OperationRegistry;
-use alkcall::registry::spec::{AccessResult, Visibility};
-use futures::stream::BoxStream;
-use serde_json::Value;
-
-const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
-
-const SERVICES_SCHEMA: &str = "services/schema";
-
-/// The shared dispatch spine over the registry, invoking operations for
-/// the neutral `ResponseEnvelope` result shape both gateway projections
-/// map to their wire formats. Identity arrives per-call as
-/// `Option<Identity>` — bearer resolution happens upstream in the auth
-/// middleware, not here.
-pub struct GatewayDispatch {
-    registry: Arc<OperationRegistry>,
-    invoke_count: AtomicUsize,
-}
-
-impl GatewayDispatch {
-    /// Assemble a dispatch spine over a registry.
-    pub fn new(registry: Arc<OperationRegistry>) -> Self {
-        Self {
-            registry,
-            invoke_count: AtomicUsize::new(0),
-        }
-    }
-
-    /// The registry operations resolve against.
-    pub fn registry(&self) -> &Arc<OperationRegistry> {
-        &self.registry
-    }
-
-    /// How many [`GatewayDispatch::invoke`] calls this spine has
-    /// served. A test-spy accessor: the over-cap batch tests assert it
-    /// stays at zero to prove no dispatch happened before the cap
-    /// rejection (review-002 PRJ-22).
-    pub fn invoke_count(&self) -> usize {
-        self.invoke_count.load(Ordering::Relaxed)
-    }
-
-    /// Invoke a Query/Mutation op under the 30 s gateway deadline; a
-    /// hung handler surfaces as a `TIMEOUT` error envelope (504).
-    pub async fn invoke(
-        &self,
-        identity: Option<Identity>,
-        op: &str,
-        input: Value,
-    ) -> ResponseEnvelope {
-        self.invoke_count.fetch_add(1, Ordering::Relaxed);
-        let operation_name = strip_leading_slash(op).to_string();
-        if let Some(error) =
-            schema_via_call_denial(&self.registry, &operation_name, &input, identity.as_ref())
-        {
-            return ResponseEnvelope::error(uuid::Uuid::new_v4().to_string(), error);
-        }
-        let request_id = uuid::Uuid::new_v4().to_string();
-        let context = self.build_root_context(&request_id, &operation_name, identity);
-        let result = tokio::time::timeout(
-            DEFAULT_TIMEOUT,
-            self.registry.invoke(&operation_name, input, context),
-        )
-        .await;
-        match result {
-            Ok(envelope) => envelope,
-            Err(_elapsed) => ResponseEnvelope::error(
-                request_id,
-                CallError::timeout(format!(
-                    "operation did not complete within the {DEFAULT_TIMEOUT:?} gateway deadline"
-                )),
-            ),
-        }
-    }
-
-    /// Dispatch a Sub op: the returned stream of envelopes is unbounded
-    /// by the deadline (subscriptions are long-lived per alkcall
-    /// ADR-021); pre-handler failures surface as one error envelope.
-    pub fn invoke_streaming(
-        &self,
-        identity: Option<Identity>,
-        op: &str,
-        input: Value,
-    ) -> BoxStream<'static, ResponseEnvelope> {
-        let operation_name = strip_leading_slash(op).to_string();
-        if let Some(error) =
-            schema_via_call_denial(&self.registry, &operation_name, &input, identity.as_ref())
-        {
-            let request_id = uuid::Uuid::new_v4().to_string();
-            return Box::pin(futures::stream::once(async move {
-                ResponseEnvelope::error(request_id, error)
-            }));
-        }
-        let request_id = uuid::Uuid::new_v4().to_string();
-        let context = self.build_root_context_streaming(&request_id, &operation_name, identity);
-        self.registry
-            .invoke_streaming(&operation_name, input, context)
-    }
-
-    /// Dispatch a `Pub` operation (ADR-046, ADR-068): the
-    /// `publish_stream` is the initiator's chunk stream (one
-    /// `Ok(Value)` per published chunk); the handler's final
-    /// `ResponseEnvelope` is the result. Pre-handler failures
-    /// (not-found, forbidden, non-Pub op) surface as a single error
-    /// envelope — the same envelope the `/publish` route maps to HTTP.
-    /// The whole dispatch — chunk streaming and the handler's final
-    /// await alike — is bounded by the same 30 s gateway deadline as
-    /// the Once-op invoke (GW-17): a hung sink handler surfaces as a
-    /// `TIMEOUT` error envelope (504 under the gateway's error
-    /// mapping), not an indefinitely-held HTTP request.
-    pub async fn invoke_sink(
-        &self,
-        identity: Option<Identity>,
-        op: &str,
-        input: Value,
-        publish_stream: alkcall::registry::registration::PublishStream,
-    ) -> ResponseEnvelope {
-        let operation_name = strip_leading_slash(op).to_string();
-        let request_id = uuid::Uuid::new_v4().to_string();
-        let context = self.build_root_context_sink(&request_id, &operation_name, identity);
-        let result = tokio::time::timeout(
-            DEFAULT_TIMEOUT,
-            self.registry
-                .invoke_sink(&operation_name, input, publish_stream, context),
-        )
-        .await;
-        match result {
-            Ok(envelope) => envelope,
-            Err(_elapsed) => ResponseEnvelope::error(
-                request_id,
-                CallError::timeout(format!(
-                    "operation did not complete within the {DEFAULT_TIMEOUT:?} gateway deadline"
-                )),
-            ),
-        }
-    }
-
-    fn build_root_context_sink(
-        &self,
-        request_id: &str,
-        operation_name: &str,
-        identity: Option<Identity>,
-    ) -> OperationContext {
-        self.build_root_context_inner(request_id, operation_name, identity, false)
-    }
-
-    fn build_root_context(
-        &self,
-        request_id: &str,
-        operation_name: &str,
-        identity: Option<Identity>,
-    ) -> OperationContext {
-        self.build_root_context_inner(request_id, operation_name, identity, true)
-    }
-
-    fn build_root_context_streaming(
-        &self,
-        request_id: &str,
-        operation_name: &str,
-        identity: Option<Identity>,
-    ) -> OperationContext {
-        self.build_root_context_inner(request_id, operation_name, identity, false)
-    }
-
-    fn build_root_context_inner(
-        &self,
-        request_id: &str,
-        operation_name: &str,
-        identity: Option<Identity>,
-        bounded: bool,
-    ) -> OperationContext {
-        let registration = self.registry.registration(operation_name);
-        let (composition_authority, capabilities, scoped_env) = match registration {
-            Some(r) => (
-                r.composition_authority.clone(),
-                r.capabilities.clone(),
-                r.scoped_env.clone().unwrap_or_else(ScopedPeerEnv::empty),
-            ),
-            None => (None, Capabilities::new(), ScopedPeerEnv::empty()),
-        };
-
-        let env: Arc<dyn alkcall::registry::env::OperationEnv + Send + Sync> =
-            Arc::new(LocalOperationEnv::new(Arc::clone(&self.registry)));
-
-        OperationContext {
-            request_id: request_id.to_string(),
-            parent_request_id: None,
-            identity,
-            handler_identity: composition_authority,
-            forwarded_for: None,
-            capabilities,
-            metadata: HashMap::new(),
-            deadline: bounded.then(|| Instant::now() + DEFAULT_TIMEOUT),
-            scoped_env,
-            env,
-            abort_policy: AbortPolicy::default(),
-            internal: false,
-            ownership: None,
-        }
-    }
-}
-
-fn strip_leading_slash(operation_id: &str) -> &str {
-    operation_id.strip_prefix('/').unwrap_or(operation_id)
-}
-
-/// The `services/schema` op-path guard (review-002 PRJ-16): when the
-/// dispatched operation is the schema meta-op, apply the GET `/schema`
-/// route's visibility + AccessControl checks to the **inner** `name`
-/// input — 404 for Internal, FORBIDDEN for ACL denial — so no transport
-/// through this spine can fetch a spec the GET `/schema` route would
-/// deny for the same identity. The complete fix is the alkcall-side
-/// handler check (CF-004); this guard stays as defense-in-depth after
-/// that lands.
-fn schema_via_call_denial(
-    registry: &OperationRegistry,
-    operation: &str,
-    input: &Value,
-    identity: Option<&Identity>,
-) -> Option<CallError> {
-    let name = input.get("name").and_then(Value::as_str)?;
-    if !matches!(
-        registry.registration(operation),
-        Some(registration) if registration.spec.name == SERVICES_SCHEMA
-    ) {
-        return None;
-    }
-    schema_disclosure_denial(registry, name, identity)
-}
-
-/// The per-transport schema-disclosure check shared by the HTTP GET
-/// `/schema` route, the dispatch-spine `services/schema` guard, and the
-/// MCP `schema` tool: Internal ops are invisible (404 / NOT_FOUND
-/// regardless of caller), ACL-forbidden ops are denied (403 /
-/// FORBIDDEN). One implementation so the transports cannot drift.
-pub(crate) fn schema_disclosure_denial(
-    registry: &OperationRegistry,
-    operation: &str,
-    identity: Option<&Identity>,
-) -> Option<CallError> {
-    let name = strip_leading_slash(operation);
-    let registration = registry.registration(name)?;
-    if registration.spec.visibility == Visibility::Internal {
-        return Some(CallError::not_found(operation));
-    }
-    if let AccessResult::Forbidden(message) =
-        registration.spec.access_control.check(identity, None, None)
-    {
-        return Some(CallError::forbidden(message));
-    }
-    None
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use alkcall::registry::registration::{
-        make_handler, make_sink_handler, make_streaming_handler, HandlerKind, HandlerRegistration,
-        OperationProvenance,
-    };
-    use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
-    use futures::StreamExt;
-
-    // The tests below run against the module's real 30 s deadline; only
-    // the elapsed-time bound (< 60 s, anti-flake) is asserted by hand.
-
-    fn spec(name: &str, visibility: Visibility, op_type: OperationType) -> OperationSpec {
-        OperationSpec::new(
-            name,
-            op_type,
-            visibility,
-            serde_json::json!({}),
-            serde_json::json!({}),
-            vec![],
-            AccessControl::default(),
-            None,
-        )
-    }
-
-    fn echo_registry() -> Arc<OperationRegistry> {
-        let mut registry = OperationRegistry::new();
-        registry
-            .register(HandlerRegistration::new(
-                spec("echo/run", Visibility::External, OperationType::Query),
-                HandlerKind::Once(make_handler(|input, ctx| async move {
-                    ResponseEnvelope::ok(ctx.request_id, input)
-                })),
-                OperationProvenance::Local,
-                None,
-                None,
-                Capabilities::new(),
-            ))
-            .unwrap();
-        Arc::new(registry)
-    }
-
-    #[tokio::test]
-    async fn invoke_external_op_round_trips() {
-        let dispatch = GatewayDispatch::new(echo_registry());
-        let envelope = dispatch
-            .invoke(None, "/echo/run", serde_json::json!({ "x": 1 }))
-            .await;
-        assert!(envelope.result.is_ok(), "expected ok, got {envelope:?}");
-    }
-
-    #[tokio::test]
-    async fn invoke_unknown_op_returns_not_found() {
-        let dispatch = GatewayDispatch::new(echo_registry());
-        let envelope = dispatch
-            .invoke(None, "/missing/op", serde_json::json!({}))
-            .await;
-        match envelope.result {
-            Err(error) => assert_eq!(error.code, "NOT_FOUND"),
-            Ok(v) => panic!("expected error, got {v:?}"),
-        }
-    }
-
-    #[tokio::test]
-    async fn invoke_internal_op_returns_not_found() {
-        let mut registry = OperationRegistry::new();
-        registry
-            .register(HandlerRegistration::new(
-                spec("internal/op", Visibility::Internal, OperationType::Query),
-                HandlerKind::Once(make_handler(|_input, ctx| async move {
-                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
-                })),
-                OperationProvenance::Local,
-                None,
-                None,
-                Capabilities::new(),
-            ))
-            .unwrap();
-        let dispatch = GatewayDispatch::new(Arc::new(registry));
-        let envelope = dispatch
-            .invoke(None, "/internal/op", serde_json::json!({}))
-            .await;
-        match envelope.result {
-            Err(error) => assert_eq!(error.code, "NOT_FOUND"),
-            Ok(v) => panic!("expected error, got {v:?}"),
-        }
-    }
-
-    #[tokio::test]
-    async fn invoke_enforces_the_default_deadline_on_a_hung_handler() {
-        use std::time::Duration;
-
-        let mut registry = OperationRegistry::new();
-        registry
-            .register(HandlerRegistration::new(
-                spec("hung/op", Visibility::External, OperationType::Query),
-                HandlerKind::Once(make_handler(|_input, ctx| async move {
-                    tokio::time::sleep(Duration::from_secs(120)).await;
-                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
-                })),
-                OperationProvenance::Local,
-                None,
-                None,
-                Capabilities::new(),
-            ))
-            .unwrap();
-        let dispatch = GatewayDispatch::new(Arc::new(registry));
-        let started = std::time::Instant::now();
-        let envelope = dispatch
-            .invoke(None, "/hung/op", serde_json::json!({}))
-            .await;
-        let elapsed = started.elapsed();
-        assert!(
-            elapsed < Duration::from_secs(60),
-            "the 30 s deadline must fire well before the 120 s handler sleep, took {elapsed:?}"
-        );
-        match envelope.result {
-            Err(error) => {
-                assert_eq!(error.code, "TIMEOUT");
-                assert!(error.retryable, "the deadline error is retryable");
-            }
-            Ok(v) => panic!("expected a TIMEOUT error, got {v:?}"),
-        }
-    }
-
-    #[tokio::test]
-    async fn invoke_completes_within_the_deadline_for_a_fast_handler() {
-        let dispatch = GatewayDispatch::new(echo_registry());
-        let envelope = dispatch
-            .invoke(None, "/echo/run", serde_json::json!({}))
-            .await;
-        assert!(envelope.result.is_ok(), "a fast handler must not time out");
-    }
-
-    #[tokio::test]
-    async fn invoke_sink_enforces_the_default_deadline_on_a_hung_sink_handler() {
-        use std::time::Duration;
-
-        let mut registry = OperationRegistry::new();
-        registry
-            .register(HandlerRegistration::new(
-                spec("hung/sink", Visibility::External, OperationType::Pub),
-                HandlerKind::Sink(make_sink_handler(|_input, ctx, _chunks| async move {
-                    tokio::time::sleep(Duration::from_secs(120)).await;
-                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
-                })),
-                OperationProvenance::Local,
-                None,
-                None,
-                Capabilities::new(),
-            ))
-            .unwrap();
-        let dispatch = GatewayDispatch::new(Arc::new(registry));
-        let chunks: alkcall::registry::registration::PublishStream =
-            Box::pin(futures::stream::empty());
-        let started = std::time::Instant::now();
-        let envelope = dispatch
-            .invoke_sink(None, "/hung/sink", serde_json::json!({}), chunks)
-            .await;
-        let elapsed = started.elapsed();
-        assert!(
-            elapsed < Duration::from_secs(60),
-            "the 30 s deadline must fire well before the 120 s handler sleep, took {elapsed:?}"
-        );
-        match envelope.result {
-            Err(error) => {
-                assert_eq!(error.code, "TIMEOUT");
-                assert!(error.retryable, "the deadline error is retryable");
-            }
-            Ok(v) => panic!("expected a TIMEOUT error, got {v:?}"),
-        }
-    }
-
-    #[tokio::test]
-    async fn invoke_sink_completes_within_the_deadline_for_a_fast_sink_handler() {
-        let mut registry = OperationRegistry::new();
-        registry
-            .register(HandlerRegistration::new(
-                spec("fast/sink", Visibility::External, OperationType::Pub),
-                HandlerKind::Sink(make_sink_handler(|_input, ctx, mut chunks| async move {
-                    use futures::StreamExt;
-                    let mut count = 0u64;
-                    while let Some(chunk) = chunks.next().await {
-                        if chunk.is_ok() {
-                            count += 1;
-                        }
-                    }
-                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "count": count }))
-                })),
-                OperationProvenance::Local,
-                None,
-                None,
-                Capabilities::new(),
-            ))
-            .unwrap();
-        let dispatch = GatewayDispatch::new(Arc::new(registry));
-        let chunks: alkcall::registry::registration::PublishStream =
-            Box::pin(futures::stream::iter(vec![
-                Ok(serde_json::json!({ "n": 1 })),
-                Ok(serde_json::json!({ "n": 2 })),
-            ]));
-        let envelope = dispatch
-            .invoke_sink(None, "/fast/sink", serde_json::json!({}), chunks)
-            .await;
-        assert!(
-            envelope.result.is_ok(),
-            "a fast sink handler must not time out, got {envelope:?}"
-        );
-    }
-
-    #[tokio::test]
-    async fn streaming_sub_op_streams_envelopes() {
-        let mut registry = OperationRegistry::new();
-        registry
-            .register(HandlerRegistration::new(
-                spec("tick/stream", Visibility::External, OperationType::Sub),
-                HandlerKind::Stream(make_streaming_handler(|input, ctx| {
-                    let count = input.get("count").and_then(|v| v.as_u64()).unwrap_or(2);
-                    let request_id = ctx.request_id.clone();
-                    futures::stream::iter(0..count).map(move |i| {
-                        ResponseEnvelope::ok(request_id.clone(), serde_json::json!({ "tick": i }))
-                    })
-                })),
-                OperationProvenance::Local,
-                None,
-                None,
-                Capabilities::new(),
-            ))
-            .unwrap();
-        let dispatch = GatewayDispatch::new(Arc::new(registry));
-        let mut stream =
-            dispatch.invoke_streaming(None, "/tick/stream", serde_json::json!({ "count": 3 }));
-        let mut ticks = Vec::new();
-        while let Some(envelope) = stream.next().await {
-            ticks.push(envelope);
-        }
-        assert_eq!(ticks.len(), 3);
-    }
-
-    fn registry_with_services_schema_over(inner_ops: Vec<OperationSpec>) -> Arc<OperationRegistry> {
-        use alkcall::registry::discovery::{services_schema_handler, services_schema_spec};
-
-        let inner = Arc::new({
-            let mut registry = OperationRegistry::new();
-            for op_spec in &inner_ops {
-                registry
-                    .register(HandlerRegistration::new(
-                        op_spec.clone(),
-                        HandlerKind::Once(make_handler(|input, ctx| async move {
-                            ResponseEnvelope::ok(ctx.request_id, input)
-                        })),
-                        OperationProvenance::Local,
-                        None,
-                        None,
-                        Capabilities::new(),
-                    ))
-                    .unwrap();
-            }
-            registry
-        });
-        let mut registry = OperationRegistry::new();
-        for op_spec in &inner_ops {
-            registry
-                .register(HandlerRegistration::new(
-                    op_spec.clone(),
-                    HandlerKind::Once(make_handler(|input, ctx| async move {
-                        ResponseEnvelope::ok(ctx.request_id, input)
-                    })),
-                    OperationProvenance::Local,
-                    None,
-                    None,
-                    Capabilities::new(),
-                ))
-                .unwrap();
-        }
-        registry
-            .register(HandlerRegistration::new(
-                services_schema_spec(),
-                HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
-                OperationProvenance::Local,
-                None,
-                None,
-                Capabilities::new(),
-            ))
-            .unwrap();
-        Arc::new(registry)
-    }
-
-    #[tokio::test]
-    async fn invoke_of_services_schema_with_internal_inner_name_is_blocked_pre_dispatch() {
-        let registry = registry_with_services_schema_over(vec![spec(
-            "secret/op",
-            Visibility::Internal,
-            OperationType::Query,
-        )]);
-        let dispatch = GatewayDispatch::new(registry);
-        let envelope = dispatch
-            .invoke(
-                None,
-                "services/schema",
-                serde_json::json!({ "name": "secret/op" }),
-            )
-            .await;
-        match envelope.result {
-            Err(error) => {
-                assert_eq!(error.code, "NOT_FOUND");
-                assert!(error.message.contains("secret/op"));
-            }
-            Ok(v) => panic!("the internal op spec must not be returned, got {v:?}"),
-        }
-    }
-
-    #[tokio::test]
-    async fn invoke_of_services_schema_with_authorized_inner_name_still_projects() {
-        let registry = registry_with_services_schema_over(vec![spec(
-            "public/op",
-            Visibility::External,
-            OperationType::Query,
-        )]);
-        let dispatch = GatewayDispatch::new(registry);
-        let envelope = dispatch
-            .invoke(
-                None,
-                "services/schema",
-                serde_json::json!({ "name": "public/op" }),
-            )
-            .await;
-        assert!(
-            envelope.result.is_ok(),
-            "an allowed inner name must still project, got {envelope:?}"
-        );
-        assert_eq!(
-            envelope
-                .result
-                .as_ref()
-                .ok()
-                .and_then(|v| v.get("name"))
-                .and_then(Value::as_str),
-            Some("public/op")
-        );
-    }
-
-    #[tokio::test]
-    async fn invoke_of_services_schema_with_forbidden_inner_name_is_denied() {
-        let restricted = OperationSpec::new(
-            "admin/op",
-            OperationType::Query,
-            Visibility::External,
-            serde_json::json!({}),
-            serde_json::json!({}),
-            vec![],
-            AccessControl {
-                required_scopes: vec!["admin".to_string()],
-                ..Default::default()
-            },
-            None,
-        );
-        let registry = registry_with_services_schema_over(vec![restricted]);
-        let dispatch = GatewayDispatch::new(registry);
-        let envelope = dispatch
-            .invoke(
-                Some(Identity {
-                    id: "user".to_string(),
-                    scopes: vec!["user".to_string()],
-                    resources: HashMap::new(),
-                }),
-                "services/schema",
-                serde_json::json!({ "name": "admin/op" }),
-            )
-            .await;
-        match envelope.result {
-            Err(error) => assert_eq!(error.code, "FORBIDDEN"),
-            Ok(v) => panic!("the ACL-denied spec must not be returned, got {v:?}"),
-        }
-    }
-
-    #[tokio::test]
-    async fn invoke_streaming_of_services_schema_with_internal_inner_name_is_blocked() {
-        let registry = registry_with_services_schema_over(vec![spec(
-            "secret/op",
-            Visibility::Internal,
-            OperationType::Query,
-        )]);
-        let dispatch = GatewayDispatch::new(registry);
-        let mut stream = dispatch.invoke_streaming(
-            None,
-            "services/schema",
-            serde_json::json!({ "name": "secret/op" }),
-        );
-        let envelopes: Vec<ResponseEnvelope> = stream.by_ref().collect().await;
-        assert_eq!(envelopes.len(), 1, "the guard error is the only item");
-        match &envelopes[0].result {
-            Err(error) => assert_eq!(error.code, "NOT_FOUND"),
-            Ok(v) => panic!("the internal op spec must not be returned, got {v:?}"),
-        }
-    }
-}

src/gateway/mod.rs

@@ -1,11 +1,7 @@
-pub mod dispatch;
 pub mod error;
 pub mod routes;
-pub(crate) mod schema_cache;
 
-#[cfg(feature = "mcp")]
+pub use alkcall::gateway::{schema_disclosure_denial, GatewayDispatch, DEFAULT_DEADLINE};
-pub(crate) use dispatch::schema_disclosure_denial;
-pub use dispatch::GatewayDispatch;
 pub use routes::CallRequest;
 
 /// The maximum number of operations accepted in a single batch,

src/gateway/routes.rs

@@ -43,10 +43,8 @@ use jsonschema::Validator;
 use serde::Deserialize;
 use serde_json::{json, Value};
 
-use super::dispatch::GatewayDispatch;
 use super::error::call_error_to_http_response_with_identity;
-use super::schema_cache::{CompileFailed, PublishSchemaCache};
+use super::{GatewayDispatch, MAX_BATCH_OPERATIONS};
-use super::MAX_BATCH_OPERATIONS;
 use crate::server::auth::ResolvedIdentity;
 use crate::server::state::RouterState;
 
@@ -84,18 +82,11 @@ type ByteStream = futures::stream::BoxStream<'static, Result<Bytes, axum::Error>
 #[derive(Clone)]
 pub(crate) struct GatewayState {
     registry: Arc<OperationRegistry>,
-    publish_schemas: crate::gateway::schema_cache::PublishSchemaCache,
 }
 
 impl GatewayState {
-    pub(crate) fn new(
+    pub(crate) fn new(registry: Arc<OperationRegistry>) -> Self {
-        registry: Arc<OperationRegistry>,
+        Self { registry }
-        publish_schemas: crate::gateway::schema_cache::PublishSchemaCache,
-    ) -> Self {
-        Self {
-            registry,
-            publish_schemas,
-        }
     }
 
     fn dispatch(&self) -> GatewayDispatch {
@@ -105,7 +96,7 @@ impl GatewayState {
 
 impl FromRef<RouterState> for GatewayState {
     fn from_ref(state: &RouterState) -> Self {
-        GatewayState::new(Arc::clone(&state.registry), state.publish_schemas.clone())
+        GatewayState::new(Arc::clone(&state.registry))
     }
 }
 
@@ -255,13 +246,13 @@ pub type SubscribeStream = BoxStream<'static, Result<Event, Infallible>>;
 /// 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
+/// against the op's `publish_schema` compiled at registration time
-/// [`PublishSchemaCache`]. A schema that **fails to compile** is
+/// (`OperationRegistry::publish_validator`, alkcall CF-003). The
-/// fail-closed, not fail-open (review-001 post-remediation follow-up):
+/// compile happens once, before the op is callable at all — an
-/// the chunk stream terminates with `INTERNAL` before any chunk reaches
+/// un-compilable schema can never reach a registry, so the HTTP path
-/// the handler — the unvalidated-ingest hole the per-request
+/// has no compile step and no fail-open window. A chunk that fails
-/// `warn`-and-skip opened is shut. The compile error itself is logged
+/// validation is a terminal `INVALID_INPUT` before any chunk reaches
-/// at error level and never echoed on the wire.
+/// the handler.
 pub(crate) async fn publish_handler(
     State(state): State<GatewayState>,
     ResolvedIdentity(identity): ResolvedIdentity,
@@ -300,7 +291,6 @@ pub(crate) async fn publish_handler(
     let chunks = NdjsonChunkStream::new(
         header_lines,
         Arc::clone(&state.registry),
-        state.publish_schemas.clone(),
         operation.clone(),
         Some(first_chunk),
     );
@@ -313,59 +303,48 @@ pub(crate) async fn publish_handler(
 
 /// 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
+/// the registry's [`alkcall::registry::registration::OperationRegistry::publish_validator`]
-/// `invoke_sink`'s pre-checks have admitted the operation (an unknown,
+/// cache on its first poll, which runs only after `invoke_sink`'s
-/// internal, or wrong-type op is rejected before any schema resolution
+/// pre-checks have admitted the operation (an unknown, internal, or
-/// or compile happens, preserving the GW-11 dispatch-owned pre-check
+/// wrong-type op is rejected before any schema resolution happens,
-/// order and keeping 404/403/422 intact).
+/// preserving the GW-11 dispatch-owned pre-check order and keeping
+/// 404/403/422 intact). The validator is compiled at **registration
+/// time** (alkcall CF-003): an un-compilable schema can never reach a
+/// registry, so there is no compile here and no fail-open window —
+/// every `/publish` chunk validates or the op has no schema.
 enum PublishSchemaState {
-    /// Not yet resolved: the stream holds the registry + cache until
+    /// Not yet resolved: the stream holds the registry until the first
-    /// the first chunk is polled.
+    /// 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>),
+    Validated(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) {
+        *self = match registry.publish_validator(&operation) {
-            Ok(None) => Self::Unvalidated,
+            Some(validator) => Self::Validated(validator),
-            Ok(Some(validator)) => Self::Validated(validator),
+            None => Self::Unvalidated,
-            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
+    /// Validate one chunk. An `Err` from this stream is terminal — the
-    /// stream-end after the first error item (an `Err` from this stream
+    /// handler sees the error and the stream closes, mirroring the wire
-    /// is terminal — the handler sees the error and the stream closes,
+    /// pump's `send(Err)` + `break`.
-    /// mirroring the wire pump's `send(Err)` + `break`).
     fn validate_chunk(
         &mut self,
         value: Value,
@@ -383,11 +362,6 @@ impl PublishSchemaState {
                     .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"),
         }
     }
@@ -582,9 +556,9 @@ impl LineError {
 /// JSON, a schema violation, or a body-read failure yields a terminal
 /// `Err` item — the same shape an initiator-side `call.error` produces
 /// on the wire, and equally terminal: like the wire pump's
-/// `send(Err)` + `break`, the first `Err` ends the stream. A schema
+/// `send(Err)` + `break`, the first `Err` ends the stream. The
-/// that failed to compile yields that terminal `Err(INTERNAL)` on the
+/// validator itself is compiled at registration time (alkcall CF-003),
-/// first chunk — fail-closed (review-001 post-remediation follow-up).
+/// so there is no compile on this path at all.
 struct NdjsonChunkStream {
     lines: std::pin::Pin<Box<BufferedLines>>,
     schema_state: PublishSchemaState,
@@ -596,7 +570,6 @@ impl NdjsonChunkStream {
     fn new(
         lines: BufferedLines,
         registry: Arc<OperationRegistry>,
-        cache: PublishSchemaCache,
         operation: String,
         pending_first: Option<Value>,
     ) -> Self {
@@ -604,7 +577,6 @@ impl NdjsonChunkStream {
             lines: Box::pin(lines),
             schema_state: PublishSchemaState::Unresolved {
                 registry,
-                cache,
                 operation,
             },
             pending_first,
@@ -1165,7 +1137,6 @@ 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()
@@ -2952,139 +2923,44 @@ mod tests {
 
     // --- publish_schema fail-closed (GW-01 follow-up) ----------------------
 
-    fn registry_with_broken_publish_schema() -> Arc<OperationRegistry> {
+    /// CF-003 (alkcall ledger): an un-compilable `publish_schema` is a
+    /// registration error — the op never becomes callable, so there is
+    /// no `/publish` request to fail and no fail-open window.
+    #[test]
+    fn uncompilable_publish_schema_is_rejected_at_registration() {
         let mut registry = OperationRegistry::new();
-        registry
+        let result = registry.register(HandlerRegistration::new(
-            .register(HandlerRegistration::new(
+            OperationSpec::new(
-                OperationSpec::new(
+                "ingest/broken",
-                    "ingest/broken",
+                OperationType::Pub,
-                    OperationType::Pub,
+                Visibility::External,
-                    Visibility::External,
+                json!({}),
-                    json!({}),
+                json!({}),
-                    json!({}),
+                vec![],
-                    vec![],
+                AccessControl::default(),
-                    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(),
+            .with_publish_schema(json!({ "required": "n" })), // `required` must be an array
-            ))
+            HandlerKind::Sink(make_sink_handler(|_input, ctx, _chunks| async move {
-            .unwrap();
+                ResponseEnvelope::ok(ctx.request_id, Value::Null)
-        Arc::new(registry)
+            })),
-    }
+            OperationProvenance::Local,
-
+            None,
-    #[tokio::test]
+            None,
-    async fn publish_with_uncompilable_schema_fails_closed_with_internal() {
+            Capabilities::new(),
-        let router = build_router(registry_with_broken_publish_schema(), unused_provider());
+        ));
-        let body = ndjson(&[
+        assert!(
-            json!({ "operation": "ingest/broken", "chunk": { "n": 1 } }),
+            result.is_err(),
-            json!({ "n": 2 }),
+            "an un-compilable publish_schema must be rejected at registration"
-        ]);
-        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")
+            registry.registration("ingest/broken").is_none(),
-                .and_then(|m| m.as_str())
+            "the un-compilable op must not be registered"
-                .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(
@@ -3114,14 +2990,14 @@ mod tests {
             ))
             .unwrap();
 
-        let v1 = cache
+        let v1 = registry
-            .validator(&registry, "ingest/typed")
+            .publish_validator("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.
+        // `additionalProperties`; the registry's per-op validator cache
+        // must follow the new registration.
         registry
             .register(HandlerRegistration::new(
                 OperationSpec::new(
@@ -3149,14 +3025,9 @@ mod tests {
             ))
             .unwrap();
 
-        let v2 = cache
+        let v2 = registry
-            .validator(&registry, "ingest/typed")
+            .publish_validator("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

@@ -1,258 +0,0 @@
-//! 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

@@ -136,7 +136,6 @@ 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 {
@@ -166,7 +165,6 @@ 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
@@ -188,7 +186,6 @@ 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);
@@ -212,7 +209,6 @@ 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
@@ -244,7 +240,6 @@ 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
@@ -1156,7 +1151,6 @@ 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

@@ -48,10 +48,6 @@ 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 {
@@ -100,7 +96,6 @@ 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/websocket/review-001-ws-eof-signal.md

@@ -84,11 +84,11 @@ and `held_session_drop_during_call_registration…` cover the CON-02 race
 asserts the sweep resolves it retryable; the original
 `connection_drop_fails_in_flight_calls_retryable_no_hang` stays green.
 One tolerated outcome note: the held-session race can also resolve via
-alkcall's write-failure path (`failed to write request frame`,
+alkcall's write-failure path — since alkcall CF-001 landed
-`INTERNAL` — the mux dies between registration and write), which is
+(`CallError::connection_closed`), that path now resolves
-prompt-but-not-retryable; that path is accepted in the race tests (the
+retryable `CONNECTION_CLOSED` too, so the race tests assert
-retryable assertion lives with the sweep and pre-drop tests where the
+retryable-only (updated in the alkcall-0.2 bump; the retryable
-call is guaranteed in-flight).
+assertion no longer needs the sweep-test carve-out).
 
 Verification: cargo test 219 ok; cargo test --features wss 231 ok (3x
 flake check); cargo clippy --all-targets -- -D warnings (default;

tests/ws_overlay_ops.rs

@@ -367,14 +367,12 @@ async fn ws_close_mid_call_to_browser_op_aborts_call() {
         )
     };
 
-    let failed = conn
+    let failed =
-        .pending()
+        conn.pending()
-        .lock()
+            .lock()
-        .fail_all(alkcall::protocol::wire::CallError::new(
+            .fail_all(alkcall::protocol::wire::CallError::connection_closed(
-            "CONNECTION_CLOSED",
+                "ws dropped",
-            "ws dropped",
+            ));
-            true,
-        ));
     assert!(failed.contains(&"hub-call-inflight".to_string()));
 
     let result = tokio::time::timeout(Duration::from_millis(100), rx).await;