feat(websocket): data-channel + op/register wiring (review 006 Unit 2+3)

The WS path wires the alkcall 0.3 per-session mechanisms — the OQ-05
deferred half (review-003 WS-20/21/22/25/26, the decisions WS-24/WS-25
resolved upstream):

- install_channel_zero reworked per the ADR-047 §4 amendment #2 shape:
  fork the deployment's base registry, register the generic channel
  ops (channel/close, channel/control, channel/resources/subscribe —
  WS-21), the deployment's openable ALPNs (ChannelCore::register_openable,
  WS-22), the bootstrap discovery set closed over the fork
  (install_bootstrap_discovery, F-06), and op/register (WS-25; the
  collision set is the fork per review-005 G-03), then dispatch over
  the fork. The session's ChannelsPolicy rides the hook (one policy
  instance across open wrappers and the demux teardown path).
- OpenableAlpn { spec, open_handler } + HttpAdapter::with_ws_openable_alpns,
  threaded RouterState -> SessionState -> hook, with the OpenableAlpns
  request-extension fallback (mirroring ChannelsPolicy/WsTimeouts).
- WsSessions retains the channel-0 Arc<CallConnection> (WS-26) with a
  self-removing guard (ConnectionGuard); live_connections() is the
  deployment-visible surface.
- UP-01: ALREADY_EXISTS maps to 409 Conflict in the gateway error map.
- from_wss import excludes the protocol-session ops (bootstrap set +
  channel lifecycle ops): the fork serves them per session, and proxying
  session-scoped machinery (e.g. channel/close across sessions) would be
  nonsense. Discovery runs first, the filter is the listing minus those
  names.
- adapter_install_channel_zero cfg matches its caller (WS-27); it
  inherits the reworked hook (session ops now served in the from_wss
  test-server producer too).

Gates (Unit 3, tests/ws_upgrade_session.rs; the WS-23 e2e set):
open -> channel_id -> discoverable in services/list -> chunks both
ways -> handler sees bytes; channel/close resolves + ledger decrement;
cap denial (channel:-prefixed); mid-open disconnect teardown; TooLarge
demux resync through the WS path (16 MiB + 1 skip consumed);
op/register announce + overlay-collision + serving-registry-collision
ALREADY_EXISTS through the WS path.

call_and_await now filters by request id and tolerates data-channel
chunks (a prior Sub's trailing call.completed may interleave).

Verification: cargo test 454 (default) / 582 (all-features), clippy
both sides -D warnings clean, fmt clean, doc clean.

src/adapters/from_wss.rs

@@ -323,6 +323,28 @@ impl WssSession {
     }
 }
 
+/// The per-session protocol ops a channels session serves on channel 0
+/// (alkcall ADR-022 amendment's bootstrap set + the channel lifecycle
+/// set): a `from_wss` import must not proxy these — they are
+/// session-scoped machinery, not domain operations. Importing
+/// `channel/close` would forward a close for *this* session's channel
+/// ids to the remote (nonsense: ids are per-session), and
+/// `op/register`/`services/list-peers` are registration/discovery
+/// machinery whose proxy duplicates the import itself. The
+/// `services/list` + `services/schema` discovery pair is excluded
+/// from import for the same reason (the importer calls them on every
+/// import; a proxy of them is dead weight), matching what the pure
+/// `from_call` path does — its own discovery dials the real ones.
+fn is_protocol_session_op(name: &str) -> bool {
+    name == "services/list"
+        || name == "services/schema"
+        || name == "services/list-peers"
+        || name == alkcall::registry::op_register::OP_REGISTER_NAME
+        || name == alkcall::channels::operations::OP_CHANNEL_CLOSE
+        || name == alkcall::channels::operations::OP_CHANNEL_CONTROL
+        || name == alkcall::channels::operations::OP_CHANNEL_RESOURCES_SUBSCRIBE
+}
+
 #[async_trait::async_trait]
 impl OperationAdapter for FromWss {
     async fn import(
@@ -334,10 +356,15 @@ impl OperationAdapter for FromWss {
             self.allow_plaintext,
         )
         .await?;
+        // The protocol-session ops ride the listing (the session fork
+        // serves them) but are excluded: the importer builds an
+        // operation_filter from the listing minus those names.
         let config = match &self.namespace {
             Some(ns) => FromCallConfig::new().with_namespace_prefix(ns),
             None => FromCallConfig::new(),
         };
+        let config = config
+            .with_operation_filter(protocol_session_ops_from(&session.call_connection).await?);
         let bundles = import_from_call(&session.call_connection, config).await;
         // Fire-and-forget: the imported handlers keep working off the
         // session's Arc'd CallConnection; the session's tasks are
@@ -348,6 +375,36 @@ impl OperationAdapter for FromWss {
     }
 }
 
+/// The remote's `services/list` names minus the protocol-session ops —
+/// the `operation_filter` the import runs with. A listing failure is
+/// `DiscoveryFailed` from `from_call` itself (the real error path); a
+/// parse failure of the names array is a transport-level parse error.
+async fn protocol_session_ops_from(
+    connection: &CallConnection,
+) -> Result<std::collections::HashSet<String>, AdapterError> {
+    let response = connection
+        .call("services/list", serde_json::json!({}))
+        .await;
+    let output = response.result.map_err(|e| AdapterError::DiscoveryFailed {
+        message: format!("services/list failed: {} ({})", e.code, e.message),
+    })?;
+    let ops = output
+        .get("operations")
+        .and_then(|v| v.as_array())
+        .ok_or_else(|| AdapterError::SchemaParse {
+            message: "services/list response missing 'operations' array".to_string(),
+        })?;
+    let mut filter = std::collections::HashSet::new();
+    for op in ops {
+        if let Some(name) = op.get("name").and_then(|v| v.as_str()) {
+            if !is_protocol_session_op(name) {
+                filter.insert(name.to_string());
+            }
+        }
+    }
+    Ok(filter)
+}
+
 #[cfg(all(test, feature = "server"))]
 mod tests {
     use super::*;

src/gateway/error.rs

@@ -2,9 +2,11 @@
 //!
 //! Protocol-level vs operation-level code distinction: protocol codes
 //! (`NOT_FOUND`, `FORBIDDEN`, `INVALID_INPUT`, `INVALID_OPERATION_TYPE`,
-//! `TIMEOUT`, `INTERNAL`) map to fixed statuses; operation-level codes
-//! imported from external HTTP APIs are prefixed `HTTP_<status>` and map
-//! to their declared status.
+//! `TIMEOUT`, `ALREADY_EXISTS` — the seventh code, alkcall ADR-022's
+//! `op/register` collision policy, mapped 409 Conflict, review-006
+//! UP-01) map to fixed statuses; operation-level codes imported from
+//! external HTTP APIs are prefixed `HTTP_<status>` and map to their
+//! declared status.
 //!
 //! The identity-aware variant maps the ambiguous protocol codes
 //! (`FORBIDDEN`, `INVALID_OPERATION_TYPE`) to `401` when no token
@@ -27,6 +29,7 @@ const PROTOCOL_CODE_FORBIDDEN: &str = "FORBIDDEN";
 const PROTOCOL_CODE_INVALID_INPUT: &str = "INVALID_INPUT";
 const PROTOCOL_CODE_INVALID_OPERATION_TYPE: &str = "INVALID_OPERATION_TYPE";
 const PROTOCOL_CODE_TIMEOUT: &str = "TIMEOUT";
+const PROTOCOL_CODE_ALREADY_EXISTS: &str = "ALREADY_EXISTS";
 const PROTOCOL_CODE_INTERNAL: &str = "INTERNAL";
 
 const HTTP_PREFIX: &str = "HTTP_";
@@ -36,6 +39,7 @@ const STATUS_UNAUTHORIZED: u16 = 401;
 const STATUS_FORBIDDEN: u16 = 403;
 const STATUS_UNPROCESSABLE: u16 = 422;
 const STATUS_TIMEOUT: u16 = 504;
+const STATUS_CONFLICT: u16 = 409;
 const STATUS_INTERNAL: u16 = 500;
 
 const RETRY_AFTER_STATUSES: &[u16] = &[429, 503];
@@ -72,6 +76,7 @@ pub fn call_error_to_http_status_with_identity(
             }
         }
         PROTOCOL_CODE_TIMEOUT => STATUS_TIMEOUT,
+        PROTOCOL_CODE_ALREADY_EXISTS => STATUS_CONFLICT,
         PROTOCOL_CODE_INTERNAL => STATUS_INTERNAL,
         code if code.starts_with(HTTP_PREFIX) => code[HTTP_PREFIX.len()..]
             .parse::<u16>()
@@ -182,6 +187,12 @@ mod tests {
         assert_eq!(call_error_to_http_status(&error), 504);
     }
 
+    #[test]
+    fn already_exists_maps_to_409() {
+        let error = CallError::already_exists("name taken");
+        assert_eq!(call_error_to_http_status(&error), 409);
+    }
+
     #[test]
     fn internal_maps_to_500() {
         let error = CallError::internal("boom");

src/gateway/routes.rs

@@ -1137,6 +1137,7 @@ mod tests {
                 crate::websocket::DEFAULT_WS_MAX_SESSIONS,
             )),
             ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
+            ws_openable_alpns: None,
         };
         let auth_state = Arc::clone(&provider);
         gateway_router()

