feat: promote dispatch spine to gateway module (ADR-048, feature-gated)

Promote alkhttp's transport-neutral dispatch spine into alkcall as
alkcall::gateway behind the opt-in gateway cargo feature (default off;
adds no dependencies):

- GatewayDispatch: deadline-bounded invoke spine over OperationRegistry
  (invoke / invoke_streaming / invoke_sink) with the root-context
  discipline (internal: false, forwarded_for: None) hubs and spokes
  relaying calls (ADR-042 translate path) need identically to alkhttp's
  HTTP gateway. The 30 s deadline becomes a constructor knob
  (with_deadline).
- schema_disclosure_denial: the shared is-internal + ACL check for
  services/schema inner-name disclosure; ACL denial returns FORBIDDEN
  (identity-aware refinement), Internal visibility returns spec-404.
  One implementation so transports cannot drift (CF-004).
- MAX_BATCH_OPERATIONS / CallRequest / HTTP error mapping stay in
  alkhttp (projection + transport concerns); alkhttp migrates to this
  module in a follow-up session and drops its local copy.

Docs: ADR-048 (decision + divergence rationale), ADR index entry,
CHANGELOG.

Verification: 574 tests pass with --features gateway (16 new), 558 pass
default, clippy -D warnings clean both feature sets, --all-features
clean, fmt clean, wasm32 target clean, rustdoc warning-free.

CHANGELOG.md

@@ -10,6 +10,25 @@ Consumer-findings remediation (CF-001..004 from
 `docs/reviews/consumer-findings-ledger.md` — alkhttp as the first real
 consumer). One behavior change noted below; otherwise additive.
 
+### Added
+
+- **`gateway` feature: the transport-neutral dispatch spine**
+  (ADR-048). New `alkcall::gateway` module behind the opt-in `gateway`
+  cargo feature (default off; adds no dependencies). `GatewayDispatch`
+  is the deadline-bounded, re-rooted-context invoke spine over
+  `OperationRegistry` (`invoke` / `invoke_streaming` / `invoke_sink`)
+  promoted from alkhttp's gateway after it proved transport-agnostic —
+  hubs and spokes relaying calls (ADR-042 translate path) need the
+  identical root-context discipline (`internal: false`,
+  `forwarded_for: None`) without any HTTP. `schema_disclosure_denial`
+  is the shared is-internal + ACL check for `services/schema`
+  inner-op-name disclosure (one implementation so transports cannot
+  drift; ACL denial returns `FORBIDDEN`, Internal visibility returns
+  spec-404 — see ADR-048 for the split from the wire handler's
+  conservative spec-404). The handler deadline is a constructor knob
+  (`with_deadline`); the default remains 30 s. alkhttp migrates to
+  this module in a follow-up and drops its local copy.
+
 ### Changed
 
 - **Un-compilable `publish_schema` values are rejected at registration

Cargo.toml

@@ -16,6 +16,7 @@ name = "alkcall"
 
 [features]
 default = []
+gateway = []
 
 [dependencies]
 jsonschema = { version = "0.46", default-features = false }

docs/architecture/README.md

@@ -100,6 +100,7 @@ are wire-stable and unchanged — see ADR-004.
 | [045](decisions/045-alknetclient-native-dial-seam.md) | AlknetClient Dial Seam | spawn_dispatch / from_connection take-over; dial in consumer |
 | [046](decisions/046-publish-operation-type-and-handler-kind-sink.md) | Publish Operation Type and HandlerKind::Sink | `OperationType::Pub` (producer→consumer streaming); `SinkHandler` + `HandlerKind::Sink`; `call.published` wire event; `invoke_sink()` dispatch; `Subscription` renamed to `Sub` |
 | [047](decisions/047-openable-alpns-are-operations.md) | Openable ALPNs Are Operations | `channel/open` dissolves into per-ALPN ops `channels/<alpn>/sub`/`pub`; `channel_open` marker on `OperationSpec`; `ChannelCore` wrapper; extension-trait `ChannelOperationEnv`; connection-owner allocates `channel_id`; opener ledger (Gap 2 fix); ALPNs are call apps |
+| [048](decisions/048-dispatch-spine-gateway-module.md) | Dispatch Spine (feature-gated `gateway` module) | `alkcall::gateway` behind the `gateway` feature; `GatewayDispatch` invoke spine (deadline knob, re-rooted context) + `schema_disclosure_denial` (FORBIDDEN for ACL deny, spec-404 for Internal); promoted from alkhttp for hub/spoke reuse |
 
 ## Relevant Open Questions
 

docs/architecture/decisions/048-dispatch-spine-gateway-module.md

