feat(review 008 Unit 3b): the hub-leg install template (ADR-051 §5)

- src/channels/hub_leg.rs: HubLegImports (the Clone discover/stash,
  bundles split by the channel_open marker, with the per-consumer
  filter) and HubLegTemplate (the hub-leg install hook: per-connection
  fork + generic channel ops + plain bundles as-is + marked specs via
  ChannelRelay::register_relay_openable + bootstrap discovery closed
  over the fork (review 004 F-06) + serving-identity resolution
  (CF-005) + the single-stream dispatch loop).
- The loud assembly posture holds at install: a Pub-typed (or
  unregistrable) marked spec ends the leg's install task — channel 0
  never dispatches, never a silent stub (ADR-051 §6).
- Assembly rule surfaced by the gate e2e: from_call discovers the
  spoke's bootstrap discovery ops like any plain op; the template
  skips BOOTSTRAP_DISCOVERY_OPS (new pub const, discovery.rs) when
  re-registering plain bundles — the template's own install, closed
  over the fork, supersedes the imported copies.
- Gate tests (7): stash split + filter, the full template e2e
  (services/list shows the re-exposed ops, a plain imported op
  round-trips through the hub, the marked op opens through the relay
  with bound surviving, data both directions), serving-identity
  precedence (scope-bearing override resolves, scope-less fails
  closed), the loud Pub-typed posture, and the filtered-stash subset
  run (NOT_FOUND for the filtered-out marked op).
- Docs: ADR-051 status (3a + 3b implemented, 3c remains); review 008
  Unit 3b landed note (the bootstrap-op skip rule).

Verification: cargo test (665 passed), clippy --all-targets
-D warnings, fmt --check, doc --no-deps — all clean.

docs/architecture/decisions/051-channel-relay-and-hub-leg.md

@@ -9,7 +9,10 @@ and 2 (both landed, alkcall 0.7.2): the relay's establisher projects
 the spoke reply's extra fields via `Establishment` reply fields
 (ADR-049 amendment 3), and the spoke's flavor-form open ops arrive
 reconstructed WITH the `channel_open` marker through discovery
-(ADR-047 amendment 3).
+(ADR-047 amendment 3). Unit 3a (the `ChannelRelay` component, §1–§4,
+§6) and Unit 3b (the hub-leg install template, §5) are implemented —
+`src/channels/relay.rs` and `src/channels/hub_leg.rs`; Unit 3c (the
+review's gate-2 e2e harness) remains.
 
 ## Context
 

docs/reviews/008-tunnels-graduation-upstream-review.md

@@ -402,6 +402,18 @@ Verification gates:
 
 ### Unit 3b — the hub-leg install template (the call-half support)
 
+Landed (2026-09-17): `src/channels/hub_leg.rs` — `HubLegImports`
+(the `Clone` stash split by the marker) and `HubLegTemplate`
+(the §5 composition exported as the `InstallChannelZero` hook;
+`serving identity` via the CF-005 seam). One assembly rule the
+planning sketch missed, surfaced by the template's gate e2e:
+`from_call` discovers the spoke's bootstrap discovery ops like any
+plain op, so the template skips
+`registry::discovery::BOOTSTRAP_DISCOVERY_OPS` when re-registering
+plain bundles — the template's own install (closed over the fork,
+review 004 F-06) supersedes the imported copies. Unit 3c (the
+gate-2 e2e) remains.
+
 The hub-side composition every test hand-rolls
 (`make_install_channel_zero`-shaped), exported in-tree (ADR-051 §5 —
 two-way-door API shape):

src/channels/hub_leg.rs