src/server/adapter.rs

@@ -98,6 +98,7 @@ pub struct HttpAdapter {
     ws_max_sessions: usize,
     ws_session_slots: Arc<tokio::sync::Semaphore>,
     ws_idle_timeout: Option<Duration>,
+    ws_openable_alpns: Option<Arc<[crate::websocket::OpenableAlpn]>>,
 }
 
 impl HttpAdapter {
@@ -136,6 +137,7 @@ impl HttpAdapter {
             ws_sessions: Arc::clone(&ws_sessions),
             ws_session_slots: Arc::clone(&ws_session_slots),
             ws_idle_timeout,
+            ws_openable_alpns: None,
         };
         let router = build_router(state, None);
         Self {
@@ -150,6 +152,7 @@ impl HttpAdapter {
             ws_max_sessions,
             ws_session_slots,
             ws_idle_timeout,
+            ws_openable_alpns: None,
         }
     }
 
@@ -165,6 +168,7 @@ impl HttpAdapter {
             ws_sessions: Arc::clone(&self.ws_sessions),
             ws_session_slots: Arc::clone(&self.ws_session_slots),
             ws_idle_timeout: self.ws_idle_timeout,
+            ws_openable_alpns: self.ws_openable_alpns.clone(),
         };
         // `extra_routes` is borrowed, not consumed (SRV-05): a builder
         // call after `with_extra_routes` must keep the custom routes in
@@ -186,6 +190,7 @@ impl HttpAdapter {
             ws_sessions: Arc::clone(&self.ws_sessions),
             ws_session_slots: Arc::clone(&self.ws_session_slots),
             ws_idle_timeout: self.ws_idle_timeout,
+            ws_openable_alpns: self.ws_openable_alpns.clone(),
         };
         self.router = build_router(state, Some(routes.clone()));
         self.extra_routes = Some(routes);
@@ -209,6 +214,7 @@ impl HttpAdapter {
             ws_sessions: Arc::clone(&self.ws_sessions),
             ws_session_slots: Self::rebuild_session_slots(max_sessions),
             ws_idle_timeout: self.ws_idle_timeout,
+            ws_openable_alpns: self.ws_openable_alpns.clone(),
         };
         self.router = build_router(state, self.extra_routes.clone());
         self
@@ -240,6 +246,7 @@ impl HttpAdapter {
             ws_sessions: Arc::clone(&self.ws_sessions),
             ws_session_slots: Arc::clone(&self.ws_session_slots),
             ws_idle_timeout: self.ws_idle_timeout,
+            ws_openable_alpns: self.ws_openable_alpns.clone(),
         };
         self.router = build_router(state, self.extra_routes.clone());
         self
@@ -253,6 +260,44 @@ impl HttpAdapter {
         Arc::clone(&self.ws_sessions)
     }
 
+    /// The openable-ALPN set for WS sessions (WS-22, review 006
+    /// Unit 2): one [`OpenableAlpn`](crate::websocket::OpenableAlpn)
+    /// per openable data-channel ALPN — the open-op spec (with the
+    /// `channel_open` marker) and the ALPN-specific
+    /// [`OpenHandler`](crate::websocket::OpenHandler). Each WS
+    /// session's per-session fork registers the set (plus the generic
+    /// channel ops, bootstrap discovery, and `op/register`), so a WS
+    /// client can open data channels exactly as any channels consumer
+    /// (ADR-067: the browser reaches what a Rust consumer on an
+    /// in-line connection would reach).
+    ///
+    /// The ALPN-specific `OpenHandler` implementations stay in the
+    /// ALPN crates (alktty et al.); this adapter only ferries the
+    /// registrations onto the session fork. Bare-registry / custom
+    /// upgrade routes pass their own set via the
+    /// [`OpenableAlpns`](crate::websocket::OpenableAlpns) request
+    /// extension instead.
+    ///
+    /// Default: no openables (channel 0 only — the pre-Unit-2 shape).
+    pub fn with_ws_openable_alpns(
+        mut self,
+        openables: Vec<crate::websocket::OpenableAlpn>,
+    ) -> Self {
+        self.ws_openable_alpns = Some(openables.into());
+        let state = RouterState {
+            registry: Arc::clone(&self.registry),
+            identity_provider: Arc::clone(&self.identity_provider),
+            decoy: self.decoy.clone(),
+            openapi_doc: self.openapi_doc.clone(),
+            ws_sessions: Arc::clone(&self.ws_sessions),
+            ws_session_slots: Arc::clone(&self.ws_session_slots),
+            ws_idle_timeout: self.ws_idle_timeout,
+            ws_openable_alpns: self.ws_openable_alpns.clone(),
+        };
+        self.router = build_router(state, self.extra_routes.clone());
+        self
+    }
+
     /// A fresh semaphore for the new cap; the retained handles of
     /// already-open sessions are unaffected (they hold their permits).
     fn rebuild_session_slots(max_sessions: usize) -> Arc<tokio::sync::Semaphore> {
@@ -1153,6 +1198,7 @@ mod tests {
                 crate::websocket::DEFAULT_WS_MAX_SESSIONS,
             )),
             ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
+            ws_openable_alpns: None,
         }
     }
 