@@ -0,0 +1,151 @@
+# ADR-048: Dispatch Spine (feature-gated `gateway` module)
+
+## Status
+
+Accepted
+
+## Context
+
+The first real downstream consumer of this crate — alkhttp — built its
+HTTP gateway (`POST /call`, `/subscribe`, `/publish`, the MCP `call`
+tool, the to_openapi projections) on a small internal component it
+calls the **dispatch spine**: a thin struct over
+`Arc<OperationRegistry>` that owns the *non-HTTP* half of gateway
+dispatch. The HTTP layer resolves bearer tokens to an `Identity`,
+frames NDJSON/SSE, maps `CallError` to HTTP statuses, and wraps axum
+handlers; the spine does everything that happens after that:
+
+- constructs the root `OperationContext` identically for every
+  transport (`internal: false` — ACL runs against the caller's
+  identity, not a handler's composition authority; `forwarded_for:
+  None` — wire-ingress only; identity supplied per-call),
+- resolves the registration's `composition_authority` /
+  `capabilities` / `scoped_env` into that context,
+- bounds Once-ops and sink dispatch with a deadline while leaving
+  streaming subscriptions unbounded (ADR-021: subscriptions are
+  long-lived),
+- and applies the `services/schema` disclosure guard when the
+  dispatched operation's *input* names another operation.
+
+This is transport-neutral work. It contains no HTTP concepts: no
+statuses, no headers, no body framing. And it is not HTTP-shaped by
+accident — the same shape is exactly what a **hub** needs when it
+terminates channel 0 on both legs and relays calls to spokes
+(ADR-042's "translate, not forward" rule): the hub must re-root the
+context at itself (its own identity, `internal: false`, no
+`forwarded_for`), re-resolve its capabilities for the outbound leg, and
+bound the relay so a hung spoke does not wedge the browser-facing
+connection. alkhttp needed it; the hub relay needs the same thing;
+any protocol crate that exposes a call surface to a less-trusted
+in-transport caller (a WS-native relay, a CLI bridge, a test harness)
+will need it again.
+
+Duplicating it per consumer is the failure mode this crate already
+paid for once: the spine's `services/schema` guard existed because the
+registry handler and the HTTP route were written against different
+disclosure rules (CF-004, filed from alkhttp's consumer review). One
+shared implementation is the fix; a second copy in a second crate
+would re-open the drift.
+
+ alkhttp is not yet published. This is the cheapest moment to move the
+component into alkcall (behind a feature, so the base crate stays lean
+and the surface is opt-in) and have alkhttp consume it rather than
+carry its own copy.
+
+## Decision
+
+Promote the dispatch spine into alkcall as a new `gateway` module,
+**feature-gated**:
+
+- `alkcall/gateway` behind the `gateway` cargo feature (default off —
+  the base crate stays lean; the module adds no dependencies, the gate
+  exists to keep the audit surface explicit and opt-in).
+- `GatewayDispatch` — the spine struct: `invoke()`,
+  `invoke_streaming()`, `invoke_sink()`, registry access, and root
+  context construction. The 30 s deadline alkhttp hardcodes becomes a
+  constructor knob (`with_deadline`) so consumers keep their own
+  policy; the default remains 30 s for drop-in equivalence.
+- `schema_disclosure_denial()` — the shared check that a spec the
+  caller could not invoke is not disclosed: `NOT_FOUND` for
+  Internal-visibility ops, **`FORBIDDEN` for ACL-denied ops**.
+- `CallRequest` stays in alkhttp (payload framing is transport
+  business); `MAX_BATCH_OPERATIONS` stays in alkhttp (batch is a
+  projection concern).
+
+### ACL denial returns FORBIDDEN, not spec-404
+
+The wire-path `services/schema` handler (CF-004, discovery.rs) returns
+spec-404 for both Internal visibility and ACL denial — the right
+answer for an unauthenticated wire caller, where even acknowledging
+the op's existence is a leak vector. The spine's guard is invoked by a
+transport that has *already* resolved the caller's identity and often
+already admitted the op elsewhere (the alkhttp GET `/schema` route
+answers `403` for ACL-denied ops, deliberately). The spine therefore
+keeps alkhttp's split:
+
+- Internal visibility → `NOT_FOUND` (never acknowledged, at any
+  authority).
+- ACL denial → `FORBIDDEN` (informative for a caller whose identity
+  the transport resolved; alkcall's own registry `invoke()` produces
+  the same code for the same caller state, so the guard's answer is
+  never *more* restrictive or *less* restrictive than the invoke that
+  would follow it).
+
+The two layers are consistent by construction: the wire handler's
+spec-404 is the conservative outer bound, the spine's FORBIDDEN is the
+identity-aware refinement. When an unauthenticated caller hits both,
+`FORBIDDEN` and `NOT_FOUND` differ only in which exists — and
+`AccessControl::check` with no identity returns
+`"authentication required"`, which HTTP-side mappers translate to 401.
+ alkhttp consumes this module in a later session and drops its local
+copy; until then the two implementations coexist (byte-identical in
+behavior, one in each crate).
+
+### What the spine deliberately does NOT include
+
+- **No HTTP mapping.** `CallError` → status/body is the consumer's
+  (alkhttp `gateway::error`). The neutral wire error is the module's
+  lowest-level vocabulary.
+- **No body framing, limits, or timeouts beyond the deadline knob.**
+  NDJSON/SSE framing, body caps, keep-alive intervals are transport
+  concerns (GW-15/GW-16 in alkhttp's ledger).
+- **No batch semantics.** `MAX_BATCH_OPERATIONS` and the batch
+  envelope are projections of the registry onto HTTP/MCP payloads.
+- **No `CallRequest` type.** The spine takes `(op, input)` — parsing
+  `{ operation, input }` is transport framing.
+
+## Consequences
+
+- The deadline policy moves from "alkhttp's 30 s HTTP convention" to
+  "spine policy, configured per-consumer" (default 30 s). Sink
+  dispatch is bounded by the same deadline as Once-ops — the wire
+  path's Pub dispatch is unbounded (`deadline: None`); consumers who
+  want the wire behavior pass `Duration::ZERO`-style no-deadline
+  configuration via `with_deadline(None)`. Documented divergence, a
+  choice the *consumer* now owns.
+- alkhttp migrates to `alkcall::gateway` in a follow-up session; its
+  local spine is deleted then, not deprecated in place (unpublished
+  crate — no compat window needed).
+- The feature adds no dependencies; `cargo test --no-default-features`
+  and `--all-features` both pass (the wasm-clean baseline is
+  untouched).
+- Future transports (WS-native relays, protocol crates) reuse the
+  spine instead of re-deriving the context/guard discipline.
+
+## Alternatives considered
+
+- **Fold into `registry` as a method set on `OperationRegistry`.**
+  Rejected: the spine is a *policy* wrapper (deadline, root-context
+  shape, disclosure guard) a consumer opts into, not the registry's
+  dispatch core. `GatewayDispatch::invoke` ≠ `OperationRegistry::invoke`
+  — conflating them invites wire paths to pick up transport-shaped
+  policy accidentally.
+- **Wait for alkhttp to publish first.** Rejected: promotes a
+  duplicate into the wild that immediately has to be deprecated; the
+  cheapest moment to move the type is before either crate's first
+  release.
+- **Promote the whole gateway module.** Rejected: routes (SSE/NDJSON
+  framing, body limits) and error mapping (`IntoResponse`, status
+  mapping) are HTTP by definition; moving them would drag axum/hyper
+  as optional dependencies into a protocol crate for no reuse —
+  alkhttp is their only consumer.

src/gateway/dispatch.rs

@@ -0,0 +1,755 @@
+//! The transport-neutral dispatch spine ([`GatewayDispatch`]) and the
+//! shared `services/schema` disclosure guard
+//! ([`schema_disclosure_denial`]) — the non-HTTP half of alkhttp's
+//! gateway, promoted for reuse by hubs, relays, and protocol crates
+//! that expose a call surface to a less-trusted in-transport caller.
+//!
+//! The spine constructs the root [`OperationContext`] identically for
+//! every transport (`internal: false` — ACL runs against the caller's
+//! identity, not a handler's composition authority; `forwarded_for:
+//! None` — wire-ingress only), resolves the registration's
+//! composition authority / capabilities / scoped env into it, and
+//! bounds Once-ops and sink dispatch with a configurable deadline
+//! while leaving streaming subscriptions unbounded (ADR-021:
+//! subscriptions are long-lived).
+//!
+//! There are no HTTP concepts here: no statuses, no headers, no body
+//! framing. CallError → HTTP mapping, NDJSON/SSE framing, body caps,
+//! and batch envelopes are the consumer's (see alkhttp's gateway).
+//!
+//! See ADR-048 for the promotion decision and the divergence note on
+//! ACL-denial codes (`FORBIDDEN` here, spec-404 in the wire handler).
+
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use crate::core::auth::Identity;
+use crate::core::types::Capabilities;
+use crate::protocol::wire::{CallError, ResponseEnvelope};
+use crate::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
+use crate::registry::env::LocalOperationEnv;
+use crate::registry::registration::OperationRegistry;
+use crate::registry::spec::{AccessResult, Visibility};
+use futures::stream::BoxStream;
+use serde_json::Value;
+
+const SERVICES_SCHEMA: &str = "services/schema";
+
+/// The default handler deadline for Once-ops and sink dispatch: 30 s.
+/// Override with [`GatewayDispatch::with_deadline`].
+pub const DEFAULT_DEADLINE: Duration = Duration::from_secs(30);
+
+/// The transport-neutral dispatch spine over an
+/// [`OperationRegistry`](crate::registry::registration::OperationRegistry):
+/// invokes operations for the neutral `ResponseEnvelope` result shape
+/// any transport maps to its own wire format. Identity arrives
+/// per-call as `Option<Identity>` — bearer resolution, transport
+/// framing, and error presentation happen upstream in the consumer.
+///
+/// See the [module docs](crate::gateway) and ADR-048.
+pub struct GatewayDispatch {
+    registry: Arc<OperationRegistry>,
+    deadline: Option<Duration>,
+    invoke_count: AtomicUsize,
+}
+
+impl GatewayDispatch {
+    /// Assemble a dispatch spine over a registry with the 30 s default
+    /// deadline ([`DEFAULT_DEADLINE`]).
+    pub fn new(registry: Arc<OperationRegistry>) -> Self {
+        Self {
+            registry,
+            deadline: Some(DEFAULT_DEADLINE),
+            invoke_count: AtomicUsize::new(0),
+        }
+    }
+
+    /// Override the Once/sink deadline: `Some(duration)` bounds every
+    /// [`GatewayDispatch::invoke`] and [`GatewayDispatch::invoke_sink`]
+    /// dispatch (a hung handler surfaces as a retryable `TIMEOUT` error
+    /// envelope); `None` removes the bound entirely (the wire path's
+    /// Pub dispatch behavior). Streaming subscriptions are unbounded
+    /// either way (ADR-021).
+    pub fn with_deadline(mut self, deadline: Option<Duration>) -> Self {
+        self.deadline = deadline;
+        self
+    }
+
+    /// 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: consumers' over-cap batch tests
+    /// assert it stays at zero to prove no dispatch happened before the
+    /// cap rejection.
+    pub fn invoke_count(&self) -> usize {
+        self.invoke_count.load(Ordering::Relaxed)
+    }
+
+    /// Invoke a Query/Mutation op under the configured deadline; a hung
+    /// handler surfaces as a retryable `TIMEOUT` error envelope.
+    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 fut = self.registry.invoke(&operation_name, input, context);
+        match self.deadline {
+            Some(deadline) => match tokio::time::timeout(deadline, fut).await {
+                Ok(envelope) => envelope,
+                Err(_elapsed) => ResponseEnvelope::error(
+                    request_id,
+                    CallError::timeout(format!(
+                        "operation did not complete within the {deadline:?} dispatch deadline"
+                    )),
+                ),
+            },
+            None => fut.await,
+        }
+    }
+
+    /// Dispatch a Sub op: the returned stream of envelopes is unbounded
+    /// by the deadline (subscriptions are long-lived per 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). The `publish_stream` is
+    /// the initiator's chunk stream (each item one published chunk, or
+    /// an initiator-side error). The dispatch — chunk pacing and the
+    /// sink handler's final completion alike — is bounded by the
+    /// configured deadline: a hung sink handler surfaces as a retryable
+    /// `TIMEOUT` error envelope, not an indefinitely-held initiator.
+    pub async fn invoke_sink(
+        &self,
+        identity: Option<Identity>,
+        op: &str,
+        input: Value,
+        publish_stream: crate::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 fut = self
+            .registry
+            .invoke_sink(&operation_name, input, publish_stream, context);
+        match self.deadline {
+            Some(deadline) => match tokio::time::timeout(deadline, fut).await {
+                Ok(envelope) => envelope,
+                Err(_elapsed) => ResponseEnvelope::error(
+                    request_id,
+                    CallError::timeout(format!(
+                        "operation did not complete within the {deadline:?} dispatch deadline"
+                    )),
+                ),
+            },
+            None => fut.await,
+        }
+    }
+
+    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 crate::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_some(self.deadline)
+                .flatten()
+                .map(|deadline| Instant::now() + deadline),
+            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 (ADR-048): when the dispatched
+/// operation is the schema meta-op, apply the disclosure checks to the
+/// **inner** `name` input before dispatching, so no transport through
+/// this spine can fetch a spec that would be denied for the same
+/// identity elsewhere.
+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 schema-disclosure check shared by every transport surface:
+/// Internal ops are invisible (`NOT_FOUND` regardless of caller);
+/// ACL-forbidden ops are denied with `FORBIDDEN`. One implementation so
+/// the transports cannot drift (CF-004's shared-implementation fix; the
+/// wire-path handler stays the conservative spec-404 outer bound).
+pub 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 crate::registry::registration::{
+        make_handler, make_sink_handler, make_streaming_handler, HandlerKind, HandlerRegistration,
+        OperationProvenance,
+    };
+    use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
+    use futures::StreamExt;
+
+    // The deadline tests use a short custom deadline instead of the
+    // module default, so no wall-clock assertion by hand is needed.
+
+    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_configured_deadline_on_a_hung_handler() {
+        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)).with_deadline(Some(Duration::from_millis(50)));
+        let started = std::time::Instant::now();
+        let envelope = dispatch
+            .invoke(None, "/hung/op", serde_json::json!({}))
+            .await;
+        assert!(
+            started.elapsed() < Duration::from_secs(10),
+            "the 50 ms deadline must fire well before the 120 s handler sleep"
+        );
+        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_with_no_deadline_completes_a_slow_handler() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec("slow/op", Visibility::External, OperationType::Query),
+                HandlerKind::Once(make_handler(|_input, ctx| async move {
+                    tokio::time::sleep(Duration::from_millis(80)).await;
+                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let dispatch = GatewayDispatch::new(Arc::new(registry)).with_deadline(None);
+        let envelope = dispatch
+            .invoke(None, "/slow/op", serde_json::json!({}))
+            .await;
+        assert!(
+            envelope.result.is_ok(),
+            "a no-deadline spine must not time out, got {envelope:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn invoke_sink_enforces_the_configured_deadline_on_a_hung_sink_handler() {
+        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)).with_deadline(Some(Duration::from_millis(50)));
+        let chunks: crate::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;
+        assert!(
+            started.elapsed() < Duration::from_secs(10),
+            "the 50 ms deadline must fire well before the 120 s handler sleep"
+        );
+        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 {
+                    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: crate::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);
+    }
+
+    #[tokio::test]
+    async fn invoke_count_counts_invokes_only() {
+        let dispatch = GatewayDispatch::new(echo_registry());
+        let _ = dispatch
+            .invoke(None, "/echo/run", serde_json::json!({}))
+            .await;
+        let _ = dispatch.invoke_streaming(None, "/echo/run", serde_json::json!({}));
+        assert_eq!(dispatch.invoke_count(), 1);
+    }
+
+    fn registry_with_services_schema_over(inner_ops: Vec<OperationSpec>) -> Arc<OperationRegistry> {
+        use crate::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:?}"),
+        }
+    }
+
+    #[test]
+    fn schema_disclosure_denial_hides_internal_ops() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec("secret/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 error = schema_disclosure_denial(&registry, "secret/op", None).unwrap();
+        assert_eq!(error.code, "NOT_FOUND");
+        assert!(schema_disclosure_denial(&registry, "missing/op", None).is_none());
+    }
+
+    #[test]
+    fn schema_disclosure_denial_allows_unrestricted_ops_without_identity() {
+        let error = schema_disclosure_denial(&echo_registry(), "echo/run", None);
+        assert!(
+            error.is_none(),
+            "default-ACL ops are fetchable unauthenticated"
+        );
+    }
+}