@@ -0,0 +1,292 @@
+//! The hub-leg install template (ADR-051 §5; review 008 Unit 3b).
+//!
+//! A relay hub accepts consumer-leg `alk/channels` connections and
+//! serves channel 0 over a per-connection fork of its base registry:
+//! the fork carries the generic channel ops, the bootstrap discovery
+//! ops (closed over the fork, so `services/list` sees the re-exposed
+//! ops — review 004 F-06), the stashed plain bundles as-is (the
+//! from_call forwarding stubs), and each stashed marked spec via the
+//! relay registration ([`ChannelRelay::register_relay_openable`]).
+//! The template exports that composition as an in-tree
+//! [`InstallChannelZero`] hook — the composition every hub-side test
+//! hand-rolled before this module existed.
+//!
+//! The spoke side needs nothing new: a spoke serving ops through a hub
+//! is the existing connect-side serving shape
+//! (`ChannelClient::from_connection_with_serving`) plus the
+//! producer-leg registration it already does.
+//!
+//! Two-phase seam (ADR-051 §4): discovery of the producer leg's ops
+//! happens once (the [`HubLegImports`] stash, `Clone`, one discovered
+//! set serves any number of consumer legs); registration happens per
+//! consumer-leg connection inside the install hook. Per-consumer
+//! op-subset filtering composes by filtering the stash before handing
+//! it to the template (the fork is per consumer leg); ACL layering per
+//! ADR-051 §5: the imported spec's own `AccessControl` gates the
+//! re-exposed op on the consumer leg, the spoke's ACL sees only the
+//! hub identity, the end consumer's identity rides `forwarded_for` as
+//! metadata and is never consulted by any `AccessControl::check`.
+
+use std::collections::HashSet;
+use std::sync::Arc;
+
+use crate::core::auth::{Identity, IdentityProvider};
+use crate::protocol::connection::{split_single_stream, CallConnection};
+use crate::protocol::dispatch::Dispatcher;
+use crate::registry::discovery::install_bootstrap_discovery;
+use crate::registry::registration::{HandlerRegistration, OperationRegistry};
+
+use super::operations::ChannelCore;
+use super::relay::ChannelRelay;
+use super::{adapter::InstallChannelZero, policy::ChannelLifecyclePolicy};
+
+/// The discover/stash half of the two-phase registration seam
+/// (ADR-051 §4 phase 1): the from_call-imported bundles split by the
+/// `channel_open` marker — marked specs become relay openables, plain
+/// bundles register as-is (the forwarding stubs). `Clone`, so one
+/// discovered set serves any number of consumer legs (and any number
+/// of per-consumer filtered variants of it).
+#[derive(Clone, Default)]
+pub struct HubLegImports {
+    /// Marked specs (reconstructed WITH the `channel_open` marker,
+    /// ADR-047 amendment 3) — registered per consumer leg via the
+    /// relay registration.
+    marked: Vec<HandlerRegistration>,
+    /// Plain bundles — registered per consumer leg as-is.
+    plain: Vec<HandlerRegistration>,
+}
+
+impl HubLegImports {
+    /// Split from_call-imported bundles by the marker. A discovered
+    /// `Pub`-typed marked spec is kept here (the stash is data, not
+    /// assembly) — [`HubLegTemplate::install_hook`] surfaces it as the
+    /// loud assembly error (ADR-051 §6) when a consumer leg installs.
+    pub fn from_bundles(bundles: Vec<HandlerRegistration>) -> Self {
+        let mut imports = Self::default();
+        for bundle in bundles {
+            if bundle.spec.channel_open.is_some() {
+                imports.marked.push(bundle);
+            } else {
+                imports.plain.push(bundle);
+            }
+        }
+        imports
+    }
+
+    /// The marked specs (relay openables).
+    pub fn marked(&self) -> &[HandlerRegistration] {
+        &self.marked
+    }
+
+    /// The plain bundles (forwarding stubs).
+    pub fn plain(&self) -> &[HandlerRegistration] {
+        &self.plain
+    }
+
+    /// Keep only the ops whose names satisfy `keep` — the per-consumer
+    /// op-subset filter (ADR-051 §4's composition note; a hub
+    /// re-exposing different op subsets to different consumers filters
+    /// the stash per fork).
+    #[must_use]
+    pub fn filtered(self, keep: impl Fn(&str) -> bool) -> Self {
+        Self {
+            marked: self
+                .marked
+                .into_iter()
+                .filter(|b| keep(&b.spec.name))
+                .collect(),
+            plain: self
+                .plain
+                .into_iter()
+                .filter(|b| keep(&b.spec.name))
+                .collect(),
+        }
+    }
+
+    /// Keep only the named ops (the common filter shape).
+    #[must_use]
+    pub fn only(self, names: &[&str]) -> Self {
+        let allowed: HashSet<&str> = names.iter().copied().collect();
+        self.filtered(|name| allowed.contains(name))
+    }
+}
+
+/// The hub-leg install template (ADR-051 §5): composes the fork +
+/// generic channel ops + bootstrap discovery + plain bundles + relay
+/// openables + serving identity, and returns the
+/// [`InstallChannelZero`] hook to hand [`super::adapter::ChannelsAdapter`].
+///
+/// The producer-leg surface is shared (ADR-051 §2): the
+/// [`ChannelRelay`] (holding `Arc<CallConnection>` + producer-leg
+/// `ChannelManager`) plus the consumer-leg `ChannelCore` are closed
+/// over per install; the template holds no extra claim on the
+/// producer leg.
+pub struct HubLegTemplate {
+    relay: Arc<ChannelRelay>,
+    imports: HubLegImports,
+    /// The consumer-leg channel lifecycle policy — the per-identity
+    /// cap the consumer leg's open-op wrapper enforces. Shared across
+    /// consumer legs (per-identity, not per-connection, ADR-041).
+    policy: Arc<dyn ChannelLifecyclePolicy>,
+    /// Serving-identity resolution for the dispatch (CF-005
+    /// precedence: payload token → explicit override → transport
+    /// identity). `identity` mirrors
+    /// [`crate::channels::client::ServingConfig::identity`] — an
+    /// explicit override for the peer the transport authenticated;
+    /// `None` falls through to the channel-0 connection's identity.
+    identity_provider: Arc<dyn IdentityProvider>,
+    identity: Option<Identity>,
+}
+
+impl HubLegTemplate {
+    /// Construct the template over a relay (the producer leg) and the
+    /// stashed imports. The policy defaults to the crate default
+    /// (256/identity, ADR-041); identity resolution defaults to the
+    /// noop provider with no override.
+    pub fn new(relay: ChannelRelay, imports: HubLegImports) -> Self {
+        Self {
+            relay: Arc::new(relay),
+            imports,
+            policy: super::policy::default_policy(),
+            identity_provider: Arc::new(crate::core::auth::NoopIdentityProvider),
+            identity: None,
+        }
+    }
+
+    /// Set the consumer-leg channel lifecycle policy.
+    pub fn with_policy(mut self, policy: Arc<dyn ChannelLifecyclePolicy>) -> Self {
+        self.policy = policy;
+        self
+    }
+
+    /// Set the identity provider for the serving dispatch (payload
+    /// `auth_token` resolution, ADR-017 §7).
+    pub fn with_identity_provider(mut self, provider: Arc<dyn IdentityProvider>) -> Self {
+        self.identity_provider = provider;
+        self
+    }
+
+    /// Set an explicit serving identity (the CF-005 seam — wins over
+    /// the transport connection's identity).
+    pub fn with_identity(mut self, identity: Identity) -> Self {
+        self.identity = Some(identity);
+        self
+    }
+
+    /// Build the `install_channel_zero` hook (ADR-051 §5 — the
+    /// composition the hub/spoke family all share). Per consumer-leg
+    /// connection:
+    ///
+    /// 1. Fork the (empty) per-leg base — the fork is the dispatch
+    ///    registry, registered fresh per connection because the
+    ///    openables close over that connection's `ChannelCore`
+    ///    (ADR-047 §4 amendment).
+    /// 2. Register the generic channel ops
+    ///    (`ChannelOperations::register_on`, closed over the consumer
+    ///    leg's manager + policy).
+    /// 3. Register the stashed plain bundles as-is.
+    /// 4. Register each stashed marked spec via
+    ///    [`ChannelRelay::register_relay_openable`] on the consumer
+    ///    leg's `ChannelCore` (the full open-op wrapper machinery —
+    ///    ACL, cap/ledger, establishment bound, teardown). A
+    ///    Pub-typed marked spec surfaces here as the loud assembly
+    ///    error (ADR-051 §6) — the install task ends and the leg's
+    ///    channel 0 never dispatches (never a silent stub).
+    /// 5. Install the bootstrap discovery ops closed over the fork
+    ///    (review 004 F-06 — `services/list` sees the re-exposed ops).
+    /// 6. Resolve the serving identity (payload token → explicit
+    ///    override → transport identity) and run the single-stream
+    ///    dispatch loop until the transport EOF.
+    pub fn install_hook(&self) -> InstallChannelZero {
+        let relay = Arc::clone(&self.relay);
+        let imports = self.imports.clone();
+        let policy = Arc::clone(&self.policy);
+        let identity_provider = Arc::clone(&self.identity_provider);
+        let identity = self.identity.clone();
+        Arc::new(move |consumer_manager, channel0_conn, auth| {
+            let relay = Arc::clone(&relay);
+            let imports = imports.clone();
+            let policy = Arc::clone(&policy);
+            let identity_provider = Arc::clone(&identity_provider);
+            let identity = identity.clone();
+            tokio::spawn(async move {
+                // The CF-005 seam (review 004 F-05 / 0.7.0): an
+                // explicit override wins; else the transport
+                // connection's identity propagates to channel 0 and
+                // the dispatch's fallback resolves through it.
+                // `set_identity` is once-only; silently skip when the
+                // adapter pre-set one.
+                if let Some(identity) = identity {
+                    let _ = channel0_conn.set_identity(identity);
+                }
+                let Ok(channel0_bidi) = channel0_conn.accept_bi().await else {
+                    return;
+                };
+                let (writer, reader) = split_single_stream(channel0_bidi);
+                let registry = Arc::new(OperationRegistry::new());
+
+                let operations = super::operations::ChannelOperations::new(
+                    consumer_manager.clone(),
+                    Arc::clone(&policy),
+                );
+                if let Err(e) = operations.register_on(&registry) {
+                    tracing::warn!(error = %e, "hub-leg template: generic channel ops failed to register");
+                    return;
+                }
+
+                let consumer_core = ChannelCore::new(consumer_manager, Arc::clone(&policy));
+                for bundle in &imports.plain {
+                    // The bootstrap discovery ops the spoke serves are
+                    // re-discovered by from_call like any plain op;
+                    // the template's own install (closed over this
+                    // fork, review 004 F-06) supersedes the imported
+                    // copies — re-registering both would double-book
+                    // the names (the fork's discovery must see the
+                    // fork's ops, not the spoke's).
+                    if crate::registry::discovery::BOOTSTRAP_DISCOVERY_OPS
+                        .contains(&bundle.spec.name.as_str())
+                    {
+                        continue;
+                    }
+                    if let Err(e) = registry.register(bundle.clone()) {
+                        tracing::warn!(error = %e, "hub-leg template: plain bundle registration failed");
+                        return;
+                    }
+                }
+                let install_auth = auth.clone();
+                for bundle in &imports.marked {
+                    if let Err(e) = relay.register_relay_openable(
+                        &consumer_core,
+                        &registry,
+                        bundle.spec.clone(),
+                        install_auth.clone(),
+                    ) {
+                        // The loud assembly posture (ADR-051 §6): a
+                        // Pub-typed (or otherwise unregistrable)
+                        // marked spec never becomes a silent stub —
+                        // the leg's channel 0 never dispatches.
+                        tracing::warn!(error = %e, "hub-leg template: relay openable registration failed");
+                        return;
+                    }
+                }
+
+                if let Err(e) = install_bootstrap_discovery(&registry) {
+                    tracing::warn!(error = %e, "hub-leg template: bootstrap discovery install failed");
+                    return;
+                }
+
+                let call_connection = Arc::new(CallConnection::new_single_stream(
+                    channel0_conn,
+                    Arc::clone(&writer),
+                ));
+                Dispatcher::new(registry, identity_provider)
+                    .run_loop_single_stream(call_connection, reader, writer)
+                    .await;
+            })
+        })
+    }
+}
+
+#[cfg(test)]
+#[path = "hub_leg_tests.rs"]
+mod tests;