src/server/state.rs

@@ -48,6 +48,11 @@ 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>,
+    /// The openable-ALPN set for WS sessions (WS-22): the per-ALPN
+    /// open-op specs + handlers registered on each session's fork.
+    /// `None` (the default) declares no openables — a WS session then
+    /// carries no data-channel open ops (the pre-Unit-2 shape).
+    pub(crate) ws_openable_alpns: Option<Arc<[crate::websocket::OpenableAlpn]>>,
 }
 
 impl axum::extract::FromRef<RouterState> for crate::websocket::SessionState {
@@ -57,6 +62,7 @@ impl axum::extract::FromRef<RouterState> for crate::websocket::SessionState {
             Arc::clone(&state.ws_sessions),
             Arc::clone(&state.ws_session_slots),
             state.ws_idle_timeout,
+            state.ws_openable_alpns.clone(),
         )
     }
 }
@@ -96,6 +102,7 @@ mod tests {
                 crate::websocket::DEFAULT_WS_MAX_SESSIONS,
             )),
             ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
+            ws_openable_alpns: None,
         };
         let extracted: DecoyConfig = axum::extract::FromRef::from_ref(&state);
         assert!(matches!(extracted, DecoyConfig::Redirect { .. }));

src/websocket/mod.rs

@@ -30,15 +30,16 @@ pub use byte_adapter::{
     split_ws_to_bytes, split_ws_to_bytes_idle, split_ws_to_bytes_idle_with_write,
 };
 
+#[cfg(feature = "server")]
+pub use alkcall::channels::operations::ChannelCore;
 #[cfg(any(test, feature = "wss"))]
 pub use byte_adapter::split_tungstenite_to_bytes;
-#[cfg(feature = "server")]
-#[allow(unused_imports)]
+#[cfg(all(test, feature = "server", feature = "wss"))]
 pub(crate) use upgrade::adapter_install_channel_zero;
 #[cfg(feature = "server")]
 pub use upgrade::{
-    run_channels_session, ws_bearer_auth, ws_upgrade_handler, ChannelsPolicy, SessionState,
-    WsSessions, WsTimeouts, DEFAULT_WS_MAX_SESSIONS,
+    run_channels_session, ws_bearer_auth, ws_upgrade_handler, ChannelsPolicy, OpenableAlpn,
+    OpenableAlpns, SessionState, WsSessions, WsTimeouts, DEFAULT_WS_MAX_SESSIONS,
 };
 
 #[cfg(all(any(test, feature = "test-support"), feature = "server"))]

src/websocket/upgrade.rs

@@ -6,9 +6,14 @@
 //! `resolve_from_token` path as any HTTP request), then upgrade →
 //! WS↔byte-stream adapter → `Connection::from_bidi(ws_stream,
 //! b"alk/channels")` → alkcall `ChannelsAdapter` (the channels accept
-//! path). The `install_channel_zero` hook constructs channel 0's
+//! path). The `install_channel_zero` hook forks the deployment's base
+//! registry, registers the per-session ops on the fork (the generic
+//! channel lifecycle ops, the deployment's openable ALPNs, the
+//! bootstrap discovery set, and `op/register` — alkcall ADR-047 §4
+//! amendment #2 + ADR-022 amendment), constructs channel 0's
 //! `CallConnection` (the identity rides the channels-layer
-//! `Connection`) and runs the shared `Dispatcher::run_loop_single_stream`.
+//! `Connection`), retains it in the session registry (WS-26), and
+//! runs the shared `Dispatcher::run_loop_single_stream` over the fork.
 //!
 //! ## Detached task lifetime semantics (WS-10)
 //!
@@ -41,10 +46,12 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 
 use alkcall::channels::adapter::ChannelsAdapter;
+use alkcall::channels::operations::{ChannelCore, OpenHandler};
 use alkcall::channels::policy::{ChannelLifecyclePolicy, NoCap};
 use alkcall::core::auth::{AuthContext, Identity};
 use alkcall::core::types::{Connection, ProtocolHandler};
 use alkcall::registry::registration::OperationRegistry;