src/gateway/mod.rs

@@ -0,0 +1,6 @@
+//! The transport-neutral dispatch spine and the shared
+//! `services/schema` disclosure guard (ADR-048).
+
+pub(crate) mod dispatch;
+
+pub use dispatch::{schema_disclosure_denial, GatewayDispatch, DEFAULT_DEADLINE};

src/lib.rs

@@ -17,6 +17,10 @@
 //!   going forward.
 //! - **Registry** ([`registry`]): operation specs, context, dispatch, and
 //!   the operation registry — the call half's dispatch core.
+//! - **Gateway** ([`gateway`], feature `gateway`): the transport-neutral
+//!   dispatch spine and `services/schema` disclosure guard — the
+//!   deadline-bounded, re-rooted-context invoke surface for HTTP
+//!   gateways, hub relays, and other transport front-ends (ADR-048).
 //! - **Protocol** ([`protocol`]): wire format, streams, adapter, dispatch
 //!   loop, pending requests, abort cascade — the call half's wire layer.
 //! - **Client** ([`client`]): `CallClient`, `from_call`, `OperationAdapter`
@@ -71,6 +75,8 @@
 pub mod channels;
 pub mod client;
 pub mod core;
+#[cfg(feature = "gateway")]
+pub mod gateway;
 pub mod protocol;
 pub mod registry;