src/channels/hub_leg_tests.rs

@@ -0,0 +1,635 @@
+//! Unit 3b gates (review 008; ADR-051 §5 / gate 5).
+//!
+//! - The stash split (marked vs plain) and the per-consumer filter.
+//! - A plain imported op round-trips through the hub (the call-half
+//!   forwarding path exercised for real).
+//! - The full template e2e: `services/list` on the consumer leg shows
+//!   the re-exposed ops (marked + plain + generic), a flavor-form
+//!   marked op opens through the relay with `bound` surviving, data
+//!   flows both directions.
+//! - A Pub-typed marked spec in the stash is the loud assembly
+//!   posture: the consumer leg's channel 0 never dispatches.
+//! - Serving-identity resolution (CF-005): an explicit override
+//!   authenticates the consumer leg's calls.
+
+use super::{HubLegImports, HubLegTemplate};
+use crate::channels::adapter::ChannelsAdapter;
+use crate::channels::operations::{ChannelCore, Establishment, OpenEstablisher, OpenHandler};
+use crate::channels::policy::{ChannelLifecyclePolicy, NoCap, PerIdentityChannelPolicy};
+use crate::channels::relay::{ChannelRelay, ProducerLeg};
+use crate::client::{from_call, FromCallConfig};
+use crate::core::auth::{AuthContext, Identity, IdentityProvider};
+use crate::core::types::Connection;
+use crate::protocol::connection::{split_single_stream, CallConnection};
+use crate::protocol::dispatch::Dispatcher;
+use crate::protocol::wire::ResponseEnvelope;
+use crate::registry::registration::{
+    make_handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry,
+};
+use crate::registry::spec::{
+    AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility,
+};
+use serde_json::json;
+use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+use std::sync::Arc;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+const TEST_ADDR: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321);
+const OP: &str = "channels/tunnel/direct";
+const ALPN: &str = "alk/tunnel";
+const PLAIN_OP: &str = "fs/read";
+
+fn identity(id: &str, scopes: &[&str]) -> Identity {
+    Identity {
+        id: id.to_string(),
+        scopes: scopes.iter().map(|s| (*s).to_string()).collect(),
+        resources: Default::default(),
+    }
+}
+
+fn relay_spec(name: &str, op_type: OperationType, alpn: Option<String>) -> OperationSpec {
+    let mut spec = OperationSpec::new(
+        name,
+        op_type,
+        Visibility::External,
+        json!({ "type": "object" }),
+        json!({ "type": "object" }),
+        vec![],
+        AccessControl::default(),
+        None,
+    );
+    if let Some(alpn) = alpn {
+        spec = spec.with_channel_open(ChannelOpenSpec::new(alpn));
+    }
+    spec
+}
+
+struct NoopIdProvider;
+impl IdentityProvider for NoopIdProvider {
+    fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
+        None
+    }
+    fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option<Identity> {
+        None
+    }
+}
+
+struct StaticIdProvider(Identity);
+impl IdentityProvider for StaticIdProvider {
+    fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
+        None
+    }
+    fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option<Identity> {
+        None
+    }
+}
+
+impl StaticIdProvider {
+    #[allow(dead_code)]
+    fn inner(&self) -> &Identity {
+        &self.0
+    }
+}
+
+// --- the stash (ADR-051 §4 phase 1) ---------------------------------------
+
+#[test]
+fn stash_splits_bundles_by_marker() {
+    let marked = HandlerRegistration::new(
+        relay_spec(OP, OperationType::Sub, Some(ALPN.to_string())),
+        HandlerKind::Once(make_handler(|input, ctx| async move {
+            ResponseEnvelope::ok(ctx.request_id, input)
+        })),
+        OperationProvenance::FromCall,
+        None,
+        None,
+        crate::core::types::Capabilities::new(),
+    );
+    let plain = HandlerRegistration::new(
+        relay_spec(PLAIN_OP, OperationType::Query, None),
+        HandlerKind::Once(make_handler(|input, ctx| async move {
+            ResponseEnvelope::ok(ctx.request_id, input)
+        })),
+        OperationProvenance::FromCall,
+        None,
+        None,
+        crate::core::types::Capabilities::new(),
+    );
+    let imports = HubLegImports::from_bundles(vec![marked, plain]);
+    assert_eq!(imports.marked().len(), 1);
+    assert_eq!(imports.marked()[0].spec.name, OP);
+    assert_eq!(imports.plain().len(), 1);
+    assert_eq!(imports.plain()[0].spec.name, PLAIN_OP);
+}
+
+#[test]
+fn stash_filter_keeps_named_ops_only() {
+    let bundle = |name: &str, marked: bool| {
+        let spec = if marked {
+            relay_spec(name, OperationType::Sub, Some(ALPN.to_string()))
+        } else {
+            relay_spec(name, OperationType::Query, None)
+        };
+        HandlerRegistration::new(
+            spec,
+            HandlerKind::Once(make_handler(|input, ctx| async move {
+                ResponseEnvelope::ok(ctx.request_id, input)
+            })),
+            OperationProvenance::FromCall,
+            None,
+            None,
+            crate::core::types::Capabilities::new(),
+        )
+    };
+    let imports = HubLegImports::from_bundles(vec![
+        bundle("a/marked", true),
+        bundle("b/plain", false),
+        bundle("c/other", false),
+    ]);
+    let filtered = imports.only(&["a/marked", "b/plain"]);
+    assert_eq!(filtered.marked().len(), 1);
+    assert_eq!(filtered.plain().len(), 1);
+    assert_eq!(filtered.plain()[0].spec.name, "b/plain");
+}
+
+// --- the spoke (producer leg) serving the direct + plain ops --------------
+
+fn spoke_open_handler(data_tx: tokio::sync::mpsc::Sender<String>) -> OpenHandler {
+    Arc::new(move |_input, _plan, channel_conn, _auth| {
+        let data_tx = data_tx.clone();
+        tokio::spawn(async move {
+            let Ok(mut bidi) = channel_conn.accept_bi().await else {
+                return;
+            };
+            let mut buf = Vec::new();
+            if tokio::io::AsyncReadExt::read_to_end(&mut bidi, &mut buf)
+                .await
+                .is_err()
+            {
+                return;
+            }
+            let _ = data_tx
+                .send(String::from_utf8_lossy(&buf).to_string())
+                .await;
+            use tokio::io::AsyncWriteExt;
+            let _ = bidi.write_all(buf.to_ascii_uppercase().as_slice()).await;
+            let _ = bidi.shutdown().await;
+        })
+    })
+}
+
+/// The spoke's install hook: registers the marked direct op (with a
+/// `bound`-contributing establisher and an echo data-plane handler)
+/// and a plain echo op, plus bootstrap discovery, and serves channel 0.
+fn make_spoke_install(
+    data_tx: tokio::sync::mpsc::Sender<String>,
+) -> crate::channels::adapter::InstallChannelZero {
+    Arc::new(move |manager, channel0_conn, auth| {
+        let data_tx = data_tx.clone();
+        tokio::spawn(async move {
+            let Ok(channel0_bidi) = channel0_conn.accept_bi().await else {
+                return;
+            };
+            let (writer, reader) = split_single_stream(channel0_bidi);
+            let core = ChannelCore::new(manager, crate::channels::policy::default_policy());
+            let registry = Arc::new(OperationRegistry::new());
+            crate::registry::discovery::install_bootstrap_discovery(&registry)
+                .expect("spoke bootstrap discovery");
+            let establisher: OpenEstablisher = Arc::new(|_input, _auth| {
+                Box::pin(async {
+                    Ok(Establishment::default()
+                        .with_reply_field("bound", json!({ "host": "203.0.113.9", "port": 42113 })))
+                })
+            });
+            let open_handler = spoke_open_handler(data_tx);
+            core.register_openable_with_establisher(
+                relay_spec(OP, OperationType::Sub, Some(ALPN.to_string())),
+                Some(establisher),
+                open_handler,
+                &registry,
+                auth.clone(),
+                None,
+            )
+            .expect("spoke registers the direct open op");
+            registry
+                .register(HandlerRegistration::new(
+                    relay_spec(PLAIN_OP, OperationType::Query, None),
+                    HandlerKind::Once(make_handler(|input, ctx| async move {
+                        ResponseEnvelope::ok(
+                            ctx.request_id,
+                            json!({ "echo": input, "served_by": "spoke" }),
+                        )
+                    })),
+                    OperationProvenance::Local,
+                    None,
+                    None,
+                    crate::core::types::Capabilities::new(),
+                ))
+                .expect("spoke registers the plain op");
+            let call_connection = Arc::new(CallConnection::new_single_stream(
+                channel0_conn,
+                Arc::clone(&writer),
+            ));
+            Dispatcher::new(
+                registry,
+                Arc::new(NoopIdProvider) as Arc<dyn IdentityProvider>,
+            )
+            .run_loop_single_stream(call_connection, reader, writer)
+            .await;
+        })
+    })
+}
+
+/// The full harness: consumer —(alk/channels)— hub —(alk/channels)—
+/// spoke. Returns the consumer's client plus the observability
+/// channels.
+async fn start_hub(
+    consumer_policy: Option<Arc<dyn ChannelLifecyclePolicy>>,
+    imports_override: Option<HubLegImports>,
+    serving_identity: Option<Identity>,
+    scope_gate_stash: bool,
+) -> crate::channels::client::ChannelClient {
+    let (data_tx, _data_rx) = tokio::sync::mpsc::channel::<String>(4);
+
+    // ---- the spoke ----
+    let (hub_spoke_client_end, hub_spoke_server_end) = tokio::io::duplex(64 * 1024);
+    let hub_conn_to_spoke = Connection::from_bidi(
+        hub_spoke_client_end,
+        b"alk/channels".to_vec(),
+        Some(TEST_ADDR),
+    );
+    let spoke_conn = Connection::from_bidi(
+        hub_spoke_server_end,
+        b"alk/channels".to_vec(),
+        Some(TEST_ADDR),
+    );
+    let spoke_adapter = ChannelsAdapter::new(make_spoke_install(data_tx.clone()), Arc::new(NoCap));
+    let _spoke_task = tokio::spawn(async move {
+        let auth = AuthContext::anonymous(b"alk/channels");
+        let _ =
+            crate::core::types::ProtocolHandler::handle(&spoke_adapter, spoke_conn, &auth).await;
+    });
+    drop(data_tx);
+
+    // ---- the hub: dial the spoke, discover, stash ----
+    let hub_client_to_spoke =
+        crate::channels::client::ChannelClient::from_connection(hub_conn_to_spoke)
+            .await
+            .expect("hub's channel client to the spoke");
+    let producer_call = hub_client_to_spoke
+        .take_call_connection()
+        .await
+        .expect("hub takes its channel-0 CallConnection");
+    let bundles = from_call(&producer_call, FromCallConfig::new())
+        .await
+        .expect("from_call discovers the spoke's ops");
+    let producer_leg = ProducerLeg {
+        call: producer_call,
+        manager: hub_client_to_spoke.manager().clone(),
+    };
+    drop(hub_client_to_spoke);
+
+    let imports = match imports_override {
+        Some(imports) => imports,
+        None => {
+            let mut imports = HubLegImports::from_bundles(bundles);
+            assert!(
+                imports.marked().iter().any(|b| b.spec.name == OP),
+                "the flavor-form direct op is stashed marked"
+            );
+            assert!(
+                imports.plain().iter().any(|b| b.spec.name == PLAIN_OP),
+                "the plain op is stashed unmarked"
+            );
+            if scope_gate_stash {
+                // The hub-side ACL composition (ADR-051 §5): the
+                // re-exposed op's gate on the consumer leg is the
+                // imported spec's own AccessControl (+ hub policy) —
+                // scope-gate the stashed spec, not the spoke's.
+                let mut gated = Vec::new();
+                for bundle in imports.plain() {
+                    let mut bundle = bundle.clone();
+                    if bundle.spec.name == PLAIN_OP {
+                        bundle.spec.access_control = AccessControl {
+                            required_scopes: vec!["fs:read".to_string()],
+                            ..AccessControl::default()
+                        };
+                    }
+                    gated.push(bundle);
+                }
+                gated.extend(imports.marked().to_vec());
+                imports = HubLegImports::from_bundles(gated);
+            }
+            imports
+        }
+    };
+
+    let relay = ChannelRelay::new(producer_leg);
+    let mut template = HubLegTemplate::new(relay, imports);
+    if let Some(policy) = consumer_policy {
+        template = template.with_policy(policy);
+    }
+    if let Some(identity) = serving_identity {
+        template = template.with_identity(identity);
+    }
+    let consumer_install = template.install_hook();
+
+    // ---- the hub's consumer leg (the template) ----
+    let (consumer_end, hub_consumer_end) = tokio::io::duplex(64 * 1024);
+    let consumer_conn =
+        Connection::from_bidi(consumer_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
+    let hub_consumer_side =
+        Connection::from_bidi(hub_consumer_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
+    let consumer_adapter = ChannelsAdapter::new(consumer_install, Arc::new(NoCap));
+    let _hub_task = tokio::spawn(async move {
+        let auth = AuthContext::anonymous(b"alk/channels");
+        let _ = crate::core::types::ProtocolHandler::handle(
+            &consumer_adapter,
+            hub_consumer_side,
+            &auth,
+        )
+        .await;
+    });
+
+    crate::channels::client::ChannelClient::from_connection(consumer_conn)
+        .await
+        .expect("consumer's channel client")
+}
+
+// --- gate: the template e2e (ADR-051 gate 5) ------------------------------
+
+/// The template composes fork + generic ops + bootstrap discovery +
+/// plain bundles + relay openables + serving identity and dispatches
+/// channel 0: `services/list` on the consumer leg shows the re-exposed
+/// ops (the marked direct op and the plain op), a plain imported op
+/// round-trips through the hub, the marked op opens through the relay
+/// with `bound` surviving, and data flows both directions.
+#[tokio::test]
+async fn template_dispatches_channel_zero_and_reexposes_ops() {
+    let consumer = start_hub(None, None, None, false).await;
+
+    // 1. services/list on the consumer leg shows the re-exposed ops.
+    let listing = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        consumer.call_open_op("services/list", json!({})),
+    )
+    .await
+    .expect("services/list timed out");
+    let ops: Vec<String> = listing
+        .result
+        .expect("services/list ok")
+        .get("operations")
+        .and_then(|v| v.as_array())
+        .expect("operations array")
+        .iter()
+        .filter_map(|o| o.get("name").and_then(|n| n.as_str()).map(String::from))
+        .collect();
+    assert!(
+        ops.iter().any(|n| n == OP),
+        "the re-exposed marked op is listed: {ops:?}"
+    );
+    assert!(
+        ops.iter().any(|n| n == PLAIN_OP),
+        "the re-exposed plain op is listed: {ops:?}"
+    );
+    assert!(
+        ops.iter().any(|n| n == "channel/close"),
+        "the generic channel ops are registered on the fork: {ops:?}"
+    );
+
+    // 2. A plain imported op round-trips through the hub (the
+    //    call-half forwarding path — the forwarding stub rides the
+    //    fork as-is).
+    let response = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        consumer.call_open_op(PLAIN_OP, json!({ "path": "/etc/motd" })),
+    )
+    .await
+    .expect("plain op round trip timed out");
+    assert!(
+        response.result.is_ok(),
+        "the plain op resolves through the hub, got {:?}",
+        response.result
+    );
+    assert_eq!(
+        response.result.unwrap(),
+        json!({ "echo": { "path": "/etc/motd" }, "served_by": "spoke" }),
+        "the call-half forwarding path executed on the spoke"
+    );
+
+    // 3. The marked op opens through the relay: hub-allocated
+    //    channel_id, `bound` survives, data both directions.
+    let (consumer_id, reply, mut send, mut recv) = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        consumer.open_channel_with_reply(OP, json!({ "target": "example.internal:443" }), ALPN),
+    )
+    .await
+    .expect("relay open timed out")
+    .expect("open through the template");
+    assert!(consumer_id > 0);
+    assert_eq!(reply["channel_id"], json!(consumer_id));
+    assert_eq!(
+        reply["bound"],
+        json!({ "host": "203.0.113.9", "port": 42113 }),
+        "the spoke establisher's `bound` survives the relay through the template"
+    );
+    send.write_all(b"hub-template-bytes").await.expect("write");
+    send.shutdown().await.expect("shutdown");
+    drop(send);
+    let mut echoed = Vec::new();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        recv.read_to_end(&mut echoed),
+    )
+    .await
+    .expect("echo timed out")
+    .expect("read echo");
+    assert_eq!(echoed, b"HUB-TEMPLATE-BYTES");
+}
+
+// --- gate: serving identity (CF-005 precedence on the template) -----------
+
+/// An explicit serving identity authenticates the consumer leg's
+/// calls: a scope-gated plain op (imported through the hub) resolves
+/// FORBIDDEN without the override and with a scope-less override, and
+/// resolves with the scope-bearing override (the ACL layering per
+/// ADR-051 §5 — the hub-side gate is the imported spec's own ACL on
+/// the consumer leg).
+#[tokio::test]
+async fn template_serving_identity_resolves_by_precedence() {
+    let consumer = start_hub(
+        None,
+        None,
+        Some(identity("hub-end-consumer", &["fs:read"])),
+        true,
+    )
+    .await;
+
+    let response = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        consumer.call_open_op(PLAIN_OP, json!({})),
+    )
+    .await
+    .expect("plain op timed out");
+    assert!(
+        response.result.is_ok(),
+        "the scope-bearing serving override authenticates the call, got {:?}",
+        response.result
+    );
+}
+
+/// The scope-less serving override fails the scope-gated op (fail
+/// closed) — the explicit override, not the transport identity, is
+/// what the ACL sees.
+#[tokio::test]
+async fn template_scope_less_identity_fails_scope_gated_op() {
+    let consumer = start_hub(None, None, Some(identity("hub-end-consumer", &[])), true).await;
+    let response = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        consumer.call_open_op(PLAIN_OP, json!({})),
+    )
+    .await
+    .expect("plain op timed out");
+    match response.result {
+        Err(e) => assert_eq!(e.code, "FORBIDDEN", "fail closed without the scope"),
+        Ok(_) => panic!("the scope-gated op must fail closed without the scope"),
+    }
+}
+
+// --- gate: the loud assembly posture (ADR-051 §6) -------------------------
+
+/// A Pub-typed marked spec in the stash is the loud assembly error at
+/// install: the consumer leg's channel 0 never dispatches (a client
+/// call to any op — including the bootstrap discovery ops — does not
+/// resolve). Never a silent stub.
+#[tokio::test]
+async fn template_pub_typed_marked_spec_is_loud_at_install() {
+    let pub_bundle = HandlerRegistration::new(
+        relay_spec(
+            "channels/x/pub",
+            OperationType::Pub,
+            Some("alk/x".to_string()),
+        ),
+        HandlerKind::Once(make_handler(|input, ctx| async move {
+            ResponseEnvelope::ok(ctx.request_id, input)
+        })),
+        OperationProvenance::FromCall,
+        None,
+        None,
+        crate::core::types::Capabilities::new(),
+    );
+    let imports = HubLegImports::from_bundles(vec![pub_bundle]);
+    let consumer = start_hub(None, Some(imports), None, false).await;
+    let response = tokio::time::timeout(
+        std::time::Duration::from_secs(3),
+        consumer.call_open_op("services/list", json!({})),
+    )
+    .await;
+    assert!(
+        response.is_err(),
+        "the leg's channel 0 never dispatches after the loud assembly error, got {:?}",
+        response
+    );
+}
+
+// --- gate: the per-consumer filter + the per-identity cap -----------------
+
+/// The per-consumer op-subset filter composes: a consumer leg
+/// installed with a stash filtered to the plain op serves it but
+/// re-exposes no marked op (its open resolves NOT_FOUND), while
+/// `services/list` still shows the filtered set.
+#[tokio::test]
+async fn template_filtered_stash_serves_subset_only() {
+    let (data_tx, _data_rx) = tokio::sync::mpsc::channel::<String>(4);
+    let policy = Arc::new(PerIdentityChannelPolicy::new(8));
+    let policy_for_assert = Arc::clone(&policy);
+    let policy_dyn: Arc<dyn ChannelLifecyclePolicy> = policy;
+
+    // Re-run the harness with a filtered stash. Build the spoke +
+    // hub inline (start_hub's default imports would carry both ops).
+    let (hub_spoke_client_end, hub_spoke_server_end) = tokio::io::duplex(64 * 1024);
+    let hub_conn_to_spoke = Connection::from_bidi(
+        hub_spoke_client_end,
+        b"alk/channels".to_vec(),
+        Some(TEST_ADDR),
+    );
+    let spoke_conn = Connection::from_bidi(
+        hub_spoke_server_end,
+        b"alk/channels".to_vec(),
+        Some(TEST_ADDR),
+    );
+    let spoke_adapter = ChannelsAdapter::new(make_spoke_install(data_tx), Arc::new(NoCap));
+    let _spoke_task = tokio::spawn(async move {
+        let auth = AuthContext::anonymous(b"alk/channels");
+        let _ =
+            crate::core::types::ProtocolHandler::handle(&spoke_adapter, spoke_conn, &auth).await;
+    });
+
+    let hub_client_to_spoke =
+        crate::channels::client::ChannelClient::from_connection(hub_conn_to_spoke)
+            .await
+            .expect("hub's channel client to the spoke");
+    let producer_call = hub_client_to_spoke
+        .take_call_connection()
+        .await
+        .expect("hub takes its channel-0 CallConnection");
+    let bundles = from_call(&producer_call, FromCallConfig::new())
+        .await
+        .expect("discovery");
+    let producer_leg = ProducerLeg {
+        call: producer_call,
+        manager: hub_client_to_spoke.manager().clone(),
+    };
+    drop(hub_client_to_spoke);
+
+    let imports = HubLegImports::from_bundles(bundles).only(&[PLAIN_OP]);
+    assert!(imports.marked().is_empty(), "the marked op is filtered out");
+    let template =
+        HubLegTemplate::new(ChannelRelay::new(producer_leg), imports).with_policy(policy_dyn);
+    let consumer_install = template.install_hook();
+
+    let (consumer_end, hub_consumer_end) = tokio::io::duplex(64 * 1024);
+    let consumer_conn =
+        Connection::from_bidi(consumer_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
+    let hub_consumer_side =
+        Connection::from_bidi(hub_consumer_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
+    let consumer_adapter = ChannelsAdapter::new(consumer_install, Arc::new(NoCap));
+    let _hub_task = tokio::spawn(async move {
+        let auth = AuthContext::anonymous(b"alk/channels");
+        let _ = crate::core::types::ProtocolHandler::handle(
+            &consumer_adapter,
+            hub_consumer_side,
+            &auth,
+        )
+        .await;
+    });
+    let consumer = crate::channels::client::ChannelClient::from_connection(consumer_conn)
+        .await
+        .expect("consumer's channel client");
+
+    // The plain op still round-trips through the hub.
+    let response = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        consumer.call_open_op(PLAIN_OP, json!({ "q": 1 })),
+    )
+    .await
+    .expect("plain op timed out");
+    assert!(response.result.is_ok(), "the filtered-in plain op resolves");
+
+    // The filtered-out marked op is gone from the fork.
+    let response = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        consumer.call_open_op(OP, json!({})),
+    )
+    .await
+    .expect("filtered-out op timed out");
+    match response.result {
+        Err(e) => assert_eq!(e.code, "NOT_FOUND", "the filtered-out op is not re-exposed"),
+        Ok(_) => panic!("the filtered-out marked op must not be re-exposed"),
+    }
+    assert_eq!(
+        policy_for_assert.count_for(&identity("anonymous", &[])),
+        0,
+        "no channel was ever opened on the consumer leg"
+    );
+}