+use alkcall::registry::spec::OperationSpec;
 use axum::extract::ws::WebSocketUpgrade;
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
@@ -71,6 +78,7 @@ use super::byte_adapter::{
 pub struct WsSessions {
     counter: Arc<AtomicU64>,
     sessions: Arc<Mutex<HashMap<u64, Arc<WsPumps>>>>,
+    connections: Arc<Mutex<HashMap<u64, Arc<alkcall::protocol::connection::CallConnection>>>>,
 }
 
 /// Default bound on concurrent WS sessions (WS-09): the semaphore
@@ -94,6 +102,8 @@ pub struct SessionState {
     session_slots: Arc<tokio::sync::Semaphore>,
     /// Idle-read timeout (WS-01): `None` disables the knob.
     idle_timeout: Option<std::time::Duration>,
+    /// The openable-ALPN set (WS-22): `None` declares no openables.
+    openable_alpns: Option<Arc<[OpenableAlpn]>>,
 }
 
 impl SessionState {
@@ -107,24 +117,27 @@ impl SessionState {
             sessions: Arc::new(WsSessions::new()),
             session_slots: Arc::new(tokio::sync::Semaphore::new(DEFAULT_WS_MAX_SESSIONS)),
             idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
+            openable_alpns: None,
         }
     }
 
     /// From the adapter's router state: the shared [`WsSessions`]
-    /// instance the assembly layer can hold for eviction, and the
+    /// instance the assembly layer can hold for eviction, the
     /// pre-built session-cap semaphore (one per `HttpAdapter`, shared
-    /// across requests).
+    /// across requests), and the deployment's openable-ALPN set.
     pub(crate) fn new(
         registry: Arc<OperationRegistry>,
         sessions: Arc<WsSessions>,
         session_slots: Arc<tokio::sync::Semaphore>,
         idle_timeout: Option<std::time::Duration>,
+        openable_alpns: Option<Arc<[OpenableAlpn]>>,
     ) -> Self {
         Self {
             registry,
             sessions,
             session_slots,
             idle_timeout,
+            openable_alpns,
         }
     }
 
@@ -139,6 +152,10 @@ impl SessionState {
     pub(crate) fn idle_timeout(&self) -> Option<std::time::Duration> {
         self.idle_timeout
     }
+
+    pub(crate) fn openable_alpns(&self) -> Option<Arc<[OpenableAlpn]>> {
+        self.openable_alpns.clone()
+    }
 }
 
 /// `FromRef` chain: a bare `Arc<OperationRegistry>` router state lifts
@@ -173,6 +190,26 @@ impl WsSessions {
         self.sessions.lock().is_empty()
     }
 
+    /// The live session's channel-0 [`CallConnection`] handles (WS-26):
+    /// the deployment-visible surface the assembly layer (or the hub)
+    /// reaches browser/session-side ops through — connection-local
+    /// overlay composition, `announce_op`-style pushes, session
+    /// enumeration. The handle is registered when the session's
+    /// channel-0 connection is built and removed when the channel-0
+    /// task ends (session teardown of any path), so the registry only
+    /// holds live sessions.
+    ///
+    /// Un-registered (default) — no handles are retained; the upgrade
+    /// runs exactly as the pre-WS-26 shape.
+    pub fn live_connections(&self) -> Vec<Arc<alkcall::protocol::connection::CallConnection>> {
+        self.connections.lock().values().cloned().collect()
+    }
+
+    /// Number of live channel-0 connection handles currently tracked.
+    pub fn live_connection_count(&self) -> usize {
+        self.connections.lock().len()
+    }
+
     fn insert(&self, pumps: Arc<WsPumps>) -> u64 {
         let id = self.counter.fetch_add(1, Ordering::Relaxed);
         self.sessions.lock().insert(id, pumps);
@@ -182,6 +219,41 @@ impl WsSessions {
     fn remove(&self, id: u64) {
         self.sessions.lock().remove(&id);
     }
+
+    fn insert_connection(&self, conn: Arc<alkcall::protocol::connection::CallConnection>) -> u64 {
+        let id = self.counter.fetch_add(1, Ordering::Relaxed);
+        self.connections.lock().insert(id, conn);
+        id
+    }
+
+    fn remove_connection(&self, id: u64) {
+        self.connections.lock().remove(&id);
+    }
+}
+
+/// One deployment-declared openable ALPN for the WS path (WS-22):
+/// the per-ALPN open-op `OperationSpec` (with the `channel_open`
+/// marker set via `OperationSpec::with_channel_open`) and the
+/// ALPN-specific [`OpenHandler`] the data-plane protocol runs on the
+/// allocated channel's `Connection`. The ALPN-specific handler stays
+/// in the ALPN crates (alktty et al.); this crate only ferries the
+/// registration onto each session's fork.
+#[derive(Clone)]
+pub struct OpenableAlpn {
+    /// The open-op spec (Query/Mutation/Sub with the `channel_open`
+    /// marker; a `Pub` open op resolves `channel:pub_open_not_implemented`
+    /// per the upstream C-08 stub until channel adoption lands).
+    pub spec: OperationSpec,
+    /// The data-plane protocol handler spawned on the allocated
+    /// channel's `Connection`.
+    pub open_handler: OpenHandler,
+}
+
+impl OpenableAlpn {
+    /// Declare one openable ALPN.
+    pub fn new(spec: OperationSpec, open_handler: OpenHandler) -> Self {
+        Self { spec, open_handler }
+    }
 }
 
 /// The channels session for an upgraded socket: adapt → `Connection`
@@ -193,7 +265,8 @@ impl WsSessions {
 /// closes the read with 1001 after `d` without inbound chunk
 /// progress. `write_timeout` bounds one outbound WS send (WS-18):
 /// `None` = the crate default window, `Some(d)` a deployment-set
-/// window.
+/// window. `openable_alpns` (WS-22) is the deployment's openable-ALPN
+/// set registered per session (see [`install_channel_zero`]).
 #[allow(clippy::too_many_arguments)]
 pub async fn run_channels_session(
     socket: axum::extract::ws::WebSocket,
@@ -203,20 +276,27 @@ pub async fn run_channels_session(
     sessions: Option<WsSessions>,
     idle_timeout: Option<std::time::Duration>,
     write_timeout: Option<std::time::Duration>,
+    openable_alpns: Option<Arc<[OpenableAlpn]>>,
 ) {
     let (byte_stream, pumps) =
         split_ws_to_bytes_idle_with_write(socket, idle_timeout, write_timeout);
     let pumps = Arc::new(pumps);
 
-    let _guard = sessions.map(|sessions| {
-        let id = sessions.insert(Arc::clone(&pumps));
-        SessionGuard { sessions, id }
+    let _guard = sessions.as_ref().map(|s| {
+        let id = s.insert(Arc::clone(&pumps));
+        SessionGuard {
+            sessions: s.clone(),
+            id,
+        }
     });
 
     let conn = Connection::from_bidi(byte_stream, b"alk/channels".to_vec(), None);
     let _ = conn.set_identity(identity.clone());
 
-    let adapter = ChannelsAdapter::new(install_channel_zero(registry), policy);
+    let adapter = ChannelsAdapter::new(
+        install_channel_zero(registry, sessions, Arc::clone(&policy), openable_alpns),
+        policy,
+    );
     let auth = AuthContext {
         identity: Some(identity),
         alpn: b"alk/channels".to_vec(),
@@ -239,10 +319,45 @@ impl Drop for SessionGuard {
     }
 }
 
-/// The `install_channel_zero` hook: split channel 0's `BiStream` into
-/// the shared writer + reader, construct the single-stream
-/// `CallConnection`, and run the shared `Dispatcher`'s single-stream
-/// loop (the alkcall `channels/client.rs` accept-side wiring pattern).
+/// The channel-0 connection handle's self-removing guard (WS-26): the
+/// drop removes the retained [`CallConnection`] from the session
+/// registry, so the registry only holds live sessions' handles.
+struct ConnectionGuard {
+    sessions: WsSessions,
+    id: u64,
+}
+
+impl Drop for ConnectionGuard {
+    fn drop(&mut self) {
+        self.sessions.remove_connection(self.id);
+    }
+}
+
+/// The `install_channel_zero` hook (WS-20/21/22/25/26, review 006
+/// Unit 2): fork the deployment's base registry, register the
+/// per-session ops on the fork — the generic channel lifecycle ops
+/// (`ChannelOperations::register_on`: `channel/close`,
+/// `channel/control`, `channel/resources/subscribe`), the
+/// deployment's openable ALPNs (`ChannelCore::register_openable`),
+/// the bootstrap discovery set closed over the fork
+/// (`install_bootstrap_discovery`, so `services/list` sees the
+/// session's own openables — the F-06 shape), and `op/register`
+/// (alkcall ADR-022 amendment; the collision set is the fork) — then
+/// run the dispatcher over the fork (alkcall ADR-047 §4 amendment #2,
+/// the per-session-fork mechanism; the alkcall e2e reference shape).
+///
+/// `policy` is the session's channel cap policy (the `ChannelsPolicy`
+/// extension resolution, `NoCap` default): the open-op wrappers the
+/// hook registers consult it (`check_open` per identity, ledger
+/// decrement on teardown), and the adapter's demux loop decrements it
+/// on connection-drop teardown — one policy instance across both
+/// halves.
+///
+/// The hook also retains the channel-0 `CallConnection` in the
+/// session registry (WS-26) when a shared [`WsSessions`] instance is
+/// in play: the deployment reaches the session's connection-local
+/// overlay and can hub→session-call announced or imported ops. The
+/// handle is removed when the channel-0 task ends (any teardown path).
 ///
 /// The dispatcher's token resolver is a no-op: the WS identity is
 /// attached to the connection at upgrade time and
@@ -251,9 +366,15 @@ impl Drop for SessionGuard {
 /// identity was established at upgrade time, not per-call).
 fn install_channel_zero(
     registry: Arc<OperationRegistry>,
+    sessions: Option<WsSessions>,
+    policy: Arc<dyn ChannelLifecyclePolicy>,
+    openable_alpns: Option<Arc<[OpenableAlpn]>>,
 ) -> alkcall::channels::adapter::InstallChannelZero {
-    Arc::new(move |_manager, channel0_conn, auth| {
+    Arc::new(move |manager, channel0_conn, auth| {
         let registry = Arc::clone(&registry);
+        let sessions = sessions.clone();
+        let policy = Arc::clone(&policy);
+        let openable_alpns = openable_alpns.clone();
         tokio::spawn(async move {
             // The WS identity rides the upgrade request; propagate it to
             // channel 0's `CallConnection` so the dispatcher's
@@ -276,8 +397,69 @@ fn install_channel_zero(
                     Arc::clone(&writer),
                 ),
             );
+
+            // The per-session dispatch registry (WS-24's decided
+            // mechanism): fork the base, register the session's ops on
+            // the fork, dispatch over the fork. Registration failures
+            // (a spec whose publish_schema fails to compile — the only
+            // error class the registry can produce here, since the
+            // openable specs were accepted at declaration) kill the
+            // session before the loop starts: a session that would
+            // mis-discover must not silently lose the failing op.
+            let fork = Arc::new(registry.fork());
+
+            let register_result: Result<(), String> = (|| {
+                alkcall::channels::operations::ChannelOperations::new(
+                    manager.clone(),
+                    Arc::clone(&policy),
+                )
+                .register_on(&fork)?;
+                if let Some(openables) = openable_alpns.as_ref() {
+                    let core = ChannelCore::new(manager, Arc::clone(&policy));
+                    for openable in openables.iter() {
+                        core.register_openable(
+                            openable.spec.clone(),
+                            Arc::clone(&openable.open_handler),
+                            &fork,
+                            auth.clone(),
+                        )?;
+                    }
+                }
+                alkcall::registry::discovery::install_bootstrap_discovery(&fork)?;
+                fork.register(alkcall::registry::registration::HandlerRegistration::new(
+                    alkcall::registry::op_register::op_register_spec(
+                        alkcall::registry::spec::AccessControl::default(),
+                    ),
+                    alkcall::registry::registration::HandlerKind::Once(
+                        alkcall::registry::op_register::op_register_handler(
+                            Arc::clone(&call_connection),
+                            Arc::clone(&fork),
+                        ),
+                    ),
+                    alkcall::registry::registration::OperationProvenance::Local,
+                    None,
+                    None,
+                    alkcall::core::types::Capabilities::new(),
+                ))?;
+                Ok(())
+            })();
+            if let Err(e) = register_result {
+                tracing::error!(error = %e, "channel-0 session registry setup failed");
+                return;
+            }
+
+            if let Some(sessions) = &sessions {
+                let id = sessions.insert_connection(Arc::clone(&call_connection));
+                let _conn_guard = ConnectionGuard {
+                    sessions: sessions.clone(),
+                    id,
+                };
+                // The guard lives for this task — its drop removes the
+                // handle when the dispatcher loop returns.
+            }
+
             let dispatcher = alkcall::protocol::dispatch::Dispatcher::new(
-                registry,
+                fork,
                 std::sync::Arc::new(NoopProvider),
             );
             dispatcher
@@ -287,11 +469,14 @@ fn install_channel_zero(
     })
 }
 
-#[allow(dead_code)]
+/// The bare-registry variant of the hook for the from_wss test
+/// server: no session registry (no WS-26 handle retention — the test
+/// server's pump slot is its own teardown lever) and no openables.
+#[cfg(all(test, feature = "server", feature = "wss"))]
 pub(crate) fn adapter_install_channel_zero(
     registry: Arc<OperationRegistry>,
 ) -> alkcall::channels::adapter::InstallChannelZero {
-    install_channel_zero(registry)
+    install_channel_zero(registry, None, Arc::new(NoCap), None)
 }
 
 struct NoopProvider;
@@ -341,6 +526,16 @@ pub struct WsTimeouts {
     pub write: Option<std::time::Duration>,
 }
 
+/// Per-request openable-ALPN override (WS-22): a deployment inserts
+/// `OpenableAlpns` into the request extensions (a route layer on the WS
+/// route, mirroring [`ChannelsPolicy`] / [`WsTimeouts`]) to set the
+/// session's openable set per route instead of the router-state
+/// default. Without the extension the [`SessionState`] value applies —
+/// [`HttpAdapter::with_ws_openable_alpns`](crate::server::HttpAdapter::with_ws_openable_alpns)
+/// for the built-in surface; bare-registry routes carry none.
+#[derive(Clone)]
+pub struct OpenableAlpns(pub Arc<[OpenableAlpn]>);
+
 /// The upgrade handler. Requires the resolved identity in request
 /// extensions (stashed by [`ws_bearer_auth`]) — a WS session without
 /// an identity cannot run `AccessControl::check`.
@@ -362,6 +557,11 @@ pub struct WsTimeouts {
 /// above; a deployment overriding them inserts the `WsTimeouts`
 /// extension on its upgrade route.
 ///
+/// The openable-ALPN set (WS-22) comes from the [`OpenableAlpns`]
+/// request extension when present, else from the router state
+/// (`HttpAdapter::with_ws_openable_alpns`); the default is no
+/// openables (channel 0 only).
+///
 /// Session cap (WS-09): one semaphore permit is acquired per upgrade,
 /// post-auth and pre-upgrade; when the configured cap
 /// (`HttpAdapter::with_ws_max_sessions`, default
@@ -374,6 +574,7 @@ pub async fn ws_upgrade_handler(
     axum::Extension(identity): axum::Extension<Identity>,
     policy: Option<axum::Extension<ChannelsPolicy>>,
     timeouts: Option<axum::Extension<WsTimeouts>>,
+    openables: Option<axum::Extension<OpenableAlpns>>,
     ws_upgrade: WebSocketUpgrade,
 ) -> Response {
     let Ok(permit) = state.session_slots.clone().try_acquire_owned() else {
@@ -397,6 +598,9 @@ pub async fn ws_upgrade_handler(
         None => state.idle_timeout(),
     };
     let write_timeout = timeouts.and_then(|t| t.write);
+    let openable_alpns = openables
+        .map(|axum::Extension(o)| o.0)
+        .or_else(|| state.openable_alpns());
     ws_upgrade
         .max_frame_size(INBOUND_WS_FRAME_CAP)
         .max_message_size(INBOUND_WS_MESSAGE_CAP)
@@ -410,6 +614,7 @@ pub async fn ws_upgrade_handler(
                 sessions,
                 idle_timeout,
                 Some(write_timeout.unwrap_or(crate::websocket::DEFAULT_WS_WRITE_TIMEOUT)),
+                openable_alpns,
             )
             .await
         })

tests/ws_upgrade_session.rs

@@ -4,7 +4,7 @@
 //! Grows from the ws-byte-adapter POC's validated suite.
 
 use alkcall::core::auth::{Identity, IdentityProvider};
-use alkcall::protocol::wire::{EventEnvelope, ResponseEnvelope, EVENT_RESPONDED};
+use alkcall::protocol::wire::{EventEnvelope, ResponseEnvelope, EVENT_ERROR, EVENT_RESPONDED};
 use alkcall::registry::discovery::{services_list_handler, services_list_spec};
 use alkcall::registry::registration::{
     make_handler, Handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry,
@@ -250,16 +250,23 @@ async fn call_and_await(
             tokio::time::Instant::now() < deadline,
             "timed out waiting for response to {request_id}"
         );
+        // A prior request's trailing `call.completed` (a Sub pump's
+        // natural end) may still be in flight — skip frames for other
+        // request ids. Data-channel chunks (id != 0) are skipped too:
+        // this helper asserts only channel-0 responses.
         if let Some(env) = frames.next_frame() {
-            return env;
+            if env.id == request_id {
+                return env;
+            }
         }
         let bin = ws.next_binary(std::time::Duration::from_millis(500)).await;
         match bin {
             Some(bytes) => {
                 chunks.push(&bytes);
                 while let Some((channel_id, payload)) = chunks.next_chunk() {
-                    assert_eq!(channel_id, 0, "response must ride channel 0");
-                    frames.push(&payload);
+                    if channel_id == 0 {
+                        frames.push(&payload);
+                    }
                 }
             }
             None => panic!("ws closed unexpectedly while awaiting {request_id}"),
@@ -969,3 +976,531 @@ async fn ws_timeouts_extension_sets_the_idle_window_on_a_bare_registry_route() {
     );
     ws.close().await;
 }
+
+// --- Unit 2/3 gates (review 006): the data-channel + op/register wiring ---
+
+use alkcall::channels::operations::OpenHandler;
+
+/// The data-plane echo the `OpenHandler` runs on the allocated
+/// channel's `Connection`: echo every inbound byte block back, then
+/// drain until EOF. The WS-test client writes raw (channel_id,
+/// payload) chunks; the handler reads the channel's BiStream and
+/// writes the echo — the exact data-plane shape ADR-067 promises.
+fn echo_open_handler() -> OpenHandler {
+    Arc::new(move |_input, channel_conn, _auth| {
+        tokio::spawn(async move {
+            let mut stream = match channel_conn.accept_bi().await {
+                Ok(s) => s,
+                Err(_) => return,
+            };
+            use tokio::io::{AsyncReadExt, AsyncWriteExt};
+            let mut buf = [0u8; 4096];
+            loop {
+                match stream.read(&mut buf).await {
+                    Ok(0) | Err(_) => break,
+                    Ok(n) => {
+                        if stream.write_all(&buf[..n]).await.is_err() {
+                            break;
+                        }
+                    }
+                }
+            }
+            let _ = stream.shutdown().await;
+        })
+    })
+}
+
+/// The open-op spec for the test's openable ALPN (`channels/echo/sub`,
+/// opening the `alk/echo` data plane).
+fn echo_open_spec() -> OperationSpec {
+    OperationSpec::new(
+        "channels/echo/sub",
+        OperationType::Sub,
+        Visibility::External,
+        serde_json::json!({
+            "type": "object",
+            "properties": {},
+            "additionalProperties": false
+        }),
+        serde_json::json!({
+            "type": "object",
+            "properties": { "channel_id": { "type": "integer" } }
+        }),
+        vec![],
+        AccessControl::default(),
+        None,
+    )
+    .with_channel_open(alkcall::registry::spec::ChannelOpenSpec::new("alk/echo"))
+}
+
+/// Spawn a WS server whose adapter declares the echo openable ALPN and
+/// a small per-identity channel cap; returns the WS URL and the shared
+/// policy (for the ledger assertions).
+async fn spawn_openable_ws_server(
+    registry: Arc<OperationRegistry>,
+    cap: usize,
+) -> (
+    String,
+    Arc<alkcall::channels::policy::PerIdentityChannelPolicy>,
+) {
+    use alkhttp::websocket::OpenableAlpns;
+
+    let policy = Arc::new(alkcall::channels::policy::PerIdentityChannelPolicy::new(
+        cap,
+    ));
+    let policy_dyn: Arc<dyn alkcall::channels::policy::ChannelLifecyclePolicy> =
+        Arc::clone(&policy) as Arc<dyn alkcall::channels::policy::ChannelLifecyclePolicy>;
+
+    let openables = OpenableAlpns(Arc::from([alkhttp::websocket::OpenableAlpn::new(
+        echo_open_spec(),
+        echo_open_handler(),
+    )]));
+
+    let app = axum::Router::new()
+        .route(
+            "/alk/channels",
+            axum::routing::get(alkhttp::websocket::ws_upgrade_handler),
+        )
+        .layer(axum::middleware::from_fn_with_state(
+            provider_with(vec![("tok-1", identity("alice", &[]))]),
+            alkhttp::websocket::ws_bearer_auth,
+        ))
+        .layer(axum::Extension(openables))
+        .layer(axum::Extension(alkhttp::websocket::ChannelsPolicy(
+            policy_dyn,
+        )))
+        .with_state(registry);
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = format!("ws://{}", listener.local_addr().unwrap());
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+    (format!("{addr}/alk/channels"), policy)
+}
+
+/// Drain channel-0 chunks until a complete envelope shows up (shared
+/// with `call_and_await`'s loop but tolerant of interleaved
+/// non-zero-channel chunks).
+async fn await_envelope(ws: &mut WsClient, request_id: &str) -> EventEnvelope {
+    let mut chunks = ChunkAssembler::new();
+    let mut frames = FrameAssembler::new();
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
+    loop {
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "timed out waiting for {request_id}"
+        );
+        if let Some(env) = frames.next_frame() {
+            if env.id == request_id {
+                return env;
+            }
+        }
+        let bin = ws.next_binary(std::time::Duration::from_millis(500)).await;
+        match bin {
+            Some(bytes) => {
+                chunks.push(&bytes);
+                while let Some((channel_id, payload)) = chunks.next_chunk() {
+                    assert_eq!(channel_id, 0, "response must ride channel 0");
+                    frames.push(&payload);
+                }
+            }
+            None => panic!("ws closed unexpectedly while awaiting {request_id}"),
+        }
+    }
+}
+
+/// Unit 3 scenario 1+2 (review 006): the open op resolves with a
+/// `channel_id`, the per-session openable is discoverable through
+/// `services/list`, and data chunks flow both directions on the
+/// channel — the `OpenHandler`'s `Connection` sees the bytes.
+#[tokio::test]
+async fn data_channel_open_discoverable_and_bytes_round_trip() {
+    let (url, _policy) = spawn_openable_ws_server(
+        echo_registry(),
+        alkcall::channels::policy::DEFAULT_CHANNEL_CAP,
+    )
+    .await;
+    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
+
+    // Discovery first: the fork's bootstrap discovery lists the
+    // session's own openable (the F-06 shape).
+    let env = call_and_await(&mut ws, "req-list", "services/list", serde_json::json!({})).await;
+    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
+    let names: Vec<String> = env.payload["output"]["operations"]
+        .as_array()
+        .expect("operations array")
+        .iter()
+        .filter_map(|o| o["name"].as_str().map(String::from))
+        .collect();
+    assert!(
+        names.contains(&"channels/echo/sub".to_string()),
+        "per-session openable discoverable: {names:?}"
+    );
+    assert!(
+        names.contains(&"channel/close".to_string()) && names.contains(&"op/register".to_string()),
+        "generic channel ops + op/register served: {names:?}"
+    );
+
+    // The open op resolves with a channel_id.
+    let env = call_and_await(
+        &mut ws,
+        "req-open",
+        "channels/echo/sub",
+        serde_json::json!({}),
+    )
+    .await;
+    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
+    let channel_id = env.payload["output"]["channel_id"]
+        .as_u64()
+        .expect("channel_id in open response");
+    assert!(channel_id > 0, "even id from the accept side: {channel_id}");
+
+    // Data-plane echo: send one chunk on the data channel and read the
+    // handler's echo back. Data-channel payloads are opaque (ADR-035:
+    // the handler owns its framing) — the echo handler echoes
+    // byte-for-byte per read.
+    let payload = b"hello-data-plane";
+    let framed = frame_data_channel(channel_id as u32, payload);
+    ws.send_binary(framed).await;
+
+    let echoed = read_data_channel_block(&mut ws, channel_id as u32).await;
+    assert_eq!(echoed, payload, "handler echoed the bytes");
+    ws.close().await;
+}
+
+/// Frame a raw data-channel chunk as a WS binary message: the 8-byte
+/// chunk header (channel_id BE + length BE) + payload — the same wire
+/// format channel 0 uses, the opaque data-plane unit (ADR-035).
+fn frame_data_channel(channel_id: u32, block: &[u8]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(8 + block.len());
+    out.extend_from_slice(&channel_id.to_be_bytes());
+    out.extend_from_slice(&(block.len() as u32).to_be_bytes());
+    out.extend_from_slice(block);
+    out
+}
+
+/// Read one echoed chunk's payload off a data channel, skipping
+/// channel-0 traffic.
+async fn read_data_channel_block(ws: &mut WsClient, channel_id: u32) -> Vec<u8> {
+    let mut chunks = ChunkAssembler::new();
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
+    loop {
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "timed out waiting for data-channel {channel_id}"
+        );
+        let bin = ws.next_binary(std::time::Duration::from_millis(500)).await;
+        match bin {
+            Some(bytes) => {
+                chunks.push(&bytes);
+                while let Some((cid, payload)) = chunks.next_chunk() {
+                    if cid == channel_id {
+                        return payload;
+                    }
+                }
+            }
+            None => panic!("ws closed unexpectedly while awaiting data channel {channel_id}"),
+        }
+    }
+}
+
+/// Unit 3 scenario 2 (review 006): `channel/close` tears the handler
+/// stream down and the opener ledger decrements (the policy count
+/// drops) — plus the close op itself resolves `{"closed": true}`.
+#[tokio::test]
+async fn channel_close_resolves_and_decrements_the_opener_ledger() {
+    let (url, policy) = spawn_openable_ws_server(
+        echo_registry(),
+        alkcall::channels::policy::DEFAULT_CHANNEL_CAP,
+    )
+    .await;
+    let alice = identity("alice", &[]);
+    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
+
+    let env = call_and_await(
+        &mut ws,
+        "req-open",
+        "channels/echo/sub",
+        serde_json::json!({}),
+    )
+    .await;
+    let channel_id = env.payload["output"]["channel_id"].as_u64().unwrap() as u32;
+    assert_eq!(policy.count_for(&alice), 1, "open incremented the ledger");
+
+    let env = call_and_await(
+        &mut ws,
+        "req-close",
+        "channel/close",
+        serde_json::json!({ "channel_id": channel_id }),
+    )
+    .await;
+    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
+    assert_eq!(
+        env.payload["output"]["closed"], true,
+        "close op resolves: {}",
+        env.payload["output"]
+    );
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
+    loop {
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "ledger decrement never landed"
+        );
+        if policy.count_for(&alice) == 0 {
+            break;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+    }
+    ws.close().await;
+}
+
+/// Unit 3 scenario 3: the per-identity cap denies the open over the
+/// limit with the `channel:` error prefix (the policy is the
+/// `ChannelsPolicy` extension; the open wrapper consults it before
+/// allocation).
+#[tokio::test]
+async fn channel_cap_denies_open_over_the_limit() {
+    let (url, policy) = spawn_openable_ws_server(echo_registry(), 1).await;
+    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
+
+    let env = call_and_await(
+        &mut ws,
+        "req-open-1",
+        "channels/echo/sub",
+        serde_json::json!({}),
+    )
+    .await;
+    assert!(
+        env.payload["output"]["channel_id"].is_u64(),
+        "first open admitted"
+    );
+
+    let env = call_and_await(
+        &mut ws,
+        "req-open-2",
+        "channels/echo/sub",
+        serde_json::json!({}),
+    )
+    .await;
+    assert_eq!(
+        env.r#type, EVENT_ERROR,
+        "cap denial: got {} payload {}",
+        env.r#type, env.payload
+    );
+    let error = &env.payload;
+    assert!(
+        error["code"]
+            .as_str()
+            .is_some_and(|c| c.starts_with("channel:")),
+        "channel-prefixed denial code: {error}"
+    );
+    assert_eq!(policy.count_for(&identity("alice", &[])), 1);
+    ws.close().await;
+}
+
+/// Unit 3 gate (the WS-24/WS-25 shape): `op/register` is served per
+/// session and its collision policy binds the session fork. Announce
+/// lands in the connection overlay (re-announce without `replace`
+/// rejects with the overlay gate's `ALREADY_EXISTS`); announcing a name
+/// the serving side itself registers is rejected with
+/// `ALREADY_EXISTS` regardless of `replace` (review 005 G-03's gate,
+/// through the WS path).
+#[tokio::test]
+async fn op_register_served_per_session_and_collision_is_already_exists() {
+    use alkcall::registry::op_register::OpRegisterRequest;
+
+    let addr = spawn_ws_server(
+        echo_registry(),
+        provider_with(vec![("tok-1", identity("alice", &[]))]),
+    )
+    .await;
+    let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1")
+        .await
+        .unwrap();
+
+    // A distinct name announces cleanly into the connection overlay.
+    let spec = OperationSpec::new(
+        "consumer/exec",
+        OperationType::Query,
+        Visibility::External,
+        serde_json::json!({}),
+        serde_json::json!({}),
+        vec![],
+        AccessControl::default(),
+        None,
+    );
+    let request = OpRegisterRequest {
+        spec,
+        replace: false,
+    };
+    let frame = EventEnvelope::requested(
+        "req-announce",
+        serde_json::json!({
+            "operationId": "op/register",
+            "input": request.to_json(),
+        }),
+    );
+    ws.send_binary(frame_channel0_chunk(&frame)).await;
+    let env = await_envelope(&mut ws, &frame.id).await;
+    assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type);
+    assert_eq!(env.payload["output"]["registered"], true);
+
+    // The announced op landed in the overlay: a second announce of the
+    // same name without `replace` hits the overlay collision gate.
+    let spec = OperationSpec::new(
+        "consumer/exec",
+        OperationType::Query,
+        Visibility::External,
+        serde_json::json!({}),
+        serde_json::json!({}),
+        vec![],
+        AccessControl::default(),
+        None,
+    );
+    let request = OpRegisterRequest {
+        spec,
+        replace: false,
+    };
+    let frame = EventEnvelope::requested(
+        "req-reannounce",
+        serde_json::json!({
+            "operationId": "op/register",
+            "input": request.to_json(),
+        }),
+    );
+    ws.send_binary(frame_channel0_chunk(&frame)).await;
+    let env = await_envelope(&mut ws, &frame.id).await;
+    assert_eq!(env.r#type, EVENT_ERROR, "got {}", env.r#type);
+    assert_eq!(
+        env.payload["code"], "ALREADY_EXISTS",
+        "overlay collision gate: {}",
+        env.payload
+    );
+
+    // Collision: announcing the serving side's own op name rejects with
+    // ALREADY_EXISTS regardless of replace (review 005 G-03's gate,
+    // through the WS path).
+    let spec = OperationSpec::new(
+        "echo/run",
+        OperationType::Query,
+        Visibility::External,
+        serde_json::json!({}),
+        serde_json::json!({}),
+        vec![],
+        AccessControl::default(),
+        None,
+    );
+    let request = OpRegisterRequest {
+        spec,
+        replace: true,
+    };
+    let frame = EventEnvelope::requested(
+        "req-collide",
+        serde_json::json!({
+            "operationId": "op/register",
+            "input": request.to_json(),
+        }),
+    );
+    ws.send_binary(frame_channel0_chunk(&frame)).await;
+    let env = await_envelope(&mut ws, &frame.id).await;
+    assert_eq!(env.r#type, EVENT_ERROR, "got {}", env.r#type);
+    assert_eq!(
+        env.payload["code"], "ALREADY_EXISTS",
+        "serving-registry collision rejected: {}",
+        env.payload
+    );
+    ws.close().await;
+}
+
+/// Unit 3 scenario 4: a mid-open disconnect tears the session down
+/// without leaking — the ledger count for the opener drops to zero
+/// (the channels adapter's teardown decrements), the WS close is
+/// observed, and no further open succeeds (the socket is gone).
+#[tokio::test]
+async fn disconnect_mid_open_tears_down_and_decrements() {
+    let (url, policy) = spawn_openable_ws_server(
+        echo_registry(),
+        alkcall::channels::policy::DEFAULT_CHANNEL_CAP,
+    )
+    .await;
+    let alice = identity("alice", &[]);
+    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
+
+    let env = call_and_await(
+        &mut ws,
+        "req-open",
+        "channels/echo/sub",
+        serde_json::json!({}),
+    )
+    .await;
+    assert!(env.payload["output"]["channel_id"].is_u64());
+    assert_eq!(policy.count_for(&alice), 1);
+
+    // Hard disconnect (drop without close frame — the pump sees EOF).
+    drop(ws);
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
+    loop {
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "teardown never decremented the ledger"
+        );
+        if policy.count_for(&alice) == 0 {
+            break;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+    }
+}
+
+/// Unit 3 scenario 5: a `TooLarge` chunk on a data channel survives —
+/// the demux skips it and resyncs (alkcall's hardening, asserted
+/// through the WS path). The oversized chunk is skipped by the demux
+/// (TooLarge), the following chunks on other channels still route.
+#[tokio::test]
+async fn too_large_data_channel_chunk_survives_demux_resync() {
+    use alkhttp::websocket::MAX_CHUNK_LEN;
+
+    let (url, _policy) = spawn_openable_ws_server(
+        echo_registry(),
+        alkcall::channels::policy::DEFAULT_CHANNEL_CAP,
+    )
+    .await;
+    let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap();
+
+    let env = call_and_await(
+        &mut ws,
+        "req-open",
+        "channels/echo/sub",
+        serde_json::json!({}),
+    )
+    .await;
+    let channel_id = env.payload["output"]["channel_id"].as_u64().unwrap() as u32;
+
+    // An oversized declared chunk on the data channel: the demux skips
+    // it (bounded by its 256 MiB budget) and stays in sync. The skip
+    // consumes exactly the declared payload length off the transport,
+    // so we must actually send that many filler bytes (16 MiB + 1 is
+    // over MAX_CHUNK_LEN but under the WS message caps, sent as
+    // separate WS messages).
+    let oversized_len = MAX_CHUNK_LEN + 1;
+    let mut header = Vec::with_capacity(8);
+    header.extend_from_slice(&channel_id.to_be_bytes());
+    header.extend_from_slice(&oversized_len.to_be_bytes());
+    ws.send_binary(header).await;
+    let filler_len = oversized_len as usize;
+    let filler = vec![0u8; filler_len];
+    let mut sent = 0usize;
+    while sent < filler_len {
+        let take = (filler_len - sent).min(256 * 1024);
+        ws.send_binary(filler[sent..sent + take].to_vec()).await;
+        sent += take;
+    }
+
+    // The next small chunk on the data channel still routes and the
+    // handler echoes it — resync proven.
+    let payload = b"after-skip";
+    ws.send_binary(frame_data_channel(channel_id, payload))
+        .await;
+    let echoed = read_data_channel_block(&mut ws, channel_id).await;
+    assert_eq!(echoed, payload, "demux resynced after the TooLarge skip");
+    ws.close().await;
+}