src/channels/mod.rs

@@ -31,6 +31,10 @@
 //!   as amended by ADR-051): translate hop + `pump_bidi` byte-forward
 //!   hop for from_call-imported marked specs re-exposed on a hub's
 //!   consumer-leg channel-0 registry.
+//! - [`hub_leg`]: `HubLegTemplate` — the hub-leg install template
+//!   (ADR-051 §5): fork + generic channel ops + bootstrap discovery +
+//!   plain bundles + relay openables + serving identity, composed into
+//!   the `install_channel_zero` hook every hub side needs.
 //! - [`client`]: `ChannelClient` — transport-agnostic
 //!   `from_connection` (ADR-043).
 //! - [`self::env`]: `ChannelOperationEnv` extension trait (ADR-047 §4 —
@@ -41,6 +45,7 @@
 pub mod adapter;
 pub mod client;
 pub mod env;
+pub mod hub_leg;
 pub mod manager;
 pub mod mux;
 pub mod operations;

src/registry/discovery.rs

@@ -14,6 +14,16 @@ const NAME_SERVICES_LIST: &str = "services/list";
 const NAME_SERVICES_LIST_PEERS: &str = "services/list-peers";
 const NAME_SERVICES_SCHEMA: &str = "services/schema";
 
+/// The bootstrap discovery op names — the set a hub-leg assembly
+/// skips when re-registering imported plain bundles (the template's
+/// own `install_bootstrap_discovery`, closed over its fork,
+/// supersedes the imported copies; ADR-051 §5).
+pub const BOOTSTRAP_DISCOVERY_OPS: [&str; 3] = [
+    NAME_SERVICES_LIST,
+    NAME_SERVICES_LIST_PEERS,
+    NAME_SERVICES_SCHEMA,
+];
+
 pub fn services_list_spec() -> OperationSpec {
     OperationSpec::new(
         NAME_SERVICES_LIST,